"""Custom OpenAPI documentation page for the RemoteTerm API.""" from __future__ import annotations import json from html import escape from typing import Any from fastapi import FastAPI from fastapi.openapi.utils import get_openapi from fastapi.responses import HTMLResponse API_DESCRIPTION = ( "RemoteTerm exposes the MeshCore companion radio as a local REST and WebSocket API.\n\n" "REST endpoints are mounted below `/api`. The live WebSocket stream is available at " "`/api/ws` for health, message, raw-packet, contact, and telemetry events.\n\n" "**Trusted network note:** RemoteTerm is designed for trusted local networks. Optional " "HTTP Basic auth can be enabled for the whole app, but operators should pair it with " "HTTPS when credentials cross the network." ) API_TAGS_METADATA: list[dict[str, Any]] = [ { "name": "health", "description": "Connection state, build info, database size, radio stats, and fanout health.", }, { "name": "debug", "description": "Support snapshots for logs, live radio probes, drift audits, and version data.", }, { "name": "radio", "description": "Radio configuration, connection lifecycle, discovery, trace, and advert commands.", }, { "name": "contacts", "description": "Mesh contacts, analytics, read state, route overrides, and path discovery.", }, { "name": "repeaters", "description": "Repeater login, telemetry, ACL, owner info, radio settings, and CLI commands.", }, { "name": "rooms", "description": "Room-server login, status, telemetry, and ACL operations.", }, { "name": "channels", "description": "Channel creation, metadata, read state, flood scope, and path-hash overrides.", }, { "name": "messages", "description": "Message history, direct sends, channel sends, and channel resend workflows.", }, { "name": "packets", "description": "Raw packet inspection, historical decryption, undecrypted counts, and maintenance.", }, { "name": "read-state", "description": "Server-side unread counters, mention flags, and mark-all-read operations.", }, { "name": "settings", "description": "App settings, favorites, muted channels, block lists, and telemetry tracking.", }, { "name": "push", "description": "Browser Web Push subscriptions, per-device preferences, tests, and conversations.", }, { "name": "fanout", "description": "MQTT, bots, webhooks, Apprise, SQS, Home Assistant, and map upload integrations.", }, { "name": "statistics", "description": "Aggregated mesh, message, packet, channel, and contact statistics.", }, ] SWAGGER_UI_CSS_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css" SWAGGER_UI_BUNDLE_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js" SWAGGER_UI_PRESET_URL = ( "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-standalone-preset.js" ) COMMON_ERROR_RESPONSES: dict[str, str] = { "400": "Bad request", "401": "Authentication required", "403": "Forbidden", "404": "Not found", "408": "Request timed out", "409": "Conflict", "422": "Validation or command error", "423": "Radio unavailable or locked", "500": "Server error", } ERROR_RESPONSE_SCHEMA: dict[str, Any] = { "type": "object", "properties": { "detail": { "description": "Human-readable error detail or structured validation detail.", "oneOf": [ {"type": "string"}, {"type": "array", "items": {}}, {"type": "object", "additionalProperties": True}, ], } }, } def _relative_openapi_url(app: FastAPI) -> str: """Keep docs usable behind reverse-proxy path prefixes.""" openapi_url = app.openapi_url or "/openapi.json" return openapi_url.lstrip("/") def _error_response(description: str) -> dict[str, Any]: return { "description": description, "content": { "application/json": { "schema": ERROR_RESPONSE_SCHEMA, } }, } def _add_common_error_responses(openapi_schema: dict[str, Any]) -> None: paths = openapi_schema.get("paths") if not isinstance(paths, dict): return for path_item in paths.values(): if not isinstance(path_item, dict): continue for operation in path_item.values(): if not isinstance(operation, dict): continue responses = operation.setdefault("responses", {}) if not isinstance(responses, dict): continue for status_code, description in COMMON_ERROR_RESPONSES.items(): responses.setdefault(status_code, _error_response(description)) def install_custom_openapi(app: FastAPI) -> None: """Install OpenAPI metadata polishing used by the custom docs page.""" original_openapi = app.openapi def custom_openapi() -> dict[str, Any]: if app.openapi_schema: return app.openapi_schema openapi_schema = get_openapi( title=app.title, version=app.version, openapi_version=app.openapi_version, description=app.description, routes=app.routes, tags=app.openapi_tags, servers=app.servers, ) _add_common_error_responses(openapi_schema) app.openapi_schema = openapi_schema return app.openapi_schema # Keep a reference for debuggability and for tests that may introspect it. app.state.default_openapi = original_openapi app.openapi = custom_openapi # type: ignore[method-assign] def _build_swagger_docs_html(app: FastAPI) -> str: title = escape(app.title) version = escape(app.version) openapi_url = json.dumps(_relative_openapi_url(app)) return f""" {title}

RemoteTerm API

{title}

Explore radio control, messaging, packet inspection, push notifications, and fanout integration endpoints from one interactive console.

GET /api/health OpenAPI JSON WebSocket /api/ws
Version {version} REST base /api Auth optional Basic
""" def register_api_docs_routes(app: FastAPI) -> None: """Register the custom Swagger UI route.""" install_custom_openapi(app) @app.get("/docs", include_in_schema=False) async def swagger_ui_html() -> HTMLResponse: return HTMLResponse(_build_swagger_docs_html(app))