1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
"""
api_layers/port_registry.py
PortRegistry — ensures one ArduinoLayer per physical serial port.
Problem: if both an AnalogInputDevice and a DigitalIODevice are
configured for the same Arduino (/dev/ttyUSB0), they would each
create their own ArduinoLayer and both try to open the same serial
port → "device reports readiness to read but returned no data".
Solution: all devices call PortRegistry.get_layer(port, baud) instead
of constructing ArduinoLayer directly. The registry returns an
existing layer if one is already open on that port, or creates a new
one. All reads/writes go through the single shared layer.
The registry also merges analog_pins across devices so the layer
caches values for all channels on the Arduino.
Usage (handled automatically by AnalogInputDevice and DigitalIODevice):
from api_layers.port_registry import port_registry
layer = port_registry.get_layer("/dev/ttyUSB0", 115200, simulate=False)
layer.connect_if_needed()
"""
from __future__ import annotations
import threading
from typing import Dict, List, Optional, Tuple
class PortRegistry:
"""
Global registry of shared ArduinoLayer instances.
Thread-safe singleton accessed via the module-level `port_registry`.
"""
def __init__(self):
self._lock: threading.Lock = threading.Lock()
self._layers: Dict[Tuple[str, int], object] = {} # (port, baud) → ArduinoLayer
self._ref_counts: Dict[Tuple[str, int], int] = {}
def get_layer(self, port: str, baud: int,
simulate: bool = False,
extra_pins: List[str] = None) -> "ArduinoLayer": # type: ignore
"""
Return a shared ArduinoLayer for (port, baud).
Creates one if it doesn't exist yet.
If simulate=True, a simulation layer is returned (always separate,
so simulated devices don't share state with each other).
"""
from api_layers.arduino_layer import ArduinoLayer
if simulate:
# Simulated devices get their own independent layer
layer = ArduinoLayer(port=port, baud=baud, simulate=True)
if extra_pins:
layer.analog_pins = list(extra_pins)
return layer
key = (port, baud)
with self._lock:
if key not in self._layers:
layer = ArduinoLayer(port=port, baud=baud, simulate=False)
if extra_pins:
layer.analog_pins = list(extra_pins)
self._layers[key] = layer
self._ref_counts[key] = 0
else:
layer = self._layers[key]
# Merge any new pin names so the shared cache covers them
if extra_pins:
existing = set(layer.analog_pins)
for p in extra_pins:
if p not in existing:
layer.analog_pins.append(p)
self._ref_counts[key] += 1
return layer
def release(self, port: str, baud: int) -> None:
"""
Decrement ref count for a port. Disconnects and removes the
layer when no devices are using it any more.
"""
key = (port, baud)
with self._lock:
if key not in self._ref_counts:
return
self._ref_counts[key] -= 1
if self._ref_counts[key] <= 0:
layer = self._layers.pop(key, None)
self._ref_counts.pop(key, None)
if layer:
try:
layer.disconnect()
except Exception:
pass
def connect_all(self) -> None:
"""Connect any layers that haven't been connected yet."""
with self._lock:
layers = list(self._layers.values())
for layer in layers:
if not layer.is_connected:
layer.connect()
def all_layers(self):
with self._lock:
return list(self._layers.values())
def clear(self) -> None:
"""Disconnect and remove all managed layers."""
with self._lock:
layers = list(self._layers.values())
self._layers.clear()
self._ref_counts.clear()
for layer in layers:
try:
layer.disconnect()
except Exception:
pass
# Module-level singleton
port_registry = PortRegistry()
|
