#!/usr/bin/env python3 """Плавный циклический переход A -> B -> A через USB-RS485 (DMX512). Нужен преобразователь USB-RS485 с аппаратным автопереключением направления, поддержкой 250000 бод, 8N2 и BREAK. Устройство должно работать в DMX-режиме. COM-порт автоматически ищется заново после каждого сбоя или переподключения. Установка: python -m pip install pyserial """ from __future__ import annotations import logging import math import time import serial from serial.tools import list_ports # ======================== НАСТРОЙКИ ======================== # Порты, которые точно НЕ относятся к DMX-адаптеру. Регистр не учитывается: # "COM3", "com3" и "CoM3" обозначают один и тот же порт. EXCLUDED_PORTS = ["COM1", "COM2"] # В Windows оставлять только USB-последовательные порты с указанным USB VID. # Если адаптер не находится, можно установить False. USB_PORTS_ONLY = True # Каналы одного DMX-прибора, начиная с DMX_START_CHANNEL. A = (80, 0, 255) B = (30, 0, 200) #A = (255, 0, 255) #B = (0, 0, 0) PERIOD_SECONDS = 2.0 # Полный цикл A -> B -> A, в секундах DMX_START_CHANNEL = 1 # DMX-адрес первого канала прибора: 1...512 FPS = 30 # Частота обновления значений, кадров/сек BREAK_SECONDS = 0.001 # BREAK >= 92 мкс; запас для неточной ОС MAB_SECONDS = 0.001 # Mark After Break >= 12 мкс TX_MARGIN_SECONDS = 0.002 WRITE_TIMEOUT_SECONDS = 0.5 RECONNECT_SECONDS = 1.0 # =========================================================== BAUDRATE = 250_000 DMX_CHANNELS = 512 DMX_FRAME_LENGTH = 1 + DMX_CHANNELS # 0x00 start code + 512 каналов BITS_PER_SLOT = 11 # 1 старт + 8 данных + 2 стоповых FRAME_WIRE_SECONDS = DMX_FRAME_LENGTH * BITS_PER_SLOT / BAUDRATE log = logging.getLogger("dmx") class PortSelectionError(Exception): """Нет однозначного подходящего COM-порта.""" def find_dmx_port() -> str: """Выбрать единственный не исключённый COM-порт, не угадывая при неоднозначности.""" excluded = {name.strip().casefold() for name in EXCLUDED_PORTS} available = sorted(list_ports.comports(), key=lambda info: info.device.casefold()) candidates = [ info for info in available if info.device.strip().casefold() not in excluded and (not USB_PORTS_ONLY or info.vid is not None) ] if len(candidates) == 1: return candidates[0].device all_names = ", ".join(info.device for info in available) or "нет" if not candidates: raise PortSelectionError( f"Подходящие COM-порты не найдены (обнаружены: {all_names}; " f"исключены: {EXCLUDED_PORTS}; USB_PORTS_ONLY={USB_PORTS_ONLY})" ) names = ", ".join(f"{info.device} ({info.description})" for info in candidates) raise PortSelectionError( f"Несколько возможных DMX-портов: {names}. " "Добавьте заведомо неподходящие порты в EXCLUDED_PORTS; " "автоматически выбирать случайный порт небезопасно" ) def validate_configuration() -> None: if not isinstance(A, tuple) or not isinstance(B, tuple): raise ValueError("A и B должны быть кортежами (tuple)") if not A or len(A) != len(B): raise ValueError("A и B должны иметь одинаковую ненулевую длину") if not all(type(v) is int and 0 <= v <= 255 for v in A + B): raise ValueError("Все значения A и B должны быть целыми от 0 до 255") if not (1 <= DMX_START_CHANNEL <= DMX_CHANNELS): raise ValueError("DMX_START_CHANNEL должен быть от 1 до 512") if DMX_START_CHANNEL + len(A) - 1 > DMX_CHANNELS: raise ValueError("Каналы прибора выходят за пределы DMX-universe") if not math.isfinite(PERIOD_SECONDS) or PERIOD_SECONDS <= 0: raise ValueError("PERIOD_SECONDS должен быть положительным числом") if not isinstance(EXCLUDED_PORTS, (list, tuple)) or any( not isinstance(name, str) or not name.strip() for name in EXCLUDED_PORTS ): raise ValueError("EXCLUDED_PORTS должен быть списком непустых имён портов") if not isinstance(USB_PORTS_ONLY, bool): raise ValueError("USB_PORTS_ONLY должен быть True или False") if not math.isfinite(FPS) or FPS <= 0: raise ValueError("FPS должен быть положительным числом") if not (0 < BREAK_SECONDS < 1 and 0 < MAB_SECONDS < 1): raise ValueError("Некорректная длительность BREAK или MAB") if (1 / FPS) < (BREAK_SECONDS + MAB_SECONDS + FRAME_WIRE_SECONDS + TX_MARGIN_SECONDS): log.warning("FPS слишком велик для полного DMX-кадра; реальная частота будет ниже") def interpolate(elapsed_seconds: float) -> list[int]: """Значения на треугольной временной кривой с периодом PERIOD_SECONDS.""" half = PERIOD_SECONDS / 2 phase = (elapsed_seconds % PERIOD_SECONDS) / half if phase <= 1.0: start, end, fraction = A, B, phase else: start, end, fraction = B, A, phase - 1.0 return [round(x + (y - x) * fraction) for x, y in zip(start, end)] def send_dmx_frame(port: serial.Serial, frame: bytearray) -> None: """Один пакет DMX: BREAK, MAB, нулевой start code, 512 значений.""" port.send_break(BREAK_SECONDS) time.sleep(MAB_SECONDS) written = port.write(frame) if written != len(frame): raise OSError(f"DMX-кадр записан не полностью: {written}/{len(frame)}") # Не начинаем следующий BREAK, пока UART должен передавать текущий кадр. # Здесь намеренно нет port.flush(): на некоторых драйверах он может # зависнуть на неопределённое время после отключения USB-устройства. time.sleep(FRAME_WIRE_SECONDS + TX_MARGIN_SECONDS) def run() -> None: validate_configuration() current = list(A) # Требуемый список той же длины, что A и B frame = bytearray(DMX_FRAME_LENGTH) # Нулевой start code, прочие каналы = 0 first_frame_time: float | None = None frame_interval = 1.0 / FPS while True: try: selected_port = find_dmx_port() # Каждый раз перечитываем порты Windows with serial.Serial( port=selected_port, baudrate=BAUDRATE, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_TWO, timeout=0.1, write_timeout=WRITE_TIMEOUT_SECONDS, xonxoff=False, rtscts=False, dsrdtr=False, ) as port: log.info("Подключен %s; передача DMX началась", selected_port) next_frame_time = time.monotonic() while True: now = time.monotonic() if first_frame_time is None: current[:] = A # Первый реально переданный кадр — точно A else: current[:] = interpolate(now - first_frame_time) # Индекс 0 — DMX start code; канал N находится по индексу N. frame[DMX_START_CHANNEL: DMX_START_CHANNEL + len(current)] = bytes(current) send_dmx_frame(port, frame) if first_frame_time is None: first_frame_time = now next_frame_time += frame_interval delay = next_frame_time - time.monotonic() if delay > 0: time.sleep(delay) else: # Не отправлять устаревшие кадры пачкой после задержки. next_frame_time = time.monotonic() except KeyboardInterrupt: log.info("Остановлено пользователем (Ctrl+C)") return except Exception: # Включая SerialException, тайм-ауты, пропажу USB и неожиданные # ошибки в рабочем цикле: закрываем порт и пробуем заново всегда. log.exception("Сбой DMX. Повторное подключение через %.1f с", RECONNECT_SECONDS) time.sleep(RECONNECT_SECONDS) if __name__ == "__main__": logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) run()