Over The Air device config feature

docs/conf.py

@@ -11,7 +11,7 @@
 from unittest.mock import Mock
 
 # Mock _ok module
-sys.modules['_ok'] = Mock()
+sys.modules["_ok"] = Mock()
 
 project = "miniscope-io"
 copyright = "2023, Jonny"
@@ -49,6 +49,7 @@
     "numpy": ("https://numpy.org/doc/stable/", None),
     "pandas": ("https://pandas.pydata.org/docs/", None),
     "rich": ("https://rich.readthedocs.io/en/stable/", None),
+    "pyserial": ("https://pyserial.readthedocs.io/en/stable/", None),
 }
 
 # ----------

docs/device/update_controller.md

@@ -0,0 +1,7 @@
+# Update controller
+
+**Under Construction:** This section will be populated when these devices are released.
+
+This page will document equipments needed to use the `miniscope_io.device_update` module (or `mio update` interface.) 
+## prerequisite
+- A custom FTDI chip based IR transmitter (details will be released soon)

docs/index.md

@@ -15,6 +15,7 @@ cli/main
 :maxdepth: 1
 
 device/test_device
+device/update_controller
 ```
 
 ```{toctree}

miniscope_io/cli/main.py

@@ -5,6 +5,7 @@
 import click
 
 from miniscope_io.cli.stream import stream
+from miniscope_io.cli.update import device, update
 
 
 @click.group()
@@ -18,3 +19,5 @@ def cli(ctx: click.Context) -> None:
 
 
 cli.add_command(stream)
+cli.add_command(update)
+cli.add_command(device)

miniscope_io/cli/update.py

@@ -0,0 +1,107 @@
+"""
+CLI for updating device over IR or UART.
+"""
+
+import click
+import yaml
+
+from miniscope_io.device_update import device_update
+from miniscope_io.models.devupdate import DeviceCommand
+
+
+@click.command()
+@click.option(
+    "-p",
+    "--port",
+    required=False,
+    help="Serial port to connect to. Needed if multiple FTDI devices are connected.",
+)
+@click.option(
+    "-i",
+    "--device_id",
+    required=False,
+    default=0,
+    type=int,
+    help="[EXPERIMENTAL FEATURE] ID of the device to update. 0 (default) will update all devices.",
+)
+@click.option(
+    "-k",
+    "--key",
+    required=False,
+    type=click.Choice(["LED", "GAIN", "ROI_X", "ROI_Y"]),
+    help="key to update. Cannot be used with --restart.",
+)
+@click.option(
+    "-v",
+    "--value",
+    required=False,
+    type=int,
+    help="Value to set. Must be used with --key and cannot be used with --restart.",
+)
+@click.option(
+    "-b",
+    "--batch",
+    required=False,
+    type=click.Path(exists=True, dir_okay=False),
+    help=(
+        "[EXPERIMENTAL FEATURE] YAML file that works as a batch file to update."
+        "Specify key and value pairs in the file."
+    ),
+)
+def update(port: str, key: str, value: int, device_id: int, batch: str) -> None:
+    """
+    Update device configuration.
+    """
+
+    # Check mutual exclusivity
+    if (key and not value) or (value and not key):
+        raise click.UsageError("Both --key and --value are required if one is specified.")
+
+    if batch and (key or value):
+        raise click.UsageError(
+            "Options --key/--value and --restart" " and --batch are mutually exclusive."
+        )
+    if key and value:
+        device_update(port=port, key=key, value=value, device_id=device_id)
+    elif batch:
+        with open(batch) as f:
+            batch_file = yaml.safe_load(f)
+        for key, value in batch_file:
+            device_update(port=port, key=key, value=value, device_id=device_id)
+    else:
+        raise click.UsageError("Either --key with --value or --restart must be specified.")
+
+
+@click.command()
+@click.option(
+    "-p",
+    "--port",
+    required=False,
+    help="Serial port to connect to. Needed if multiple FTDI devices are connected.",
+)
+@click.option(
+    "-i",
+    "--device_id",
+    required=False,
+    default=0,
+    type=int,
+    help="[EXPERIMENTAL FEATURE] ID of the device to update. 0 (default) will update all devices.",
+)
+@click.option(
+    "--reboot",
+    is_flag=True,
+    type=bool,
+    help="Restart the device. Cannot be used with --key or --value.",
+)
+def device(port: str, device_id: int, reboot: bool) -> None:
+    """
+    Send device commands (e.g., reboot)
+    """
+
+    # Check mutual exclusivity
+    if reboot:
+        device_update(
+            port=port, key="DEVICE", value=DeviceCommand.REBOOT.value, device_id=device_id
+        )
+    else:
+        raise click.UsageError("Only --reboot is currently implemented.")

miniscope_io/device_update.py

@@ -1,144 +1,105 @@
 """
-Update miniscope device configuration.
-
-.. todo::
-
-    What kind of devices does this apply to?
-
+Update miniscope device configuration, such as LED, GAIN, etc.
 """
 
-import argparse
-import sys
 import time
+from typing import Optional
 
-import numpy as np
 import serial
+import serial.tools.list_ports
 
-from miniscope_io import init_logger
+from miniscope_io.logging import init_logger
+from miniscope_io.models.devupdate import DevUpdateCommand, UpdateCommandDefinitions
 
-# Parsers for update LED
-updateDeviceParser = argparse.ArgumentParser("updateDevice")
-updateDeviceParser.add_argument("port", help="serial port")
-updateDeviceParser.add_argument("baudrate", help="baudrate")
-updateDeviceParser.add_argument("module", help="module to update")
-updateDeviceParser.add_argument("value", help="LED value")
+logger = init_logger(name="device_update", level="INFO")
+FTDI_VENDOR_ID = 0x0403
+FTDI_PRODUCT_ID = 0x6001
 
 
-def updateDevice() -> None:
+def device_update(
+    target: str,
+    value: int,
+    device_id: int,
+    port: Optional[str] = None,
+) -> None:
     """
-    Script to update hardware settings over a generic UART-USB converter.
-    This script currently supports updating the excitation LED brightness and
-    electrical wetting lens driver gain.
-
-    .. note::
+    Remote update of device configuration.
 
-        Not tested after separating from stream_daq.py.
+    Args:
+        device_id: ID of the device. 0 will update all devices.
+        port: Serial port to which the device is connected.
+        target: What to update on the device (e.g., LED, GAIN).
+        value: Value to which the target should be updated.
 
-    Examples
-    --------
-    >>> updateDevice [COM port] [baudrate] [module] [value]
-
-    ..todo::
-        Test to see if changing package structure broke anything.
+    Returns:
+        None
     """
-    logger = init_logger("streamDaq")
-
-    args = updateDeviceParser.parse_args()
-    moduleList = ["LED", "EWL"]
-
-    ledMAX = 100
-    ledMIN = 0
-
-    ewlMAX = 255
-    ewlMIN = 0
-
-    ledDeviceTag = 0  # 2-bits each for now
-    ewlDeviceTag = 1  # 2-bits each for now
-
-    deviceTagPos = 4
-    preamblePos = 6
-
-    Preamble = [2, 1]  # 2-bits each for now
-
-    uartPayload = 4
-    uartRepeat = 5
-    uartTimeGap = 0.01
-
-    try:
-        assert len(vars(args)) == 4
-    except AssertionError as msg:
-        logger.exception("Usage: updateDevice [COM port] [baudrate] [module] [value]")
-        raise msg
-
-    try:
-        comport = str(args.port)
-    except (ValueError, IndexError) as e:
-        logger.exception(e)
-        raise e
-
-    try:
-        baudrate = int(args.baudrate)
-    except (ValueError, IndexError) as e:
-        logger.exception(e)
-        raise e
-
-    try:
-        module = str(args.module)
-        assert module in moduleList
-    except AssertionError as msg:
-        err_str = "Available modules:\n"
-        for module in moduleList:
-            err_str += "\t" + module + "\n"
-        logger.exception(err_str)
-        raise msg
 
