""" devices/digital_io.py Digital I/O device module. Backends: NI-DAQmx or Arduino. NI backend – uses nidaqmx digital line tasks (P0.0..P0.7) Arduino – uses ArduinoLayer digital pin reads; writes via W:Dxx:val """ import random import time from typing import Any, Dict from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QCheckBox, QPushButton, QLabel, QFormLayout, QComboBox, QLineEdit, ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus _IN_COLORS = ["#00d4ff", "#4cc9f0", "#90e0ef", "#caf0f8", "#0077b6", "#023e8a", "#48cae4", "#ade8f4"] _OUT_COLORS = ["#ff6b35", "#ffcc00", "#f77f00", "#fcbf49", "#d62828", "#e63946", "#fb8500", "#ffd166"] class DigitalIODevice(BaseDevice): DEVICE_TYPE = "digital_io" ICON = "⬛" def __init__( self, device_id: str = "dio_0", num_inputs: int = 8, num_outputs: int = 8, simulate: bool = True, backend: str = "nidaqmx", # "nidaqmx" | "arduino" ni_device: str = "Dev1", ard_port: str = "COM3", ard_baud: int = 115200, ): self.simulate = simulate self.backend = backend self._ni_device = ni_device self._ard_port = ard_port self._ard_baud = ard_baud channels = [] for i in range(num_inputs): channels.append(ChannelConfig( channel_id=f"di{i}", name=f"DI {i}", unit="", min_value=0.0, max_value=1.0, color=_IN_COLORS[i % len(_IN_COLORS)], )) for i in range(num_outputs): channels.append(ChannelConfig( channel_id=f"do{i}", name=f"DO {i}", unit="", min_value=0.0, max_value=1.0, color=_OUT_COLORS[i % len(_OUT_COLORS)], )) info = DeviceInfo( device_id=device_id, name=f"Digital I/O ({backend.upper()})", device_type=self.DEVICE_TYPE, description="Digital input/output module", icon=self.ICON, channels=channels, ) super().__init__(info) self._output_state: Dict[str, int] = { f"do{i}": 0 for i in range(num_outputs) } self._sim_toggle: Dict[str, int] = {} self._sim_state: Dict[str, int] = {} # Hardware task placeholders self._ni_in_task = None self._ni_out_task = None self._ard_layer = None # ── BaseDevice ────────────────────────────────────────────────────── def connect(self) -> bool: if self.simulate: self.status = DeviceStatus.SIMULATED return True if self.backend == "nidaqmx": return self._ni_connect() else: return self._ard_connect() def _ni_connect(self) -> bool: try: import nidaqmx # type: ignore from nidaqmx.constants import LineGrouping # type: ignore n_in = sum(1 for c in self.info.channels if c.channel_id.startswith("di")) n_out = sum(1 for c in self.info.channels if c.channel_id.startswith("do")) if n_in: self._ni_in_task = nidaqmx.Task() for i in range(n_in): self._ni_in_task.di_channels.add_di_chan( f"{self._ni_device}/port0/line{i}", line_grouping=LineGrouping.CHAN_PER_LINE, ) self._ni_in_task.start() if n_out: self._ni_out_task = nidaqmx.Task() for i in range(n_out): self._ni_out_task.do_channels.add_do_chan( f"{self._ni_device}/port1/line{i}", line_grouping=LineGrouping.CHAN_PER_LINE, ) self._ni_out_task.start() self.status = DeviceStatus.CONNECTED return True except Exception as e: print(f"[DigitalIODevice] NI connect failed: {e}") self.status = DeviceStatus.ERROR return False def _ard_connect(self) -> bool: from api_layers.arduino_layer import ArduinoLayer self._ard_layer = ArduinoLayer( port=self._ard_port, baud=self._ard_baud, digital_pins=[c.channel_id for c in self.info.channels if c.channel_id.startswith("di")], simulate=False, ) ok = self._ard_layer.connect() self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR return ok def disconnect(self) -> None: if self._ni_in_task: try: self._ni_in_task.stop(); self._ni_in_task.close() except Exception: pass if self._ni_out_task: try: self._ni_out_task.stop(); self._ni_out_task.close() except Exception: pass if self._ard_layer: self._ard_layer.disconnect() self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: if self.simulate: return self._sim_read() if self.backend == "nidaqmx": return self._ni_read() return self._ard_read() def _ni_read(self) -> Dict[str, float]: result = {} try: if self._ni_in_task: vals = self._ni_in_task.read() for i, ch in enumerate(c for c in self.info.channels if c.channel_id.startswith("di")): result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) except Exception as e: print(f"[DigitalIODevice] NI read failed: {e}") for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) return result def _ard_read(self) -> Dict[str, float]: if not self._ard_layer: return {} raw = self._ard_layer.read() result = {} for ch in self.info.channels: if ch.channel_id in raw: result[ch.channel_id] = raw[ch.channel_id] elif ch.channel_id.startswith("do"): result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) return result def _sim_read(self) -> Dict[str, float]: result = {} for ch in self.info.channels: if ch.channel_id.startswith("di"): cnt = self._sim_toggle.get(ch.channel_id, 0) + 1 if cnt >= random.randint(8, 40): self._sim_state[ch.channel_id] = 1 - self._sim_state.get(ch.channel_id, 0) cnt = 0 self._sim_toggle[ch.channel_id] = cnt result[ch.channel_id] = float(self._sim_state.get(ch.channel_id, 0)) else: result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) return result def write_channel(self, channel_id: str, value: Any) -> bool: self._output_state[channel_id] = int(bool(value)) if not self.simulate: if self.backend == "nidaqmx" and self._ni_out_task: try: out_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] idx = next((i for i, c in enumerate(out_chs) if c.channel_id == channel_id), None) if idx is not None: states = [self._output_state.get(c.channel_id, 0) for c in out_chs] self._ni_out_task.write(states) except Exception as e: print(f"[DigitalIODevice] NI write failed: {e}") elif self.backend == "arduino" and self._ard_layer: self._ard_layer.write(channel_id, int(bool(value))) return True def get_config_widget(self) -> QWidget: return DigitalIOConfigWidget(self) class DigitalIOConfigWidget(QWidget): def __init__(self, device: DigitalIODevice): super().__init__() self.device = device self._buttons: Dict[str, QPushButton] = {} self._build() def _build(self): root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) self.setMinimumWidth(380) be_grp = QGroupBox("Backend") be_form = QFormLayout(be_grp) self.be_cb = QComboBox() self.be_cb.addItems(["nidaqmx", "arduino"]) self.be_cb.setCurrentText(self.device.backend) be_form.addRow("Backend:", self.be_cb) self.sim_chk = QCheckBox("Simulate") self.sim_chk.setChecked(self.device.simulate) be_form.addRow(self.sim_chk) root.addWidget(be_grp) out_grp = QGroupBox("Digital Outputs") out_lay = QVBoxLayout(out_grp) for ch in (c for c in self.device.info.channels if c.channel_id.startswith("do")): row = QHBoxLayout() lbl = QLabel(ch.name) lbl.setMinimumWidth(50) btn = QPushButton("OFF") btn.setCheckable(True) btn.setChecked(bool(self.device._output_state.get(ch.channel_id, 0))) btn.setObjectName("digitalOutBtn") cid = ch.channel_id def _tog(checked, c=cid, b=btn): self.device.write_channel(c, checked) b.setText("ON" if checked else "OFF") btn.toggled.connect(_tog) row.addWidget(lbl) row.addWidget(btn) out_lay.addLayout(row) self._buttons[ch.channel_id] = btn root.addWidget(out_grp) root.addStretch()