summaryrefslogtreecommitdiff
path: root/devices/digital_io.py
diff options
context:
space:
mode:
authorChristian Kolset <christian.kolset@gmail.com>2026-04-21 22:58:30 -0600
committerChristian Kolset <christian.kolset@gmail.com>2026-04-21 22:58:30 -0600
commit4db119fb9c04428e7aef6f1a278c1639aa867baf (patch)
treee0e0645559637e593f1bbfb9d713256028668527 /devices/digital_io.py
parent97e15f69b835af03b071c4eeb4547c65a02e6f80 (diff)
Major update to Output control
Including: - Control panel - Change status of simulation after devices creation. - Arduino serial conflict with multiple channels. - Claude instructions. - Update to arduino on board code.
Diffstat (limited to 'devices/digital_io.py')
-rw-r--r--devices/digital_io.py73
1 files changed, 67 insertions, 6 deletions
diff --git a/devices/digital_io.py b/devices/digital_io.py
index 53014b5..3d24192 100644
--- a/devices/digital_io.py
+++ b/devices/digital_io.py
@@ -126,12 +126,18 @@ class DigitalIODevice(BaseDevice):
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")],
+ from api_layers.port_registry import port_registry
+ # Get shared layer — won't open a second connection if AnalogInputDevice
+ # is already connected on the same port
+ self._ard_layer = port_registry.get_layer(
+ port=self._ard_port,
+ baud=self._ard_baud,
simulate=False,
)
+ if self._ard_layer.is_connected:
+ # Port already open — just register ourselves and return OK
+ self.status = DeviceStatus.CONNECTED
+ return True
ok = self._ard_layer.connect()
self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR
return ok
@@ -143,8 +149,10 @@ class DigitalIODevice(BaseDevice):
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()
+ if self._ard_layer and self.backend == "arduino" and not self.simulate:
+ from api_layers.port_registry import port_registry
+ port_registry.release(self._ard_port, self._ard_baud)
+ self._ard_layer = None
self.status = DeviceStatus.DISCONNECTED
def read_channels(self) -> Dict[str, float]:
@@ -226,6 +234,24 @@ class DigitalIODevice(BaseDevice):
return self._ard_layer.set_parameter(name, value)
return False
+ def switch_backend(self, backend: str, simulate: bool,
+ ni_device: str, ard_port: str, ard_baud: int) -> None:
+ was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED)
+ if was_running:
+ self.disconnect()
+
+ self.backend = backend
+ self.simulate = simulate
+ self._ni_device = ni_device
+ self._ard_port = ard_port
+ self._ard_baud = ard_baud
+ self._ard_layer = None
+ self._ni_in_task = None
+ self._ni_out_task = None
+
+ if was_running:
+ self.connect()
+
def get_config_widget(self) -> QWidget:
return DigitalIOConfigWidget(self)
@@ -251,7 +277,27 @@ class DigitalIOConfigWidget(QWidget):
self.sim_chk = QCheckBox("Simulate")
self.sim_chk.setChecked(self.device.simulate)
be_form.addRow(self.sim_chk)
+
+ self.ni_dev_edit = QLineEdit(self.device._ni_device)
+ be_form.addRow("NI Device:", self.ni_dev_edit)
+
+ self.ard_port_edit = QLineEdit(self.device._ard_port)
+ self.ard_port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3")
+ be_form.addRow("Arduino Port:", self.ard_port_edit)
+
+ self.ard_baud_cb = QComboBox()
+ self.ard_baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"])
+ self.ard_baud_cb.setCurrentText(str(self.device._ard_baud))
+ be_form.addRow("Baud Rate:", self.ard_baud_cb)
+
+ apply_btn = QPushButton("Apply & Reconnect")
+ apply_btn.setObjectName("applyButton")
+ apply_btn.clicked.connect(self._apply)
+ be_form.addRow(apply_btn)
+
+ self.be_cb.currentTextChanged.connect(self._update_visibility)
root.addWidget(be_grp)
+ self._update_visibility(self.device.backend)
out_grp = QGroupBox("Digital Outputs")
out_lay = QVBoxLayout(out_grp)
@@ -277,3 +323,18 @@ class DigitalIOConfigWidget(QWidget):
root.addWidget(out_grp)
root.addStretch()
+
+ def _update_visibility(self, backend: str = ""):
+ backend = backend or self.be_cb.currentText()
+ self.ni_dev_edit.setVisible(backend == "nidaqmx")
+ self.ard_port_edit.setVisible(backend == "arduino")
+ self.ard_baud_cb.setVisible(backend == "arduino")
+
+ def _apply(self):
+ self.device.switch_backend(
+ backend=self.be_cb.currentText(),
+ simulate=self.sim_chk.isChecked(),
+ ni_device=self.ni_dev_edit.text().strip() or "Dev1",
+ ard_port=self.ard_port_edit.text().strip(),
+ ard_baud=int(self.ard_baud_cb.currentText()),
+ )