#!/usr/bin/env python3 from __future__ import annotations import argparse import json import mimetypes import sqlite3 import threading import uuid from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlparse ROOT = Path(__file__).resolve().parent STATIC = ROOT / "static" DATA = ROOT / "data" DATABASE = DATA / "limbforge.sqlite3" DATABASE_LOCK = threading.Lock() MEASUREMENTS = [ { "key": "cosmetic_envelope_length", "name": "Cosmetic envelope length", "instruction": ( "Measure between the specialist-agreed proximal reference " "and the intended distal end of the cosmetic cover." ), "guide": "lf_guide_length.png", "focus": "full", "oi_only": False, }, { "key": "proximal_circumference", "name": "Proximal circumference", "instruction": ( "Measure around the full cosmetic envelope at the confirmed " "proximal reference line." ), "guide": "lf_guide_circumference.png", "focus": "circ-proximal", "oi_only": False, }, { "key": "mid_circumference", "name": "Mid-envelope circumference", "instruction": ( "Measure around the centre of the cosmetic envelope while " "keeping the tape level." ), "guide": "lf_guide_circumference.png", "focus": "circ-mid", "oi_only": False, }, { "key": "distal_circumference", "name": "Distal circumference", "instruction": ( "Measure around the specialist-agreed distal cosmetic-cover " "reference line." ), "guide": "lf_guide_circumference.png", "focus": "circ-distal", "oi_only": False, }, { "key": "maximum_width", "name": "Maximum medial-lateral width", "instruction": ( "Measure side to side at the specialist-agreed reference level." ), "guide": "lf_guide_width_depth.png", "focus": "width", "oi_only": False, }, { "key": "maximum_depth", "name": "Maximum anterior-posterior depth", "instruction": ( "Measure front to back at the specialist-agreed reference level." ), "guide": "lf_guide_width_depth.png", "focus": "depth", "oi_only": False, }, { "key": "joint_to_distal", "name": "Joint or hinge centre to distal end", "instruction": ( "Measure from the confirmed joint or hinge centreline to the " "planned distal end of the cosmetic cover." ), "guide": "lf_guide_length.png", "focus": "full", "oi_only": False, }, { "key": "component_clearance", "name": "Component clearance", "instruction": ( "Record the smallest specialist-verified clearance around the " "hardware, hinge, connector or release mechanism." ), "guide": "lf_guide_width_depth.png", "focus": "width", "oi_only": False, }, { "key": "oi_no_contact_radius", "name": "OI no-contact radius", "instruction": ( "Enter only the no-contact radius confirmed by the prosthetic " "specialist." ), "guide": "lf_guide_oi.png", "focus": "oi-radius", "oi_only": True, }, { "key": "oi_access_length", "name": "OI access opening length", "instruction": ( "Record the confirmed length of the cleaning, inspection and " "emergency-access opening." ), "guide": "lf_guide_oi.png", "focus": "oi-opening", "oi_only": True, }, { "key": "oi_access_width", "name": "OI access opening width", "instruction": ( "Record the confirmed width of the cleaning, inspection and " "emergency-access opening." ), "guide": "lf_guide_oi.png", "focus": "oi-opening", "oi_only": True, }, ] MEASUREMENT_BY_KEY = { item["key"]: item for item in MEASUREMENTS } def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def connect() -> sqlite3.Connection: connection = sqlite3.connect(DATABASE) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") return connection def initialise_database() -> None: DATA.mkdir(parents=True, exist_ok=True) with DATABASE_LOCK, connect() as database: database.executescript( """ CREATE TABLE IF NOT EXISTS projects ( id TEXT PRIMARY KEY, name TEXT NOT NULL, reference_id TEXT NOT NULL DEFAULT '', side TEXT NOT NULL, cover_class TEXT NOT NULL, osseointegration INTEGER NOT NULL DEFAULT 0, specialist_name TEXT NOT NULL DEFAULT '', clinic_name TEXT NOT NULL DEFAULT '', exclusion_zones TEXT NOT NULL DEFAULT '', component_notes TEXT NOT NULL DEFAULT '', review_notes TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS measurements ( project_id TEXT NOT NULL, measurement_key TEXT NOT NULL, value_mm REAL NOT NULL, notes TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'completed', updated_at TEXT NOT NULL, PRIMARY KEY (project_id, measurement_key), FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE ); """ ) def row_to_project( row: sqlite3.Row, measurements: list[sqlite3.Row] | None = None, ) -> dict: project = dict(row) project["osseointegration"] = bool(project["osseointegration"]) if measurements is not None: project["measurements"] = { item["measurement_key"]: { "value_mm": item["value_mm"], "notes": item["notes"], "status": item["status"], "updated_at": item["updated_at"], } for item in measurements } return project def list_projects() -> list[dict]: with DATABASE_LOCK, connect() as database: rows = database.execute( """ SELECT * FROM projects ORDER BY updated_at DESC """ ).fetchall() return [row_to_project(row) for row in rows] def load_project(project_id: str) -> dict | None: with DATABASE_LOCK, connect() as database: project = database.execute( "SELECT * FROM projects WHERE id = ?", (project_id,), ).fetchone() if project is None: return None measurements = database.execute( """ SELECT * FROM measurements WHERE project_id = ? ORDER BY updated_at """, (project_id,), ).fetchall() return row_to_project(project, measurements) def create_project(payload: dict) -> dict: project_id = str(uuid.uuid4()) timestamp = utc_now() name = str(payload.get("name", "")).strip() side = str(payload.get("side", "")).strip() cover_class = str(payload.get("cover_class", "")).strip() if not name: raise ValueError("Project name is required.") if side not in {"Right", "Left", "Bilateral"}: raise ValueError("Select Right, Left or Bilateral.") if not cover_class: raise ValueError("Cover class is required.") oi = bool(payload.get("osseointegration", False)) with DATABASE_LOCK, connect() as database: database.execute( """ INSERT INTO projects ( id, name, reference_id, side, cover_class, osseointegration, specialist_name, clinic_name, exclusion_zones, component_notes, review_notes, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, '', '', '', '', '', ?, ?) """, ( project_id, name, str(payload.get("reference_id", "")).strip(), side, cover_class, int(oi), timestamp, timestamp, ), ) return load_project(project_id) def update_project(project_id: str, payload: dict) -> dict: existing = load_project(project_id) if existing is None: raise LookupError("Project not found.") name = str(payload.get("name", existing["name"])).strip() if not name: raise ValueError("Project name is required.") side = str(payload.get("side", existing["side"])).strip() if side not in {"Right", "Left", "Bilateral"}: raise ValueError("Select Right, Left or Bilateral.") cover_class = str( payload.get("cover_class", existing["cover_class"]) ).strip() if not cover_class: raise ValueError("Cover class is required.") oi = bool( payload.get( "osseointegration", existing["osseointegration"], ) ) timestamp = utc_now() with DATABASE_LOCK, connect() as database: database.execute( """ UPDATE projects SET name = ?, reference_id = ?, side = ?, cover_class = ?, osseointegration = ?, specialist_name = ?, clinic_name = ?, exclusion_zones = ?, component_notes = ?, review_notes = ?, updated_at = ? WHERE id = ? """, ( name, str( payload.get( "reference_id", existing["reference_id"], ) ).strip(), side, cover_class, int(oi), str( payload.get( "specialist_name", existing["specialist_name"], ) ).strip(), str( payload.get( "clinic_name", existing["clinic_name"], ) ).strip(), str( payload.get( "exclusion_zones", existing["exclusion_zones"], ) ).strip(), str( payload.get( "component_notes", existing["component_notes"], ) ).strip(), str( payload.get( "review_notes", existing["review_notes"], ) ).strip(), timestamp, project_id, ), ) return load_project(project_id) def save_measurement( project_id: str, measurement_key: str, payload: dict, ) -> dict: project = load_project(project_id) if project is None: raise LookupError("Project not found.") definition = MEASUREMENT_BY_KEY.get(measurement_key) if definition is None: raise ValueError("Unknown measurement.") if definition["oi_only"] and not project["osseointegration"]: raise ValueError( "This measurement is available only for " "osseointegration projects." ) try: value = float(payload.get("value_mm")) except (TypeError, ValueError): raise ValueError("Enter a valid measurement in millimetres.") if value <= 0 or value > 2000: raise ValueError( "Measurement must be greater than zero and below 2000 mm." ) status = str(payload.get("status", "completed")) if status not in {"completed", "specialist_review"}: raise ValueError("Invalid measurement status.") timestamp = utc_now() with DATABASE_LOCK, connect() as database: database.execute( """ INSERT INTO measurements ( project_id, measurement_key, value_mm, notes, status, updated_at ) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(project_id, measurement_key) DO UPDATE SET value_mm = excluded.value_mm, notes = excluded.notes, status = excluded.status, updated_at = excluded.updated_at """, ( project_id, measurement_key, value, str(payload.get("notes", "")).strip(), status, timestamp, ), ) database.execute( """ UPDATE projects SET updated_at = ? WHERE id = ? """, (timestamp, project_id), ) return load_project(project_id) def delete_measurement( project_id: str, measurement_key: str, ) -> dict: project = load_project(project_id) if project is None: raise LookupError("Project not found.") if measurement_key not in MEASUREMENT_BY_KEY: raise ValueError("Unknown measurement.") with DATABASE_LOCK, connect() as database: database.execute( """ DELETE FROM measurements WHERE project_id = ? AND measurement_key = ? """, (project_id, measurement_key), ) database.execute( """ UPDATE projects SET updated_at = ? WHERE id = ? """, (utc_now(), project_id), ) return load_project(project_id) class LimbForgeHandler(BaseHTTPRequestHandler): server_version = "SHiRE-LimbForge/0.1" def log_message(self, message: str, *args) -> None: print( f"{self.address_string()} " f"[{self.log_date_time_string()}] " f"{message % args}", flush=True, ) def send_json( self, payload: dict | list, status: HTTPStatus = HTTPStatus.OK, ) -> None: encoded = json.dumps( payload, indent=2, ensure_ascii=False, ).encode("utf-8") self.send_response(status) self.send_header( "Content-Type", "application/json; charset=utf-8", ) self.send_header("Content-Length", str(len(encoded))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(encoded) def read_json(self) -> dict: length = int(self.headers.get("Content-Length", "0")) if length <= 0: return {} raw = self.rfile.read(length) try: value = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): raise ValueError("Request body must contain valid JSON.") if not isinstance(value, dict): raise ValueError("Request body must be a JSON object.") return value def handle_api_error(self, error: Exception) -> None: if isinstance(error, LookupError): status = HTTPStatus.NOT_FOUND elif isinstance(error, ValueError): status = HTTPStatus.BAD_REQUEST else: status = HTTPStatus.INTERNAL_SERVER_ERROR self.send_json( {"error": str(error)}, status, ) def do_GET(self) -> None: path = urlparse(self.path).path try: if path == "/api/health": self.send_json( { "status": "ok", "service": "limbforge-shared-workspace", "database": str(DATABASE), } ) return if path == "/api/definitions": self.send_json(MEASUREMENTS) return if path == "/api/projects": self.send_json(list_projects()) return parts = [ unquote(part) for part in path.strip("/").split("/") if part ] if ( len(parts) == 3 and parts[0] == "api" and parts[1] == "projects" ): project = load_project(parts[2]) if project is None: raise LookupError("Project not found.") self.send_json(project) return self.serve_static(path) except Exception as error: self.handle_api_error(error) def do_POST(self) -> None: path = urlparse(self.path).path try: if path == "/api/projects": self.send_json( create_project(self.read_json()), HTTPStatus.CREATED, ) return self.send_json( {"error": "Route not found."}, HTTPStatus.NOT_FOUND, ) except Exception as error: self.handle_api_error(error) def do_PUT(self) -> None: path = urlparse(self.path).path parts = [ unquote(part) for part in path.strip("/").split("/") if part ] try: if ( len(parts) == 3 and parts[0] == "api" and parts[1] == "projects" ): self.send_json( update_project( parts[2], self.read_json(), ) ) return if ( len(parts) == 5 and parts[0] == "api" and parts[1] == "projects" and parts[3] == "measurements" ): self.send_json( save_measurement( parts[2], parts[4], self.read_json(), ) ) return self.send_json( {"error": "Route not found."}, HTTPStatus.NOT_FOUND, ) except Exception as error: self.handle_api_error(error) def do_DELETE(self) -> None: path = urlparse(self.path).path parts = [ unquote(part) for part in path.strip("/").split("/") if part ] try: if ( len(parts) == 5 and parts[0] == "api" and parts[1] == "projects" and parts[3] == "measurements" ): self.send_json( delete_measurement( parts[2], parts[4], ) ) return self.send_json( {"error": "Route not found."}, HTTPStatus.NOT_FOUND, ) except Exception as error: self.handle_api_error(error) def serve_static(self, path: str) -> None: if path in {"", "/"}: relative = Path("index.html") else: relative = Path(unquote(path.lstrip("/"))) if ".." in relative.parts: self.send_error(HTTPStatus.BAD_REQUEST) return target = (STATIC / relative).resolve() try: target.relative_to(STATIC.resolve()) except ValueError: self.send_error(HTTPStatus.FORBIDDEN) return if not target.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return content = target.read_bytes() mime, _ = mimetypes.guess_type(target.name) self.send_response(HTTPStatus.OK) self.send_header( "Content-Type", mime or "application/octet-stream", ) self.send_header("Content-Length", str(len(content))) if target.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}: self.send_header( "Cache-Control", "public, max-age=86400", ) else: self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(content) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument( "--port", type=int, default=8790, ) arguments = parser.parse_args() initialise_database() server = ThreadingHTTPServer( ("127.0.0.1", arguments.port), LimbForgeHandler, ) print( f"LimbForge shared workspace listening on " f"http://127.0.0.1:{arguments.port}", flush=True, ) server.serve_forever() if __name__ == "__main__": main()