Blame for neural_forge/engine.py

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"],
        )