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