indi_nexus.web

The FastAPI web bridge: one shared upstream IndiClient relayed to browsers as typed JSON over a WebSocket, plus a REST snapshot and the bundled panel.

App

indi_nexus.web.app

FastAPI application factory for the INDINexus web bridge.

:func:create_app builds a FastAPI app that serves, on top of one shared :class:~indi_nexus.client.IndiClient:

• GET / - the built reference panel if present, else the debug inspector page;
• GET /debug - the self-contained debug inspector page;
• GET /health - liveness and upstream connection state;
• GET /api/devices and /api/devices/{device}[/{name}] - a read-only JSON snapshot of the property cache;
• WS /ws - the live bridge: a snapshot on connect, then streamed updates, with browser-sent frames forwarded upstream.

The client is injectable so tests drive the app over an in-memory transport; by default a real TCP client to indiserver is created.

create_app

create_app(*, client: IndiClient | None = None, indi_host: str = 'localhost', indi_port: int = 7624) -> FastAPI

Build the web-bridge FastAPI application.

Parameters:

Name	Type	Description	Default
client	IndiClient	An existing client to relay (used by tests); if omitted, a real :class:IndiClient to indi_host/indi_port is created.	None
indi_host	str	Upstream indiserver host (when client is not given).	'localhost'
indi_port	int	Upstream indiserver port (when client is not given).	7624

Returns:

Name	Type	Description
app	FastAPI	The configured application; its lifespan starts and stops the bridge.

Source code in src/indi_nexus/web/app.py

def create_app(
    *,
    client: IndiClient | None = None,
    indi_host: str = "localhost",
    indi_port: int = 7624,
) -> FastAPI:
    """Build the web-bridge FastAPI application.

    Parameters
    ----------
    client : IndiClient, optional
        An existing client to relay (used by tests); if omitted, a real
        :class:`IndiClient` to ``indi_host``/``indi_port`` is created.
    indi_host : str, optional
        Upstream ``indiserver`` host (when ``client`` is not given).
    indi_port : int, optional
        Upstream ``indiserver`` port (when ``client`` is not given).

    Returns
    -------
    app : FastAPI
        The configured application; its lifespan starts and stops the bridge.
    """
    indi_client = client or IndiClient(indi_host, indi_port)
    bridge = Bridge(indi_client)

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        """Start the bridge on startup and close it on shutdown."""
        await bridge.start()
        try:
            yield
        finally:
            await bridge.aclose()

    app = FastAPI(title="INDINexus web bridge", lifespan=lifespan)
    app.state.bridge = bridge
    app.state.client = indi_client

    @app.get("/health")
    async def health() -> dict[str, Any]:
        """Report liveness and whether the upstream client is connected."""
        return {"status": "ok", "connected": indi_client.connected}

    @app.get("/api/devices")
    async def list_devices() -> list[str]:
        """Return the names of all known devices."""
        return indi_client.store.devices()

    @app.get("/api/devices/{device}")
    async def device_properties(device: str) -> dict[str, Any]:
        """Return one device's properties as JSON, keyed by property name."""
        props = indi_client.store.device(device)
        if not props:
            raise HTTPException(status_code=404, detail=f"unknown device {device!r}")
        return {name: vec.model_dump(mode="json") for name, vec in props.items()}

    @app.get("/api/devices/{device}/{name}")
    async def one_property(device: str, name: str) -> dict[str, Any]:
        """Return a single property vector as JSON."""
        vec = indi_client.store.get(device, name)
        if vec is None:
            raise HTTPException(status_code=404, detail=f"unknown property {device}.{name}")
        return vec.model_dump(mode="json")

    @app.websocket("/ws")
    async def ws(websocket: WebSocket) -> None:
        """Stream live updates to a browser and forward its frames upstream."""
        await websocket.accept()
        for frame in bridge.snapshot():
            await websocket.send_text(frame)
        await websocket.send_text(bridge.connection_frame(indi_client.connected))
        bridge.add_sink(websocket.send_text)
        try:
            while True:
                await bridge.handle_incoming(await websocket.receive_text())
        except WebSocketDisconnect:
            pass
        finally:
            bridge.remove_sink(websocket.send_text)

    @app.get("/debug")
    async def debug_page() -> FileResponse:
        """Serve the self-contained debug inspector page."""
        return FileResponse(_STATIC / "debug.html")

    # Serve the built reference panel at the root when it is present (produced by
    # ``pnpm --filter @indi-nexus/panel build``); otherwise fall back to the debug
    # page. The static mount is added last so the API/WS/debug routes above win.
    if (_PANEL / "index.html").is_file():
        app.mount("/", StaticFiles(directory=_PANEL, html=True), name="panel")
    else:

        @app.get("/")
        async def index() -> FileResponse:
            """Serve the debug page when the reference panel is not built."""
            return FileResponse(_STATIC / "debug.html")

    return app

Bridge

indi_nexus.web.bridge

Bridge: fan one shared IndiClient out to many browser WebSockets.

