motian_mavbridge/motian_bridge/cli.py

"""Production entry point using paho-mqtt and pyserial."""

import argparse
import logging
import signal
import threading

from .bridge import Bridge
from .config import Config


def main() -> None:
    parser = argparse.ArgumentParser(description="Motian MQTT / MAVLink telemetry and serial bridge")
    parser.add_argument("--config", required=True, help="path to JSON configuration")
    parser.add_argument("--debug", action="store_true")
    args = parser.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO,
                        format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    import paho.mqtt.client as mqtt
    import serial

    config = Config.from_file(args.config)
    client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
                         client_id=config.client_id, protocol=mqtt.MQTTv311)
    if config.mqtt_username is not None:
        client.username_pw_set(config.mqtt_username, config.mqtt_password())
    if config.mqtt_tls:
        client.tls_set(ca_certs=config.mqtt_ca_file,
                       certfile=config.mqtt_cert_file, keyfile=config.mqtt_key_file)
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.max_queued_messages_set(1)
    bridge = Bridge(config, client, serial.serial_for_url)
    done = threading.Event()

    def shutdown(signum: int, frame: object) -> None:
        done.set()

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)
    bridge.start()
    try:
        done.wait()
    finally:
        bridge.stop()


if __name__ == "__main__":
    main()