LIMBFORGE EDIT-ANY-MEASUREMENT API ANCHORS ============================================================================== IMPORTS AND MODULE CONSTANTS ============================================================================== 00001: #!/usr/bin/env python3 00002: 00003: from __future__ import annotations 00004: 00005: import argparse 00006: import json 00007: import mimetypes 00008: import sqlite3 00009: import threading 00010: import uuid 00011: 00012: from datetime import datetime, timezone 00013: from http import HTTPStatus 00014: from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer 00015: from pathlib import Path 00016: from urllib.parse import unquote, urlparse 00017: 00018: ROOT = Path(__file__).resolve().parent 00019: STATIC = ROOT / "static" 00020: DATA = ROOT / "data" 00021: DATABASE = DATA / "limbforge.sqlite3" 00022: 00023: DATABASE_LOCK = threading.Lock() 00024: 00025: MEASUREMENTS = [ 00026: { 00027: "key": "cosmetic_envelope_length", 00028: "name": "Cosmetic envelope length", 00029: "instruction": ( 00030: "Measure between the specialist-agreed proximal reference " 00031: "and the intended distal end of the cosmetic cover." 00032: ), 00033: "guide": "lf_guide_length.png", 00034: "focus": "full", 00035: "oi_only": False, 00036: }, 00037: { 00038: "key": "proximal_circumference", 00039: "name": "Proximal circumference", 00040: "instruction": ( 00041: "Measure around the full cosmetic envelope at the confirmed " 00042: "proximal reference line." 00043: ), 00044: "guide": "lf_guide_circumference.png", 00045: "focus": "circ-proximal", 00046: "oi_only": False, 00047: }, 00048: { 00049: "key": "mid_circumference", 00050: "name": "Mid-envelope circumference", 00051: "instruction": ( 00052: "Measure around the centre of the cosmetic envelope while " 00053: "keeping the tape level." 00054: ), 00055: "guide": "lf_guide_circumference.png", 00056: "focus": "circ-mid", 00057: "oi_only": False, 00058: }, 00059: { 00060: "key": "distal_circumference", 00061: "name": "Distal circumference", 00062: "instruction": ( 00063: "Measure around the specialist-agreed distal cosmetic-cover " 00064: "reference line." 00065: ), 00066: "guide": "lf_guide_circumference.png", 00067: "focus": "circ-distal", 00068: "oi_only": False, 00069: }, 00070: { 00071: "key": "maximum_width", 00072: "name": "Maximum medial-lateral width", 00073: "instruction": ( 00074: "Measure side to side at the specialist-agreed reference level." 00075: ), 00076: "guide": "lf_guide_width_depth.png", 00077: "focus": "width", 00078: "oi_only": False, 00079: }, 00080: { 00081: "key": "maximum_depth", 00082: "name": "Maximum anterior-posterior depth", 00083: "instruction": ( 00084: "Measure front to back at the specialist-agreed reference level." 00085: ), 00086: "guide": "lf_guide_width_depth.png", 00087: "focus": "depth", 00088: "oi_only": False, 00089: }, 00090: { 00091: "key": "joint_to_distal", 00092: "name": "Joint or hinge centre to distal end", 00093: "instruction": ( 00094: "Measure from the confirmed joint or hinge centreline to the " 00095: "planned distal end of the cosmetic cover." 00096: ), 00097: "guide": "lf_guide_length.png", 00098: "focus": "full", 00099: "oi_only": False, 00100: }, 00101: { 00102: "key": "component_clearance", 00103: "name": "Component clearance", 00104: "instruction": ( 00105: "Record the smallest specialist-verified clearance around the " 00106: "hardware, hinge, connector or release mechanism." 00107: ), 00108: "guide": "lf_guide_width_depth.png", 00109: "focus": "width", 00110: "oi_only": False, 00111: }, 00112: { 00113: "key": "oi_no_contact_radius", 00114: "name": "OI no-contact radius", 00115: "instruction": ( 00116: "Enter only the no-contact radius confirmed by the prosthetic " 00117: "specialist." 00118: ), 00119: "guide": "lf_guide_oi.png", 00120: "focus": "oi-radius", 00121: "oi_only": True, 00122: }, 00123: { 00124: "key": "oi_access_length", 00125: "name": "OI access opening length", 00126: "instruction": ( 00127: "Record the confirmed length of the cleaning, inspection and " 00128: "emergency-access opening." 00129: ), 00130: "guide": "lf_guide_oi.png", 00131: "focus": "oi-opening", 00132: "oi_only": True, 00133: }, 00134: { 00135: "key": "oi_access_width", 00136: "name": "OI access opening width", 00137: "instruction": ( 00138: "Record the confirmed width of the cleaning, inspection and " 00139: "emergency-access opening." 00140: ), 00141: "guide": "lf_guide_oi.png", 00142: "focus": "oi-opening", 00143: "oi_only": True, 00144: }, 00145: ] 00146: 00147: MEASUREMENT_BY_KEY = { 00148: item["key"]: item 00149: for item in MEASUREMENTS 00150: } 00151: 00152: 00153: def utc_now() -> str: ============================================================================== FUNCTION: utc_now LINES 153-154 ============================================================================== 00153: def utc_now() -> str: 00154: return datetime.now(timezone.utc).isoformat(timespec="seconds") ============================================================================== FUNCTION: row_to_project LINES 202-220 ============================================================================== 00202: def row_to_project( 00203: row: sqlite3.Row, 00204: measurements: list[sqlite3.Row] | None = None, 00205: ) -> dict: 00206: project = dict(row) 00207: project["osseointegration"] = bool(project["osseointegration"]) 00208: 00209: if measurements is not None: 00210: project["measurements"] = { 00211: item["measurement_key"]: { 00212: "value_mm": item["value_mm"], 00213: "notes": item["notes"], 00214: "status": item["status"], 00215: "updated_at": item["updated_at"], 00216: } 00217: for item in measurements 00218: } 00219: 00220: return project ============================================================================== FUNCTION: list_projects LINES 223-233 ============================================================================== 00223: def list_projects() -> list[dict]: 00224: with DATABASE_LOCK, connect() as database: 00225: rows = database.execute( 00226: """ 00227: SELECT * 00228: FROM projects 00229: ORDER BY updated_at DESC 00230: """ 00231: ).fetchall() 00232: 00233: return [row_to_project(row) for row in rows] ============================================================================== FUNCTION: load_project LINES 236-256 ============================================================================== 00236: def load_project(project_id: str) -> dict | None: 00237: with DATABASE_LOCK, connect() as database: 00238: project = database.execute( 00239: "SELECT * FROM projects WHERE id = ?", 00240: (project_id,), 00241: ).fetchone() 00242: 00243: if project is None: 00244: return None 00245: 00246: measurements = database.execute( 00247: """ 00248: SELECT * 00249: FROM measurements 00250: WHERE project_id = ? 00251: ORDER BY updated_at 00252: """, 00253: (project_id,), 00254: ).fetchall() 00255: 00256: return row_to_project(project, measurements) ============================================================================== FUNCTION: create_project LINES 259-310 ============================================================================== 00259: def create_project(payload: dict) -> dict: 00260: project_id = str(uuid.uuid4()) 00261: timestamp = utc_now() 00262: 00263: name = str(payload.get("name", "")).strip() 00264: side = str(payload.get("side", "")).strip() 00265: cover_class = str(payload.get("cover_class", "")).strip() 00266: 00267: if not name: 00268: raise ValueError("Project name is required.") 00269: 00270: if side not in {"Right", "Left", "Bilateral"}: 00271: raise ValueError("Select Right, Left or Bilateral.") 00272: 00273: if not cover_class: 00274: raise ValueError("Cover class is required.") 00275: 00276: oi = bool(payload.get("osseointegration", False)) 00277: 00278: with DATABASE_LOCK, connect() as database: 00279: database.execute( 00280: """ 00281: INSERT INTO projects ( 00282: id, 00283: name, 00284: reference_id, 00285: side, 00286: cover_class, 00287: osseointegration, 00288: specialist_name, 00289: clinic_name, 00290: exclusion_zones, 00291: component_notes, 00292: review_notes, 00293: created_at, 00294: updated_at 00295: ) 00296: VALUES (?, ?, ?, ?, ?, ?, '', '', '', '', '', ?, ?) 00297: """, 00298: ( 00299: project_id, 00300: name, 00301: str(payload.get("reference_id", "")).strip(), 00302: side, 00303: cover_class, 00304: int(oi), 00305: timestamp, 00306: timestamp, 00307: ), 00308: ) 00309: 00310: return load_project(project_id) ============================================================================== FUNCTION: update_project LINES 313-409 ============================================================================== 00313: def update_project(project_id: str, payload: dict) -> dict: 00314: existing = load_project(project_id) 00315: 00316: if existing is None: 00317: raise LookupError("Project not found.") 00318: 00319: name = str(payload.get("name", existing["name"])).strip() 00320: 00321: if not name: 00322: raise ValueError("Project name is required.") 00323: 00324: side = str(payload.get("side", existing["side"])).strip() 00325: 00326: if side not in {"Right", "Left", "Bilateral"}: 00327: raise ValueError("Select Right, Left or Bilateral.") 00328: 00329: cover_class = str( 00330: payload.get("cover_class", existing["cover_class"]) 00331: ).strip() 00332: 00333: if not cover_class: 00334: raise ValueError("Cover class is required.") 00335: 00336: oi = bool( 00337: payload.get( 00338: "osseointegration", 00339: existing["osseointegration"], 00340: ) 00341: ) 00342: 00343: timestamp = utc_now() 00344: 00345: with DATABASE_LOCK, connect() as database: 00346: database.execute( 00347: """ 00348: UPDATE projects 00349: SET 00350: name = ?, 00351: reference_id = ?, 00352: side = ?, 00353: cover_class = ?, 00354: osseointegration = ?, 00355: specialist_name = ?, 00356: clinic_name = ?, 00357: exclusion_zones = ?, 00358: component_notes = ?, 00359: review_notes = ?, 00360: updated_at = ? 00361: WHERE id = ? 00362: """, 00363: ( 00364: name, 00365: str( 00366: payload.get( 00367: "reference_id", 00368: existing["reference_id"], 00369: ) 00370: ).strip(), 00371: side, 00372: cover_class, 00373: int(oi), 00374: str( 00375: payload.get( 00376: "specialist_name", 00377: existing["specialist_name"], 00378: ) 00379: ).strip(), 00380: str( 00381: payload.get( 00382: "clinic_name", 00383: existing["clinic_name"], 00384: ) 00385: ).strip(), 00386: str( 00387: payload.get( 00388: "exclusion_zones", 00389: existing["exclusion_zones"], 00390: ) 00391: ).strip(), 00392: str( 00393: payload.get( 00394: "component_notes", 00395: existing["component_notes"], 00396: ) 00397: ).strip(), 00398: str( 00399: payload.get( 00400: "review_notes", 00401: existing["review_notes"], 00402: ) 00403: ).strip(), 00404: timestamp, 00405: project_id, 00406: ), 00407: ) 00408: 00409: return load_project(project_id) ============================================================================== FUNCTION: save_measurement LINES 412-488 ============================================================================== 00412: def save_measurement( 00413: project_id: str, 00414: measurement_key: str, 00415: payload: dict, 00416: ) -> dict: 00417: project = load_project(project_id) 00418: 00419: if project is None: 00420: raise LookupError("Project not found.") 00421: 00422: definition = MEASUREMENT_BY_KEY.get(measurement_key) 00423: 00424: if definition is None: 00425: raise ValueError("Unknown measurement.") 00426: 00427: if definition["oi_only"] and not project["osseointegration"]: 00428: raise ValueError( 00429: "This measurement is available only for " 00430: "osseointegration projects." 00431: ) 00432: 00433: try: 00434: value = float(payload.get("value_mm")) 00435: except (TypeError, ValueError): 00436: raise ValueError("Enter a valid measurement in millimetres.") 00437: 00438: if value <= 0 or value > 2000: 00439: raise ValueError( 00440: "Measurement must be greater than zero and below 2000 mm." 00441: ) 00442: 00443: status = str(payload.get("status", "completed")) 00444: 00445: if status not in {"completed", "specialist_review"}: 00446: raise ValueError("Invalid measurement status.") 00447: 00448: timestamp = utc_now() 00449: 00450: with DATABASE_LOCK, connect() as database: 00451: database.execute( 00452: """ 00453: INSERT INTO measurements ( 00454: project_id, 00455: measurement_key, 00456: value_mm, 00457: notes, 00458: status, 00459: updated_at 00460: ) 00461: VALUES (?, ?, ?, ?, ?, ?) 00462: ON CONFLICT(project_id, measurement_key) 00463: DO UPDATE SET 00464: value_mm = excluded.value_mm, 00465: notes = excluded.notes, 00466: status = excluded.status, 00467: updated_at = excluded.updated_at 00468: """, 00469: ( 00470: project_id, 00471: measurement_key, 00472: value, 00473: str(payload.get("notes", "")).strip(), 00474: status, 00475: timestamp, 00476: ), 00477: ) 00478: 00479: database.execute( 00480: """ 00481: UPDATE projects 00482: SET updated_at = ? 00483: WHERE id = ? 00484: """, 00485: (timestamp, project_id), 00486: ) 00487: 00488: return load_project(project_id) ============================================================================== FUNCTION: delete_measurement LINES 491-522 ============================================================================== 00491: def delete_measurement( 00492: project_id: str, 00493: measurement_key: str, 00494: ) -> dict: 00495: project = load_project(project_id) 00496: 00497: if project is None: 00498: raise LookupError("Project not found.") 00499: 00500: if measurement_key not in MEASUREMENT_BY_KEY: 00501: raise ValueError("Unknown measurement.") 00502: 00503: with DATABASE_LOCK, connect() as database: 00504: database.execute( 00505: """ 00506: DELETE FROM measurements 00507: WHERE project_id = ? 00508: AND measurement_key = ? 00509: """, 00510: (project_id, measurement_key), 00511: ) 00512: 00513: database.execute( 00514: """ 00515: UPDATE projects 00516: SET updated_at = ? 00517: WHERE id = ? 00518: """, 00519: (utc_now(), project_id), 00520: ) 00521: 00522: return load_project(project_id) ============================================================================== HTTP HANDLER CLASS: LimbForgeHandler START LINE 525 ============================================================================== ============================================================================== HTTP METHOD: LimbForgeHandler.send_json LINES 536-555 ============================================================================== 00536: def send_json( 00537: self, 00538: payload: dict | list, 00539: status: HTTPStatus = HTTPStatus.OK, 00540: ) -> None: 00541: encoded = json.dumps( 00542: payload, 00543: indent=2, 00544: ensure_ascii=False, 00545: ).encode("utf-8") 00546: 00547: self.send_response(status) 00548: self.send_header( 00549: "Content-Type", 00550: "application/json; charset=utf-8", 00551: ) 00552: self.send_header("Content-Length", str(len(encoded))) 00553: self.send_header("Cache-Control", "no-store") 00554: self.end_headers() 00555: self.wfile.write(encoded) ============================================================================== HTTP METHOD: LimbForgeHandler.read_json LINES 557-573 ============================================================================== 00557: def read_json(self) -> dict: 00558: length = int(self.headers.get("Content-Length", "0")) 00559: 00560: if length <= 0: 00561: return {} 00562: 00563: raw = self.rfile.read(length) 00564: 00565: try: 00566: value = json.loads(raw.decode("utf-8")) 00567: except (UnicodeDecodeError, json.JSONDecodeError): 00568: raise ValueError("Request body must contain valid JSON.") 00569: 00570: if not isinstance(value, dict): 00571: raise ValueError("Request body must be a JSON object.") 00572: 00573: return value ============================================================================== HTTP METHOD: LimbForgeHandler.do_GET LINES 588-632 ============================================================================== 00588: def do_GET(self) -> None: 00589: path = urlparse(self.path).path 00590: 00591: try: 00592: if path == "/api/health": 00593: self.send_json( 00594: { 00595: "status": "ok", 00596: "service": "limbforge-shared-workspace", 00597: "database": str(DATABASE), 00598: } 00599: ) 00600: return 00601: 00602: if path == "/api/definitions": 00603: self.send_json(MEASUREMENTS) 00604: return 00605: 00606: if path == "/api/projects": 00607: self.send_json(list_projects()) 00608: return 00609: 00610: parts = [ 00611: unquote(part) 00612: for part in path.strip("/").split("/") 00613: if part 00614: ] 00615: 00616: if ( 00617: len(parts) == 3 00618: and parts[0] == "api" 00619: and parts[1] == "projects" 00620: ): 00621: project = load_project(parts[2]) 00622: 00623: if project is None: 00624: raise LookupError("Project not found.") 00625: 00626: self.send_json(project) 00627: return 00628: 00629: self.serve_static(path) 00630: 00631: except Exception as error: 00632: self.handle_api_error(error) ============================================================================== HTTP METHOD: LimbForgeHandler.do_POST LINES 634-651 ============================================================================== 00634: def do_POST(self) -> None: 00635: path = urlparse(self.path).path 00636: 00637: try: 00638: if path == "/api/projects": 00639: self.send_json( 00640: create_project(self.read_json()), 00641: HTTPStatus.CREATED, 00642: ) 00643: return 00644: 00645: self.send_json( 00646: {"error": "Route not found."}, 00647: HTTPStatus.NOT_FOUND, 00648: ) 00649: 00650: except Exception as error: 00651: self.handle_api_error(error) ============================================================================== HTTP METHOD: LimbForgeHandler.do_PUT LINES 653-696 ============================================================================== 00653: def do_PUT(self) -> None: 00654: path = urlparse(self.path).path 00655: parts = [ 00656: unquote(part) 00657: for part in path.strip("/").split("/") 00658: if part 00659: ] 00660: 00661: try: 00662: if ( 00663: len(parts) == 3 00664: and parts[0] == "api" 00665: and parts[1] == "projects" 00666: ): 00667: self.send_json( 00668: update_project( 00669: parts[2], 00670: self.read_json(), 00671: ) 00672: ) 00673: return 00674: 00675: if ( 00676: len(parts) == 5 00677: and parts[0] == "api" 00678: and parts[1] == "projects" 00679: and parts[3] == "measurements" 00680: ): 00681: self.send_json( 00682: save_measurement( 00683: parts[2], 00684: parts[4], 00685: self.read_json(), 00686: ) 00687: ) 00688: return 00689: 00690: self.send_json( 00691: {"error": "Route not found."}, 00692: HTTPStatus.NOT_FOUND, 00693: ) 00694: 00695: except Exception as error: 00696: self.handle_api_error(error) ============================================================================== HTTP METHOD: LimbForgeHandler.do_DELETE LINES 698-727 ============================================================================== 00698: def do_DELETE(self) -> None: 00699: path = urlparse(self.path).path 00700: parts = [ 00701: unquote(part) 00702: for part in path.strip("/").split("/") 00703: if part 00704: ] 00705: 00706: try: 00707: if ( 00708: len(parts) == 5 00709: and parts[0] == "api" 00710: and parts[1] == "projects" 00711: and parts[3] == "measurements" 00712: ): 00713: self.send_json( 00714: delete_measurement( 00715: parts[2], 00716: parts[4], 00717: ) 00718: ) 00719: return 00720: 00721: self.send_json( 00722: {"error": "Route not found."}, 00723: HTTPStatus.NOT_FOUND, 00724: ) 00725: 00726: except Exception as error: 00727: self.handle_api_error(error)