import asyncio import json import logging import math import time import uuid from abc import ABC, abstractmethod from collections import deque from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from typing import Any, Callable, Literal import cv2 import numpy as np from fastapi import WebSocket from pydantic import BaseModel, Field, field_validator from neural_forge import signals from neural_forge.models import SETTINGS, FaceData, HandData, PoseData, PoseIdx, SessionTracker logger = logging.getLogger("neural_forge.engine") EXECUTOR = ThreadPoolExecutor(max_workers=SETTINGS.worker_threads, thread_name_prefix="neural-forge") FindingStatus = Literal["normal", "borderline", "abnormal", "inconclusive"] Urgency = Literal["none", "routine", "soon", "urgent"] Likelihood = Literal["low", "moderate", "high"] ViewType = Literal["face", "face_close", "mouth", "hands", "hand_close", "upper_body", "full_body", "skin"] TrackerName = Literal["face", "hands", "pose"] CONTROL_ACTIONS = {"start", "ready", "pause", "resume", "skip", "stop", "add_area", "next"} def json_safe(value: Any) -> Any: if isinstance(value, dict): return {str(key): json_safe(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [json_safe(item) for item in value] if isinstance(value, np.ndarray): return json_safe(value.tolist()) if isinstance(value, (bool, np.bool_)): return bool(value) if isinstance(value, np.integer): return int(value) if isinstance(value, (float, np.floating)): number = float(value) return number if math.isfinite(number) else None return value class Measurement(BaseModel): name: str value: float | None = None unit: str = "" reference: str = "" side: Literal["right", "left", "both", "none"] = "none" @field_validator("value", mode="before") @classmethod def clean_value(cls, value: Any) -> float | None: if value is None: return None try: number = float(value) except (TypeError, ValueError): return None if not math.isfinite(number): return None return round(number, 4) class Finding(BaseModel): code: str title: str status: FindingStatus confidence: float = 0.0 measurements: list[Measurement] = Field(default_factory=list) explanation: str = "" red_flag: bool = False evidence_image: str | None = None @field_validator("confidence", mode="before") @classmethod def clean_confidence(cls, value: Any) -> float: try: number = float(value) except (TypeError, ValueError): return 0.0 if not math.isfinite(number): return 0.0 return round(min(1.0, max(0.0, number)), 3) class ConditionRule(BaseModel): name: str all_of: list[str] = Field(default_factory=list) any_of: list[str] = Field(default_factory=list) min_any: int = 1 likelihood: Likelihood = "moderate" urgency: Urgency = "routine" recommendation: str = "" class Step(BaseModel): id: str title: str display_text: str speech_text: str tips: str = "" safety_note: str = "" view: ViewType trackers: list[TrackerName] = Field(default_factory=list) hands_required: int = 0 min_seconds: float = 3.0 max_seconds: float = 20.0 min_valid_frames: int = 15 get_ready_seconds: int = 3 requires_ready: bool = True skippable: bool = False repeatable: bool = False collect: bool = True strict_quality: bool = True check_blur: bool = True def public(self, number: int, total: int) -> dict[str, Any]: return { "id": self.id, "number": number, "total": total, "title": self.title, "display_text": self.display_text, "speech_text": self.speech_text, "tips": self.tips, "safety_note": self.safety_note, "view": self.view, "min_seconds": self.min_seconds, "max_seconds": self.max_seconds, "requires_ready": self.requires_ready, "skippable": self.skippable, "repeatable": self.repeatable, } class PatientContext(BaseModel): age: int | None = Field(default=None, ge=0, le=120) sex: Literal["male", "female", "other", "unspecified"] = "unspecified" symptoms: list[str] = Field(default_factory=list, max_length=30) medical_history: list[str] = Field(default_factory=list, max_length=30) notes: str = Field(default="", max_length=1000) class StepRecord(BaseModel): step_id: str title: str status: Literal["completed", "incomplete", "skipped"] attempts: int valid_frames: int duration_seconds: float note: str = "" QUALITY = { "min_brightness": 50.0, "max_brightness": 220.0, "max_overexposed_fraction": 0.15, "min_sharpness_face": 25.0, "min_sharpness_hands": 20.0, "min_sharpness_frame": 12.0, "face_min_width": 0.18, "face_max_width": 0.85, "face_close_min_width": 0.42, "mouth_min_width": 0.35, "max_yaw_degrees": 20.0, "max_pitch_degrees": 20.0, "hand_min_size": 0.10, "hand_close_min_size": 0.22, "full_body_min_height": 0.45, "edge_margin_pixels": 4, "good_frames_required": 6, } QUALITY_TEXT = { "too_dark": "The picture is too dark. Please move to a brighter place or turn on a light in front of you.", "too_bright": "There is too much light. Please avoid bright light or a window behind you.", "blurry": "The picture is blurry. Please keep the camera and your body still.", "no_face": "I cannot see your face. Please look at the camera.", "face_too_far": "Please move closer to the camera.", "face_too_close": "Please move a little further away from the camera.", "need_closer_eyes": "Please bring your face closer so it fills most of the screen.", "need_closer_mouth": "Please bring the camera closer to your mouth.", "head_turned": "Please face the camera directly.", "head_tilted": "Please keep your head level and look straight at the camera.", "hands_missing_one": "Please show your hand clearly to the camera.", "hands_missing_two": "Please show both hands clearly to the camera.", "hand_too_far": "Please bring your hand closer to the camera.", "upper_body_missing": "Please move back so your head, shoulders and both arms are visible.", "full_body_missing": "Please step back so your whole body, from head to feet, is visible.", "full_body_too_small": "Please come a little closer while keeping your whole body visible.", "full_body_cut": "Part of your body is outside the picture. Please step back a little.", "no_frames": "I am not receiving video from your camera. Please check that the camera is on.", } @dataclass class QualityResult: ok: bool code: str = "ok" message: str = "" @staticmethod def issue(code: str) -> "QualityResult": return QualityResult(ok=False, code=code, message=QUALITY_TEXT.get(code, "Please adjust your position.")) @dataclass class Frame: image_bgr: np.ndarray timestamp_s: float face: FaceData | None = None hands: dict[str, HandData] = field(default_factory=dict) pose: PoseData | None = None quality: QualityResult | None = None _gray: np.ndarray | None = field(default=None, repr=False) _lab_raw: np.ndarray | None = field(default=None, repr=False) _lab_balanced: np.ndarray | None = field(default=None, repr=False) @property def width(self) -> int: return int(self.image_bgr.shape[1]) @property def height(self) -> int: return int(self.image_bgr.shape[0]) def gray(self) -> np.ndarray: if self._gray is None: self._gray = cv2.cvtColor(self.image_bgr, cv2.COLOR_BGR2GRAY) return self._gray def lab(self, balanced: bool = False) -> np.ndarray: if balanced: if self._lab_balanced is None: self._lab_balanced = signals.to_lab(signals.grey_world_balance(self.image_bgr)) return self._lab_balanced if self._lab_raw is None: self._lab_raw = signals.to_lab(self.image_bgr) return self._lab_raw def _union_box(boxes: list[tuple[int, int, int, int]]) -> tuple[int, int, int, int]: return ( min(box[0] for box in boxes), min(box[1] for box in boxes), max(box[2] for box in boxes), max(box[3] for box in boxes), ) def assess_quality(frame: Frame, step: Step) -> QualityResult: small = signals.downscale_to_width(frame.gray(), 640) brightness = float(small.mean()) if brightness < QUALITY["min_brightness"]: return QualityResult.issue("too_dark") if brightness > QUALITY["max_brightness"] or float((small >= 250).mean()) > QUALITY["max_overexposed_fraction"]: return QualityResult.issue("too_bright") focus_box: tuple[int, int, int, int] | None = None sharpness_minimum = QUALITY["min_sharpness_frame"] view = step.view if view in ("face", "face_close", "mouth"): face = frame.face if face is None: return QualityResult.issue("no_face") if view == "face": if face.width_fraction < QUALITY["face_min_width"]: return QualityResult.issue("face_too_far") if face.width_fraction > QUALITY["face_max_width"]: return QualityResult.issue("face_too_close") elif view == "face_close": if face.width_fraction < QUALITY["face_close_min_width"]: return QualityResult.issue("need_closer_eyes") elif face.width_fraction < QUALITY["mouth_min_width"]: return QualityResult.issue("need_closer_mouth") if abs(face.yaw) > QUALITY["max_yaw_degrees"]: return QualityResult.issue("head_turned") if view != "mouth" and abs(face.pitch) > QUALITY["max_pitch_degrees"]: return QualityResult.issue("head_tilted") focus_box = face.box sharpness_minimum = QUALITY["min_sharpness_face"] elif view in ("hands", "hand_close"): required = max(1, step.hands_required) if len(frame.hands) < required: return QualityResult.issue("hands_missing_two" if required >= 2 else "hands_missing_one") minimum_size = QUALITY["hand_close_min_size"] if view == "hand_close" else QUALITY["hand_min_size"] for hand in frame.hands.values(): if hand.size_fraction < minimum_size: return QualityResult.issue("hand_too_far") focus_box = _union_box([hand.box for hand in frame.hands.values()]) sharpness_minimum = QUALITY["min_sharpness_hands"] elif view == "upper_body": pose = frame.pose required_points = ( PoseIdx.NOSE, PoseIdx.LEFT_SHOULDER, PoseIdx.RIGHT_SHOULDER, PoseIdx.LEFT_ELBOW, PoseIdx.RIGHT_ELBOW, ) if pose is None or not all(pose.visible(index) for index in required_points): return QualityResult.issue("upper_body_missing") if step.hands_required and len(frame.hands) < step.hands_required: return QualityResult.issue("hands_missing_two" if step.hands_required >= 2 else "hands_missing_one") elif view == "full_body": pose = frame.pose required_points = ( PoseIdx.LEFT_SHOULDER, PoseIdx.RIGHT_SHOULDER, PoseIdx.LEFT_HIP, PoseIdx.RIGHT_HIP, PoseIdx.LEFT_KNEE, PoseIdx.RIGHT_KNEE, PoseIdx.LEFT_ANKLE, PoseIdx.RIGHT_ANKLE, ) if pose is None or not all(pose.visible(index) for index in required_points): return QualityResult.issue("full_body_missing") margin = QUALITY["edge_margin_pixels"] if pose.box[1] <= margin or pose.box[3] >= frame.height - 1 - margin: return QualityResult.issue("full_body_cut") if pose.height_fraction < QUALITY["full_body_min_height"]: return QualityResult.issue("full_body_too_small") if step.check_blur: region = frame.gray() if focus_box is None else signals.crop(frame.gray(), focus_box) if region is None: return QualityResult.issue("blurry") region = signals.downscale_to_width(region, 640) if signals.sharpness(region) < sharpness_minimum: return QualityResult.issue("blurry") return QualityResult(ok=True) class ExamBase(ABC): exam_id: str = "" title: str = "" description: str = "" condition_guide: str = "" condition_rules: list[ConditionRule] = [] steps: list[Step] = [] def __init__(self, patient: PatientContext) -> None: self.patient = patient self.samples: dict[str, list[dict[str, Any]]] = {step.id: [] for step in self.steps} self.valid_frames: dict[str, int] = {step.id: 0 for step in self.steps} self.state: dict[str, Any] = {} @classmethod def get_step(cls, step_id: str) -> Step: for step in cls.steps: if step.id == step_id: return step raise KeyError(step_id) @classmethod def summary(cls, include_steps: bool = True) -> dict[str, Any]: total = len(cls.steps) seconds = sum(step.min_seconds + step.get_ready_seconds + 8.0 for step in cls.steps) data: dict[str, Any] = { "exam_id": cls.exam_id, "title": cls.title, "description": cls.description, "total_steps": total, "estimated_minutes": round(seconds / 60.0, 1), } if include_steps: data["steps"] = [step.public(index + 1, total) for index, step in enumerate(cls.steps)] return data def accept(self, frame: Frame, step: Step) -> None: self.valid_frames[step.id] = self.valid_frames.get(step.id, 0) + 1 sample = self.measure(frame, step) if sample is None: return sample.setdefault("t", frame.timestamp_s) self.samples.setdefault(step.id, []).append(sample) @abstractmethod def measure(self, frame: Frame, step: Step) -> dict[str, Any] | None: raise NotImplementedError @abstractmethod def analyze(self, vision: Any) -> list[Finding]: raise NotImplementedError def step_data(self, step_id: str) -> list[dict[str, Any]]: return self.samples.get(step_id, []) def progress(self, step: Step) -> float: needed = max(1, step.min_valid_frames) return min(1.0, len(self.samples.get(step.id, [])) / needed) def is_step_complete(self, step: Step, elapsed_seconds: float) -> bool: return elapsed_seconds >= step.min_seconds and self.progress(step) >= 1.0 def repeat_reason(self, step: Step) -> str: return "I could not capture enough clear images for this step." def live_reading(self, step: Step) -> dict[str, Any] | None: return None def reset_step(self, step: Step) -> None: self.samples[step.id] = [] self.valid_frames[step.id] = 0 def max_repeats(self, step: Step) -> int: return 1 def begin_repeat(self, step: Step) -> None: return None @staticmethod def inconclusive(code: str, title: str, reason: str) -> Finding: return Finding(code=code, title=title, status="inconclusive", confidence=0.0, explanation=reason) class Session: def __init__(self, exam_id: str, patient: PatientContext, camera_mirrored: bool) -> None: self.session_id = uuid.uuid4().hex self.exam_id = exam_id self.patient = patient self.camera_mirrored = camera_mirrored self.state = "created" self.report: dict[str, Any] | None = None self.step_records: list[StepRecord] = [] self.connected = False self.created_at = time.time() self.updated_at = self.created_at def touch(self) -> None: self.updated_at = time.time() def status(self) -> dict[str, Any]: return { "session_id": self.session_id, "exam_id": self.exam_id, "state": self.state, "connected": self.connected, "created_at": self.created_at, "updated_at": self.updated_at, "report_available": self.report is not None, "completed_steps": sum(1 for record in self.step_records if record.status == "completed"), "steps": [record.model_dump() for record in self.step_records], } class SessionStore: def __init__(self) -> None: self._sessions: dict[str, Session] = {} def create(self, exam_id: str, patient: PatientContext, camera_mirrored: bool) -> Session: session = Session(exam_id, patient, camera_mirrored) self._sessions[session.session_id] = session return session def get(self, session_id: str) -> Session | None: return self._sessions.get(session_id) def delete(self, session_id: str) -> bool: return self._sessions.pop(session_id, None) is not None def cleanup_expired(self, ttl_seconds: float) -> int: now = time.time() expired = [ session_id for session_id, session in self._sessions.items() if not session.connected and now - session.updated_at > ttl_seconds ] for session_id in expired: self._sessions.pop(session_id, None) return len(expired) def count(self) -> int: return len(self._sessions) @dataclass class FramePayload: data: str | bytes timestamp_ms: float class ExamInterrupted(Exception): pass class ExamRunner: PROGRESS_INTERVAL_SECONDS = 0.5 QUALITY_FEEDBACK_INTERVAL_SECONDS = 3.0 NO_FRAME_TIMEOUT_SECONDS = 4.0 POSITION_WAIT_LIMIT_SECONDS = 90.0 READY_WAIT_LIMIT_SECONDS = 300.0 ADD_AREA_WAIT_SECONDS = 25.0 STEP_PAUSE_SECONDS = 1.5 def __init__(self, websocket: WebSocket, session: Session, exam: ExamBase, reasoner: Any) -> None: self.websocket = websocket self.session = session self.exam = exam self.reasoner = reasoner self.tracker = SessionTracker() self.frame_queue: asyncio.Queue[FramePayload] = asyncio.Queue(maxsize=6) self.control_queue: asyncio.Queue[str] = asyncio.Queue() self.current_step: Step | None = None self.collecting_step_id: str | None = None self.latest_quality: QualityResult | None = None self.last_frame_monotonic = 0.0 self.good_streak = 0 self.processed_times: deque[float] = deque(maxlen=120) self.stopped = False self.disconnected = asyncio.Event() self.send_lock = asyncio.Lock() self.loop: asyncio.AbstractEventLoop | None = None async def send(self, message_type: str, **fields: Any) -> None: if self.disconnected.is_set(): return payload = json_safe({"type": message_type, "session_id": self.session.session_id, "timestamp": time.time(), **fields}) async with self.send_lock: try: await self.websocket.send_json(payload) except Exception: self.disconnected.set() async def run(self) -> None: self.loop = asyncio.get_running_loop() self.session.connected = True self.session.touch() receiver = asyncio.create_task(self._receive_loop()) processor = asyncio.create_task(self._process_loop()) try: await self._run_examination() except ExamInterrupted: logger.info("Session %s interrupted", self.session.session_id) finally: self.collecting_step_id = None self.current_step = None receiver.cancel() processor.cancel() await asyncio.gather(receiver, processor, return_exceptions=True) await self.loop.run_in_executor(EXECUTOR, self.tracker.close) self.session.connected = False if self.session.state in ("running", "created"): self.session.state = "interrupted" self.session.touch() def _raise_if_disconnected(self) -> None: if self.disconnected.is_set(): raise ExamInterrupted() async def _sleep_alive(self, seconds: float) -> None: end = time.monotonic() + seconds while time.monotonic() < end: self._raise_if_disconnected() await asyncio.sleep(min(0.1, max(0.0, end - time.monotonic()))) async def _receive_loop(self) -> None: try: while True: message = await self.websocket.receive() if message.get("type") == "websocket.disconnect": break binary = message.get("bytes") if binary is not None: if len(binary) > SETTINGS.max_frame_bytes: await self.send("error", code="frame_too_large", display_text="The video frame is too large.") continue self._enqueue_frame(FramePayload(binary, time.monotonic() * 1000.0)) continue text = message.get("text") if text is not None: await self._handle_text_message(text) except Exception: logger.debug("Receive loop ended", exc_info=True) finally: self.disconnected.set() async def _handle_text_message(self, text: str) -> None: try: message = json.loads(text) except json.JSONDecodeError: await self.send("error", code="invalid_json", display_text="The message could not be read.") return if not isinstance(message, dict): await self.send("error", code="invalid_message", display_text="The message format is not valid.") return message_type = message.get("type") if message_type == "frame": data = message.get("image") if not isinstance(data, str) or not data: await self.send("error", code="invalid_frame", display_text="The video frame is missing.") return if len(data) > SETTINGS.max_frame_bytes * 4 // 3 + 64: await self.send("error", code="frame_too_large", display_text="The video frame is too large.") return timestamp = message.get("timestamp_ms") if isinstance(timestamp, (int, float)) and not isinstance(timestamp, bool) and math.isfinite(float(timestamp)): timestamp_ms = float(timestamp) else: timestamp_ms = time.monotonic() * 1000.0 self._enqueue_frame(FramePayload(data, timestamp_ms)) elif message_type == "control": action = str(message.get("action", "")).strip().lower() if action not in CONTROL_ACTIONS: await self.send("error", code="unknown_action", display_text=f"Unknown action: {action}") return await self.control_queue.put(action) elif message_type == "ping": await self.send("pong") else: await self.send("error", code="unknown_message_type", display_text="Unknown message type.") def _enqueue_frame(self, payload: FramePayload) -> None: if self.frame_queue.full(): try: self.frame_queue.get_nowait() except asyncio.QueueEmpty: pass try: self.frame_queue.put_nowait(payload) except asyncio.QueueFull: pass async def _process_loop(self) -> None: while True: payload = await self.frame_queue.get() try: frame = await self.loop.run_in_executor(EXECUTOR, self._process_frame_sync, payload) except asyncio.CancelledError: raise except Exception: logger.exception("Frame processing failed") continue if frame is None: continue now = time.monotonic() self.last_frame_monotonic = now self.processed_times.append(now) self.latest_quality = frame.quality if frame.quality is not None and frame.quality.ok: self.good_streak += 1 else: self.good_streak = 0 def _process_frame_sync(self, payload: FramePayload) -> Frame | None: image = signals.decode_image(payload.data) if image is None: return None if self.session.camera_mirrored: image = cv2.flip(image, 1) image = signals.resize_max_width(image, SETTINGS.max_frame_width) frame = Frame(image_bgr=image, timestamp_s=payload.timestamp_ms / 1000.0) step = self.current_step if step is None: frame.quality = QualityResult(ok=False, code="idle") return frame face, hands, pose = self.tracker.process(image, payload.timestamp_ms, set(step.trackers)) frame.face = face frame.hands = hands frame.pose = pose frame.quality = assess_quality(frame, step) if step.collect and self.collecting_step_id == step.id: tracked = (not step.trackers) or face is not None or bool(hands) or pose is not None if frame.quality.ok or (not step.strict_quality and tracked): try: self.exam.accept(frame, step) except Exception: logger.exception("Measurement failed in step %s", step.id) return frame def _drain_actions(self) -> list[str]: actions: list[str] = [] while True: try: actions.append(self.control_queue.get_nowait()) except asyncio.QueueEmpty: return actions def _clear_stale_actions(self) -> None: for action in self._drain_actions(): if action == "stop": self.stopped = True async def _wait_for_action(self, allowed: set[str], limit_seconds: float | None) -> str | None: deadline = None if limit_seconds is None else time.monotonic() + limit_seconds while True: self._raise_if_disconnected() try: action = await asyncio.wait_for(self.control_queue.get(), timeout=0.5) except asyncio.TimeoutError: if deadline is not None and time.monotonic() >= deadline: return None continue if action == "stop": self.stopped = True return "stop" if action in allowed: return action def _fps(self) -> float: now = time.monotonic() recent = [moment for moment in self.processed_times if now - moment <= 2.0] return round(len(recent) / 2.0, 1) async def _run_examination(self) -> None: steps = self.exam.steps total = len(steps) self.session.state = "running" self.session.step_records = [] summary = self.exam.summary(include_steps=False) await self.send( "session_ready", exam_id=self.exam.exam_id, exam_title=self.exam.title, total_steps=total, estimated_minutes=summary["estimated_minutes"], display_text=f"{self.exam.title}. Press start when you are ready.", speech_text=f"Welcome to the {self.exam.title}. I will guide you step by step. Press start when you are ready.", ) await self._wait_for_action({"start"}, None) if not self.stopped: for index, step in enumerate(steps): await self._run_step(index, step, total) if self.stopped: break if self.stopped: await self.send( "examination_stopped", display_text="The examination was stopped.", speech_text="The examination has been stopped.", ) await self._analyse() async def _run_step(self, index: int, step: Step, total: int) -> None: attempts = 0 repeat_count = 0 started = time.monotonic() reset_needed = True while True: attempts += 1 if reset_needed: self.exam.reset_step(step) reset_needed = True self._clear_stale_actions() if self.stopped: return self.current_step = step self.good_streak = 0 await self.send( "step_preview", step=step.public(index + 1, total), repeat_number=repeat_count, display_text=step.display_text, speech_text=step.speech_text, ) if step.requires_ready: allowed = {"ready", "skip"} if step.skippable else {"ready"} await self.send( "waiting_for_ready", step_id=step.id, display_text="Press ready when you are in position.", speech_text="Press ready when you are in position.", ) action = await self._wait_for_action(allowed, self.READY_WAIT_LIMIT_SECONDS) if action == "stop": return if action == "skip": await self._finish_step(step, index, total, "skipped", attempts, started, "Skipped by the patient.") return if action is None: await self._finish_step(step, index, total, "incomplete", attempts, started, "The patient did not confirm readiness.") return gate = await self._wait_for_position(step) if gate == "stop": return if gate == "skip": await self._finish_step(step, index, total, "skipped", attempts, started, "Skipped by the patient.") return if gate == "timeout": await self._finish_step(step, index, total, "incomplete", attempts, started, "The required camera view could not be achieved.") return await self._countdown(step) result = await self._collect(step, index, total) if result == "stop": return if result == "skip": await self._finish_step(step, index, total, "skipped", attempts, started, "Skipped by the patient.") return if result == "complete": if step.repeatable and repeat_count + 1 < self.exam.max_repeats(step): await self.send( "area_complete", step_id=step.id, areas_captured=repeat_count + 1, display_text="Done. Press add area to show another area, or next to continue.", speech_text="Done. To show another area, press add area. Otherwise, press next.", ) choice = await self._wait_for_action({"add_area", "next"}, self.ADD_AREA_WAIT_SECONDS) if choice == "stop": return if choice == "add_area": repeat_count += 1 attempts = 0 self.exam.begin_repeat(step) reset_needed = False continue await self._finish_step(step, index, total, "completed", attempts, started, "") return if attempts < 2: reason = self.exam.repeat_reason(step) await self.send( "step_repeat", step_id=step.id, reason=reason, display_text=f"{reason} Let's try this step once more.", speech_text=f"{reason} Let's try this step once more.", ) await self._sleep_alive(self.STEP_PAUSE_SECONDS) continue await self._finish_step(step, index, total, "incomplete", attempts, started, self.exam.repeat_reason(step)) return async def _wait_for_position(self, step: Step) -> str: started = time.monotonic() last_feedback = 0.0 while True: self._raise_if_disconnected() for action in self._drain_actions(): if action == "stop": self.stopped = True return "stop" if action == "skip" and step.skippable: return "skip" now = time.monotonic() if self.good_streak >= QUALITY["good_frames_required"]: await self.send( "position_ok", step_id=step.id, display_text="Good. Hold that position.", speech_text="Good. Hold that position.", ) return "ok" if now - started >= 1.5 and now - last_feedback >= self.QUALITY_FEEDBACK_INTERVAL_SECONDS: await self._send_quality_feedback(step) last_feedback = now if now - started >= self.POSITION_WAIT_LIMIT_SECONDS: return "timeout" await asyncio.sleep(0.1) async def _send_quality_feedback(self, step: Step) -> None: if time.monotonic() - self.last_frame_monotonic > self.NO_FRAME_TIMEOUT_SECONDS: code, text = "no_frames", QUALITY_TEXT["no_frames"] elif self.latest_quality is not None and not self.latest_quality.ok and self.latest_quality.code != "idle": code, text = self.latest_quality.code, self.latest_quality.message else: return await self.send("quality_feedback", step_id=step.id, code=code, display_text=text, speech_text=text) async def _countdown(self, step: Step) -> None: for remaining in range(step.get_ready_seconds, 0, -1): await self.send( "get_ready", step_id=step.id, seconds=remaining, display_text=str(remaining), speech_text=str(remaining), ) await self._sleep_alive(1.0) await self.send( "step_started", step_id=step.id, min_seconds=step.min_seconds, max_seconds=step.max_seconds, display_text="Start now.", speech_text="Start now.", ) def _combined_progress(self, step: Step, elapsed: float) -> float: data_progress = self.exam.progress(step) time_progress = 1.0 if step.min_seconds <= 0 else min(1.0, elapsed / step.min_seconds) return max(0.0, min(data_progress, time_progress)) async def _collect(self, step: Step, index: int, total: int) -> str: self.collecting_step_id = step.id start = time.monotonic() paused_at: float | None = None paused_total = 0.0 last_progress = 0.0 last_feedback = 0.0 bad_since: float | None = None try: while True: self._raise_if_disconnected() for action in self._drain_actions(): now = time.monotonic() if action == "stop": self.stopped = True return "stop" if action == "skip" and step.skippable: return "skip" if action == "pause" and paused_at is None: paused_at = now self.collecting_step_id = None await self.send("paused", step_id=step.id, display_text="Paused.", speech_text="Paused.") if action == "resume" and paused_at is not None: paused_total += now - paused_at paused_at = None self.collecting_step_id = step.id await self.send("resumed", step_id=step.id, display_text="Continuing.", speech_text="Continuing.") if paused_at is not None: await asyncio.sleep(0.2) continue now = time.monotonic() elapsed = now - start - paused_total if now - last_progress >= self.PROGRESS_INTERVAL_SECONDS: await self.send( "step_progress", step_id=step.id, step_number=index + 1, total_steps=total, progress_percent=int(round(100 * self._combined_progress(step, elapsed))), elapsed_seconds=round(elapsed, 1), seconds_remaining=round(max(0.0, step.max_seconds - elapsed), 1), quality_ok=bool(self.latest_quality is not None and self.latest_quality.ok), frames_per_second=self._fps(), live=self.exam.live_reading(step), ) last_progress = now no_frames = now - self.last_frame_monotonic > self.NO_FRAME_TIMEOUT_SECONDS quality_bad = self.latest_quality is not None and not self.latest_quality.ok and step.strict_quality if no_frames or quality_bad: if bad_since is None: bad_since = now if now - bad_since >= 1.0 and now - last_feedback >= self.QUALITY_FEEDBACK_INTERVAL_SECONDS: await self._send_quality_feedback(step) last_feedback = now else: bad_since = None if self.exam.is_step_complete(step, elapsed): return "complete" if elapsed >= step.max_seconds: return "timeout" await asyncio.sleep(0.1) finally: self.collecting_step_id = None async def _finish_step( self, step: Step, index: int, total: int, status: str, attempts: int, started: float, note: str, ) -> None: record = StepRecord( step_id=step.id, title=step.title, status=status, attempts=max(1, attempts), valid_frames=self.exam.valid_frames.get(step.id, 0), duration_seconds=round(time.monotonic() - started, 1), note=note, ) self.session.step_records.append(record) self.session.touch() next_step = self.exam.steps[index + 1] if index + 1 < total else None if status == "completed": message_type, text = "step_complete", "Well done." elif status == "skipped": message_type, text = "step_skipped", "Step skipped." else: message_type, text = "step_incomplete", "We could not complete this step. Let's move on." speech = f"{text} Next: {next_step.title}." if next_step is not None else text await self.send( message_type, step_id=step.id, status=status, note=note, next_step=next_step.public(index + 2, total) if next_step is not None else None, display_text=text, speech_text=speech, ) await self._sleep_alive(self.STEP_PAUSE_SECONDS) async def _analyse(self) -> None: completed = [record for record in self.session.step_records if record.status == "completed"] if not completed: self.session.state = "failed" await self.send( "examination_incomplete", display_text="No steps were completed, so results cannot be produced.", speech_text="No steps were completed, so I cannot produce results. Please try the examination again.", ) return if not self.stopped: await self.send( "examination_complete", display_text="Examination complete.", speech_text="Thank you. The examination is complete.", ) self.session.state = "analysing" await self.send( "analysis_started", display_text="Analysing your results. This may take up to a minute.", speech_text="Please wait while your results are analysed.", ) loop = self.loop def notify(stage: str) -> None: if loop is not None and not loop.is_closed(): asyncio.run_coroutine_threadsafe(self.send("analysis_progress", stage=stage, display_text=stage), loop) try: report = await loop.run_in_executor(EXECUTOR, self.reasoner.build_report, self.exam, self.session, notify) except Exception: logger.exception("Analysis failed for session %s", self.session.session_id) self.session.state = "failed" await self.send( "error", code="analysis_failed", display_text="The analysis could not be completed. Please try again.", speech_text="Sorry, the analysis could not be completed.", ) return report["stopped_early"] = self.stopped self.session.report = report self.session.state = "completed" self.session.touch() await self.send( "report_ready", report=report, display_text=report["interpretation"]["patient_message"], speech_text=report["speech_summary"], )