| """Render ``tui_gateway/contracts`` into TypeScript and OpenRPC. |
| |
| Python-only (the Python CI lane has no Node): Pydantic's ``model_json_schema()`` output is walked |
| by a deliberately small JSON-Schema-subset renderer — object/properties/required, primitives, |
| enum, const, anyOf-with-null, array/items, ``$ref``, oneOf + discriminator, additionalProperties. |
| Anything else raises at generation time so an unsupported model is fixed at the model, never |
| worked around in the output. Prettier runs on the TS when a node_modules binary is present |
| (output is already in the repo's prettier style; the Python CI lane regenerates and diffs it). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| import subprocess |
| import sys |
| from collections import OrderedDict |
| from pathlib import Path |
| from typing import Any |
|
|
| from pydantic import TypeAdapter |
| from pydantic.json_schema import GenerateJsonSchema |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
| from tui_gateway import contracts |
| from tui_gateway.contracts.registry import EVENTS, METHODS, SERVER_REQUESTS |
|
|
| TS_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.generated.ts" |
| OPENRPC_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.openrpc.json" |
|
|
| HEADER = ( |
| "// GENERATED by scripts/gen_gateway_contracts.py from tui_gateway/contracts — DO NOT EDIT.\n" |
| "// Regenerate: .venv/bin/python scripts/gen_gateway_contracts.py\n" |
| "// tests/tui_gateway/contracts/test_generated.py fails when this file is stale.\n" |
| ) |
|
|
|
|
| class _Schema(GenerateJsonSchema): |
| """Stable ``$defs`` naming: the model's class name (no module qualifiers).""" |
|
|
| def normalize_name(self, name: str) -> str: |
| return re.sub(r"[^A-Za-z0-9_]", "_", name) |
|
|
|
|
| def _schema_for(models: list[type]) -> tuple[dict[str, dict], list[dict]]: |
| """One shared ``$defs`` for every model, plus each model's own schema (a ``$ref`` in practice).""" |
| from pydantic.json_schema import models_json_schema |
|
|
| defs, top = models_json_schema( |
| [(m, "serialization") for m in models], schema_generator=_Schema, ref_template="#/$defs/{model}" |
| ) |
| return top.get("$defs", {}), [defs[(m, "serialization")] for m in models] |
|
|
|
|
| |
|
|
|
|
| class Renderer: |
| def __init__(self, defs: dict[str, dict]): |
| self.defs = defs |
| self.emitted: OrderedDict[str, str] = OrderedDict() |
|
|
| def ref_name(self, ref: str) -> str: |
| assert ref.startswith("#/$defs/"), ref |
| return ref[len("#/$defs/"):] |
|
|
| def type_of(self, schema: dict, *, inline_depth: int = 0) -> str: |
| if "$ref" in schema: |
| name = self.ref_name(schema["$ref"]) |
| self.ensure(name) |
| return name |
| if "const" in schema: |
| return _lit(schema["const"]) |
| if "enum" in schema: |
| return " | ".join(_lit(v) for v in schema["enum"]) |
| if "anyOf" in schema or "oneOf" in schema: |
| variants = schema.get("anyOf") or schema.get("oneOf") or [] |
| rendered = list(dict.fromkeys(self.type_of(v, inline_depth=inline_depth) for v in variants)) |
| return " | ".join(rendered) |
| t = schema.get("type") |
| if isinstance(t, list): |
| return " | ".join(self.type_of({**schema, "type": x}, inline_depth=inline_depth) for x in t) |
| if t == "string": |
| return "string" |
| if t in ("integer", "number"): |
| return "number" |
| if t == "boolean": |
| return "boolean" |
| if t == "null": |
| return "null" |
| if t == "array": |
| items = schema.get("items") |
| if items is None: |
| return "unknown[]" |
| if "prefixItems" in schema: |
| return "[" + ", ".join(self.type_of(x) for x in schema["prefixItems"]) + "]" |
| inner = self.type_of(items, inline_depth=inline_depth) |
| return f"({inner})[]" if " | " in inner else f"{inner}[]" |
| if t == "object" or "properties" in schema or "additionalProperties" in schema: |
| return self.object_literal(schema, inline_depth) |
| if not schema or set(schema) <= {"title", "description", "default"}: |
| return "unknown" |
| raise ValueError(f"unsupported JSON-Schema construct: {json.dumps(schema)[:200]}") |
|
|
| def object_literal(self, schema: dict, depth: int) -> str: |
| props = schema.get("properties") |
| extra = schema.get("additionalProperties") |
| if not props: |
| if extra is False: |
| return "Record<string, never>" |
| if extra in (None, True): |
| return "Record<string, unknown>" |
| return f"Record<string, {self.type_of(extra, inline_depth=depth + 1)}>" |
| required = set(schema.get("required", ())) |
| lines = ["{"] |
| for key, sub in props.items(): |
| opt = "" if key in required else "?" |
| lines.append(f" {_prop(key)}{opt}: {self.type_of(sub, inline_depth=depth + 1)}") |
| if extra not in (None, False): |
| lines.append(f" [key: string]: {'unknown' if extra is True else self.type_of(extra)}") |
| lines.append("}") |
| return "\n".join(lines) |
|
|
| def ensure(self, name: str) -> None: |
| if name in self.emitted: |
| return |
| self.emitted[name] = "" |
| schema = self.defs[name] |
| doc = _doc(schema.get("description")) |
| if "enum" in schema: |
| body = f"export type {name} = {self.type_of({'enum': schema['enum']})}\n" |
| elif schema.get("properties"): |
| body = f"export interface {name} {self.object_literal(schema, 0)}\n" |
| else: |
| body = f"export type {name} = {self.type_of(schema)}\n" |
| self.emitted[name] = doc + body |
|
|
|
|
| _IDENT = re.compile(r"^[A-Za-z_$][A-Za-z0-9_$]*$") |
|
|
|
|
| def _prop(key: str) -> str: |
| return key if _IDENT.match(key) else _lit(key) |
|
|
|
|
| def _const_items(names: list[str]) -> str: |
| return ",\n".join(f" {_lit(n)}" for n in names) + "\n" |
|
|
|
|
| def _lit(value) -> str: |
| """A TS literal in the repo's prettier style (single quotes) so the committed file needs no |
| Node-side formatting pass — the Python CI lane regenerates and diffs it.""" |
| if isinstance(value, str): |
| return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" |
| return json.dumps(value) |
|
|
|
|
| def _doc(text: str | None, indent: str = "") -> str: |
| if not text: |
| return "" |
| clean = " ".join(text.split()) |
| return f"{indent}/** {clean} */\n" |
|
|
|
|
| def _pascal(name: str) -> str: |
| return "".join(p[:1].upper() + p[1:] for p in re.split(r"[._]", name)) |
|
|
|
|
| def render_ts() -> str: |
| models: list[type] = [] |
| for m in METHODS.values(): |
| models += [m.params, m.result] |
| for r in SERVER_REQUESTS.values(): |
| models += [r.params, r.result] |
| for e in EVENTS.values(): |
| if e.payload is not None: |
| models.append(e.payload) |
| |
| seen: dict[type, None] = OrderedDict() |
| for m in models: |
| seen.setdefault(m) |
| models = list(seen) |
| defs, tops = _schema_for(models) |
| r = Renderer(defs) |
| name_of = {m: r.ref_name(t["$ref"]) for m, t in zip(models, tops)} |
| for m in models: |
| r.ensure(name_of[m]) |
|
|
| out = [HEADER, "/* eslint-disable */\n", "// ── Types ──\n"] |
| out.extend(r.emitted.values()) |
|
|
| out.append("\n// ── Client→server methods ──\n") |
| out.append("export interface RpcMethods {\n") |
| for m in sorted(METHODS.values(), key=lambda x: x.name): |
| out.append(_doc(m.doc, " ")) |
| out.append(f" {_prop(m.name)}: {{ params: {name_of[m.params]}; result: {name_of[m.result]} }}\n") |
| out.append("}\n") |
| out.append("export type RpcMethod = keyof RpcMethods\n") |
| out.append("export const RPC_METHODS = [\n" + _const_items(sorted(METHODS)) + "] as const satisfies readonly RpcMethod[]\n") |
|
|
| out.append("\n// ── Server→client requests ──\n") |
| out.append("export interface ServerRequestMap {\n") |
| for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name): |
| out.append(_doc(s.doc, " ")) |
| out.append(f" {_prop(s.name)}: {{ params: {name_of[s.params]}; result: {name_of[s.result]} }}\n") |
| out.append("}\n") |
| out.append("export type ServerRequestMethod = keyof ServerRequestMap\n") |
| out.append("export const SERVER_REQUEST_METHODS = [\n" + _const_items(sorted(SERVER_REQUESTS)) |
| + "] as const satisfies readonly ServerRequestMethod[]\n") |
|
|
| out.append("\n// ── Notifications (`event` frames) ──\n") |
| out.append("export interface BackendGatewayEventMap {\n") |
| for e in sorted(EVENTS.values(), key=lambda x: x.name): |
| out.append(_doc(e.doc, " ")) |
| payload = name_of[e.payload] if e.payload is not None else "Record<string, never>" |
| out.append(f" {_prop(e.name)}: {payload}\n") |
| out.append("}\n") |
| out.append("export type BackendGatewayEventName = keyof BackendGatewayEventMap\n") |
| out.append("export const GATEWAY_EVENT_TYPES = [\n" + _const_items(sorted(EVENTS)) |
| + "] as const satisfies readonly BackendGatewayEventName[]\n") |
| return "".join(out) |
|
|
|
|
| def _tidy(text: str) -> str: |
| """No trailing whitespace, single trailing newline (matches `git diff --check` + prettier).""" |
| return "\n".join(line.rstrip() for line in text.splitlines()).rstrip("\n") + "\n" |
|
|
|
|
| |
|
|
|
|
| def _openrpc_schema(model: type) -> dict: |
| schema = TypeAdapter(model).json_schema(schema_generator=_Schema, ref_template="#/components/schemas/{model}") |
| schema.pop("$defs", None) |
| return schema |
|
|
|
|
| def render_openrpc() -> str: |
| components: dict[str, dict] = {} |
| all_models: list[type] = [] |
| for m in METHODS.values(): |
| all_models += [m.params, m.result] |
| for r in SERVER_REQUESTS.values(): |
| all_models += [r.params, r.result] |
| for e in EVENTS.values(): |
| if e.payload is not None: |
| all_models.append(e.payload) |
| from pydantic.json_schema import models_json_schema |
|
|
| seen: dict[type, None] = OrderedDict() |
| for m in all_models: |
| seen.setdefault(m) |
| _, top = models_json_schema( |
| [(m, "serialization") for m in seen], schema_generator=_Schema, |
| ref_template="#/components/schemas/{model}", |
| ) |
| components = top.get("$defs", {}) |
|
|
| def ref(model: type) -> dict: |
| return {"$ref": f"#/components/schemas/{model.__name__}"} |
|
|
| doc = { |
| "openrpc": "1.3.2", |
| "info": {"title": "Hermes TUI/Desktop gateway", "version": "1", |
| "description": "Generated from tui_gateway/contracts by scripts/gen_gateway_contracts.py."}, |
| "methods": [ |
| {"name": m.name, "summary": " ".join(m.doc.split()), |
| "params": [{"name": "params", "schema": ref(m.params)}], |
| "result": {"name": "result", "schema": ref(m.result)}} |
| for m in sorted(METHODS.values(), key=lambda x: x.name) |
| ], |
| "components": {"schemas": components}, |
| "x-server-requests": [ |
| {"name": s.name, "summary": " ".join(s.doc.split()), |
| "params": [{"name": "params", "schema": ref(s.params)}], |
| "result": {"name": "result", "schema": ref(s.result)}} |
| for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name) |
| ], |
| "x-notifications": [ |
| {"name": e.name, "summary": " ".join(e.doc.split()), |
| "params": [{"name": "payload", "schema": ref(e.payload) if e.payload is not None |
| else {"type": "object", "additionalProperties": False}}]} |
| for e in sorted(EVENTS.values(), key=lambda x: x.name) |
| ], |
| } |
| return json.dumps(doc, indent=2, sort_keys=False) + "\n" |
|
|
|
|
| def render_all() -> dict[Path, str]: |
| return {TS_OUT: _tidy(render_ts()), OPENRPC_OUT: render_openrpc()} |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| args = argv if argv is not None else sys.argv[1:] |
| check = "--check" in args |
| stale = [] |
| for path, text in render_all().items(): |
| current = path.read_text(encoding="utf-8") if path.exists() else None |
| if current == text: |
| continue |
| if check: |
| stale.append(path) |
| else: |
| path.write_text(text, encoding="utf-8") |
| print(f"wrote {path.relative_to(ROOT)}") |
| if stale: |
| for p in stale: |
| print(f"stale: {p.relative_to(ROOT)} — run scripts/gen_gateway_contracts.py", file=sys.stderr) |
| return 1 |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
