summaryrefslogtreecommitdiff
path: root/daq_system/devices/digital_io.py
blob: b78b5cf1a07cd6084fcdd31626428ca191844b0a (plain)
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
"""
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