-    try:
-        value = int(args.value)
-    except Exception as e:
-        logger.exception("Value needs to be an integer")
-        raise e
+    if port:
+        logger.info(f"Using port {port}")
+    else:
+        ftdi_port_list = find_ftdi_device()
+        if len(ftdi_port_list) == 0:
+            raise ValueError("No FTDI devices found.")
+        if len(ftdi_port_list) > 1:
+            raise ValueError("Multiple FTDI devices found. Please specify the port.")
+        if len(ftdi_port_list) == 1:
+            port = ftdi_port_list[0]
+            logger.info(f"Using port {port}")
+
+    command = DevUpdateCommand(device_id=device_id, port=port, target=target, value=value)
+    logger.info(f"Updating {target} to {value} on port {port}")
 
     try:
-        if module == "LED":
-            assert value <= ledMAX and value >= ledMIN
-        if module == "EWL":
-            assert value <= ewlMAX and value >= ewlMIN
-    except AssertionError as msg:
-        if module == "LED":
-            logger.exception("LED value need to be a integer within 0-100")
-        if module == "EWL":
-            logger.exception("EWL value need to be an integer within 0-255")
-        raise msg
-
-    if module == "LED":
-        deviceTag = ledDeviceTag << deviceTagPos
-    elif module == "EWL":
-        deviceTag = ewlDeviceTag << deviceTagPos
-
-    command = [0, 0]
-
-    command[0] = int(
-        Preamble[0] * 2**preamblePos + deviceTag + np.floor(value / (2**uartPayload))
-    ).to_bytes(1, "big")
-    command[1] = int(
-        Preamble[1] * 2**preamblePos + deviceTag + value % (2**uartPayload)
-    ).to_bytes(1, "big")
-
-    # set up serial port
-    try:
-        serial_port = serial.Serial(port=comport, baudrate=baudrate, timeout=5, stopbits=1)
+        serial_port = serial.Serial(port=command.port, baudrate=2400, timeout=5, stopbits=2)
     except Exception as e:
         logger.exception(e)
         raise e
     logger.info("Open serial port")
 
-    for uartCommand in command:
-        for _ in range(uartRepeat):
-            # read UART data until preamble and put into queue
-            serial_port.write(uartCommand)
-            time.sleep(uartTimeGap)
+    try:
+        id_command = (command.device_id + UpdateCommandDefinitions.id_header) & 0xFF
+        serial_port.write(id_command.to_bytes(1, "big"))
+        logger.debug(f"Command: {format(id_command, '08b')}; Device ID: {command.device_id}")
+        time.sleep(0.1)
+
+        target_command = (command.target.value + UpdateCommandDefinitions.target_header) & 0xFF
+        serial_port.write(target_command.to_bytes(1, "big"))
+        logger.debug(f"Command: {format(target_command, '08b')}; Target: {command.target.name}")
+        time.sleep(0.1)
+
+        value_LSB_command = (
+            (command.value & UpdateCommandDefinitions.LSB_value_mask)
+            + UpdateCommandDefinitions.LSB_header
+        ) & 0xFF
+        serial_port.write(value_LSB_command.to_bytes(1, "big"))
+        logger.debug(f"Command: {format(value_LSB_command, '08b')}; Value: {command.value} (LSB)")
+        time.sleep(0.1)
+
+        value_MSB_command = (
+            ((command.value & UpdateCommandDefinitions.MSB_value_mask) >> 6)
+            + UpdateCommandDefinitions.MSB_header
+        ) & 0xFF
+        serial_port.write(value_MSB_command.to_bytes(1, "big"))
+        logger.debug(f"Command: {format(value_MSB_command, '08b')}; Value: {command.value} (MSB)")
+        time.sleep(0.1)
+
+        serial_port.write(UpdateCommandDefinitions.reset_byte.to_bytes(1, "big"))
+
+    finally:
+        serial_port.close()
+        logger.info("Closed serial port")
+
+
+def find_ftdi_device() -> list[str]:
+    """
+    Find FTDI devices connected to the computer.
+    """
+    ports = serial.tools.list_ports.comports()
+    ftdi_ports = []
+
+    for port in ports:
+        if port.vid == FTDI_VENDOR_ID and port.pid == FTDI_PRODUCT_ID:
+            ftdi_ports.append(port.device)
 
-    serial_port.close()
-    logger.info("\t" + module + ": " + str(value))
-    logger.info("Close serial port")
-    sys.exit(1)
+    return ftdi_ports