summaryrefslogtreecommitdiff
path: root/devices/digital_io.py
diff options
context:
space:
mode:
Diffstat (limited to 'devices/digital_io.py')
-rw-r--r--devices/digital_io.py261
1 files changed, 261 insertions, 0 deletions
diff --git a/devices/digital_io.py b/devices/digital_io.py
new file mode 100644
index 0000000..23a345f
--- /dev/null
+++ b/devices/digital_io.py
@@ -0,0 +1,261 @@
+"""
+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(0, 0, 0, 0)
+
+ 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()
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-13 20:37:17 +0000