LIMBFORGE EDIT-ANY-MEASUREMENT EXACT SOURCE ============================================ MODULE IMPORTS AND CONSTANTS — LINES 1-80 ============================================ 1 #!/usr/bin/env python3 2 3 from __future__ import annotations 4 5 import argparse 6 import json 7 import mimetypes 8 import sqlite3 9 import threading 10 import uuid 11 12 from datetime import datetime, timezone 13 from http import HTTPStatus 14 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer 15 from pathlib import Path 16 from urllib.parse import unquote, urlparse 17 18 ROOT = Path(__file__).resolve().parent 19 STATIC = ROOT / "static" 20 DATA = ROOT / "data" 21 DATABASE = DATA / "limbforge.sqlite3" 22 23 DATABASE_LOCK = threading.Lock() 24 25 MEASUREMENTS = [ 26 { 27 "key": "cosmetic_envelope_length", 28 "name": "Cosmetic envelope length", 29 "instruction": ( 30 "Measure between the specialist-agreed proximal reference " 31 "and the intended distal end of the cosmetic cover." 32 ), 33 "guide": "lf_guide_length.png", 34 "focus": "full", 35 "oi_only": False, 36 }, 37 { 38 "key": "proximal_circumference", 39 "name": "Proximal circumference", 40 "instruction": ( 41 "Measure around the full cosmetic envelope at the confirmed " 42 "proximal reference line." 43 ), 44 "guide": "lf_guide_circumference.png", 45 "focus": "circ-proximal", 46 "oi_only": False, 47 }, 48 { 49 "key": "mid_circumference", 50 "name": "Mid-envelope circumference", 51 "instruction": ( 52 "Measure around the centre of the cosmetic envelope while " 53 "keeping the tape level." 54 ), 55 "guide": "lf_guide_circumference.png", 56 "focus": "circ-mid", 57 "oi_only": False, 58 }, 59 { 60 "key": "distal_circumference", 61 "name": "Distal circumference", 62 "instruction": ( 63 "Measure around the specialist-agreed distal cosmetic-cover " 64 "reference line." 65 ), 66 "guide": "lf_guide_circumference.png", 67 "focus": "circ-distal", 68 "oi_only": False, 69 }, 70 { 71 "key": "maximum_width", 72 "name": "Maximum medial-lateral width", 73 "instruction": ( 74 "Measure side to side at the specialist-agreed reference level." 75 ), 76 "guide": "lf_guide_width_depth.png", 77 "focus": "width", 78 "oi_only": False, 79 }, 80 { TIME AND DATABASE HELPERS — LINES 145-165 ============================================ 145 ] 146 147 MEASUREMENT_BY_KEY = { 148 item["key"]: item 149 for item in MEASUREMENTS 150 } 151 152 153 def utc_now() -> str: 154 return datetime.now(timezone.utc).isoformat(timespec="seconds") 155 156 157 def connect() -> sqlite3.Connection: 158 connection = sqlite3.connect(DATABASE) 159 connection.row_factory = sqlite3.Row 160 connection.execute("PRAGMA foreign_keys = ON") 161 return connection 162 163 164 def initialise_database() -> None: 165 DATA.mkdir(parents=True, exist_ok=True) PROJECT SERIALISATION AND LOAD — LINES 202-256 ============================================ 202 def row_to_project( 203 row: sqlite3.Row, 204 measurements: list[sqlite3.Row] | None = None, 205 ) -> dict: 206 project = dict(row) 207 project["osseointegration"] = bool(project["osseointegration"]) 208 209 if measurements is not None: 210 project["measurements"] = { 211 item["measurement_key"]: { 212 "value_mm": item["value_mm"], 213 "notes": item["notes"], 214 "status": item["status"], 215 "updated_at": item["updated_at"], 216 } 217 for item in measurements 218 } 219 220 return project 221 222 223 def list_projects() -> list[dict]: 224 with DATABASE_LOCK, connect() as database: 225 rows = database.execute( 226 """ 227 SELECT * 228 FROM projects 229 ORDER BY updated_at DESC 230 """ 231 ).fetchall() 232 233 return [row_to_project(row) for row in rows] 234 235 236 def load_project(project_id: str) -> dict | None: 237 with DATABASE_LOCK, connect() as database: 238 project = database.execute( 239 "SELECT * FROM projects WHERE id = ?", 240 (project_id,), 241 ).fetchone() 242 243 if project is None: 244 return None 245 246 measurements = database.execute( 247 """ 248 SELECT * 249 FROM measurements 250 WHERE project_id = ? 251 ORDER BY updated_at 252 """, 253 (project_id,), 254 ).fetchall() 255 256 return row_to_project(project, measurements) MEASUREMENT SAVE AND DELETE — LINES 412-522 ============================================ 412 def save_measurement( 413 project_id: str, 414 measurement_key: str, 415 payload: dict, 416 ) -> dict: 417 project = load_project(project_id) 418 419 if project is None: 420 raise LookupError("Project not found.") 421 422 definition = MEASUREMENT_BY_KEY.get(measurement_key) 423 424 if definition is None: 425 raise ValueError("Unknown measurement.") 426 427 if definition["oi_only"] and not project["osseointegration"]: 428 raise ValueError( 429 "This measurement is available only for " 430 "osseointegration projects." 431 ) 432 433 try: 434 value = float(payload.get("value_mm")) 435 except (TypeError, ValueError): 436 raise ValueError("Enter a valid measurement in millimetres.") 437 438 if value <= 0 or value > 2000: 439 raise ValueError( 440 "Measurement must be greater than zero and below 2000 mm." 441 ) 442 443 status = str(payload.get("status", "completed")) 444 445 if status not in {"completed", "specialist_review"}: 446 raise ValueError("Invalid measurement status.") 447 448 timestamp = utc_now() 449 450 with DATABASE_LOCK, connect() as database: 451 database.execute( 452 """ 453 INSERT INTO measurements ( 454 project_id, 455 measurement_key, 456 value_mm, 457 notes, 458 status, 459 updated_at 460 ) 461 VALUES (?, ?, ?, ?, ?, ?) 462 ON CONFLICT(project_id, measurement_key) 463 DO UPDATE SET 464 value_mm = excluded.value_mm, 465 notes = excluded.notes, 466 status = excluded.status, 467 updated_at = excluded.updated_at 468 """, 469 ( 470 project_id, 471 measurement_key, 472 value, 473 str(payload.get("notes", "")).strip(), 474 status, 475 timestamp, 476 ), 477 ) 478 479 database.execute( 480 """ 481 UPDATE projects 482 SET updated_at = ? 483 WHERE id = ? 484 """, 485 (timestamp, project_id), 486 ) 487 488 return load_project(project_id) 489 490 491 def delete_measurement( 492 project_id: str, 493 measurement_key: str, 494 ) -> dict: 495 project = load_project(project_id) 496 497 if project is None: 498 raise LookupError("Project not found.") 499 500 if measurement_key not in MEASUREMENT_BY_KEY: 501 raise ValueError("Unknown measurement.") 502 503 with DATABASE_LOCK, connect() as database: 504 database.execute( 505 """ 506 DELETE FROM measurements 507 WHERE project_id = ? 508 AND measurement_key = ? 509 """, 510 (project_id, measurement_key), 511 ) 512 513 database.execute( 514 """ 515 UPDATE projects 516 SET updated_at = ? 517 WHERE id = ? 518 """, 519 (utc_now(), project_id), 520 ) 521 522 return load_project(project_id) HTTP HANDLER AND ROUTES — LINES 525-727 ============================================ 525 class LimbForgeHandler(BaseHTTPRequestHandler): 526 server_version = "SHiRE-LimbForge/0.1" 527 528 def log_message(self, message: str, *args) -> None: 529 print( 530 f"{self.address_string()} " 531 f"[{self.log_date_time_string()}] " 532 f"{message % args}", 533 flush=True, 534 ) 535 536 def send_json( 537 self, 538 payload: dict | list, 539 status: HTTPStatus = HTTPStatus.OK, 540 ) -> None: 541 encoded = json.dumps( 542 payload, 543 indent=2, 544 ensure_ascii=False, 545 ).encode("utf-8") 546 547 self.send_response(status) 548 self.send_header( 549 "Content-Type", 550 "application/json; charset=utf-8", 551 ) 552 self.send_header("Content-Length", str(len(encoded))) 553 self.send_header("Cache-Control", "no-store") 554 self.end_headers() 555 self.wfile.write(encoded) 556 557 def read_json(self) -> dict: 558 length = int(self.headers.get("Content-Length", "0")) 559 560 if length <= 0: 561 return {} 562 563 raw = self.rfile.read(length) 564 565 try: 566 value = json.loads(raw.decode("utf-8")) 567 except (UnicodeDecodeError, json.JSONDecodeError): 568 raise ValueError("Request body must contain valid JSON.") 569 570 if not isinstance(value, dict): 571 raise ValueError("Request body must be a JSON object.") 572 573 return value 574 575 def handle_api_error(self, error: Exception) -> None: 576 if isinstance(error, LookupError): 577 status = HTTPStatus.NOT_FOUND 578 elif isinstance(error, ValueError): 579 status = HTTPStatus.BAD_REQUEST 580 else: 581 status = HTTPStatus.INTERNAL_SERVER_ERROR 582 583 self.send_json( 584 {"error": str(error)}, 585 status, 586 ) 587 588 def do_GET(self) -> None: 589 path = urlparse(self.path).path 590 591 try: 592 if path == "/api/health": 593 self.send_json( 594 { 595 "status": "ok", 596 "service": "limbforge-shared-workspace", 597 "database": str(DATABASE), 598 } 599 ) 600 return 601 602 if path == "/api/definitions": 603 self.send_json(MEASUREMENTS) 604 return 605 606 if path == "/api/projects": 607 self.send_json(list_projects()) 608 return 609 610 parts = [ 611 unquote(part) 612 for part in path.strip("/").split("/") 613 if part 614 ] 615 616 if ( 617 len(parts) == 3 618 and parts[0] == "api" 619 and parts[1] == "projects" 620 ): 621 project = load_project(parts[2]) 622 623 if project is None: 624 raise LookupError("Project not found.") 625 626 self.send_json(project) 627 return 628 629 self.serve_static(path) 630 631 except Exception as error: 632 self.handle_api_error(error) 633 634 def do_POST(self) -> None: 635 path = urlparse(self.path).path 636 637 try: 638 if path == "/api/projects": 639 self.send_json( 640 create_project(self.read_json()), 641 HTTPStatus.CREATED, 642 ) 643 return 644 645 self.send_json( 646 {"error": "Route not found."}, 647 HTTPStatus.NOT_FOUND, 648 ) 649 650 except Exception as error: 651 self.handle_api_error(error) 652 653 def do_PUT(self) -> None: 654 path = urlparse(self.path).path 655 parts = [ 656 unquote(part) 657 for part in path.strip("/").split("/") 658 if part 659 ] 660 661 try: 662 if ( 663 len(parts) == 3 664 and parts[0] == "api" 665 and parts[1] == "projects" 666 ): 667 self.send_json( 668 update_project( 669 parts[2], 670 self.read_json(), 671 ) 672 ) 673 return 674 675 if ( 676 len(parts) == 5 677 and parts[0] == "api" 678 and parts[1] == "projects" 679 and parts[3] == "measurements" 680 ): 681 self.send_json( 682 save_measurement( 683 parts[2], 684 parts[4], 685 self.read_json(), 686 ) 687 ) 688 return 689 690 self.send_json( 691 {"error": "Route not found."}, 692 HTTPStatus.NOT_FOUND, 693 ) 694 695 except Exception as error: 696 self.handle_api_error(error) 697 698 def do_DELETE(self) -> None: 699 path = urlparse(self.path).path 700 parts = [ 701 unquote(part) 702 for part in path.strip("/").split("/") 703 if part 704 ] 705 706 try: 707 if ( 708 len(parts) == 5 709 and parts[0] == "api" 710 and parts[1] == "projects" 711 and parts[3] == "measurements" 712 ): 713 self.send_json( 714 delete_measurement( 715 parts[2], 716 parts[4], 717 ) 718 ) 719 return 720 721 self.send_json( 722 {"error": "Route not found."}, 723 HTTPStatus.NOT_FOUND, 724 ) 725 726 except Exception as error: 727 self.handle_api_error(error)