motian_mavbridge/tools/gen_dialect.py

"""Regenerate motian_bridge/dialects/v20/VKFly.py from the dialect submodule.

The generated module must never be hand-edited. Run this script after updating
the `msg_definitions` submodule, then commit both the submodule pointer and the
regenerated file.
"""

import argparse
import hashlib
import logging
import subprocess
import sys
import tempfile
from pathlib import Path

if __package__ in (None, ""):
    # Allow `python tools/gen_dialect.py` as well as `from tools import ...`.
    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

ROOT = Path(__file__).resolve().parent.parent
OUTPUT = ROOT / "motian_bridge" / "dialects" / "v20" / "VKFly.py"
SOURCE = ROOT / "msg_definitions" / "msg_definitions" / "VKFly.xml"
INCLUDES = ("common.xml", "standard.xml", "minimal.xml")

LOG = logging.getLogger("gen_dialect")


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with open(path, "rb") as stream:
        for block in iter(lambda: stream.read(1 << 20), b""):
            digest.update(block)
    return digest.hexdigest()


def check_sources() -> None:
    """Fail early with an actionable message instead of a mavgen traceback."""
    missing = [path for path in (SOURCE, *(SOURCE.parent / name for name in INCLUDES))
               if not path.is_file()]
    if missing:
        listing = "\n".join(f"  - {path}" for path in missing)
        raise SystemExit(
            "MAVLink XML sources are missing:\n"
            f"{listing}\n"
            "Initialize the dialect submodule first:\n"
            "  git submodule update --init --recursive"
        )


def mavgen(extra_args: list[str], output: Path) -> None:
    command = [
        sys.executable, "-m", "pymavlink.tools.mavgen",
        "--lang", "Python",
        "--wire-protocol", "2.0",
        "--output", str(output),
        *extra_args,
        str(SOURCE),
    ]
    LOG.debug("running: %s", " ".join(command))
    subprocess.run(command, cwd=ROOT, check=True)


def generate(dry_run: bool = False, check: bool = False, no_validate: bool = False) -> bool:
    """Generate the dialect. Returns True when the committed file was updated."""
    check_sources()
    previous = _sha256(OUTPUT) if OUTPUT.is_file() else None
    with tempfile.TemporaryDirectory(prefix="mavgen-") as directory:
        candidate = Path(directory) / "VKFly.py"
        mavgen(["--no-validate"] if no_validate else [], candidate)
        generated = candidate.read_bytes()
    if check:
        if previous is None:
            raise SystemExit(f"{OUTPUT} does not exist; run: python tools/gen_dialect.py")
        if _sha256(OUTPUT) != hashlib.sha256(generated).hexdigest():
            raise SystemExit(
                f"{OUTPUT} is stale. Regenerate with: python tools/gen_dialect.py"
            )
        LOG.info("generated dialect matches the submodule XML")
        return False
    if previous == hashlib.sha256(generated).hexdigest():
        LOG.info("generated dialect already up to date")
        return False
    if dry_run:
        LOG.info("dry run: %s would be regenerated", OUTPUT)
        return True
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_bytes(generated)
    LOG.info("wrote %s (%d bytes)", OUTPUT, len(generated))
    return True


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true",
                        help="verify the committed file matches the XML; do not write")
    parser.add_argument("--dry-run", action="store_true",
                        help="report whether regeneration is needed; do not write")
    parser.add_argument("--no-validate", action="store_true",
                        help="pass --no-validate to mavgen (skips upstream enum warnings)")
    parser.add_argument("--quiet", action="store_true")
    args = parser.parse_args()
    logging.basicConfig(
        level=logging.WARNING if args.quiet else logging.INFO,
        format="%(levelname)s %(name)s: %(message)s",
    )
    try:
        generate(dry_run=args.dry_run, check=args.check, no_validate=args.no_validate)
    except subprocess.CalledProcessError as exc:
        raise SystemExit(f"mavgen failed with exit code {exc.returncode}")


if __name__ == "__main__":
    main()