path: root/devices/digital_io.py

"""
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()