init

1 files changed, 95 insertions, 0 deletions

diff --git a/daq_system/devices/digital_io.py b/daq_system/devices/digital_io.py
new file mode 100644
index 0000000..b78b5cf
--- /dev/null
+++ b/daq_system/devices/digital_io.py
@@ -0,0 +1,95 @@
+"""
+devices/digital_io.py
+
+Digital I/O device module — reads and writes binary channels.
+"""
+
+import random
+import time
+from typing import Any, Dict
+
+from PyQt6.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QCheckBox, QGroupBox, QPushButton, QLabel
+from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus
+
+
+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):
+        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="#00d4ff" if i % 2 == 0 else "#4cc9f0"
+            ))
+        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="#ff6b35" if i % 2 == 0 else "#ffcc00"
+            ))
+        info = DeviceInfo(
+            device_id=device_id, name="Digital I/O",
+            device_type=self.DEVICE_TYPE,
+            description="Digital input/output module",
+            icon=self.ICON, channels=channels
+        )
+        super().__init__(info)
+        self.simulate = simulate
+        self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(num_outputs)}
+        self._toggle_counters = [0] * num_inputs
+
+    def connect(self) -> bool:
+        self.status = DeviceStatus.SIMULATED if self.simulate else DeviceStatus.ERROR
+        return self.simulate
+
+    def disconnect(self) -> None:
+        self.status = DeviceStatus.DISCONNECTED
+
+    def read_channels(self) -> Dict[str, float]:
+        result = {}
+        # Simulate toggling inputs randomly
+        for i, ch in enumerate(self.info.channels):
+            if ch.channel_id.startswith("di"):
+                self._toggle_counters[i] += 1
+                if self._toggle_counters[i] > random.randint(5, 30):
+                    self._toggle_counters[i] = 0
+                    result[ch.channel_id] = float(random.randint(0, 1))
+                else:
+                    result[ch.channel_id] = result.get(ch.channel_id, 0.0)
+            elif ch.channel_id.startswith("do"):
+                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:
+        if channel_id in self._output_state:
+            self._output_state[channel_id] = int(bool(value))
+            return True
+        return False
+
+    def get_config_widget(self) -> QWidget:
+        w = QWidget()
+        layout = QVBoxLayout(w)
+        grp = QGroupBox("Output Controls")
+        grp_layout = QVBoxLayout(grp)
+        for k in self._output_state:
+            row = QHBoxLayout()
+            lbl = QLabel(k.upper())
+            btn = QPushButton("OFF")
+            btn.setCheckable(True)
+            btn.setChecked(bool(self._output_state[k]))
+            btn.setText("ON" if self._output_state[k] else "OFF")
+            channel_id = k
+            def on_toggle(checked, cid=channel_id, b=btn):
+                self.write_channel(cid, checked)
+                b.setText("ON" if checked else "OFF")
+            btn.toggled.connect(on_toggle)
+            row.addWidget(lbl)
+            row.addWidget(btn)
+            grp_layout.addLayout(row)
+        layout.addWidget(grp)
+        layout.addStretch()
+        return w