TuxDMX-WebUI/tools/tuxdmx-midi-bridge/tuxdmx_midi_bridge/main.py

from __future__ import annotations

import argparse
import asyncio
import logging
from contextlib import suppress

from .client import TuxDmxMidiClient, reconnect_delay
from .config import BridgeConfig, load_config
from .logging_setup import configure_logging
from .midi import MidiEventFilter, list_input_devices, match_input_device, normalize_message

logger = logging.getLogger("tuxdmx_midi_bridge")


async def run_bridge(config: BridgeConfig, debug: bool = False) -> None:
    if debug:
        config.logging.level = "DEBUG"
        config.logging.show_midi_events = True
    configure_logging(config.logging.level)
    client = TuxDmxMidiClient(config.server)
    event_filter = MidiEventFilter(
        channel_filter=config.midi.channel_filter,
        control_change_interval_ms=config.performance.control_change_interval_ms,
        suppress_duplicate_values=config.performance.suppress_duplicate_values,
    )
    attempt = 0

    while True:
        try:
            device_name = match_input_device(config.midi.device_name)
            if device_name is None:
                raise RuntimeError(f"MIDI-input matcher ikke '{config.midi.device_name}'.")
            logger.info("Valgt MIDI-input: %s", device_name)
            import mido

            with mido.open_input(device_name) as port:
                await client.send_heartbeat(
                    {
                        "type": "heartbeat",
                        "protocol_version": 1,
                        "bridge_id": config.bridge_id,
                        "device": device_name,
                    }
                )
                logger.info("Forbundet til TuxDMX som bridge '%s'", config.bridge_id)
                attempt = 0
                heartbeat_task = asyncio.create_task(_heartbeat_loop(client, config.bridge_id, device_name))
                try:
                    while True:
                        for message in port.iter_pending():
                            if config.midi.ignore_active_sensing and getattr(message, "type", "") == "active_sensing":
                                continue
                            payload = normalize_message(message, config.bridge_id, device_name)
                            if payload is None or not event_filter.should_forward(payload):
                                continue
                            if config.logging.show_midi_events:
                                logger.info(
                                    "MIDI %s ch=%s no=%s val=%s",
                                    payload["message"]["type"],
                                    payload["message"]["channel"],
                                    payload["message"]["number"],
                                    payload["message"]["value"],
                                )
                            await client.send_event(payload)
                        await asyncio.sleep(0.01)
                finally:
                    heartbeat_task.cancel()
                    with suppress(asyncio.CancelledError):
                        await heartbeat_task
        except KeyboardInterrupt:
            logger.info("Afslutter MIDI-bridge kontrolleret.")
            return
        except Exception as exc:
            attempt += 1
            delay = reconnect_delay(attempt, config.server.reconnect_seconds)
            logger.warning("Bridge-fejl: %s. Nyt forsøg om %.1f sek.", exc, delay)
            await asyncio.sleep(delay)


async def _heartbeat_loop(client: TuxDmxMidiClient, bridge_id: str, device_name: str) -> None:
    while True:
        await asyncio.sleep(5)
        await client.send_heartbeat(
            {
                "type": "heartbeat",
                "protocol_version": 1,
                "bridge_id": bridge_id,
                "device": device_name,
            }
        )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="TuxDMX MIDI bridge")
    parser.add_argument("--config", default="config.yaml", help="Sti til YAML-konfiguration")
    parser.add_argument("--list-devices", action="store_true", help="List tilgængelige MIDI-inputs")
    parser.add_argument("--test-connection", action="store_true", help="Test forbindelse til TuxDMX")
    parser.add_argument("--debug", action="store_true", help="Aktivér debuglog")
    return parser


async def _async_main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if args.list_devices:
        for name in list_input_devices():
            print(name)
        return 0

    config = load_config(args.config)
    configure_logging("DEBUG" if args.debug else config.logging.level)

    if args.test_connection:
        client = TuxDmxMidiClient(config.server)
        device_name = match_input_device(config.midi.device_name) or config.midi.device_name
        result = await client.test_connection(config.bridge_id, device_name)
        print(result)
        return 0

    await run_bridge(config, debug=args.debug)
    return 0


def main() -> int:
    return asyncio.run(_async_main())


if __name__ == "__main__":
    raise SystemExit(main())