922 lines
33 KiB
Python
922 lines
33 KiB
Python
"""Domain models independent of OpenCV and Ultralytics."""
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BoundingBox:
|
|
x1: int
|
|
y1: int
|
|
x2: int
|
|
y2: int
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.x1 >= self.x2 or self.y1 >= self.y2:
|
|
raise ValueError("BoundingBox requires x1 < x2 and y1 < y2")
|
|
|
|
@property
|
|
def width(self) -> int:
|
|
return self.x2 - self.x1
|
|
|
|
@property
|
|
def height(self) -> int:
|
|
return self.y2 - self.y1
|
|
|
|
@property
|
|
def center_x(self) -> int:
|
|
return (self.x1 + self.x2) // 2
|
|
|
|
@property
|
|
def center_y(self) -> int:
|
|
return (self.y1 + self.y2) // 2
|
|
|
|
@property
|
|
def foot_x(self) -> int:
|
|
return self.center_x
|
|
|
|
@property
|
|
def foot_y(self) -> int:
|
|
return self.y2
|
|
|
|
@property
|
|
def area(self) -> int:
|
|
return self.width * self.height
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Detection:
|
|
class_id: int
|
|
class_name: str
|
|
confidence: float
|
|
bbox: BoundingBox
|
|
inside_evaluation_region: bool = False
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class FrameInfo:
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
width: int
|
|
height: int
|
|
measured_fps: float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PixelRegion:
|
|
x1: int
|
|
y1: int
|
|
x2: int
|
|
y2: int
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.x1 >= self.x2 or self.y1 >= self.y2:
|
|
raise ValueError("PixelRegion requires x1 < x2 and y1 < y2")
|
|
|
|
def contains(self, x: int, y: int) -> bool:
|
|
return self.x1 <= x <= self.x2 and self.y1 <= y <= self.y2
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PixelTriggerLine:
|
|
"""Trigger line in frame pixel coordinates."""
|
|
|
|
orientation: str
|
|
position: int
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.orientation not in {"vertical", "horizontal"}:
|
|
raise ValueError("orientation must be vertical or horizontal")
|
|
if self.position < 0:
|
|
raise ValueError("position must be non-negative")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TrackedPerson:
|
|
"""Library-independent snapshot of one tracker output."""
|
|
|
|
track_id: int
|
|
bbox: BoundingBox
|
|
confidence: float
|
|
class_id: int
|
|
class_name: str
|
|
center_x: float
|
|
center_y: float
|
|
foot_x: float
|
|
foot_y: float
|
|
inside_evaluation_region: bool
|
|
is_confirmed: bool
|
|
track_age_frames: int
|
|
missed_frames: int
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.track_id < 0 or self.track_age_frames < 0 or self.missed_frames < 0:
|
|
raise ValueError("track identifiers and frame counters must be non-negative")
|
|
if not 0.0 <= self.confidence <= 1.0:
|
|
raise ValueError("confidence must be in [0, 1]")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TrackPoint:
|
|
timestamp_monotonic: float
|
|
timestamp_iso: str
|
|
frame_number: int
|
|
center_x: float
|
|
center_y: float
|
|
foot_x: float
|
|
foot_y: float
|
|
bbox: BoundingBox
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.timestamp_monotonic < 0 or self.frame_number < 0:
|
|
raise ValueError("track point time and frame must be non-negative")
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TrackState:
|
|
"""Research state accumulated for one temporary tracking ID."""
|
|
|
|
track_id: int
|
|
first_seen_time: str
|
|
last_seen_time: str
|
|
first_seen_frame: int
|
|
last_seen_frame: int
|
|
entered_evaluation_region: bool = False
|
|
entry_time: str | None = None
|
|
entry_frame: int | None = None
|
|
entry_monotonic: float | None = None
|
|
exited_evaluation_region: bool = False
|
|
exit_time: str | None = None
|
|
exit_frame: int | None = None
|
|
exit_monotonic: float | None = None
|
|
trigger_crossed: bool = False
|
|
trigger_crossing_time: str | None = None
|
|
trigger_crossing_frame: int | None = None
|
|
trigger_crossing_direction: str | None = None
|
|
movement_direction: str = "unknown"
|
|
total_visible_frames: int = 0
|
|
total_missed_frames: int = 0
|
|
max_missed_frames: int = 0
|
|
trajectory: deque[TrackPoint] = field(default_factory=deque)
|
|
finalized: bool = False
|
|
is_confirmed: bool = False
|
|
first_seen_monotonic: float = 0.0
|
|
last_seen_monotonic: float = 0.0
|
|
current_inside_region: bool = False
|
|
pending_entry_frames: int = 0
|
|
pending_exit_frames: int = 0
|
|
valid_for_counting: bool = False
|
|
invalid_reason: str = "not_finalized"
|
|
|
|
@property
|
|
def duration_sec(self) -> float:
|
|
return max(0.0, self.last_seen_monotonic - self.first_seen_monotonic)
|
|
|
|
@property
|
|
def dwell_time_sec(self) -> float:
|
|
if self.entry_monotonic is None:
|
|
return 0.0
|
|
end = self.exit_monotonic if self.exit_monotonic is not None else self.last_seen_monotonic
|
|
return max(0.0, end - self.entry_monotonic)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TrackEvent:
|
|
event_type: str
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
x: float
|
|
y: float
|
|
direction: str | None = None
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
valid = {"track_created", "region_entered", "trigger_crossed", "region_exited", "track_lost", "track_finalized"}
|
|
if self.event_type not in valid:
|
|
raise ValueError(f"Unsupported track event: {self.event_type}")
|
|
if self.track_id < 0 or self.frame_number < 0:
|
|
raise ValueError("event track ID and frame must be non-negative")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class FaceROI:
|
|
"""Frame-coordinate face candidate region derived from a person box."""
|
|
|
|
track_id: int
|
|
x1: int
|
|
y1: int
|
|
x2: int
|
|
y2: int
|
|
source_bbox: BoundingBox
|
|
confidence: float
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.track_id < 0 or self.x1 >= self.x2 or self.y1 >= self.y2:
|
|
raise ValueError("FaceROI requires a non-negative track ID and valid coordinates")
|
|
if not 0.0 <= self.confidence <= 1.0:
|
|
raise ValueError("FaceROI confidence must be in [0, 1]")
|
|
|
|
@property
|
|
def width(self) -> int:
|
|
return self.x2 - self.x1
|
|
|
|
@property
|
|
def height(self) -> int:
|
|
return self.y2 - self.y1
|
|
|
|
@property
|
|
def center_x(self) -> float:
|
|
return (self.x1 + self.x2) / 2.0
|
|
|
|
@property
|
|
def center_y(self) -> float:
|
|
return (self.y1 + self.y2) / 2.0
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class FaceLandmark:
|
|
index: int
|
|
x: float
|
|
y: float
|
|
z: float
|
|
visibility: float | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.index < 0:
|
|
raise ValueError("landmark index must be non-negative")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HeadPoseEstimation:
|
|
"""Geometry-only result returned by HeadPoseEstimator."""
|
|
|
|
success: bool
|
|
pitch: float | None = None
|
|
yaw: float | None = None
|
|
roll: float | None = None
|
|
reprojection_error: float | None = None
|
|
failure_reason: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HeadPoseResult:
|
|
"""Head pose linked to a temporary tracking ID and frame."""
|
|
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
face_detected: bool
|
|
pose_estimated: bool
|
|
pitch: float | None
|
|
yaw: float | None
|
|
roll: float | None
|
|
landmarks_count: int
|
|
roi: FaceROI | None
|
|
confidence: float | None
|
|
failure_reason: str = ""
|
|
landmarks: tuple[FaceLandmark, ...] = field(default_factory=tuple, repr=False)
|
|
# Diagnostics for yaw-jitter analysis: pre-smoothing angles, PnP reprojection
|
|
# error [px], and outer-eye-corner distance [px] as a face-size indicator.
|
|
raw_pitch: float | None = None
|
|
raw_yaw: float | None = None
|
|
raw_roll: float | None = None
|
|
reprojection_error: float | None = None
|
|
interocular_px: float | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.track_id < 0 or self.frame_number < 0 or self.landmarks_count < 0:
|
|
raise ValueError("head pose identifiers and counts must be non-negative")
|
|
if self.pose_estimated and any(value is None for value in (self.pitch, self.yaw, self.roll)):
|
|
raise ValueError("successful pose estimation requires pitch, yaw, and roll")
|
|
if self.confidence is not None and not 0.0 <= self.confidence <= 1.0:
|
|
raise ValueError("face confidence must be in [0, 1]")
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FaceTrackState:
|
|
"""Per-track face statistics and Phase 4-ready recent history."""
|
|
|
|
track_id: int
|
|
total_frames: int = 0
|
|
face_detected_frames: int = 0
|
|
pose_estimated_frames: int = 0
|
|
last_pitch: float | None = None
|
|
last_yaw: float | None = None
|
|
last_roll: float | None = None
|
|
min_yaw: float | None = None
|
|
max_yaw: float | None = None
|
|
yaw_sum: float = 0.0
|
|
last_result: HeadPoseResult | None = None
|
|
history: deque[HeadPoseResult] = field(default_factory=deque, repr=False)
|
|
finalized: bool = False
|
|
|
|
@property
|
|
def mean_yaw(self) -> float | None:
|
|
return self.yaw_sum / self.pose_estimated_frames if self.pose_estimated_frames else None
|
|
|
|
@property
|
|
def face_detection_rate(self) -> float:
|
|
return self.face_detected_frames / self.total_frames if self.total_frames else 0.0
|
|
|
|
@property
|
|
def pose_estimation_rate(self) -> float:
|
|
return self.pose_estimated_frames / self.total_frames if self.total_frames else 0.0
|
|
|
|
|
|
class TurnLevel(str, Enum):
|
|
NONE = "none"
|
|
FACE_APPEARED = "face_appeared"
|
|
SUBTLE = "subtle"
|
|
WEAK = "weak"
|
|
MEDIUM = "medium"
|
|
STRONG = "strong"
|
|
NOT_EVALUABLE = "not_evaluable"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TurnFrameResult:
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
face_detected: bool
|
|
pose_estimated: bool
|
|
baseline_available: bool
|
|
baseline_yaw: float | None
|
|
current_yaw: float | None
|
|
yaw_delta_raw: float | None
|
|
yaw_delta_toward_signage: float | None
|
|
signage_direction: str
|
|
turn_candidate: bool
|
|
turn_confirmed: bool
|
|
turn_level: str
|
|
continuous_duration_sec: float
|
|
failure_reason: str = ""
|
|
bearing_deg: float | None = None
|
|
corrected_yaw: float | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.track_id < 0 or self.frame_number < 0 or self.continuous_duration_sec < 0:
|
|
raise ValueError("turn frame identifiers and duration must be non-negative")
|
|
if self.signage_direction not in {"positive", "negative"}:
|
|
raise ValueError("invalid signage direction")
|
|
if self.turn_level not in {level.value for level in TurnLevel}:
|
|
raise ValueError("invalid turn level")
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TurnTrackState:
|
|
track_id: int
|
|
total_frames: int = 0
|
|
pose_frames: int = 0
|
|
face_detected_frames: int = 0
|
|
baseline_yaw: float | None = None
|
|
baseline_frame: int | None = None
|
|
baseline_time: str | None = None
|
|
baseline_sample_count: int = 0
|
|
baseline_acquired: bool = False
|
|
first_face_detected_frame: int | None = None
|
|
first_face_detected_time: str | None = None
|
|
face_appeared_after_missing: bool = False
|
|
max_yaw_delta_toward_signage: float | None = None
|
|
max_yaw_delta_frame: int | None = None
|
|
max_yaw_delta_time: str | None = None
|
|
turn_started: bool = False
|
|
turn_start_frame: int | None = None
|
|
turn_start_time: str | None = None
|
|
turn_confirmed: bool = False
|
|
turn_confirmed_frame: int | None = None
|
|
turn_confirmed_time: str | None = None
|
|
turn_level: str = TurnLevel.NOT_EVALUABLE.value
|
|
turn_duration_sec: float = 0.0
|
|
evaluable: bool = False
|
|
valid_for_turn_analysis: bool = False
|
|
exclusion_reason: str = "not_finalized"
|
|
finalized: bool = False
|
|
yaw_history: deque[tuple[float, float]] = field(default_factory=deque, repr=False)
|
|
baseline_samples: deque[tuple[float, float, int, str]] = field(default_factory=deque, repr=False)
|
|
candidate_start_elapsed: float | None = None
|
|
last_pose_elapsed: float | None = None
|
|
missing_started_elapsed: float | None = None
|
|
strongest_level: str = TurnLevel.NONE.value
|
|
emitted_events: set[str] = field(default_factory=set, repr=False)
|
|
|
|
@property
|
|
def face_detection_rate(self) -> float:
|
|
return self.face_detected_frames / self.total_frames if self.total_frames else 0.0
|
|
|
|
@property
|
|
def pose_estimation_rate(self) -> float:
|
|
return self.pose_frames / self.total_frames if self.total_frames else 0.0
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TurnEvent:
|
|
event_type: str
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
yaw: float | None
|
|
yaw_delta_toward_signage: float | None
|
|
turn_level: str
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
valid = {"baseline_acquired", "face_appeared", "turn_candidate_started", "turn_confirmed",
|
|
"turn_level_changed", "turn_lost", "turn_track_finalized"}
|
|
if self.event_type not in valid or self.track_id < 0 or self.frame_number < 0:
|
|
raise ValueError("invalid turn event")
|
|
|
|
|
|
class VoicePromptMode(str, Enum):
|
|
DISABLED = "disabled"
|
|
PROMPT = "prompt"
|
|
CONTROL = "control"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class VoicePromptDecision:
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
mode: str
|
|
eligible: bool
|
|
should_prompt: bool
|
|
prompt_played: bool
|
|
pseudo_prompt: bool
|
|
reason: str
|
|
audio_file: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class VoicePromptEvent:
|
|
event_type: str
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
mode: str
|
|
prompt_played: bool = False
|
|
pseudo_prompt: bool = False
|
|
audio_file: str | None = None
|
|
turn_confirmed: bool = False
|
|
turn_level: str | None = None
|
|
reaction_time_sec: float | None = None
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class VoicePromptTrackState:
|
|
track_id: int
|
|
mode: str
|
|
eligible: bool = False
|
|
eligibility_frame: int | None = None
|
|
eligibility_time: str | None = None
|
|
prompt_attempted: bool = False
|
|
prompt_played: bool = False
|
|
prompt_failed: bool = False
|
|
pseudo_prompt: bool = False
|
|
prompt_frame: int | None = None
|
|
prompt_time: str | None = None
|
|
prompt_elapsed_time_sec: float | None = None
|
|
audio_file: str | None = None
|
|
response_window_started: bool = False
|
|
response_window_start_frame: int | None = None
|
|
response_window_start_time: str | None = None
|
|
response_window_end_time_sec: float | None = None
|
|
response_detected: bool = False
|
|
response_frame: int | None = None
|
|
response_time: str | None = None
|
|
reaction_time_sec: float | None = None
|
|
response_turn_level: str | None = None
|
|
max_yaw_delta_after_prompt: float | None = None
|
|
final_turn_level: str | None = None
|
|
valid_for_voice_analysis: bool = False
|
|
exclusion_reason: str = "not_finalized"
|
|
finalized: bool = False
|
|
response_window_expired: bool = False
|
|
|
|
|
|
class PersonExperimentStatus(str, Enum):
|
|
NEW = "new"
|
|
TRACKED = "tracked"
|
|
INSIDE_REGION = "inside_region"
|
|
ELIGIBLE = "eligible"
|
|
PROMPTED = "prompted"
|
|
PSEUDO_PROMPTED = "pseudo_prompted"
|
|
OBSERVING_RESPONSE = "observing_response"
|
|
RESPONDED = "responded"
|
|
NO_RESPONSE = "no_response"
|
|
COMPLETED = "completed"
|
|
EXCLUDED = "excluded"
|
|
NOT_EVALUABLE = "not_evaluable"
|
|
LOST = "lost"
|
|
|
|
|
|
class PersonExclusionReason(str, Enum):
|
|
NONE = "none"
|
|
TRACKER_DISABLED = "tracker_disabled"
|
|
VOICE_DISABLED = "voice_disabled"
|
|
TURN_DISABLED = "turn_disabled"
|
|
NO_POSE_ESTIMATED = "no_pose_estimated"
|
|
NO_FACE_DETECTED = "no_face_detected"
|
|
BASELINE_NOT_ACQUIRED = "baseline_not_acquired"
|
|
INSUFFICIENT_POSE_SAMPLES = "insufficient_pose_samples"
|
|
LOW_FACE_DETECTION_RATE = "low_face_detection_rate"
|
|
LOW_POSE_ESTIMATION_RATE = "low_pose_estimation_rate"
|
|
NOT_PROMPT_ELIGIBLE = "not_prompt_eligible"
|
|
ALREADY_TURNED_BEFORE_PROMPT = "already_turned_before_prompt"
|
|
PROMPT_FAILED = "prompt_failed"
|
|
AUDIO_MISSING = "audio_missing"
|
|
RESPONSE_NOT_EVALUABLE = "response_not_evaluable"
|
|
TRACK_TOO_SHORT = "track_too_short"
|
|
ID_SWITCH_SUSPECTED = "id_switch_suspected"
|
|
TRACK_LOST_BEFORE_PROMPT = "track_lost_before_prompt"
|
|
TRACK_LOST_DURING_RESPONSE_WINDOW = "track_lost_during_response_window"
|
|
OUTSIDE_REGION = "outside_region"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PersonStateSnapshot:
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
status: str
|
|
previous_status: str | None = None
|
|
inside_region: bool = False
|
|
track_confirmed: bool = False
|
|
track_lost: bool = False
|
|
trigger_crossed: bool = False
|
|
movement_direction: str | None = None
|
|
face_detected: bool | None = None
|
|
pose_estimated: bool | None = None
|
|
turn_level: str | None = None
|
|
turn_confirmed: bool = False
|
|
voice_mode: str | None = None
|
|
voice_eligible: bool = False
|
|
prompt_played: bool = False
|
|
pseudo_prompt: bool = False
|
|
response_window_started: bool = False
|
|
response_detected: bool = False
|
|
reaction_time_sec: float | None = None
|
|
valid_for_turn_analysis: bool | None = None
|
|
valid_for_voice_analysis: bool | None = None
|
|
exclusion_reason: str = PersonExclusionReason.NONE.value
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PersonStateEvent:
|
|
event_type: str
|
|
track_id: int
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
previous_status: str | None
|
|
new_status: str
|
|
reason: str = ""
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PersonStateSummary:
|
|
experiment_id: str
|
|
track_id: int
|
|
first_frame: int | None = None
|
|
last_frame: int | None = None
|
|
first_seen_time: str | None = None
|
|
last_seen_time: str | None = None
|
|
duration_sec: float | None = None
|
|
final_status: str = PersonExperimentStatus.COMPLETED.value
|
|
completed: bool = False
|
|
excluded: bool = False
|
|
exclusion_reason: str = PersonExclusionReason.NONE.value
|
|
inside_region_ever: bool = False
|
|
trigger_crossed: bool = False
|
|
trigger_crossing_direction: str | None = None
|
|
movement_direction: str | None = None
|
|
face_detected_ever: bool = False
|
|
pose_estimated_ever: bool = False
|
|
face_detection_rate: float | None = None
|
|
pose_estimation_rate: float | None = None
|
|
baseline_acquired: bool | None = None
|
|
turn_confirmed: bool = False
|
|
turn_level: str | None = None
|
|
max_yaw_delta_toward_signage: float | None = None
|
|
voice_mode: str | None = None
|
|
voice_eligible: bool = False
|
|
prompt_played: bool = False
|
|
pseudo_prompt: bool = False
|
|
prompt_time: str | None = None
|
|
prompt_elapsed_time_sec: float | None = None
|
|
response_window_started: bool = False
|
|
response_detected: bool = False
|
|
reaction_time_sec: float | None = None
|
|
response_turn_level: str | None = None
|
|
valid_for_turn_analysis: bool | None = None
|
|
valid_for_voice_analysis: bool | None = None
|
|
notes: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class UnifiedEvent:
|
|
experiment_id: str
|
|
session_id: str
|
|
event_id: str
|
|
source: str
|
|
event_type: str
|
|
track_id: int | None
|
|
frame_number: int
|
|
timestamp_iso: str
|
|
elapsed_time_sec: float
|
|
person_status: str | None = None
|
|
voice_mode: str | None = None
|
|
prompt_played: bool | None = None
|
|
pseudo_prompt: bool | None = None
|
|
turn_confirmed: bool | None = None
|
|
turn_level: str | None = None
|
|
response_detected: bool | None = None
|
|
reaction_time_sec: float | None = None
|
|
exclusion_reason: str | None = None
|
|
severity: str = "info"
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.source not in {"tracking", "face", "turn", "voice", "person_state", "session", "system"}:
|
|
raise ValueError("invalid unified event source")
|
|
if self.severity not in {"info", "warning", "error"}:
|
|
raise ValueError("invalid unified event severity")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PersonAnalysisRecord:
|
|
experiment_id: str
|
|
session_id: str
|
|
track_id: int
|
|
voice_mode: str | None = None
|
|
prompt_condition: str = "unknown"
|
|
prompt_played: bool = False
|
|
pseudo_prompt: bool = False
|
|
prompt_time: str | None = None
|
|
prompt_elapsed_time_sec: float | None = None
|
|
response_window_started: bool = False
|
|
response_detected: bool = False
|
|
reaction_time_sec: float | None = None
|
|
response_turn_level: str | None = None
|
|
turn_confirmed: bool = False
|
|
turn_level: str | None = None
|
|
max_yaw_delta_toward_signage: float | None = None
|
|
face_detected_ever: bool = False
|
|
pose_estimated_ever: bool = False
|
|
face_detection_rate: float | None = None
|
|
pose_estimation_rate: float | None = None
|
|
valid_for_turn_analysis: bool | None = None
|
|
valid_for_voice_analysis: bool | None = None
|
|
final_status: str = "completed"
|
|
completed: bool = False
|
|
excluded: bool = False
|
|
exclusion_reason: str = "none"
|
|
track_duration_sec: float | None = None
|
|
trigger_crossed: bool = False
|
|
trigger_crossing_direction: str | None = None
|
|
movement_direction: str | None = None
|
|
camera_position_note: str | None = None
|
|
notes: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SessionSummary:
|
|
experiment_id: str
|
|
session_id: str
|
|
started_at: str
|
|
ended_at: str | None = None
|
|
duration_sec: float | None = None
|
|
voice_mode: str | None = None
|
|
total_tracks: int = 0
|
|
completed_tracks: int = 0
|
|
excluded_tracks: int = 0
|
|
evaluable_turn_tracks: int = 0
|
|
valid_voice_analysis_tracks: int = 0
|
|
prompt_played_count: int = 0
|
|
pseudo_prompt_count: int = 0
|
|
response_detected_count: int = 0
|
|
no_response_count: int = 0
|
|
not_evaluable_count: int = 0
|
|
response_rate: float | None = None
|
|
mean_reaction_time_sec: float | None = None
|
|
median_reaction_time_sec: float | None = None
|
|
weak_or_higher_count: int = 0
|
|
subtle_count: int = 0
|
|
medium_or_higher_count: int = 0
|
|
strong_count: int = 0
|
|
average_fps: float | None = None
|
|
camera_position_note: str | None = None
|
|
warnings_count: int = 0
|
|
errors_count: int = 0
|
|
notes: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LogConsistencyIssue:
|
|
issue_type: str
|
|
severity: str
|
|
track_id: int | None
|
|
frame_number: int | None
|
|
message: str
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AnalysisConditionSummary:
|
|
condition: str; total_records: int; valid_voice_analysis_count: int; response_detected_count: int
|
|
no_response_count: int; response_rate: float | None; excluded_count: int; not_evaluable_count: int
|
|
mean_reaction_time_sec: float | None; median_reaction_time_sec: float | None; std_reaction_time_sec: float | None
|
|
mean_max_yaw_delta: float | None; median_max_yaw_delta: float | None; std_max_yaw_delta: float | None
|
|
subtle_count: int; weak_count: int; medium_count: int; strong_count: int
|
|
weak_or_higher_count: int; medium_or_higher_count: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ResponseRateTestResult:
|
|
test_name: str; prompt_success: int; prompt_total: int; control_success: int; control_total: int
|
|
prompt_rate: float | None; control_rate: float | None; rate_difference: float | None
|
|
statistic: float | None; p_value: float | None; effect_size_name: str; effect_size: float | None
|
|
odds_ratio: float | None; method_note: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class NumericComparisonResult:
|
|
metric_name: str; test_name: str; prompt_count: int; control_count: int
|
|
prompt_mean: float | None; control_mean: float | None; prompt_median: float | None; control_median: float | None
|
|
mean_difference: float | None; statistic: float | None; p_value: float | None
|
|
effect_size_name: str; effect_size: float | None; method_note: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TurnLevelDistributionResult:
|
|
condition: str; none_count: int; face_appeared_count: int; subtle_count: int; weak_count: int
|
|
medium_count: int; strong_count: int; not_evaluable_count: int; total_count: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExclusionSummary:
|
|
condition: str; exclusion_reason: str; count: int; ratio: float | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AnalysisRunSummary:
|
|
analysis_id: str; input_files_count: int; total_records: int; prompt_records: int; control_records: int
|
|
valid_prompt_records: int; valid_control_records: int; generated_at: str; output_directory: str; notes: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AnalysisResult:
|
|
condition_summaries: list[AnalysisConditionSummary]
|
|
response_rate_test: ResponseRateTestResult
|
|
numeric_comparisons: list[NumericComparisonResult]
|
|
turn_level_distributions: list[TurnLevelDistributionResult]
|
|
exclusion_summaries: list[ExclusionSummary]
|
|
run_summary: AnalysisRunSummary
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RuntimeMetrics:
|
|
"""Small, GUI-safe snapshot of the current experiment state."""
|
|
|
|
frame_number: int = 0
|
|
elapsed_time_sec: float = 0.0
|
|
fps: float | None = None
|
|
active_tracks: int = 0
|
|
total_tracks: int = 0
|
|
region_visitors: int = 0
|
|
trigger_crossings: int = 0
|
|
face_detected_tracks: int = 0
|
|
turn_confirmed_count: int = 0
|
|
voice_mode: str | None = None
|
|
prompt_played_count: int = 0
|
|
pseudo_prompt_count: int = 0
|
|
response_detected_count: int = 0
|
|
person_state_counts: dict[str, int] = field(default_factory=dict)
|
|
current_log_directory: str | None = None
|
|
latest_message: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExperimentProtocol:
|
|
protocol_id: str; title: str; version: str; created_at: str; purpose: str
|
|
target_location: str; signage_description: str; camera_position_description: str
|
|
camera_height_cm: float | None; camera_angle_description: str; walking_direction_description: str
|
|
lighting_condition: str; voice_prompt_text: str; voice_file_path: str; voice_volume: float
|
|
trigger_strategy: str; trigger_line_description: str; response_window_sec: float
|
|
turn_weak_threshold_deg: float; face_process_every_n_frames: int
|
|
prompt_condition_rule: str; control_condition_rule: str; privacy_note: str; notes: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExperimentChecklistItem:
|
|
item_id: str; category: str; description: str; required: bool = True; checked: bool = False; note: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExperimentSessionNote:
|
|
session_note_id: str; session_id: str | None; experiment_id: str; date: str; condition: str
|
|
operator: str; location: str; camera_position_note: str; camera_height_cm: float | None
|
|
camera_angle_note: str; signage_position_note: str; walking_direction_note: str; lighting_note: str
|
|
audio_file: str | None; voice_prompt_text: str; voice_volume: float | None; config_path: str
|
|
started_at: str | None; ended_at: str | None; planned_duration_min: float | None
|
|
actual_duration_sec: float | None; issues: str; notes: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LogQualityIssue:
|
|
issue_type: str; severity: str; file_path: str | None; session_id: str | None
|
|
track_id: int | None; message: str; suggested_action: str; metadata: dict[str, Any] = field(default_factory=dict)
|
|
def __post_init__(self) -> None:
|
|
if self.severity not in {"info", "warning", "error"}: raise ValueError("invalid quality issue severity")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PilotSessionReport:
|
|
report_id: str; generated_at: str; input_directory: str; output_directory: str
|
|
total_sessions: int; prompt_sessions: int; control_sessions: int; total_tracks: int
|
|
valid_voice_analysis_tracks: int; valid_voice_analysis_rate: float | None
|
|
response_rate_prompt: float | None; response_rate_control: float | None; rate_difference: float | None
|
|
mean_reaction_time_prompt: float | None; mean_reaction_time_control: float | None
|
|
high_level_findings: str; quality_notes: str; recommended_next_actions: str
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PilotDiagnosticMetric:
|
|
metric_name:str;value:float|int|str|None;threshold:float|int|str|None;passed:bool|None;severity:str;message:str;suggested_action:str
|
|
@dataclass(frozen=True, slots=True)
|
|
class ConditionBalanceDiagnostic:
|
|
prompt_total_records:int;control_total_records:int;prompt_valid_records:int;control_valid_records:int
|
|
prompt_response_rate:float|None;control_response_rate:float|None;valid_count_difference:int
|
|
valid_count_ratio:float|None;balanced:bool;message:str;suggested_action:str
|
|
@dataclass(frozen=True, slots=True)
|
|
class SessionQualitySummary:
|
|
session_id:str;condition:str;total_tracks:int;valid_voice_analysis_tracks:int;valid_voice_analysis_rate:float|None
|
|
response_detected_count:int;response_rate:float|None;not_evaluable_count:int;not_evaluable_rate:float|None
|
|
no_pose_estimated_count:int;no_pose_estimated_rate:float|None;mean_reaction_time_sec:float|None
|
|
median_reaction_time_sec:float|None;camera_position_note:str|None;quality_level:str;main_issue:str;suggested_action:str
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImprovementRecommendation:
|
|
recommendation_id:str;category:str;priority:str;title:str;reason:str;suggested_action:str
|
|
related_metrics:dict[str,Any]=field(default_factory=dict);expected_effect:str=""
|
|
@dataclass(frozen=True, slots=True)
|
|
class PilotDiagnosticReport:
|
|
report_id:str;generated_at:str;input_analysis_dir:str;input_logs_dir:str|None;overall_status:str
|
|
ready_for_main_experiment:bool;total_records:int;prompt_valid_records:int;control_valid_records:int
|
|
prompt_response_rate:float|None;control_response_rate:float|None;valid_voice_analysis_rate:float|None
|
|
not_evaluable_rate:float|None;no_pose_estimated_rate:float|None;condition_balance:ConditionBalanceDiagnostic
|
|
session_summaries:list[SessionQualitySummary];metrics:list[PilotDiagnosticMetric]
|
|
recommendations:list[ImprovementRecommendation];notes:str
|
|
@dataclass(frozen=True, slots=True)
|
|
class CalibrationChangeRecord:
|
|
change_id:str;created_at:str;operator:str;category:str;before_value:str;after_value:str;reason:str
|
|
related_report_id:str|None=None;notes:str=""
|
|
@dataclass(frozen=True,slots=True)
|
|
class MainExperimentPlan:
|
|
plan_id:str;title:str;created_at:str;target_prompt_valid_records:int;target_control_valid_records:int
|
|
minimum_prompt_valid_records:int;minimum_control_valid_records:int;planned_prompt_sessions:int;planned_control_sessions:int
|
|
condition_order_rule:str;camera_position_policy:str;voice_prompt_text:str;voice_volume:float
|
|
response_window_sec:float;turn_weak_threshold_deg:float;notes:str
|
|
@dataclass(frozen=True,slots=True)
|
|
class MainExperimentProgress:
|
|
plan_id:str;generated_at:str;prompt_total_records:int;control_total_records:int;prompt_valid_records:int
|
|
control_valid_records:int;prompt_sessions:int;control_sessions:int;prompt_completion_rate:float|None
|
|
control_completion_rate:float|None;total_valid_records:int;target_reached:bool;ready_for_final_analysis:bool
|
|
main_issue:str;suggested_action:str
|
|
@dataclass(frozen=True,slots=True)
|
|
class ExperimentSnapshot:
|
|
snapshot_id:str;created_at:str;config_path:str;merged_config_path:str|None;application_version:str
|
|
git_commit:str|None;camera_position_note:str;voice_prompt_text:str;voice_volume:float|None
|
|
trigger_strategy:str;response_window_sec:float|None;turn_weak_threshold_deg:float|None
|
|
face_process_every_n_frames:int|None;signage_yaw_direction:str|None;notes:str
|
|
@dataclass(frozen=True,slots=True)
|
|
class FinalAnalysisWorkflowResult:
|
|
workflow_id:str;generated_at:str;input_logs_dir:str;analysis_output_dir:str;final_output_dir:str
|
|
analysis_success:bool;quality_check_success:bool;diagnostics_success:bool;final_report_success:bool
|
|
presentation_assets_success:bool;output_files:list[str];warnings:list[str];errors:list[str]
|
|
@dataclass(frozen=True,slots=True)
|
|
class FinalExperimentReport:
|
|
report_id:str;generated_at:str;experiment_title:str;total_sessions:int;prompt_sessions:int;control_sessions:int
|
|
prompt_valid_records:int;control_valid_records:int;prompt_response_rate:float|None;control_response_rate:float|None
|
|
rate_difference:float|None;response_rate_p_value:float|None;response_rate_test_name:str|None
|
|
response_rate_effect_size:float|None;odds_ratio:float|None;mean_reaction_time_prompt:float|None
|
|
mean_reaction_time_control:float|None;reaction_time_p_value:float|None;mean_yaw_delta_prompt:float|None
|
|
mean_yaw_delta_control:float|None;yaw_delta_p_value:float|None;top_exclusion_reasons:list[str]
|
|
main_quality_notes:str;limitations:str;conclusion_for_presentation:str
|
|
@dataclass(frozen=True,slots=True)
|
|
class PresentationAsset:
|
|
asset_id:str;asset_type:str;title:str;file_path:str;description:str;suggested_slide:str
|