motian_mavbridge/tools/local_mqtt_broker.py

"""Start an anonymous MQTT broker for local bridge testing.

This is intentionally a development-only helper. It binds to loopback and
does not configure authentication or TLS.
"""

import argparse
import asyncio
import logging

from amqtt.broker import Broker


def build_config(host: str, port: int) -> dict:
    return {
        "listeners": {
            "default": {
                "type": "tcp",
                "bind": f"{host}:{port}",
            }
        },
        "auth": {
            "allow-anonymous": True,
        },
        "topic-check": {
            "enabled": False,
        },
    }


async def run(host: str, port: int) -> None:
    broker = Broker(build_config(host, port))
    await broker.start()
    logging.info("local MQTT broker listening on mqtt://%s:%s", host, port)
    try:
        await asyncio.Event().wait()
    finally:
        await broker.shutdown()


def main() -> None:
    parser = argparse.ArgumentParser(description="Local development MQTT broker")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", default=1883, type=int)
    args = parser.parse_args()
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    try:
        asyncio.run(run(args.host, args.port))
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()