""" arnold/api.py — FastAPI REST application. Receives an AppContext at creation time; no module-level mutable globals. Endpoints: GET / Redirect to web UI GET /web/* Static files (web interface) GET /status Device comms health + poll stats GET /io All signal states from the poll cache GET /io/{signal} Single signal state + metadata POST /io/{signal}/write Write an output signal value GET /config/signals Full signal metadata for UI bootstrap GET /sequences List sequences defined in config GET /sequences/{name} Sequence detail with step list POST /sequences/{name}/run Start a sequence → {run_id} (409 if busy) GET /runs Recent run log (last 50, most recent first) GET /runs/{run_id} Single run result (pending/running/success/failed/error) """ from __future__ import annotations import time from dataclasses import dataclass from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from .config import Config from .sequencer import Sequencer from .terminator_io import IORegistry @dataclass class AppContext: """All runtime objects the API needs, passed in at app creation.""" config: Config registry: IORegistry sequencer: Sequencer started_at: float def create_app(ctx: AppContext) -> FastAPI: app = FastAPI( title="Arnold — Terminator I/O Server", version="0.1.0", description=( "Fast-poll Modbus TCP I/O server for AutomationDirect Terminator I/O. " "Provides real-time signal state and timed sequence execution." ), ) # ------------------------------------------------------------------ # Web UI — static files + root redirect # ------------------------------------------------------------------ # Locate the web/ directory relative to this file _web_dir = Path(__file__).resolve().parent.parent / "web" if _web_dir.is_dir(): app.mount("/web", StaticFiles(directory=str(_web_dir)), name="web-static") @app.get("/", include_in_schema=False) def root_redirect(): return RedirectResponse(url="/web/index.html") # ------------------------------------------------------------------ # GET /status # ------------------------------------------------------------------ @app.get("/status", summary="Device comms health and poll statistics") def get_status() -> dict: return { "status": "ok", "uptime_s": round(time.monotonic() - ctx.started_at, 1), "devices": ctx.registry.driver_status(), "poll_stats": ctx.registry.poll_stats(), "active_run": ctx.sequencer.active_run_id(), } # ------------------------------------------------------------------ # GET /io # ------------------------------------------------------------------ @app.get("/io", summary="All signal states") def get_all_io() -> dict[str, Any]: return { name: { "value": s.value, "stale": s.stale, "updated_at": s.updated_at, } for name, s in ctx.registry.snapshot().items() } # ------------------------------------------------------------------ # GET /io/{signal} # ------------------------------------------------------------------ @app.get("/io/{signal_name}", summary="Single signal state") def get_signal(signal_name: str) -> dict: sig = ctx.config.signal(signal_name) if sig is None: raise HTTPException(404, f"Unknown signal: {signal_name!r}") s = ctx.registry.get(signal_name) return { "name": signal_name, "direction": sig.direction, "category": sig.category, "value_type": sig.value_type, "modbus_space": sig.modbus_space, "device": sig.device, "slot": sig.slot, "point": sig.point, "modbus_address": sig.modbus_address, "value": s.value if s else None, "stale": s.stale if s else True, "updated_at": s.updated_at if s else None, } # ------------------------------------------------------------------ # POST /io/{signal}/write — write an output value # ------------------------------------------------------------------ class WriteRequest(BaseModel): value: bool | int # bool for digital, int for analog @app.post("/io/{signal_name}/write", summary="Write an output signal") def write_signal(signal_name: str, req: WriteRequest) -> dict: sig = ctx.config.signal(signal_name) if sig is None: raise HTTPException(404, f"Unknown signal: {signal_name!r}") if sig.direction != "output": raise HTTPException(400, f"Signal {signal_name!r} is an input, not writable") driver = ctx.registry.driver(sig.device) if driver is None: raise HTTPException(503, f"No driver for device {sig.device!r}") if sig.value_type == "int": val = int(req.value) if val < 0 or val > 65535: raise HTTPException(400, f"Analog value must be 0–65535, got {val}") ok = driver.write_register(sig.modbus_address, val) else: val = bool(req.value) ok = driver.write_output(sig.modbus_address, val) if not ok: raise HTTPException(502, f"Write failed for {signal_name!r}") return {"signal": signal_name, "value": val, "ok": True} # ------------------------------------------------------------------ # GET /config/signals — full signal metadata for UI bootstrap # ------------------------------------------------------------------ @app.get("/config/signals", summary="All signal metadata from config") def get_config_signals() -> list[dict]: return [ { "name": s.name, "direction": s.direction, "category": s.category, "value_type": s.value_type, "modbus_space": s.modbus_space, "device": s.device, "slot": s.slot, "point": s.point, "modbus_address": s.modbus_address, "default_state": s.default_state, "default_value": s.default_value, } for s in ctx.config.logical_io ] # ------------------------------------------------------------------ # GET /sequences # ------------------------------------------------------------------ @app.get("/sequences", summary="List all sequences") def list_sequences() -> list[dict]: return [ {"name": seq.name, "description": seq.description, "steps": len(seq.steps)} for seq in ctx.config.sequences ] # ------------------------------------------------------------------ # GET /sequences/{name} # ------------------------------------------------------------------ @app.get("/sequences/{name}", summary="Sequence detail") def get_sequence(name: str) -> dict: seq = ctx.config.sequence(name) if seq is None: raise HTTPException(404, f"Unknown sequence: {name!r}") return { "name": seq.name, "description": seq.description, "steps": [ { "t_ms": step.t_ms, "action": step.action, "signal": step.signal, # Digital fields (None for analog) "state": step.state, "expected": step.expected, # Analog fields (None for digital) "value": step.value, "expected_value": step.expected_value, "tolerance": step.tolerance, # wait_input "timeout_ms": step.timeout_ms, } for step in seq.steps ], } # ------------------------------------------------------------------ # POST /sequences/{name}/run # ------------------------------------------------------------------ @app.post("/sequences/{name}/run", summary="Start a sequence", status_code=202) def run_sequence(name: str) -> dict: if ctx.config.sequence(name) is None: raise HTTPException(404, f"Unknown sequence: {name!r}") run_id, started = ctx.sequencer.start(name) if not started: active = ctx.sequencer.active_run_id() raise HTTPException( 409, f"Sequence already running (run_id={active!r}). " f"Poll GET /runs/{active} for status.", ) return {"run_id": run_id, "sequence": name, "status": "running"} # ------------------------------------------------------------------ # GET /runs/{run_id} # ------------------------------------------------------------------ @app.get("/runs/{run_id}", summary="Run result by ID") def get_run(run_id: str) -> dict: result = ctx.sequencer.get_result(run_id) if result is None: raise HTTPException(404, f"Unknown run_id: {run_id!r}") return result.to_dict() # ------------------------------------------------------------------ # GET /runs # ------------------------------------------------------------------ @app.get("/runs", summary="Recent run history") def list_runs(limit: int = 50) -> list[dict]: return ctx.sequencer.recent_runs(min(limit, 200)) return app