diff options
Diffstat (limited to 'devices/arduino_device.py')
| -rw-r--r-- | devices/arduino_device.py | 136 |
1 files changed, 84 insertions, 52 deletions
diff --git a/devices/arduino_device.py b/devices/arduino_device.py index 1641eaa..5c954c8 100644 --- a/devices/arduino_device.py +++ b/devices/arduino_device.py @@ -15,7 +15,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _ANALOG_COLORS = [ @@ -91,26 +91,11 @@ class ArduinoDevice(BaseDevice): self._di_pins: List[str] = list(di_pins) self._do_pins: List[str] = list(do_pins) - channels = [] - for i, pin in enumerate(self._analog_pins): - channels.append(ChannelConfig( - channel_id=pin, name=pin, unit="V", - min_value=0.0, max_value=5.0, - color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], - )) - for i in range(len(self._di_pins)): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", - min_value=0.0, max_value=1.0, - color=_DI_COLORS[i % len(_DI_COLORS)], - )) - for i in range(len(self._do_pins)): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", - min_value=0.0, max_value=1.0, - writable=True, - color=_DO_COLORS[i % len(_DO_COLORS)], - )) + self._ai_map = {f"ai{i}": pin for i, pin in enumerate(self._analog_pins)} + self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} + self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + + channels = self._build_channels() _desc = ArduinoLayer.describe_port(port) if not simulate else "" if _desc: @@ -121,22 +106,45 @@ class ArduinoDevice(BaseDevice): device_id=device_id, name=_name, device_type=self.DEVICE_TYPE, - description=(f"Arduino — {len(self._analog_pins)} analog, " - f"{len(self._di_pins)} DI, {len(self._do_pins)} DO"), + description=self._description(), manufacturer="Arduino", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(len(self._do_pins))} - self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} - - self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} - self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} self._layer = self._make_layer() + def _description(self) -> str: + return (f"Arduino — {len(self._analog_pins)} analog, " + f"{len(self._di_pins)} DI, {len(self._do_pins)} DO") + + def _build_channels(self) -> List[SignalConfig]: + channels = [] + for i, pin in enumerate(self._analog_pins): + channels.append(SignalConfig( + signal_id=f"ai{i}", name=pin, unit="V", + min_value=0.0, max_value=5.0, + color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], + )) + for i in range(len(self._di_pins)): + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", + min_value=0.0, max_value=1.0, + color=_DI_COLORS[i % len(_DI_COLORS)], + )) + for i in range(len(self._do_pins)): + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", + min_value=0.0, max_value=1.0, + writable=True, + color=_DO_COLORS[i % len(_DO_COLORS)], + )) + return channels + def _make_layer(self): from api_layers.port_registry import port_registry return port_registry.get_layer( @@ -177,10 +185,12 @@ class ArduinoDevice(BaseDevice): def read_channels(self) -> Dict[str, float]: raw = self._layer.read() result = {} - for ch in self.info.channels: - cid = ch.channel_id - if cid in raw: - result[cid] = raw[cid] + for ch in self.info.signals: + cid = ch.signal_id + if cid.startswith("ai"): + pin = self._ai_map.get(cid) + if pin and pin in raw: + result[cid] = raw[pin] elif cid.startswith("di"): pin = self._di_map.get(cid) if pin and pin in raw: @@ -244,15 +254,25 @@ class ArduinoDevice(BaseDevice): di_pins: List[str] = None, do_pins: List[str] = None, ) -> None: - """Update active pin mapping and send config commands if hardware-connected.""" + """Update active pin mapping, rebuild channels, and send config + commands if hardware-connected. Caller is responsible for emitting + a `device_reconfigured` signal so the acquisition engine, profile, + and UI pick up the new channel list (see ui/windows/config_window.py).""" if analog_pins is not None: self._analog_pins = list(analog_pins) + self._ai_map = {f"ai{i}": p for i, p in enumerate(analog_pins)} if di_pins is not None: self._di_pins = list(di_pins) self._di_map = {f"di{i}": p for i, p in enumerate(di_pins)} if do_pins is not None: self._do_pins = list(do_pins) self._do_map = {f"do{i}": p for i, p in enumerate(do_pins)} + + self.info.signals = self._build_channels() + self.info.description = self._description() + self._output_state = {f"do{i}": self._output_state.get(f"do{i}", 0) for i in range(len(self._do_pins))} + self._di_state = {f"di{i}": self._di_state.get(f"di{i}", 0) for i in range(len(self._di_pins))} + if self.status == DeviceStatus.CONNECTED: self._layer.configure_pins( di_pins=self._di_pins, @@ -329,27 +349,40 @@ class ArduinoConfigWidget(QWidget): # ── Pin Mapping ─────────────────────────────────────────────────── pin_grp = QGroupBox("Pin Mapping") - pin_form = QFormLayout(pin_grp) - pin_form.setContentsMargins(10, 16, 10, 10) + pin_outer = QVBoxLayout(pin_grp) + pin_outer.setContentsMargins(10, 16, 10, 10) + pin_outer.setSpacing(8) + + pin_cols = QHBoxLayout() + pin_cols.setSpacing(12) + analog_grp = QGroupBox("Analog Pins") + analog_form = QFormLayout(analog_grp) self.analog_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._analog_pins)) - self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3 (indices into A0–A5)") - pin_form.addRow("Analog Input Pins:", self.analog_pins_edit) + self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3") + analog_form.addRow("Input:", self.analog_pins_edit) + pin_cols.addWidget(analog_grp) + digital_grp = QGroupBox("Digital Pins") + digital_form = QFormLayout(digital_grp) self.di_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._di_pins)) self.di_pins_edit.setPlaceholderText("e.g. 2, 3, 8") - pin_form.addRow("Digital Input Pins:", self.di_pins_edit) - + digital_form.addRow("Input:", self.di_pins_edit) self.do_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._do_pins)) - self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7, 9") - pin_form.addRow("Digital Output Pins:", self.do_pins_edit) + self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7") + digital_form.addRow("Output:", self.do_pins_edit) + pin_cols.addWidget(digital_grp) + + pin_outer.addLayout(pin_cols) hint = QLabel("Changes take effect on Apply & Reconnect.\n" "Analog: enter A-pin indices (0=A0, 1=A1, …). " - "Digital: enter pin numbers.") + "Digital: enter pin numbers. " + "For analog-style output, drive a PWM-capable digital " + "output pin (e.g. D9) with a duty-cycle value.") hint.setObjectName("traceSource") hint.setWordWrap(True) - pin_form.addRow(hint) + pin_outer.addWidget(hint) root.addWidget(pin_grp) @@ -417,19 +450,18 @@ class ArduinoConfigWidget(QWidget): self.port_edit.setText(port) def _apply(self): - # Parse pin fields and update device before reconnect so connect() sends correct APIN/DPIN + # Parse pin fields and remap (rebuilds device.info.signals) before + # reconnect, so connect() sends correct APIN/DPIN and the acquisition + # engine picks up the new channel list once this dialog closes. analog = _parse_analog_pin_edit(self.analog_pins_edit.text()) di = _parse_pin_edit(self.di_pins_edit.text()) do = _parse_pin_edit(self.do_pins_edit.text()) - if analog: - self.device._analog_pins = analog - if di: - self.device._di_pins = di - self.device._di_map = {f"di{i}": p for i, p in enumerate(di)} - if do: - self.device._do_pins = do - self.device._do_map = {f"do{i}": p for i, p in enumerate(do)} + self.device.remap_pins( + analog_pins=analog or None, + di_pins=di or None, + do_pins=do or None, + ) self.device.switch_backend( simulate=self.sim_chk.isChecked(), port=self.port_edit.text().strip() or self.device._port, |
