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
|
"""
api_layers/protocols/base_protocol.py
Abstract base for serial instrument protocol layers.
Background thread calls _poll() at poll_interval and stores results in
self._cache — same threading model as ArduinoLayer.
Simulation uses _simulate(t) instead.
"""
import threading
import time
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Tuple
class BaseProtocol(ABC):
def __init__(
self,
port: str,
baud: int,
poll_interval: float = 0.1,
simulate: bool = True,
):
self.port = port
self.baud = baud
self.poll_interval = poll_interval
self.simulate = simulate
self._cache: Dict[str, float] = {}
self._lock = threading.Lock()
self._running = False
self._thread: Optional[threading.Thread] = None
self._ser = None
# ── Public API ────────────────────────────────────────────────────────
def connect(self) -> bool:
if self.simulate:
self._running = True
self._thread = threading.Thread(target=self._sim_loop, daemon=True)
self._thread.start()
return True
try:
import serial
self._ser = serial.Serial(
self.port, self.baud, timeout=1.0,
**self._serial_kwargs(),
)
time.sleep(0.05)
self._running = True
self._thread = threading.Thread(target=self._poll_loop, daemon=True)
self._thread.start()
return True
except Exception as e:
print(f"[{type(self).__name__}] connect failed on {self.port}: {e}")
return False
def disconnect(self) -> None:
self._running = False
if self._thread:
self._thread.join(timeout=2.0)
self._thread = None
if self._ser:
try:
self._ser.close()
except Exception:
pass
self._ser = None
def read(self) -> Dict[str, float]:
with self._lock:
return dict(self._cache)
@classmethod
def list_ports(cls) -> List[Tuple[str, str]]:
try:
import serial.tools.list_ports
return [(p.device, p.description) for p in serial.tools.list_ports.comports()]
except ImportError:
return []
# ── Internal ──────────────────────────────────────────────────────────
def _serial_kwargs(self) -> dict:
return {}
def _poll_loop(self) -> None:
while self._running:
try:
result = self._poll()
if result:
with self._lock:
self._cache.update(result)
except Exception as e:
print(f"[{type(self).__name__}] poll error: {e}")
time.sleep(self.poll_interval)
def _sim_loop(self) -> None:
t0 = time.time()
while self._running:
t = time.time() - t0
with self._lock:
self._cache = self._simulate(t)
time.sleep(self.poll_interval)
# ── Abstract ──────────────────────────────────────────────────────────
@abstractmethod
def _poll(self) -> Dict[str, float]:
"""Read instrument, return {channel_id: value}."""
@abstractmethod
def _simulate(self, t: float) -> Dict[str, float]:
"""Return simulated values at elapsed time t (seconds)."""
@abstractmethod
def write(self, channel_id: str, value) -> bool:
"""Send command to instrument."""
|
