python/generate_messages.py

Rama main · réplica de lectura del repositorio original.

"""Generates fonoster/voice/messages.py (TypedDicts) from mods/common/schema/session.v1.json.

Run after changing the session schemas:  .venv/bin/python generate_messages.py
tests/test_messages.py fails while the generated file is out of date.
"""

from __future__ import annotations

import json
import keyword
from pathlib import Path

ROOT = Path(__file__).resolve().parent
SCHEMA = ROOT.parent / "mods" / "common" / "schema" / "session.v1.json"
OUT = ROOT / "fonoster-voice" / "src" / "fonoster" / "voice" / "messages.py"


def py_type(schema: dict) -> str:
    if "const" in schema:
        return f"Literal[{json.dumps(schema['const'])}]"
    if "enum" in schema:
        return "Literal[" + ", ".join(json.dumps(v) for v in schema["enum"]) + "]"
    if "oneOf" in schema:
        return " | ".join(py_type(s) for s in schema["oneOf"])
    t = schema.get("type")
    if t == "string":
        return "str"
    if t == "integer":
        return "int"
    if t == "number":
        return "float"
    if t == "boolean":
        return "bool"
    if t == "null":
        return "None"
    if t == "array":
        return f"list[{py_type(schema.get('items', {}))}]"
    if t == "object":
        return "dict[str, Any]"
    return "Any"


def class_name(message_type: str) -> str:
    if message_type == "error":
        return "ErrorMessage"  # not "Error", which would read like an exception class
    return "".join(part.capitalize() for part in message_type.replace(".", "_").split("_"))


def typed_dict(name: str, schema: dict) -> str:
    required = set(schema.get("required", []))
    props = schema.get("properties", {})
    lines = []
    for key, prop in props.items():
        t = py_type(prop)
        lines.append(f"        {json.dumps(key)}: {t if key in required else f'NotRequired[{t}]'},")
    # Functional syntax: keys such as "from" and "type" are not valid identifiers everywhere.
    return f"{name} = TypedDict(\n    {json.dumps(name)},\n    {{\n" + "\n".join(lines) + "\n    },\n)\n"


def generate() -> str:
    doc = json.loads(SCHEMA.read_text())
    parts = [
        '"""Session protocol messages (fonoster.session.v1). Generated by python/generate_messages.py',
        "from mods/common/schema/session.v1.json; do not edit by hand.",
        '"""',
        "",
        "from __future__ import annotations",
        "",
        "from typing import Any, Literal, NotRequired, TypedDict, Union",
        "",
        f"PROTOCOL = {json.dumps(doc['protocol'])}",
        "",
    ]
    names: dict[str, list[str]] = {"appBridgeToApp": [], "appToAppBridge": []}
    for direction, suffix in (("appBridgeToApp", "Event"), ("appToAppBridge", "Command")):
        for message_type, schema in doc[direction].items():
            name = class_name(message_type) + ("Event" if message_type == "mark" and suffix == "Event" else "")
            if message_type == "mark" and suffix == "Command":
                name = "MarkCommand"
            names[direction].append(name)
            parts.append("")
            parts.append(typed_dict(name, schema))
    parts.append("")
    parts.append("AppBridgeMessage = Union[" + ", ".join(names["appBridgeToApp"]) + "]")
    parts.append("AppCommand = Union[" + ", ".join(names["appToAppBridge"]) + "]")
    parts.append("")
    return "\n".join(parts)


if __name__ == "__main__":
    OUT.write_text(generate())
    print(f"wrote {OUT.relative_to(ROOT)}")