The bridge owns a single upstream connection to indiserver (via the M3 :class:~indi_nexus.client.IndiClient) and relays its activity to every connected browser as JSON. Property changes, log messages, and connection-state transitions are broadcast to all registered sinks; a browser's inbound frames are parsed back into typed models and forwarded upstream. A newly-connected browser first receives a snapshot of the current cache so it starts with full state.

Server -> browser frames are either an INDI message ({"tag": ...}, mirroring the protocol models) or a small bridge control frame ({"event": ...}) for UI affordances the INDI protocol has no message for, e.g. upstream connection state. Browser -> server frames are always INDI messages.

Bridge

Bridge(client: IndiClient)

Relay between one IndiClient and many browser WebSocket sinks.

Parameters:

Name	Type	Description	Default
client	IndiClient	The shared upstream client the bridge relays.	required

Source code in src/indi_nexus/web/bridge.py

def __init__(self, client: IndiClient) -> None:
    self._client = client
    self._sinks: set[Sink] = set()
    # INDI messages are transient (not part of the property cache), so keep a
    # bounded history to prime a newly-connected browser's log.
    self._messages: deque[str] = deque(maxlen=_MESSAGE_HISTORY)

client property

client: IndiClient

The upstream client this bridge relays.

start async

start() -> None

Subscribe to the client and open its upstream connection.

Source code in src/indi_nexus/web/bridge.py

async def start(self) -> None:
    """Subscribe to the client and open its upstream connection."""
    self._client.subscribe(self._on_event)
    self._client.on_message(self._on_message)
    self._client.on_connection(self._on_connection)
    await self._client.start()

aclose async

aclose() -> None

Close the upstream connection.

Source code in src/indi_nexus/web/bridge.py

async def aclose(self) -> None:
    """Close the upstream connection."""
    await self._client.aclose()

add_sink

add_sink(sink: Sink) -> None

Register a browser sink to receive broadcasts.

Parameters:

Name	Type	Description	Default
sink	Sink	An awaitable that sends one text frame to the browser.	required

Source code in src/indi_nexus/web/bridge.py

def add_sink(self, sink: Sink) -> None:
    """Register a browser sink to receive broadcasts.

    Parameters
    ----------
    sink : Sink
        An awaitable that sends one text frame to the browser.
    """
    self._sinks.add(sink)

remove_sink

remove_sink(sink: Sink) -> None

Remove a previously registered sink.

Parameters:

Name	Type	Description	Default
sink	Sink	The sink to remove.	required

Source code in src/indi_nexus/web/bridge.py

def remove_sink(self, sink: Sink) -> None:
    """Remove a previously registered sink.

    Parameters
    ----------
    sink : Sink
        The sink to remove.
    """
    self._sinks.discard(sink)

snapshot

snapshot() -> list[str]

Return the current cache and recent messages as JSON frames.

Returns:

Name	Type	Description
frames	list of str	One defXxxVector JSON frame per cached property, followed by the retained message frames (oldest first), for priming a newly-connected browser.

Source code in src/indi_nexus/web/bridge.py

def snapshot(self) -> list[str]:
    """Return the current cache and recent messages as JSON frames.

    Returns
    -------
    frames : list of str
        One ``defXxxVector`` JSON frame per cached property, followed by the
        retained ``message`` frames (oldest first), for priming a
        newly-connected browser.
    """
    store = self._client.store
    frames: list[str] = []
    for device in store.devices():
        for vector in store.device(device).values():
            frames.append(to_json(DefVector(vector=vector)))
    frames.extend(self._messages)
    return frames

connection_frame staticmethod

connection_frame(connected: bool) -> str

Build the control frame announcing upstream connection state.

Parameters:

Name	Type	Description	Default
connected	bool	Whether the bridge is connected to indiserver.	required

Returns:

Name	Type	Description
frame	str	A JSON control frame ({"event": "connection", ...}).

Source code in src/indi_nexus/web/bridge.py

@staticmethod
def connection_frame(connected: bool) -> str:
    """Build the control frame announcing upstream connection state.

    Parameters
    ----------
    connected : bool
        Whether the bridge is connected to ``indiserver``.

    Returns
    -------
    frame : str
        A JSON control frame (``{"event": "connection", ...}``).
    """
    return json.dumps({"event": "connection", "connected": connected})

handle_incoming async

handle_incoming(text: str) -> None

Parse a browser frame and forward it upstream.

Malformed frames are logged and dropped rather than closing the socket.

Parameters:

Name	Type	Description	Default
text	str	A JSON INDI message from the browser.	required

Source code in src/indi_nexus/web/bridge.py

async def handle_incoming(self, text: str) -> None:
    """Parse a browser frame and forward it upstream.

    Malformed frames are logged and dropped rather than closing the socket.

    Parameters
    ----------
    text : str
        A JSON INDI message from the browser.
    """
    from indi_nexus.protocol import from_json

    try:
        msg = from_json(text)
    except (ValueError, TypeError):
        logger.warning("dropping malformed inbound frame: %r", text[:200])
        return
    await self._client.send(msg)