From 0fbd21737441eb32804e662726620d3b4448d62d Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 09:56:44 +0100 Subject: [PATCH 1/6] Initial USB support --- pyproject.toml | 1 + tui/gdtf_share/gdtf.py | 2 +- tui/messages.py | 11 +- tui/rdm_search.py | 596 +++++++++++++++++++++++++++++++++++++++++ tui/screens.py | 48 +++- uv.lock | 11 + 6 files changed, 661 insertions(+), 8 deletions(-) create mode 100644 tui/rdm_search.py diff --git a/pyproject.toml b/pyproject.toml index f85d138..c2332e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ requires-python = ">=3.13" dependencies = [ "ifaddr>=0.2.0", "pymvr>=1.0.4", + "pyserial>=3.5", "requests>=2.32.5", "textual>=6.5.0", "textual-fspicker>=0.6.0", diff --git a/tui/gdtf_share/gdtf.py b/tui/gdtf_share/gdtf.py index eba893b..83ebc77 100644 --- a/tui/gdtf_share/gdtf.py +++ b/tui/gdtf_share/gdtf.py @@ -29,7 +29,7 @@ from textual.widgets import Button, Static, Input, Label, Checkbox, Select, Switch from textual import on, work, events from textual_fspicker import FileOpen, Filters -from tui.messages import Errors, DevicesDiscovered +from tui.messages import Errors, NetworkDevicesDiscovered from tui.network import get_network_cards from tui.artnet import ArtNetDiscovery from tui.share_api_client import update_data, download_files diff --git a/tui/messages.py b/tui/messages.py index be4c27a..d0d4100 100644 --- a/tui/messages.py +++ b/tui/messages.py @@ -35,7 +35,16 @@ def __init__(self, error: str | None = None) -> None: super().__init__() -class DevicesDiscovered(Message): +class NetworkDevicesDiscovered(Message): + """Message sent when monitors are fetched from the API.""" + + def __init__(self, devices: list | None = None, error: str = "") -> None: + self.devices = devices + self.error = error + super().__init__() + + +class RdmDevicesDiscovered(Message): """Message sent when monitors are fetched from the API.""" def __init__(self, devices: list | None = None, error: str = "") -> None: diff --git a/tui/rdm_search.py b/tui/rdm_search.py new file mode 100644 index 0000000..6e3f3f8 --- /dev/null +++ b/tui/rdm_search.py @@ -0,0 +1,596 @@ +import serial +import time +import struct +import random + +# Robe Universal Interface API constants +HEADER = 0xA5 +PACKET_TYPE_RDM_RESPONSE = 0x11 +PACKET_TYPE_RDM_PACKET_OUT = 0x10 +PACKET_TYPE_RDM_DISCOVERY_UNIQUE_BRANCH = 0x12 +PACKET_TYPE_RDM_DISCOVERY_RESPONSE = 0x13 +PACKET_TYPE_RDM_INFO_COMMAND = 0x14 +PACKET_TYPE_RDM_INFO_RESPONSE = 0x15 + +# RDM constants +RDM_START_CODE = 0xCC +RDM_SUB_START_CODE = 0x01 +BROADCAST_ALL_DEVICES_ID = b"\xff\xff\xff\xff\xff\xff" +# This is the UID of the Robe Universal Interface itself, acting as the controller +CONTROLLER_UID = b"\x52\x53\x02\x00\x00\x15" + +# RDM Command Class +DISCOVERY_COMMAND = 0x10 +DISCOVERY_COMMAND_RESPONSE = 0x11 +GET_COMMAND = 0x20 +GET_COMMAND_RESPONSE = 0x21 +SET_COMMAND = 0x30 +SET_COMMAND_RESPONSE = 0x31 + +# RDM Parameter IDs (PID) +DISC_UNIQUE_BRANCH = 0x0001 +DISC_MUTE = 0x0002 +DISC_UN_MUTE = 0x0003 +SUPPORTED_PARAMETERS = 0x0050 +DEVICE_INFO = 0x0060 +DEVICE_MODEL_DESCRIPTION = 0x0080 +MANUFACTURER_LABEL = 0x0081 +DEVICE_LABEL = 0x0082 +SOFTWARE_VERSION_LABEL = 0x00C0 +DMX_START_ADDRESS = 0x00F0 + +RDM_PARAMETER_NAMES = { + v: k for k, v in globals().items() if k.isupper() and isinstance(v, int) +} + +RDM_RESPONSE_TYPE_NAMES = { + 0x00: "RESPONSE_TYPE_ACK", + 0x01: "RESPONSE_TYPE_ACK_TIMER", + 0x02: "RESPONSE_TYPE_NACK_REASON", + 0x03: "RESPONSE_TYPE_ACK_OVERFLOW", +} + + +def calculate_byte_sum_crc(data: bytes) -> int: + """Calculates the 1-byte sum of all bytes.""" + return sum(data) & 0xFF + + +def build_robe_packet(packet_type: int, rdm_packet: bytes) -> bytes: + """Builds the final packet for the Robe Universal Interface.""" + data_to_wrap = rdm_packet + # As per Robe API, 4 random bytes are appended for certain packet types + if packet_type in [ + PACKET_TYPE_RDM_PACKET_OUT, + PACKET_TYPE_RDM_DISCOVERY_UNIQUE_BRANCH, + ]: + data_to_wrap += bytes(random.getrandbits(8) for _ in range(4)) + + data_len = len(data_to_wrap) + header_part = bytearray( + [HEADER, packet_type, data_len & 0xFF, (data_len >> 8) & 0xFF] + ) + header_crc = calculate_byte_sum_crc(header_part) + + packet = bytearray() + packet.extend(header_part) + packet.append(header_crc) + packet.extend(data_to_wrap) + + all_crc = calculate_byte_sum_crc(packet) + packet.append(all_crc) + + return bytes(packet) + + +def calculate_rdm_checksum(data: bytes) -> int: + """Calculates the 16-bit RDM checksum.""" + return sum(data) + + +def build_rdm_packet( + dest_uid: bytes, tn: int, cc: int, pid: int, pd: bytes = b"" +) -> bytes: + """Builds an RDM packet (the part that goes into the Robe packet).""" + pdl = len(pd) + message_length = 24 + pdl # Standard length without checksum + + packet_for_checksum = bytearray( + [RDM_START_CODE, RDM_SUB_START_CODE, message_length] + ) + packet_for_checksum.extend(dest_uid) + packet_for_checksum.extend(CONTROLLER_UID) + packet_for_checksum.extend([tn, 1, 0]) # TN, Port ID, Message Count + packet_for_checksum.extend(b"\x00\x00") # Sub-device + packet_for_checksum.append(cc) + packet_for_checksum.extend(struct.pack(">H", pid)) + packet_for_checksum.append(pdl) + packet_for_checksum.extend(pd) + + checksum = calculate_rdm_checksum(packet_for_checksum) + + # Final RDM message (without Start Code) to be wrapped in a Robe packet + rdm_message = packet_for_checksum[1:] + rdm_message.extend(struct.pack(">H", checksum)) + + return bytes(rdm_message) + + +def parse_text_response(pd: bytes): + """Parses a simple null-terminated string response.""" + try: + print(f" └─ Text: {pd.decode('utf-8', errors='ignore').strip()}") + except Exception as e: + print(f" └─ Error decoding text: {e}") + + +def parse_supported_parameters(pd: bytes): + """Parses a list of supported PIDs.""" + pids = struct.unpack(f">{len(pd) // 2}H", pd) + print(" └─ Supported PIDs:") + for pid in pids: + name = RDM_PARAMETER_NAMES.get(pid, "Unknown") + print(f" - 0x{pid:04x} ({name})") + + +def parse_device_info(pd: bytes): + """Parses the DEVICE_INFO response.""" + ( + rdm_version, + model_id, + category, + sw_version, + footprint, + personality, + dmx_address, + sub_device_count, + sensor_count, + ) = struct.unpack(">HHHIHHHHB", pd) + current_personality = personality >> 8 + total_personalities = personality & 0xFF + print(" └─ Device Info:") + print(f" - RDM Version: {rdm_version >> 8}.{rdm_version & 0xFF}") + print(f" - Device Model ID: 0x{model_id:04x}") + print(f" - Product Category: 0x{category:04x}") + print(f" - Software Version ID: 0x{sw_version:08x}") + print(f" - DMX512 Footprint: {footprint}") + print( + f" - DMX512 Personality: {current_personality} of {total_personalities}" + ) + print(f" - DMX Start Address: {dmx_address}") + print(f" - Sub-device Count: {sub_device_count}") + print(f" - Sensor Count: {sensor_count}") + + +def parse_dmx_start_address(pd: bytes): + """Parses the DMX_START_ADDRESS response.""" + address = struct.unpack(">H", pd)[0] + print(f" └─ DMX Start Address: {address}") + + +def parse_ack(pd: bytes, pid: int, cc: int): + """Parses a generic ACK response.""" + if not pd: + print(" └─ Acknowledged (no data).") + # For MUTE/UNMUTE, the PD is a 2-byte control field + elif pid in [DISC_MUTE, DISC_UN_MUTE]: + control_field = struct.unpack(">H", pd)[0] + print(f" └─ Acknowledged. Control Field: 0x{control_field:04x}") + else: + print(f" └─ Acknowledged with data: {pd.hex(' ')}") + + +def parse_discovery_response(rdm_data: bytes): + """Parses a DISC_UNIQUE_BRANCH response and returns the decoded UID.""" + print(" └─ Parsing Discovery response...") + try: + # Find the preamble separator + separator_index = rdm_data.find(b"\xaa") + if separator_index == -1: + print(" └─ Discovery response separator (0xAA) not found.") + return None + + # The EUID and ECS follow the separator + euid_ecs_data = rdm_data[separator_index + 1 :] + if len(euid_ecs_data) < 16: + print( + f" └─ Insufficient data for EUID and Checksum (found {len(euid_ecs_data)} bytes)." + ) + return None + + euid = euid_ecs_data[:12] + ecs = euid_ecs_data[12:16] + + # Decode UID + uid = bytearray() + for i in range(6): + uid.append(euid[i * 2] & euid[i * 2 + 1]) + uid = bytes(uid) + print(f" ├─ Discovered UID: {uid.hex(':')}") + + # Verify checksum + calculated_checksum = sum(euid) + + cs_msb = ecs[0] & ecs[1] + cs_lsb = ecs[2] & ecs[3] + received_checksum = (cs_msb << 8) | cs_lsb + + if calculated_checksum == received_checksum: + print(f" └─ Checksum OK (0x{received_checksum:04x})") + else: + print( + f" └─ Checksum mismatch! Calculated: 0x{calculated_checksum:04x}, Received: 0x{received_checksum:04x}" + ) + + return uid + + except Exception as e: + print(f" └─ Error parsing discovery response: {e}") + return None + + +def parse_rdm_response(rdm_data: bytes, sent_pid: int): + """Parses the core RDM response packet.""" + if not rdm_data: + print(" └─ Empty RDM data.") + return + + try: + sub_start = rdm_data[0] + msg_len = rdm_data[1] + dest_uid = rdm_data[2:8] + src_uid = rdm_data[8:14] + tn = rdm_data[14] + response_type = rdm_data[15] + msg_count = rdm_data[16] + sub_device = struct.unpack(">H", rdm_data[17:19])[0] + cc = rdm_data[19] + pid = struct.unpack(">H", rdm_data[20:22])[0] + pdl = rdm_data[22] + pd = rdm_data[23 : 23 + pdl] + checksum = struct.unpack(">H", rdm_data[-2:])[0] + + print(" ├─ RDM Response:") + print(f" │ - Source UID: {src_uid.hex(':')}") + print(f" │ - Transaction #: {tn}") + print( + f" │ - Response Type: {RDM_RESPONSE_TYPE_NAMES.get(response_type, 'Unknown')}" + ) + print( + f" │ - Command Class: 0x{cc:02x} ({RDM_PARAMETER_NAMES.get(cc, 'Unknown')}_RESPONSE)" + ) + print(f" │ - PID: 0x{pid:04x} ({RDM_PARAMETER_NAMES.get(pid, 'Unknown')})") + print(f" │ - PDL: {pdl}") + + if response_type == 0x00: # ACK + if pid == DEVICE_INFO: + parse_device_info(pd) + elif pid == SUPPORTED_PARAMETERS: + parse_supported_parameters(pd) + elif pid in [ + DEVICE_LABEL, + SOFTWARE_VERSION_LABEL, + MANUFACTURER_LABEL, + DEVICE_MODEL_DESCRIPTION, + ]: + parse_text_response(pd) + elif pid == DMX_START_ADDRESS: + parse_dmx_start_address(pd) + else: + parse_ack(pd, pid, cc) + else: + print(f" └─ Received NACK or other response type.") + + except Exception as e: + print(f" └─ Error parsing RDM response: {e} (Data: {rdm_data.hex()})") + + +def parse_robe_response(response: bytes, sent_pid: int): + """ + Parses the outer Robe packet, dispatches RDM parsing, + and returns a status tuple (type, data). + """ + print(" ├─ Parsing Robe response...") + if not response or response[0] != HEADER: + print(" └─ Invalid or empty response.") + return "error", None + + packet_type = response[1] + data_len = struct.unpack(" upper_bound: + return tn + + # Base case: If we are searching a single UID, try to mute it. + if lower_bound == upper_bound: + print(f"\n--- Checking single UID: {lower_bound:012x} ---") + uid_to_check = struct.pack(">Q", lower_bound)[2:] + + # Per RDM spec, send DISC_MUTE directly when at the lowest branch. + # A device will respond with an ACK if it exists at this UID. + rdm_packet_mute = build_rdm_packet( + uid_to_check, tn, DISCOVERY_COMMAND, DISC_MUTE + ) + robe_packet_mute = build_robe_packet( + PACKET_TYPE_RDM_PACKET_OUT, rdm_packet_mute + ) + + status, data = send_and_receive( + ser, f"Mute Check ({uid_to_check.hex()})", robe_packet_mute, DISC_MUTE + ) + tn += 1 + + if status == "ack": + uid = uid_to_check + if uid not in discovered_uids: + print(f"--- Found new device: {uid.hex(':')} ---") + discovered_uids.append(uid) + return tn + + # Recursive step for a range + print(f"\n--- Searching range: {lower_bound:012x} to {upper_bound:012x} ---") + pd = struct.pack(">Q", lower_bound)[2:] + struct.pack(">Q", upper_bound)[2:] + rdm_packet = build_rdm_packet( + BROADCAST_ALL_DEVICES_ID, tn, DISCOVERY_COMMAND, DISC_UNIQUE_BRANCH, pd + ) + robe_packet = build_robe_packet(PACKET_TYPE_RDM_DISCOVERY_UNIQUE_BRANCH, rdm_packet) + + status, data = send_and_receive( + ser, + f"Discovery Branch ({lower_bound:012x}-{upper_bound:012x})", + robe_packet, + DISC_UNIQUE_BRANCH, + ) + tn += 1 + + # If there was any kind of response (a single UID or a collision), + # we need to take action. + if status == "uid": + uid = data + if uid not in discovered_uids: + print(f"--- Found new device: {uid.hex(':')} ---") + discovered_uids.append(uid) + # Mute the device so it doesn't respond to further discovery messages + rdm_packet_mute = build_rdm_packet(uid, tn, DISCOVERY_COMMAND, DISC_MUTE) + robe_packet_mute = build_robe_packet( + PACKET_TYPE_RDM_PACKET_OUT, rdm_packet_mute + ) + send_and_receive( + ser, f"Mute Device ({uid.hex()})", robe_packet_mute, DISC_MUTE + ) + tn += 1 + + # After muting, search the same range again to find other devices. + # If the muted device was the only one, the next search will yield 'no_response'. + tn = binary_search_branch(ser, tn, lower_bound, upper_bound, discovered_uids) + + elif status == "collision": + print("--- Collision detected, branching... ---") + mid_point = (lower_bound + upper_bound) // 2 + tn = binary_search_branch(ser, tn, lower_bound, mid_point, discovered_uids) + tn = binary_search_branch(ser, tn, mid_point + 1, upper_bound, discovered_uids) + + else: # 'no_response' or other + print("--- No devices in this range. ---") + + return tn + + +def discover_all_devices(ser: serial.Serial, tn: int): + """ + Discovers all RDM devices on the line using a binary search algorithm. + Finally, it un-mutes all discovered devices. + """ + discovered_uids = [] + + # 1. Un-mute all devices to start fresh + print("\n--- Sending Un-Mute All to start discovery ---") + rdm_packet_unmute = build_rdm_packet( + BROADCAST_ALL_DEVICES_ID, tn, DISCOVERY_COMMAND, DISC_UN_MUTE + ) + robe_packet_unmute = build_robe_packet( + PACKET_TYPE_RDM_PACKET_OUT, rdm_packet_unmute + ) + send_and_receive(ser, "Un-Mute All Devices", robe_packet_unmute, DISC_UN_MUTE) + tn += 1 + + # 2. Start the recursive binary search + full_range_upper = 0xFFFFFFFFFFFF + tn = binary_search_branch(ser, tn, 0, full_range_upper, discovered_uids) + + # 3. Un-mute all discovered devices so they can be addressed normally + if discovered_uids: + print("\n--- Un-muting all discovered devices ---") + rdm_packet_unmute_final = build_rdm_packet( + BROADCAST_ALL_DEVICES_ID, tn, DISCOVERY_COMMAND, DISC_UN_MUTE + ) + robe_packet_unmute_final = build_robe_packet( + PACKET_TYPE_RDM_PACKET_OUT, rdm_packet_unmute_final + ) + send_and_receive( + ser, "Un-Mute All Devices", robe_packet_unmute_final, DISC_UN_MUTE + ) + tn += 1 + else: + print("\n--- No devices were found during discovery. ---") + + return discovered_uids, tn + + +def get_device_parameters(ser: serial.Serial, discovered_uid: bytes, tn: int): + """ + Retrieves a standard set of parameters from a discovered RDM device. + """ + print(f"\n--- Getting parameters for device: {discovered_uid.hex(':')} ---") + + # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, SUPPORTED_PARAMETERS) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive(ser, "Get Supported Parameters", robe_packet, SUPPORTED_PARAMETERS) + # tn += 1 + + rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DEVICE_INFO) + robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + send_and_receive(ser, "Get Device Info", robe_packet, DEVICE_INFO) + tn += 1 + + # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, MANUFACTURER_LABEL) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive(ser, "Get Manufacturer Label", robe_packet, MANUFACTURER_LABEL) + # tn += 1 + + rdm_packet = build_rdm_packet( + discovered_uid, tn, GET_COMMAND, DEVICE_MODEL_DESCRIPTION + ) + robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + send_and_receive( + ser, "Get Device Model Description", robe_packet, DEVICE_MODEL_DESCRIPTION + ) + tn += 1 + + # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DEVICE_LABEL) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive(ser, "Get Device Label", robe_packet, DEVICE_LABEL) + # tn += 1 + + # rdm_packet = build_rdm_packet( + # discovered_uid, tn, GET_COMMAND, SOFTWARE_VERSION_LABEL + # ) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive( + # ser, "Get Software Version Label", robe_packet, SOFTWARE_VERSION_LABEL + # ) + # tn += 1 + + # dmx_address = 1 + # pd = struct.pack(">H", dmx_address) + # rdm_packet = build_rdm_packet( + # discovered_uid, tn, SET_COMMAND, DMX_START_ADDRESS, pd + # ) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive( + # ser, f"Set DMX Start Address to {dmx_address}", robe_packet, DMX_START_ADDRESS + # ) + # tn += 1 + + # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DMX_START_ADDRESS) + # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) + # send_and_receive(ser, "Get DMX Start Address", robe_packet, DMX_START_ADDRESS) + # tn += 1 + + return tn + + +def get_device_info(device_port): + ser = serial.Serial(device_port, baudrate=250000, timeout=0.1) + robe_packet = build_robe_packet(PACKET_TYPE_RDM_INFO_COMMAND, b"") + ser.write(robe_packet) + time.sleep(0.2) + response = ser.read(ser.in_waiting) + print("response", response) + ser.close() + if response and response[0] == HEADER: + if response[1] == PACKET_TYPE_RDM_INFO_RESPONSE: + return True + + +def main(device_port): + """Main function to run the device search and communication flow.""" + try: + ser = serial.Serial(device_port, baudrate=250000, timeout=0.1) + print(f"Serial port {device_port} opened successfully.\n") + except serial.SerialException as e: + print(f"Error opening serial port: {e}") + return [] + + tn = 0 # Transaction Number + + print("--- Starting RDM Discovery ---") + discovered_uids, tn = discover_all_devices(ser, tn) + print("found this", discovered_uids) + return [ + { + "reported_ip": "", + "source_ip": "", + "short_name": uid.hex(":"), + "long_name": uid.hex(":"), + } + for uid in discovered_uids + ] + + if not discovered_uids: + print("\n--- No devices discovered. Exiting. ---") + ser.close() + return [] + + print(f"\n--- Discovery finished. Found {len(discovered_uids)} device(s). ---") + + # for uid in discovered_uids: + # tn = get_device_parameters(ser, uid, tn) + # ser.close() + + # return { + # "reported_ip": reported_ip, + # "source_ip": addr[0], + # "short_name": short_name, + # "long_name": long_name, + # } + print("\nCommunication flow complete. Serial port closed.") + + +if __name__ == "__main__": + main() diff --git a/tui/screens.py b/tui/screens.py index a969c6c..6416f0b 100644 --- a/tui/screens.py +++ b/tui/screens.py @@ -22,12 +22,15 @@ from textual.widgets import Button, Static, Input, Label, Checkbox, Select, Switch from textual import on, work, events from textual_fspicker import FileOpen, Filters -from tui.messages import Errors, DevicesDiscovered +from tui.messages import Errors, NetworkDevicesDiscovered, RdmDevicesDiscovered from tui.network import get_network_cards from tui.artnet import ArtNetDiscovery +from tui.rdm_search import get_device_info, main import re import sys import json +import serial +import serial.tools.list_ports class QuitScreen(ModalScreen[bool]): @@ -184,10 +187,33 @@ def on_mount(self): if any(ip == "0.0.0.0" for name, ip in self.networks): select_widget.value = "0.0.0.0" # for Win select_widget.refresh() # Force redraw + self.get_robe_devices() + + @work(thread=True) + def get_robe_devices(self) -> None: + """Find USB devices in a background worker.""" + devices = [] + ports = serial.tools.list_ports.comports() + for port in ports: + if port.description and "Runit WTX" in port.description: + print(f"Found port: {port.device} - {port.description}") + if get_device_info(port.device): + devices.append(port) + self.app.call_from_thread(self.update_devices, devices) + + def update_devices(self, devices: list) -> None: + """Update the Select widget with the found devices.""" + sel = self.query_one("#networks_select", Select) + options = self.networks + options += [(f"{port.product}: {port.device}", port.device) for port in devices] + sel.set_options(options) def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "do_start": - self.run_discovery() + if "dev" in self.network: + self.run_rdm_discovery() + else: + self.run_network_discovery() btn = self.query_one("#do_start") btn.disabled = True btn.label = "...discovering..." @@ -201,7 +227,17 @@ def on_select_changed(self, event: Select.Changed) -> None: self.query_one("#do_start").disabled = False @work(thread=True) - async def run_discovery(self) -> str: + async def run_rdm_discovery(self) -> str: + try: + results_widget = self.query_one("#results", Static) + results_widget.update(f"Searching...") + discovered_devices = main(self.network) + self.post_message(NetworkDevicesDiscovered(devices=discovered_devices)) + except Exception as e: + self.post_message(NetworkDevicesDiscovered(error=str(e))) + + @work(thread=True) + async def run_network_discovery(self) -> str: try: results_widget = self.query_one("#results", Static) results_widget.update( @@ -212,9 +248,9 @@ async def run_discovery(self) -> str: timeout = float(self.app.configuration.artnet_timeout) result = discovery.discover_devices(timeout=timeout) discovery.stop() # not really needed, as the thread will close... - self.post_message(DevicesDiscovered(devices=result)) + self.post_message(NetworkDevicesDiscovered(devices=result)) except Exception as e: - self.post_message(DevicesDiscovered(error=str(e))) + self.post_message(NetworkDevicesDiscovered(error=str(e))) def extract_uni_dmx(self, long_name): address = None @@ -227,7 +263,7 @@ def extract_uni_dmx(self, long_name): universe = match.group(2) return universe, address - def on_devices_discovered(self, message: DevicesDiscovered) -> None: + def on_network_devices_discovered(self, message: NetworkDevicesDiscovered) -> None: devices = [] results_widget = self.query_one("#results", Static) if message.devices: diff --git a/uv.lock b/uv.lock index e5abbea..1b1f60d 100644 --- a/uv.lock +++ b/uv.lock @@ -560,6 +560,7 @@ source = { virtual = "." } dependencies = [ { name = "ifaddr" }, { name = "pymvr" }, + { name = "pyserial" }, { name = "requests" }, { name = "textual" }, { name = "textual-fspicker" }, @@ -575,6 +576,7 @@ dev = [ requires-dist = [ { name = "ifaddr", specifier = ">=0.2.0" }, { name = "pymvr", specifier = ">=1.0.4" }, + { name = "pyserial", specifier = ">=3.5" }, { name = "requests", specifier = ">=2.32.5" }, { name = "textual", specifier = ">=6.5.0" }, { name = "textual-fspicker", specifier = ">=0.6.0" }, @@ -713,6 +715,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/27/aab12298c910492a5dd04db66e733f6341acdcc252b991b3532ed92f1b80/pymvr-1.0.4-py3-none-any.whl", hash = "sha256:4126945e15b05b2443495a0c2fa39803859713288291490871a9a45c31d2f105", size = 15009, upload-time = "2025-09-14T08:35:46.964Z" }, ] +[[package]] +name = "pyserial" +version = "3.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1e/7d/ae3f0a63f41e4d2f6cb66a5b57197850f919f59e558159a4dd3a818f5082/pyserial-3.5.tar.gz", hash = "sha256:3c77e014170dfffbd816e6ffc205e9842efb10be9f58ec16d3e8675b4925cddb", size = 159125, upload-time = "2020-11-23T03:59:15.045Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/bc/587a445451b253b285629263eb51c2d8e9bcea4fc97826266d186f96f558/pyserial-3.5-py2.py3-none-any.whl", hash = "sha256:c4451db6ba391ca6ca299fb3ec7bae67a5c55dde170964c7a14ceefec02f2cf0", size = 90585, upload-time = "2020-11-23T03:59:13.41Z" }, +] + [[package]] name = "pywin32-ctypes" version = "0.2.3" From 7fbaa9089c71495b01f66b7d691c17da4e050736 Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 11:16:27 +0100 Subject: [PATCH 2/6] Split search --- tui/rdm_search.py | 178 +++++++++++++++++++++------------------------- tui/screens.py | 80 ++++++++++++++++++--- 2 files changed, 155 insertions(+), 103 deletions(-) diff --git a/tui/rdm_search.py b/tui/rdm_search.py index 6e3f3f8..8f815a6 100644 --- a/tui/rdm_search.py +++ b/tui/rdm_search.py @@ -119,9 +119,12 @@ def build_rdm_packet( def parse_text_response(pd: bytes): """Parses a simple null-terminated string response.""" try: - print(f" └─ Text: {pd.decode('utf-8', errors='ignore').strip()}") + text = pd.decode("utf-8", errors="ignore").strip() + print(f" └─ Text: {text}") + return text except Exception as e: print(f" └─ Error decoding text: {e}") + return None def parse_supported_parameters(pd: bytes): @@ -131,6 +134,7 @@ def parse_supported_parameters(pd: bytes): for pid in pids: name = RDM_PARAMETER_NAMES.get(pid, "Unknown") print(f" - 0x{pid:04x} ({name})") + return list(pids) def parse_device_info(pd: bytes): @@ -160,24 +164,42 @@ def parse_device_info(pd: bytes): print(f" - DMX Start Address: {dmx_address}") print(f" - Sub-device Count: {sub_device_count}") print(f" - Sensor Count: {sensor_count}") + return { + "rdm_protocol_version": f"{rdm_version >> 8}.{rdm_version & 0xFF}", + "device_model_id": model_id, + "product_category": category, + "software_version_id": sw_version, + "dmx512_footprint": footprint, + "dmx_personality": { + "current": current_personality, + "count": total_personalities, + }, + "dmx_start_address": dmx_address, + "sub_device_count": sub_device_count, + "sensor_count": sensor_count, + } def parse_dmx_start_address(pd: bytes): """Parses the DMX_START_ADDRESS response.""" address = struct.unpack(">H", pd)[0] print(f" └─ DMX Start Address: {address}") + return address def parse_ack(pd: bytes, pid: int, cc: int): """Parses a generic ACK response.""" if not pd: print(" └─ Acknowledged (no data).") + return True # For MUTE/UNMUTE, the PD is a 2-byte control field elif pid in [DISC_MUTE, DISC_UN_MUTE]: control_field = struct.unpack(">H", pd)[0] print(f" └─ Acknowledged. Control Field: 0x{control_field:04x}") + return control_field else: print(f" └─ Acknowledged with data: {pd.hex(' ')}") + return pd def parse_discovery_response(rdm_data: bytes): @@ -233,7 +255,7 @@ def parse_rdm_response(rdm_data: bytes, sent_pid: int): """Parses the core RDM response packet.""" if not rdm_data: print(" └─ Empty RDM data.") - return + return None, None try: sub_start = rdm_data[0] @@ -262,27 +284,31 @@ def parse_rdm_response(rdm_data: bytes, sent_pid: int): print(f" │ - PID: 0x{pid:04x} ({RDM_PARAMETER_NAMES.get(pid, 'Unknown')})") print(f" │ - PDL: {pdl}") + response_data = None if response_type == 0x00: # ACK if pid == DEVICE_INFO: - parse_device_info(pd) + response_data = parse_device_info(pd) elif pid == SUPPORTED_PARAMETERS: - parse_supported_parameters(pd) + response_data = parse_supported_parameters(pd) elif pid in [ DEVICE_LABEL, SOFTWARE_VERSION_LABEL, MANUFACTURER_LABEL, DEVICE_MODEL_DESCRIPTION, ]: - parse_text_response(pd) + response_data = parse_text_response(pd) elif pid == DMX_START_ADDRESS: - parse_dmx_start_address(pd) + response_data = parse_dmx_start_address(pd) else: - parse_ack(pd, pid, cc) + response_data = parse_ack(pd, pid, cc) else: print(f" └─ Received NACK or other response type.") + return pid, response_data + except Exception as e: print(f" └─ Error parsing RDM response: {e} (Data: {rdm_data.hex()})") + return None, None def parse_robe_response(response: bytes, sent_pid: int): @@ -306,8 +332,11 @@ def parse_robe_response(response: bytes, sent_pid: int): rdm_data = rdm_data_with_trailer[:-4] if packet_type == PACKET_TYPE_RDM_RESPONSE: - parse_rdm_response(rdm_data, sent_pid) - return "ack", None # Assuming any standard response is an ACK for our purposes + pid, rdm_response_data = parse_rdm_response(rdm_data, sent_pid) + return "ack", ( + pid, + rdm_response_data, + ) # Assuming any standard response is an ACK for our purposes elif packet_type == PACKET_TYPE_RDM_DISCOVERY_RESPONSE: # Per the Robe API, a 4-byte data length for a discovery response # means the interface timed out waiting for a real RDM response. @@ -478,61 +507,35 @@ def get_device_parameters(ser: serial.Serial, discovered_uid: bytes, tn: int): """ print(f"\n--- Getting parameters for device: {discovered_uid.hex(':')} ---") - # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, SUPPORTED_PARAMETERS) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive(ser, "Get Supported Parameters", robe_packet, SUPPORTED_PARAMETERS) - # tn += 1 + device_data = {"uid": discovered_uid.hex(":")} - rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DEVICE_INFO) - robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - send_and_receive(ser, "Get Device Info", robe_packet, DEVICE_INFO) - tn += 1 + pids_to_get = [ + SUPPORTED_PARAMETERS, + DEVICE_INFO, + MANUFACTURER_LABEL, + DEVICE_MODEL_DESCRIPTION, + DEVICE_LABEL, + SOFTWARE_VERSION_LABEL, + DMX_START_ADDRESS, + ] - # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, MANUFACTURER_LABEL) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive(ser, "Get Manufacturer Label", robe_packet, MANUFACTURER_LABEL) - # tn += 1 + for pid in pids_to_get: + rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, pid) + robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - rdm_packet = build_rdm_packet( - discovered_uid, tn, GET_COMMAND, DEVICE_MODEL_DESCRIPTION - ) - robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - send_and_receive( - ser, "Get Device Model Description", robe_packet, DEVICE_MODEL_DESCRIPTION - ) - tn += 1 + pid_name = RDM_PARAMETER_NAMES.get(pid, f"0x{pid:04x}") + status, data = send_and_receive(ser, f"Get {pid_name}", robe_packet, pid) + tn += 1 - # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DEVICE_LABEL) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive(ser, "Get Device Label", robe_packet, DEVICE_LABEL) - # tn += 1 - - # rdm_packet = build_rdm_packet( - # discovered_uid, tn, GET_COMMAND, SOFTWARE_VERSION_LABEL - # ) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive( - # ser, "Get Software Version Label", robe_packet, SOFTWARE_VERSION_LABEL - # ) - # tn += 1 - - # dmx_address = 1 - # pd = struct.pack(">H", dmx_address) - # rdm_packet = build_rdm_packet( - # discovered_uid, tn, SET_COMMAND, DMX_START_ADDRESS, pd - # ) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive( - # ser, f"Set DMX Start Address to {dmx_address}", robe_packet, DMX_START_ADDRESS - # ) - # tn += 1 - - # rdm_packet = build_rdm_packet(discovered_uid, tn, GET_COMMAND, DMX_START_ADDRESS) - # robe_packet = build_robe_packet(PACKET_TYPE_RDM_PACKET_OUT, rdm_packet) - # send_and_receive(ser, "Get DMX Start Address", robe_packet, DMX_START_ADDRESS) - # tn += 1 + if status == "ack" and data and data[1] is not None: + # data is (pid, response_data) + returned_pid, response_data = data + pid_name_key = RDM_PARAMETER_NAMES.get( + returned_pid, f"pid_{returned_pid}" + ).lower() + device_data[pid_name_key] = response_data - return tn + return device_data, tn def get_device_info(device_port): @@ -548,49 +551,34 @@ def get_device_info(device_port): return True -def main(device_port): +def get_devices(ser): """Main function to run the device search and communication flow.""" - try: - ser = serial.Serial(device_port, baudrate=250000, timeout=0.1) - print(f"Serial port {device_port} opened successfully.\n") - except serial.SerialException as e: - print(f"Error opening serial port: {e}") - return [] tn = 0 # Transaction Number - print("--- Starting RDM Discovery ---") discovered_uids, tn = discover_all_devices(ser, tn) print("found this", discovered_uids) - return [ - { - "reported_ip": "", - "source_ip": "", - "short_name": uid.hex(":"), - "long_name": uid.hex(":"), - } - for uid in discovered_uids - ] - - if not discovered_uids: - print("\n--- No devices discovered. Exiting. ---") - ser.close() - return [] - - print(f"\n--- Discovery finished. Found {len(discovered_uids)} device(s). ---") + return discovered_uids - # for uid in discovered_uids: - # tn = get_device_parameters(ser, uid, tn) - # ser.close() - # return { - # "reported_ip": reported_ip, - # "source_ip": addr[0], - # "short_name": short_name, - # "long_name": long_name, - # } - print("\nCommunication flow complete. Serial port closed.") +if __name__ == "__main__": + try: + ser = serial.Serial("/dev/ttyUSB0", baudrate=250000, timeout=0.1) + except serial.SerialException as e: + print(f"Error opening serial port: {e}") + result = get_devices(ser) + print(f"{result=}") + tn = 0 + for uid in result: + device_data, tn = get_device_parameters(ser, uid, tn) + print(f"{device_data=}") + ser.close() -if __name__ == "__main__": - main() +def get_port(device_name): + ser = None + try: + ser = serial.Serial(device_name, baudrate=250000, timeout=0.1) + except serial.SerialException as e: + print(f"Error opening serial port: {e}") + return ser diff --git a/tui/screens.py b/tui/screens.py index 6416f0b..bf137ae 100644 --- a/tui/screens.py +++ b/tui/screens.py @@ -25,7 +25,7 @@ from tui.messages import Errors, NetworkDevicesDiscovered, RdmDevicesDiscovered from tui.network import get_network_cards from tui.artnet import ArtNetDiscovery -from tui.rdm_search import get_device_info, main +from tui.rdm_search import get_device_info, get_devices, get_port import re import sys import json @@ -187,10 +187,10 @@ def on_mount(self): if any(ip == "0.0.0.0" for name, ip in self.networks): select_widget.value = "0.0.0.0" # for Win select_widget.refresh() # Force redraw - self.get_robe_devices() + self.get_robe_usb_devices() @work(thread=True) - def get_robe_devices(self) -> None: + def get_robe_usb_devices(self) -> None: """Find USB devices in a background worker.""" devices = [] ports = serial.tools.list_ports.comports() @@ -199,9 +199,9 @@ def get_robe_devices(self) -> None: print(f"Found port: {port.device} - {port.description}") if get_device_info(port.device): devices.append(port) - self.app.call_from_thread(self.update_devices, devices) + self.app.call_from_thread(self.update_usb_devices_list, devices) - def update_devices(self, devices: list) -> None: + def update_usb_devices_list(self, devices: list) -> None: """Update the Select widget with the found devices.""" sel = self.query_one("#networks_select", Select) options = self.networks @@ -231,10 +231,12 @@ async def run_rdm_discovery(self) -> str: try: results_widget = self.query_one("#results", Static) results_widget.update(f"Searching...") - discovered_devices = main(self.network) - self.post_message(NetworkDevicesDiscovered(devices=discovered_devices)) + port = get_port(self.network) + discovered_devices = get_devices(port) + uid_list = [{"uid": uid.hex(":")} for uid in discovered_devices] + self.post_message(RdmDevicesDiscovered(devices=uid_list)) except Exception as e: - self.post_message(NetworkDevicesDiscovered(error=str(e))) + self.post_message(RdmDevicesDiscovered(error=str(e))) @work(thread=True) async def run_network_discovery(self) -> str: @@ -263,6 +265,68 @@ def extract_uni_dmx(self, long_name): universe = match.group(2) return universe, address + def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: + devices = [] + + # { + # "uid": "52:53:00:45:07:f8", + # "device_info": { + # "rdm_protocol_version": "1.0", + # "device_model_id": 75, + # "product_category": 256, + # "software_version_id": 43, + # "dmx512_footprint": 38, + # "dmx_personality": {"current": 1, "count": 4}, + # "dmx_start_address": 1, + # "sub_device_count": 0, + # "sensor_count": 3, + # }, + # "manufacturer_label": "Robe Lighting s.r.o.", + # "device_model_description": "Robin MMX Spot", + # "device_label": "aawww", + # "software_version_label": "Sw.ver. 4.3", + # "dmx_start_address": 1, + # } + results_widget = self.query_one("#results", Static) + if message.devices: + for device in message.devices: + uid = device.get("uid", None) + if uid is None: + continue + + short_name = device.get("device_model_description", uid) + device_info = device.get("device_info", {}) + universe = 0 + ip_address = "" + address = device_info.get("dmx_start_address", "") + devices.append( + SimpleNamespace( + ip_address=ip_address, + short_name=short_name, + universe=universe, + address=address, + ) + ) + result = "\n".join( + f"{item.short_name} {item.ip_address} {item.universe or ''} {item.address or ''}" + for item in devices + ) + + if devices: + result = f"[green]Found {len(devices)}:[/green]\n\n{result}" + + else: + result = f"[red]No devices found {message.error}[/red]" + + self.discovered_devices = devices + results_widget.update(result) + btn = self.query_one("#do_start") + btn.disabled = False + btn.label = "Discover" + if len(devices): + btn = self.query_one("#close_discovery") + btn.label = f"Add {len(devices)} device{'s' if len(devices) > 1 else ''} to MVR Layer" + def on_network_devices_discovered(self, message: NetworkDevicesDiscovered) -> None: devices = [] results_widget = self.query_one("#results", Static) From d6d84c282e64c034105a7ac18534cb94bcd11ef9 Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 12:08:20 +0100 Subject: [PATCH 3/6] Add RDM details --- tui/create_mvr.py | 5 +++- tui/messages.py | 26 ++++++++++++---- tui/rdm_search.py | 7 ++++- tui/screens.py | 76 +++++++++++++++++++++++++++++++++++++---------- 4 files changed, 92 insertions(+), 22 deletions(-) diff --git a/tui/create_mvr.py b/tui/create_mvr.py index 2398235..20753e0 100644 --- a/tui/create_mvr.py +++ b/tui/create_mvr.py @@ -50,7 +50,10 @@ def create_mvr(devices, mvr_layers, gdtf_map, save_to): if net_fixture.short_name in gdtf_map: fixture.gdtf_spec = gdtf_map[net_fixture.short_name] - fixture.addresses.network.append(pymvr.Network(ipv4=net_fixture.ip_address)) + if net_fixture.ip_address: + fixture.addresses.network.append( + pymvr.Network(ipv4=net_fixture.ip_address) + ) if fixture.gdtf_spec: files_to_pack.append( diff --git a/tui/messages.py b/tui/messages.py index d0d4100..8e11111 100644 --- a/tui/messages.py +++ b/tui/messages.py @@ -16,6 +16,7 @@ # along with this program. If not, see . from textual.message import Message +from dataclasses import dataclass class MvrParsed(Message): @@ -44,10 +45,25 @@ def __init__(self, devices: list | None = None, error: str = "") -> None: super().__init__() +@dataclass class RdmDevicesDiscovered(Message): - """Message sent when monitors are fetched from the API.""" + """Message with discovered devices.""" - def __init__(self, devices: list | None = None, error: str = "") -> None: - self.devices = devices - self.error = error - super().__init__() + devices: list | None = None + error: str = "" + + +@dataclass +class RdmDeviceDetailDiscovered(Message): + """Message with discovered device details.""" + + data: dict | None = None + error: str = "" + + +@dataclass +class RdmDiscoveryMessage(Message): + """Message to signal RDM discovery is complete.""" + + label: str = "" + disabled: bool = False diff --git a/tui/rdm_search.py b/tui/rdm_search.py index 8f815a6..2bda33f 100644 --- a/tui/rdm_search.py +++ b/tui/rdm_search.py @@ -558,7 +558,12 @@ def get_devices(ser): print("--- Starting RDM Discovery ---") discovered_uids, tn = discover_all_devices(ser, tn) print("found this", discovered_uids) - return discovered_uids + return discovered_uids, tn + + +def get_device_details(ser, uid, tn): + device_data, tn = get_device_parameters(ser, uid, tn) + return device_data, tn if __name__ == "__main__": diff --git a/tui/screens.py b/tui/screens.py index bf137ae..16b2e9c 100644 --- a/tui/screens.py +++ b/tui/screens.py @@ -22,10 +22,16 @@ from textual.widgets import Button, Static, Input, Label, Checkbox, Select, Switch from textual import on, work, events from textual_fspicker import FileOpen, Filters -from tui.messages import Errors, NetworkDevicesDiscovered, RdmDevicesDiscovered +from tui.messages import ( + Errors, + NetworkDevicesDiscovered, + RdmDevicesDiscovered, + RdmDeviceDetailDiscovered, + RdmDiscoveryMessage, +) from tui.network import get_network_cards from tui.artnet import ArtNetDiscovery -from tui.rdm_search import get_device_info, get_devices, get_port +from tui.rdm_search import get_device_info, get_devices, get_port, get_device_details import re import sys import json @@ -48,7 +54,7 @@ def compose(self) -> ComposeResult: Static("[bold]Are you sure you want to quit?[/bold]", id="question"), Horizontal( Button("Yes", variant="error", id="yes"), - Button("No", variant="primary", id="no"), + Button("No", "primary", id="no"), id="quit_buttons", ), id="dialog", @@ -117,8 +123,8 @@ def compose(self) -> ComposeResult: password=True, ) yield Horizontal( - Button("Save", variant="success", id="save"), - Button("Cancel", variant="error", id="cancel"), + Button("Save", "success", id="save"), + Button("Cancel", "error", id="cancel"), id="config_buttons", ) @@ -228,15 +234,27 @@ def on_select_changed(self, event: Select.Changed) -> None: @work(thread=True) async def run_rdm_discovery(self) -> str: + port = None try: results_widget = self.query_one("#results", Static) - results_widget.update(f"Searching...") + results_widget.update("Searching...") port = get_port(self.network) - discovered_devices = get_devices(port) - uid_list = [{"uid": uid.hex(":")} for uid in discovered_devices] + discovered_uids, tn = get_devices(port) + uid_list = [{"uid": uid.hex(":")} for uid in discovered_uids] self.post_message(RdmDevicesDiscovered(devices=uid_list)) + + self.post_message( + RdmDiscoveryMessage(label="...getting RDM data...", disabled=True) + ) + for uid in discovered_uids: + device_data, tn = get_device_details(port, uid, tn) + self.post_message(RdmDeviceDetailDiscovered(data=device_data)) + self.post_message(RdmDiscoveryMessage(label="Discover", disabled=False)) except Exception as e: self.post_message(RdmDevicesDiscovered(error=str(e))) + finally: + if port and port.is_open: + port.close() @work(thread=True) async def run_network_discovery(self) -> str: @@ -251,6 +269,7 @@ async def run_network_discovery(self) -> str: result = discovery.discover_devices(timeout=timeout) discovery.stop() # not really needed, as the thread will close... self.post_message(NetworkDevicesDiscovered(devices=result)) + self.post_message(RdmDiscoveryMessage(label="Discover", disabled=False)) except Exception as e: self.post_message(NetworkDevicesDiscovered(error=str(e))) @@ -265,6 +284,11 @@ def extract_uni_dmx(self, long_name): universe = match.group(2) return universe, address + def on_rdm_discovery_message(self, message: RdmDiscoveryMessage) -> None: + btn = self.query_one("#do_start") + btn.disabled = message.disabled + btn.label = message.label + def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: devices = [] @@ -301,6 +325,7 @@ def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: address = device_info.get("dmx_start_address", "") devices.append( SimpleNamespace( + uid=uid, ip_address=ip_address, short_name=short_name, universe=universe, @@ -308,7 +333,7 @@ def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: ) ) result = "\n".join( - f"{item.short_name} {item.ip_address} {item.universe or ''} {item.address or ''}" + f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'Address: {item.address}' if item.address else ''}" for item in devices ) @@ -320,13 +345,37 @@ def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: self.discovered_devices = devices results_widget.update(result) - btn = self.query_one("#do_start") - btn.disabled = False - btn.label = "Discover" + if len(devices): btn = self.query_one("#close_discovery") btn.label = f"Add {len(devices)} device{'s' if len(devices) > 1 else ''} to MVR Layer" + def on_rdm_device_detail_discovered( + self, message: RdmDeviceDetailDiscovered + ) -> None: + """Update a single device with details.""" + if message.data: + uid_to_update = message.data.get("uid") + for i, device in enumerate(self.discovered_devices): + if device.uid == uid_to_update: + device.short_name = message.data.get( + "device_model_description", device.uid + ) + device_info = message.data.get("device_info", {}) + device.address = device_info.get("dmx_start_address", "") + self.discovered_devices[i] = device + break + + # Regenerate the results text + results_text = "\n".join( + f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'Address: {item.address}' if item.address else ''}" + for item in self.discovered_devices + ) + results_widget = self.query_one("#results", Static) + results_widget.update( + f"[green]Found {len(self.discovered_devices)}:[/green]\n\n{results_text}" + ) + def on_network_devices_discovered(self, message: NetworkDevicesDiscovered) -> None: devices = [] results_widget = self.query_one("#results", Static) @@ -356,9 +405,6 @@ def on_network_devices_discovered(self, message: NetworkDevicesDiscovered) -> No self.discovered_devices = devices results_widget.update(result) - btn = self.query_one("#do_start") - btn.disabled = False - btn.label = "Discover" if len(devices): btn = self.query_one("#close_discovery") btn.label = f"Add {len(devices)} device{'s' if len(devices) > 1 else ''} to MVR Layer" From d62b1fa153e608eeb0707aa8e6738f4055ca1d66 Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 13:12:00 +0100 Subject: [PATCH 4/6] Unify listing of interfaces --- tui/network.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tui/network.py b/tui/network.py index 30d9858..1719b6a 100644 --- a/tui/network.py +++ b/tui/network.py @@ -19,7 +19,7 @@ def get_network_cards(): - all_cards = [("All interfaces 0.0.0.0", "0.0.0.0")] + all_cards = [("All Network Interfaces: 0.0.0.0", "0.0.0.0")] for adapter in ifaddr.get_adapters(): for ip in adapter.ips: if isinstance(ip.ip, tuple): # Skip IPv6 @@ -27,7 +27,7 @@ def get_network_cards(): if ip.ip.startswith("169.254."): # Skip link-local continue - label = f"{adapter.nice_name} ({ip.ip})" + label = f"{adapter.nice_name}: {ip.ip}" value = ip.ip all_cards.append((label, value)) return all_cards From 360d6949410aac6d661a445a94b8106d40a29696 Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 13:23:40 +0100 Subject: [PATCH 5/6] Improve device listing --- tui/app.css | 3 +++ tui/app.py | 2 +- tui/screens.py | 4 ++-- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/tui/app.css b/tui/app.css index 576f7b7..47bee3d 100644 --- a/tui/app.css +++ b/tui/app.css @@ -195,6 +195,9 @@ SelectCurrent:focus { border: none; } +GDTFMapping { + height: 2; +} GDTFMapping SelectCurrent, SelectCurrent:focus { border: none; diff --git a/tui/app.py b/tui/app.py index fd7bbc7..8f80bd0 100644 --- a/tui/app.py +++ b/tui/app.py @@ -47,7 +47,7 @@ def update_items(self, items): for fixture in fixtures: self.mount( Static( - f"[green]{fixture.short_name}[/green] {fixture.universe or ''} {fixture.address or ''} {fixture.ip_address} " + f"[green]{fixture.short_name}[/green] {f'IP Address: {fixture.ip_address}' if fixture.ip_address else ''} {f'Universe: {fixture.universe}' if fixture.universe else ''} {f'DMX: {fixture.address}' if fixture.address else ''}" ) ) diff --git a/tui/screens.py b/tui/screens.py index 16b2e9c..e4d2de9 100644 --- a/tui/screens.py +++ b/tui/screens.py @@ -333,7 +333,7 @@ def on_rdm_devices_discovered(self, message: RdmDevicesDiscovered) -> None: ) ) result = "\n".join( - f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'Address: {item.address}' if item.address else ''}" + f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'DMX: {item.address}' if item.address else ''}" for item in devices ) @@ -368,7 +368,7 @@ def on_rdm_device_detail_discovered( # Regenerate the results text results_text = "\n".join( - f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'Address: {item.address}' if item.address else ''}" + f"{item.short_name} {f'IP Address: {item.ip_address}' if item.ip_address else ''} {f'Universe: {item.universe}' if item.universe else ''} {f'DMX: {item.address}' if item.address else ''}" for item in self.discovered_devices ) results_widget = self.query_one("#results", Static) From 6665df309ea5820a9e1754819650dbcefde05446 Mon Sep 17 00:00:00 2001 From: vanous Date: Sat, 8 Nov 2025 17:25:29 +0100 Subject: [PATCH 6/6] Update docs, format code, update version --- README.md | 28 ++++++++++++++++++++++------ pyproject.toml | 4 ++-- run.py | 4 ++-- tui/app.py | 14 +++----------- tui/gdtf_share/gdtf.py | 12 ++---------- tui/rdm_search.py | 2 +- tui/screens.py | 15 ++++++--------- 7 files changed, 38 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index 0d1b7e4..25cafcb 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,8 @@ -[PollToMVR](https://github.com/vanous/PollToMVR) - a tool to perform network -discovery via ArtNet - ArtPoll and to save found devices in an +[PollToMVR](https://github.com/vanous/PollToMVR) - a tool to perform device +discovery via ArtNet or RDM E1.20 and to save found devices in an [MVR](https://gdtf-share.com/) (My Virtual Rig) scene file. @@ -13,13 +13,14 @@ discovery via ArtNet - ArtPoll and to save found devices in an - Uses [pymvr](https://pypi.org/project/pymvr/) to write fixtures to MVR files - Provides Graphical [Terminal User Interface](https://textual.textualize.io/) - Uses ArtPoll based device network discovery, can parse DMX Address and Universe from device description if present +- Uses RDM ANSI E1.20 via Robe RUNIT USB Interfaces for DMX line device discovery - Uses the [GDTF Share](https://gdtf-share.com) Fixture Library for fixture definitions in MVR ## FAQ ### What is this -A tool to quickly create an MVR file based on network scan. +A tool to quickly create an MVR file based on ArtPoll or RDM scan. ### What this is not @@ -36,6 +37,12 @@ No, it only uses ArtPoll and ArtPollReply. Some manufacturers put DMX Address/Universe into `long_name`. If present and parsed, these values will then be used in the MVR export. +### Can devices be discovered via RDM? + +Yes, PollToMVR can use ANSI E1.20 for device discovery and to query the device +for Device Info and Device Model Description. Supported USB interfaces are the +Robe Lighting: `Universal Interface` and `RUNIT WTX` interfaces. + ## Instalation Binary releases for Linux, macOS and Windows are available from the @@ -55,11 +62,11 @@ Hub](https://gdtf.eu/). ## Quick Start - Start the PollToMVR -- Click Discover to run discovery +- Click Discover to run network discovery - After discovery, press the "Add devices to MVR Layer", click Add - Click `Save Devices` to store the result as an MVR file -This will create a barebone MVR file with device names and their IP addresses +This will create a bare-bone MVR file with device names and their IP addresses (and Universes, DMX address, if also discovered). For more featured MVR: set user login credential in the Config, download some GDTF files and after network discovery, link the GDTFs to the discovered fixtures. After saving, the MVR @@ -78,8 +85,9 @@ will also contain the full GDTF definitions file will then be used for the created MVR file. To download the GDTF files, use the GDTF Files button on the main screen. - ### Discover - - Discover devices on the local network + - Discover devices on the local network or on DMX line - Select a network interface on which the discovery will run + - Select a USB based Runit interface for RDM based discovery - ### Add Discovered Devices - Add the discovered devices to a selected MVR layer - Create a new named MVR Layer @@ -159,6 +167,14 @@ uv run textual console uv run textual run --dev run.py ``` +## Code formatting + +All python code is to be formatted with ruff: + +``` +uv tool run ruff format +`` + ## Packaging Initial pyinstaller setup diff --git a/pyproject.toml b/pyproject.toml index c2332e2..7fff87c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "PolltoMVR" -version = "0.1.0" -description = "A tool to discover network devices via Art-Net ArtPoll and to save them into [MVR](https://gdtf-share.com/) (My Virtual Rig) scene file" +version = "0.2.0" +description = "A tool to discover network devices via Art-Net or RDM E1.20 and to save them into [MVR](https://gdtf-share.com/) (My Virtual Rig) scene file" readme = "README.md" requires-python = ">=3.13" dependencies = [ diff --git a/run.py b/run.py index 8c6ee2b..ec2c877 100644 --- a/run.py +++ b/run.py @@ -15,8 +15,8 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from tui.app import ArtPollToMVR +from tui.app import PollToMVR if __name__ == "__main__": - app = ArtPollToMVR() + app = PollToMVR() app.run() diff --git a/tui/app.py b/tui/app.py index 8f80bd0..744c535 100644 --- a/tui/app.py +++ b/tui/app.py @@ -15,22 +15,14 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -import functools import json import os -import random -import traceback -import subprocess from types import SimpleNamespace from textual.app import App, ComposeResult from textual import on, work from textual.containers import Horizontal, Vertical, VerticalScroll, Grid -from textual.widgets import Header, Footer, Input, Button, Static, Select -from textual.worker import Worker, WorkerState +from textual.widgets import Header, Button, Static, Select from tui.screens import ArtNetScreen, QuitScreen, ConfigScreen, ImportDiscovery -from textual.message import Message -from textual.reactive import reactive -from tui.messages import MvrParsed, Errors import uuid as py_uuid from tui.create_mvr import create_mvr from textual_fspicker import FileSave, Filters @@ -123,7 +115,7 @@ def on_select_changed(self, event: Select.Changed): self.app.gdtf_map[self.fixture] = event.value -class ArtPollToMVR(App): +class PollToMVR(App): """A Textual app to manage Uptime Kuma MVR.""" CSS_PATH = [ @@ -272,5 +264,5 @@ async def save_a_file(self, event: Button.Pressed) -> None: if __name__ == "__main__": - app = ArtPollToMVR() + app = PollToMVR() app.run() diff --git a/tui/gdtf_share/gdtf.py b/tui/gdtf_share/gdtf.py index 83ebc77..a3fb3d5 100644 --- a/tui/gdtf_share/gdtf.py +++ b/tui/gdtf_share/gdtf.py @@ -15,26 +15,18 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from types import SimpleNamespace from textual.message import Message from textual.screen import ModalScreen from textual.app import ComposeResult from textual.containers import ( - Grid, Horizontal, Vertical, HorizontalGroup, VerticalScroll, ) -from textual.widgets import Button, Static, Input, Label, Checkbox, Select, Switch -from textual import on, work, events -from textual_fspicker import FileOpen, Filters -from tui.messages import Errors, NetworkDevicesDiscovered -from tui.network import get_network_cards -from tui.artnet import ArtNetDiscovery +from textual.widgets import Button, Static, Input, Select +from textual import work, events from tui.share_api_client import update_data, download_files -import re -import sys from pathlib import Path import json import asyncio diff --git a/tui/rdm_search.py b/tui/rdm_search.py index 2bda33f..824b279 100644 --- a/tui/rdm_search.py +++ b/tui/rdm_search.py @@ -302,7 +302,7 @@ def parse_rdm_response(rdm_data: bytes, sent_pid: int): else: response_data = parse_ack(pd, pid, cc) else: - print(f" └─ Received NACK or other response type.") + print(" └─ Received NACK or other response type.") return pid, response_data diff --git a/tui/screens.py b/tui/screens.py index e4d2de9..999b522 100644 --- a/tui/screens.py +++ b/tui/screens.py @@ -19,11 +19,9 @@ from textual.screen import ModalScreen from textual.app import ComposeResult from textual.containers import Grid, Horizontal, Vertical -from textual.widgets import Button, Static, Input, Label, Checkbox, Select, Switch -from textual import on, work, events -from textual_fspicker import FileOpen, Filters +from textual.widgets import Button, Static, Input, Label, Select, Switch +from textual import work, events from tui.messages import ( - Errors, NetworkDevicesDiscovered, RdmDevicesDiscovered, RdmDeviceDetailDiscovered, @@ -34,7 +32,6 @@ from tui.rdm_search import get_device_info, get_devices, get_port, get_device_details import re import sys -import json import serial import serial.tools.list_ports @@ -497,12 +494,12 @@ def on_input_changed(self, event: Input.Changed): layer_names = [x[0] for x in self.app.mvr_layers] if event.value in layer_names: self.query_one("#add").disabled = True - self.notify(f"Layer name already exists", timeout=1) + self.notify("Layer name already exists", timeout=1) else: self.query_one("#add").disabled = False else: self.query_one("#add").disabled = True - self.notify(f"Must not be empty", timeout=1) + self.notify("Must not be empty", timeout=1) def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "add": @@ -521,7 +518,7 @@ def action_focus_next(self) -> None: layer_names = [x[0] for x in self.app.mvr_layers] if new_layer_name in layer_names: self.query_one("#add").disabled = True - self.notify(f"Layer name already exists", timeout=1) + self.notify("Layer name already exists", timeout=1) self.focus_next() def action_focus_previous(self) -> None: @@ -531,7 +528,7 @@ def action_focus_previous(self) -> None: layer_names = [x[0] for x in self.app.mvr_layers] if new_layer_name in layer_names: self.query_one("#add").disabled = True - self.notify(f"Layer name already exists", timeout=1) + self.notify("Layer name already exists", timeout=1) self.focus_previous() async def on_key(self, event: events.Key) -> None: