""" DrSimon Engine Python SDK Sample Client Usage: client = DrSimonClient(api_key="drs_your_api_key_here") result = client.ingest("uuid-here", "edu", [ {"time": "2026-03-04T10:30:00.000Z", "type": "session_duration", "value": 45.0, "tier": 1}, ]) Requirements: pip install httpx """ from __future__ import annotations from datetime import datetime, timezone import hashlib import json import math import re import time from typing import Any, NoReturn, cast from urllib.parse import urlsplit import httpx JsonObject = dict[str, Any] SERVER_OWNED_INGEST_FIELDS = ("factor_scores", "bn_factor_scores") RAW_MEDIA_INGEST_PRIMITIVE_FIELDS = ( "raw_media", "raw_payload", "raw_sensor_payload", "media_bytes", "video_bytes", "frame_bytes", "image_bytes", "image_base64", "audio_bytes", "screenshot_bytes", "camera_frame", "screen_capture", ) REPORT_FACTOR_QUALITY_REASON = "factor_signal_coverage_limited" GROWTH_FEEDBACK_ALLOWED_KINDS = frozenset( { "weekly_factor_growth_feedback", "weekly_test_behavior_feedback", } ) GROWTH_FEEDBACK_REQUIRED_EVIDENCE_FLAGS = ( "context_applied", "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", ) RECOMMENDED_GUIDE_ALLOWED_REASONS = frozenset( { "하락 추세", "정체", "성장 영역", } ) RECOMMENDED_GUIDE_REQUIRED_EVIDENCE_FLAGS = ( "context_applied", "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", ) REPORT_SUMMARY_REQUIRED_EVIDENCE_FLAGS = ( "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", ) REPORT_COGNITIVE_CONSTRUCT_EVIDENCE_SUMMARY_REQUIRED_FIELDS = frozenset( { "cognitive_construct_evidence_summary_v", "evidence_policy", "summary_kind", "cognitive_construct_policy", "activity_contexts", "assessment_modalities", "context_status", "observed_source_primitives", "observed_construct_count", "constructs", "weighting_policy", "product_boundaries", "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", } ) REPORT_COGNITIVE_CONSTRUCT_EVIDENCE_SUMMARY_REQUIRED_FLAGS = ( "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", ) PROTOCOL_VERSION = 1 REDACTED_DETAIL_TOKEN = "[redacted]" COLLECTION_PLAN_SENSOR_GROUPS = frozenset( { "timing", "keyboard", "app_interaction", "security_events", "camera_visual", "paper_exam_cv", "touch", "stylus", "clinical_audio", } ) COLLECTION_PLAN_SENSOR_GROUP_MIN_TIERS: dict[str, int] = { "timing": 0, "keyboard": 0, "app_interaction": 0, "security_events": 0, "camera_visual": 1, "paper_exam_cv": 2, "touch": 2, "stylus": 2, "clinical_audio": 3, } COLLECTION_PLAN_STRING_LIST_MAX_ITEMS = 64 COLLECTION_PLAN_PUBLIC_TOKEN_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") SCENE_DEF_VERSION_RE = re.compile(r"^[a-z][a-z0-9-]{0,63}$") SCENE_TRANSITION_PUBLIC_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") SCENE_TRANSITION_PRODUCT_RE = re.compile(r"^[a-z][a-z0-9_]{0,19}$") SCENE_TRANSITION_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,255}$") SCENE_TRANSITION_SCENE_TOKEN_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$") SCENE_TRANSITION_ISO_UTC_MS_RE = re.compile( r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{1,3}Z$" ) SCENE_TRANSITION_SOURCES = frozenset( {"teacher", "system", "backend", "reconciliation"} ) SCENE_TRANSITION_SCENE_KEYS = frozenset({"group", "sub"}) SCENE_TRANSITION_PRIVATE_ID_PREFIXES = ( "aws-", "aws_", "database-", "database_", "db-", "db_", "drsimon-", "drsimon_", "jwt-", "jwt_", "nextauth-", "nextauth_", "openai-", "openai_", "secret-", "secret_", ) SCENE_TRANSITION_PRIVATE_SCENE_TOKEN_PREFIXES = ( "AWS_", "DATABASE_", "DB_", "DRSIMON_", "JWT_", "NEXTAUTH_", "OPENAI_", "SECRET_", ) REPORT_REQUEST_PUBLIC_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") REPORT_REQUEST_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") REPORT_REQUEST_JURISDICTION_RE = re.compile(r"^[a-z][a-z0-9_]{1,15}$") REPORT_REQUEST_AUDIENCES = frozenset({"student", "teacher", "parent"}) REPORT_REQUEST_BOOLEAN_FIELDS = frozenset( { "guardian_consent", "cross_product_consent", "special_category_basis", "dpia_completed", "ferpa_compliant_agreement", } ) REPORT_FOCUSPANG_WEEKLY_OPTION_FIELDS = frozenset( { "tenant_id", "jurisdiction", "audience", "guardian_consent", "cross_product_consent", "special_category_basis", "dpia_completed", "ferpa_compliant_agreement", "policy_assertion_id", } ) REPORT_MOMENT_DAILY_OPTION_FIELDS = frozenset( { "tenant_id", "jurisdiction", "guardian_consent", "cross_product_consent", "special_category_basis", "dpia_completed", "policy_assertion_id", } ) QUIZPANG_RESPONSE_TIME_PRIMITIVE_TYPE = "response_time_ms" QUIZPANG_RESPONSE_TIME_MIN_MS = 100.0 QUIZPANG_RESPONSE_TIME_MAX_MS = 30_000.0 TASK_LIFECYCLE_STARTED_PRIMITIVE_TYPE = "task_started_count" TASK_LIFECYCLE_ABANDONED_PRIMITIVE_TYPE = "task_abandoned_count" TASK_LIFECYCLE_MIN_STARTED_COUNT = 5 TASK_LIFECYCLE_MAX_COUNT = 1_000 INTERACTION_TAXONOMY_SOCIAL_PRIMITIVE_TYPE = "social_interaction_count" INTERACTION_TAXONOMY_TOTAL_PRIMITIVE_TYPE = "total_interaction_count" INTERACTION_TAXONOMY_MIN_TOTAL_COUNT = 20 INTERACTION_TAXONOMY_MAX_SOCIAL_COUNT = 1_000 INTERACTION_TAXONOMY_MAX_TOTAL_COUNT = 10_000 ACTIVITY_COUNT_PRIMITIVE_TYPE = "activity_count" ACTIVITY_COUNT_MIN_COUNT = 0 ACTIVITY_COUNT_MAX_COUNT = 10_000 SEGMENT_SCENE_CONTEXT_POLICY = "max_observed_scene_overlap" SEGMENT_SCENE_CONTEXT_SOURCES = frozenset({"attention_scene_timeline", "none"}) SEGMENT_SCENE_CONTEXT_STATUSES = frozenset({"missing", "partial", "resolved"}) SEGMENT_SCENE_CONTEXT_SCENE_SOURCES = frozenset( {"teacher", "system", "backend", "reconciliation"} ) SEGMENT_SCENE_CONTEXT_CAVEATS = frozenset( { "scene_context_not_scoring_input", "scene_timeline_missing", "scene_context_partial", "scene_overlap_partial", "segment_duration_missing", } ) SEGMENT_SCENE_CONTEXT_FORBIDDEN_KEYS = frozenset( {"actorId", "actor_id", "setBy", "set_by", "teacherId", "teacher_id", "userId", "user_id"} ) STUDENT_GROWTH_DOMAINS = ( "attention", "emotion", "achievement", "interest", ) SUPPORTED_BEHAVIOR_DIMENSIONS = ( "problem_solving_rhythm", "pause_pattern", "response_impulsivity", "focus_continuity", "gaze_shift", "posture_shift", "hand_activity_shift", "screen_or_app_exit", "object_context_change", "environment_disruption", "repetitive_interaction", ) COGNITIVE_CONSTRUCT_POLICY = "cognitive_construct_evidence_v1" SUPPORTED_COGNITIVE_CONSTRUCTS = ( "attention_control", "sustained_attention", "working_memory", "cognitive_control", "processing_speed", "auditory_processing", "visual_attention", "emotion_regulation", ) COGNITIVE_CONSTRUCT_EVIDENCE_MODES = ( "direct_task_evidence", "behavioral_proxy", "insufficient_evidence", ) COGNITIVE_CONSTRUCT_CONFIDENCE_CAPS = ( "insufficient", "low", "moderate", "high", ) COGNITIVE_CONSTRUCT_WEIGHTING_POLICY = ( "renormalize_observed_evidence_with_confidence_cap" ) COGNITIVE_CONSTRUCT_PRODUCT_BOUNDARIES = ( "not_clinical_diagnosis", "not_raw_sensor_inference", "task_context_required_for_task_constructs", "missing_sensor_not_zero_score", "confidence_reflects_observed_evidence_only", ) COGNITIVE_CONSTRUCT_WARNING_CODES = ( "no_observed_construct_evidence", "missing_sensor_not_zero_score", "task_context_required_for_construct", "clinical_audio_requires_research_context", ) COGNITIVE_CONSTRUCT_SCIENCE_REFERENCES = ( "rdoc_cognitive_systems", "response_time_variability_review", "digital_phenotyping_review", "auditory_processing_working_memory_review", ) COGNITIVE_CONSTRUCT_SOURCE_PRIMITIVES: dict[str, tuple[str, ...]] = { "attention_control": ( "focus_ratio", "proctor_focus_ratio", "app_switch_count", "interaction_interval_ms", "gaze_entropy", ), "sustained_attention": ( "sustained_attention_decay_slope", "focus_ratio", "fixation_duration", "blink_rate", "response_time_variance", ), "working_memory": ( "response_time_ms", "response_time_variance", "input_error_rate", "task_abandoned_count", "interaction_interval_ms", ), "cognitive_control": ( "premature_response_rate", "proctor_security_copy", "proctor_security_paste", "app_switch_count", "window_focus_lost_count", "input_error_rate", ), "processing_speed": ( "response_time_ms", "response_latency_drift", "interaction_interval_ms", "input_speed", ), "auditory_processing": ( "voice_activity", "voice_speaking_rate", "voice_pause_distribution_entropy", "voice_prosody_pitch_variance", ), "visual_attention": ( "gaze_entropy", "gaze_deviation", "fixation_duration", "head_pose_deviation", "proctoring_face_detected", ), "emotion_regulation": ( "task_avoidance_score", "posture_drift_score", "head_pose_micro_movement_rate", "keystroke_pressure_variance", "voice_prosody_pitch_variance", ), } COGNITIVE_CONSTRUCT_SOURCE_PRIMITIVE_SET = frozenset( primitive for source_primitives in COGNITIVE_CONSTRUCT_SOURCE_PRIMITIVES.values() for primitive in source_primitives ) COGNITIVE_CONTEXTUAL_EVIDENCE_FIELDS = frozenset( { "cognitive_construct_policy", "supported_cognitive_constructs", "cognitive_construct_evidence_modes", "cognitive_construct_weighting_policy", "cognitive_construct_product_boundaries", } ) ANALYSIS_SUPPORTED_ACTIVITY_CONTEXTS = ( "unknown", "quiz", "unit_assessment", "performance_assessment", "mock_exam", "exam_digital", "exam_paper", "classroom_learning", "quizpang", "focustime", "survey", "assignment", "self_study", "homework", "digital_activity", "paper_activity", ) ANALYSIS_SUPPORTED_ASSESSMENT_MODALITIES = ("none", "digital", "paper") ANALYSIS_SUPPORTED_CONTEXT_STATUSES = ("missing", "partial", "resolved") TEST_BEHAVIOR_ACTIVITY_CONTEXTS = frozenset( { "quiz", "quizpang", "unit_assessment", "performance_assessment", "mock_exam", "exam_digital", "exam_paper", } ) ASSESSMENT_MODALITY_BY_ACTIVITY_CONTEXT = { "quiz": "digital", "quizpang": "digital", "survey": "digital", "assignment": "digital", "exam_digital": "digital", "exam_paper": "paper", "digital_activity": "digital", "paper_activity": "paper", } ANALYSIS_EVIDENCE_SOURCE_MODALITIES = ( "primitive_timeseries", "quality_metrics", "posterior_state", "bn_factors", "trend_windows", "behavior_alerts", ) ANALYSIS_EVIDENCE_SUMMARY_FIELDS = ( "scores", "confidence", "data_freshness", "quality_metrics", "bn_factors", "alerts", ) ANALYSIS_ALLOWED_DOWNSTREAM_USES = ("student_growth_support",) REPORT_PUBLIC_POSITIONING_BOUNDARIES = ( "student_growth_support_only", "supportive_environment_signal_only", "no_automated_integrity_label", "no_adverse_action", "human_review_context_only", ) EVIDENCE_BOUND_CLAIM_EXPLANATION_POLICY = "evidence_bound_behavior_signal" EVIDENCE_BOUND_REQUIRED_EXPLANATION_COMPONENTS = ( "activity_context", "assessment_modality", "context_status", "source_modalities", "behavior_dimensions", "student_growth_domains", "data_quality_reliability", "interpretation_caveats", ) DATA_QUALITY_RELIABILITY_VALUES = frozenset({"low", "moderate", "high"}) ANALYSIS_CONTEXTUAL_EVIDENCE_REQUIRED_FIELDS = frozenset( { "supported_activity_contexts", "supported_assessment_modalities", "supported_context_statuses", "source_modalities", "evidence_summary_fields", "supported_behavior_dimensions", "evidence_payload_kind", "claim_explanation_policy", "required_explanation_components", "allowed_downstream_uses", "public_positioning_boundaries", "redisclosure_boundary", "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", } ) ANALYSIS_COGNITIVE_CONSTRUCT_EVIDENCE_REQUIRED_FIELDS = frozenset( { "evidence_v", "policy_id", "activity_context", "assessment_modality", "context_status", "observed_source_primitives", "weighting_policy", "constructs", "product_boundaries", "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", } ) SESSION_ANALYSIS_CONTEXT_REQUIRED_FIELDS = frozenset( { "analysis_context_v", "activity_context", "assessment_modality", "primary_engine_role", "context_status", "student_growth_domains", "evidence_policy", "proctoring_role", } ) ANALYSIS_RESPONSE_CONTEXT_REQUIRED_FIELDS = frozenset( { "analysis_context_v", "activity_context", "assessment_modality", "primary_engine_role", "context_source", "context_status", "context_required", } ) SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES: dict[str, frozenset[str]] = { "activity_context": frozenset( { "classroom_learning", "quiz", "unit_assessment", "performance_assessment", "mock_exam", "exam_digital", "exam_paper", "quizpang", "focustime", "survey", "assignment", "self_study", "homework", "digital_activity", "paper_activity", "clinical_research", "unknown", } ), "assessment_modality": frozenset({"none", "digital", "paper"}), "primary_engine_role": frozenset( { "test_behavior_analysis", "learning_behavior_analysis", "clinical_signal_analysis", } ), "context_status": frozenset({"missing", "partial", "resolved"}), "evidence_policy": frozenset({"contextual_multimodal_evidence_required"}), "proctoring_role": frozenset({"supportive_environment_signal"}), } ANALYSIS_RESPONSE_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES: dict[str, frozenset[str]] = { "activity_context": SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "activity_context" ], "assessment_modality": SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "assessment_modality" ], "primary_engine_role": SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "primary_engine_role" ], "context_source": frozenset({"client_provided", "server_default"}), "context_status": SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "context_status" ], } COLLECTION_PLAN_CAPABILITY_SUMMARY_BOOLEAN_FIELDS = frozenset( { "profile_present", "visual_collection_allowed", "paper_exam_cv_allowed", "touch_collection_allowed", "stylus_collection_allowed", "clinical_audio_allowed", "offline_collection_required", "power_saver_required", } ) COLLECTION_PLAN_CAPABILITY_SUMMARY_LIST_FIELDS = frozenset( { "available_sensor_groups", "unavailable_sensor_groups", "capability_reason_codes", } ) RUNTIME_PAUSED_SENSOR_GROUPS = frozenset( { "camera_visual", "paper_exam_cv", "clinical_audio", } ) RUNTIME_SENSOR_PAUSE_REASONS = frozenset( { "battery_low_power_saver", "low_memory_power_saver", "thermal_critical_power_saver", } ) COLLECTION_PLAN_REQUIRED_ACTIONS_BY_REASON: dict[str, tuple[str, ...]] = { "device_profile_missing": ("send_device_profile",), "sensor_tier_unverified": ("send_device_profile",), "sensor_tier_mismatch": ("use_server_sensor_tier",), "cpu_class_tier_cap": ( "use_server_sensor_tier", "increase_upload_interval", "reduce_camera_frame_rate", ), "camera_unavailable": ("disable_camera_primitives",), "paper_exam_requires_device_profile": ("send_device_profile",), "paper_exam_requires_camera": ("switch_to_exam_digital",), "paper_exam_requires_camera_resolution": ("use_higher_resolution_camera",), "camera_resolution_insufficient": ("disable_camera_primitives",), "camera_disabled_by_plan": ("disable_camera_primitives",), "camera_visual_disabled_by_plan": ("disable_camera_primitives",), "paper_exam_requires_tier_2": ("switch_to_exam_digital",), "paper_exam_requires_camera_consent": ("request_camera_consent",), "camera_consent_missing": ("request_camera_consent",), "camera_consent_false": ("disable_camera_primitives",), "paper_exam_requires_stable_power": ("prompt_plug_in_or_degrade",), "paper_exam_requires_thermal_safe": ( "cool_device_or_degrade", "switch_to_exam_digital", ), "paper_exam_requires_network_connectivity": ( "restore_network_or_degrade", "buffer_until_online", ), "touch_unavailable": ("disable_touch_primitives",), "touch_disabled_by_plan": ("disable_touch_primitives",), "stylus_unavailable": ("disable_stylus_primitives",), "stylus_disabled_by_plan": ("disable_stylus_primitives",), "microphone_consent_missing": ("request_microphone_consent",), "microphone_unavailable": ("disable_voice_activity",), "clinical_audio_disabled_by_plan": ("disable_voice_activity",), "low_end_cpu_power_saver": ( "increase_upload_interval", "reduce_camera_frame_rate", ), "low_memory_power_saver": ( "increase_upload_interval", "reduce_memory_footprint", "pause_visual_collection", "disable_voice_activity", ), "battery_low_power_saver": ( "increase_upload_interval", "pause_visual_collection", "disable_voice_activity", "prompt_plug_in_or_degrade", ), "thermal_serious_power_saver": ( "increase_upload_interval", "reduce_camera_frame_rate", ), "thermal_critical_power_saver": ( "increase_upload_interval", "pause_visual_collection", "disable_voice_activity", ), "network_constrained_buffering": ("increase_upload_interval",), "network_offline_buffering": ("increase_upload_interval", "buffer_until_online"), "paper_exam_cv_low_resource_device": ("disable_paper_exam_cv",), "paper_exam_cv_disabled_by_plan": ("disable_paper_exam_cv",), "paper_exam_cv_thermal_throttled": ("disable_paper_exam_cv",), "paper_exam_cv_power_saver": ("disable_paper_exam_cv",), "paper_exam_cv_offline": ("disable_paper_exam_cv",), "paper_exam_cv_stack_not_enabled": ("contact_ops_cv_manifest",), "paper_exam_cv_manifest_id_missing": ("contact_ops_cv_manifest",), "paper_exam_cv_required_components_missing": ("contact_ops_cv_manifest",), "paper_exam_mode_disabled": ("contact_ops_cv_manifest",), "cv_model_manifest_missing": ("contact_ops_cv_manifest",), "cv_model_manifest_invalid": ("contact_ops_cv_manifest",), "proctoring_capability_warning_invalid": ("contact_ops_cv_manifest",), "microphone_not_allowed_for_education_surface": ("disable_voice_activity",), "research_clinical_mode_not_enabled": ( "contact_ops_research_enablement", "disable_voice_activity", ), "research_clinical_requires_clinic_product": ( "switch_to_clinic_product", "disable_voice_activity", ), "research_clinical_study_id_missing": ( "provide_irb_study_id", "disable_voice_activity", ), "research_clinical_consent_missing": ( "request_research_clinical_consent", "disable_voice_activity", ), "research_clinical_consent_false": ( "request_research_clinical_consent", "disable_voice_activity", ), "research_clinical_subject_enrollment_unverified": ( "verify_research_subject_enrollment", "disable_voice_activity", ), "device_clock_skew_high": ("sync_device_clock",), } CAMERA_VISUAL_PRIMITIVES = frozenset( { "gaze_entropy", "gaze_deviation", "head_pose_deviation", "head_deviation", "fixation_duration", "blink_rate", "head_pose", "head_pose_micro_movement_rate", "proctor_vision_gaze_deviation", "proctor_vision_head_deviation", "proctor_vision_blink_rate", "proctor_vision_face_detected", "proctoring_gaze_entropy", "proctoring_head_deviation", "proctoring_blink_rate", "proctoring_fixation_duration", "proctoring_face_detected", } ) ENHANCED_CAMERA_PRIMITIVES = frozenset( { "hand_landmark_movement", "posture_lean_angle", "posture_drift_score", "posture_alignment", } ) PAPER_EXAM_CONTEXT_CV_PRIMITIVES = frozenset( { "gaze_off_paper_ratio", "hand_off_paper_seconds_per_minute", "object_phone_detected_count", "object_extra_paper_detected_count", } ) TOUCH_PRIMITIVES = frozenset({"touch_pressure_variance", "multi_touch_event_count"}) STYLUS_PRIMITIVES = frozenset( {"stylus_pressure_variance", "stylus_velocity_variance", "stylus_lift_off_count"} ) PRESSURE_PRIMITIVES = frozenset( {"keystroke_pressure_variance", "proctor_pointer_pressure"} ) CLINICAL_AUDIO_PRIMITIVES = frozenset( { "voice_prosody_pitch_variance", "voice_speaking_rate", "voice_pause_distribution_entropy", "voice_activity", } ) class DrSimonError(Exception): """API error with status code and structured server detail when available.""" def __init__( self, status_code: int, detail: str, *, payload: JsonObject | None = None, ): self.status_code = status_code self.detail = detail self.payload = payload or {} self.error = self.payload.get("error") self.rejected_fields = list(self.payload.get("rejected_fields") or []) self.rejection_reasons = list(self.payload.get("rejection_reasons") or []) super().__init__(f"[{status_code}] {detail}") class DrSimonClient: """Synchronous client for DrSimon Engine API.""" def __init__( self, api_key: str, base_url: str = "https://engine.drsimon.ai", timeout: float = 10.0, ): self._base_url = base_url.rstrip("/") self._sensitive_values = self._sensitive_error_values(api_key, base_url) self._client = httpx.Client( base_url=self._base_url, headers={"X-API-Key": api_key}, timeout=timeout, ) self._sessions: dict[tuple[str, str], JsonObject] = {} def _request(self, method: str, path: str, **kwargs: Any) -> JsonObject: resp = self._client.request(method, path, **kwargs) if resp.status_code >= 400: payload = self._parse_error_payload(resp) if payload is not None: payload = self._redact_error_payload(payload) detail = str( payload.get("message") or payload.get("detail") or payload.get("error") or resp.text ) raise DrSimonError(resp.status_code, detail, payload=payload) raise DrSimonError( resp.status_code, self._redact_error_text(resp.text), ) return cast(JsonObject, resp.json()) @staticmethod def _sensitive_error_values(api_key: str, base_url: str) -> tuple[str, ...]: values: list[str] = [api_key] try: parts = urlsplit(base_url) except ValueError: parts = None if parts is not None: if parts.username: values.append(parts.username) if parts.password: values.append(parts.password) if "@" in parts.netloc: values.append(parts.netloc.rsplit("@", 1)[0]) return tuple(value for value in values if len(value) >= 6) def _redact_error_text(self, value: object) -> str: redacted = str(value) for sensitive in self._sensitive_values: for candidate in {sensitive, sensitive.lower(), sensitive.upper()}: redacted = redacted.replace(candidate, REDACTED_DETAIL_TOKEN) return redacted def _redact_error_payload(self, value: Any) -> Any: if isinstance(value, str): return self._redact_error_text(value) if isinstance(value, list): return [self._redact_error_payload(item) for item in value] if isinstance(value, dict): return {key: self._redact_error_payload(item) for key, item in value.items()} return value @staticmethod def _parse_error_payload(resp: httpx.Response) -> JsonObject | None: try: payload = resp.json() except ValueError: return None return cast(JsonObject, payload) if isinstance(payload, dict) else None @staticmethod def _now_iso_ms() -> str: dt = datetime.now(timezone.utc) return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z" @staticmethod def _hash16(payload: Any) -> str: encoded = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest()[:16] @classmethod def _hash16_device_profile(cls, payload: JsonObject | None) -> str: if payload is None: payload = { "device_profile_v": 1, "device_profile_missing": True, } elif "camera_max_res" not in payload: payload = {**payload, "camera_max_res": None} return cls._hash16(cls._normalize_integral_floats(payload)) @classmethod def _normalize_integral_floats(cls, value: Any) -> Any: if isinstance(value, float) and value.is_integer(): return int(value) if isinstance(value, list): return [cls._normalize_integral_floats(item) for item in value] if isinstance(value, dict): return { key: cls._normalize_integral_floats(item) for key, item in value.items() } return value @staticmethod def _hash16_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] @staticmethod def _session_key(uuid: str, product: str) -> tuple[str, str]: return (uuid, product) @classmethod def _make_session_id(cls, uuid: str, started_at: str) -> str: return f"{uuid}:{started_at}:{cls._hash16_text(uuid + started_at)[:8]}" @staticmethod def _validate_collection_plan(payload: JsonObject) -> JsonObject: plan = payload.get("collection_plan") if not isinstance(plan, dict): raise DrSimonError( 502, "session/start response missing collection_plan", payload={"error": "collection_plan_missing"}, ) if plan.get("upload_mode") != "primitive_upload": raise DrSimonError( 502, "unsafe collection_plan upload_mode", payload={"error": "collection_plan_upload_mode_unsafe"}, ) if plan.get("raw_payload_upload_allowed") is not False: raise DrSimonError( 502, "unsafe collection_plan permits raw payload upload", payload={"error": "collection_plan_raw_payload_unsafe"}, ) if plan.get("local_processing_required") is not True: raise DrSimonError( 502, "unsafe collection_plan does not require local processing", payload={"error": "collection_plan_local_processing_required"}, ) if plan.get("plan_version") != 1: raise DrSimonError( 502, "collection_plan.plan_version must be 1", payload={"error": "collection_plan_schema_invalid"}, ) sensor_tier = plan.get("sensor_tier") if ( isinstance(sensor_tier, bool) or not isinstance(sensor_tier, int) or sensor_tier < 0 or sensor_tier > 3 ): raise DrSimonError( 502, "collection_plan.sensor_tier must be an integer 0..3", payload={"error": "collection_plan_sensor_tier_invalid"}, ) response_sensor_tier = payload.get("sensor_tier") if ( isinstance(response_sensor_tier, bool) or not isinstance(response_sensor_tier, int) or response_sensor_tier < 0 or response_sensor_tier > 3 ): raise DrSimonError( 502, "session/start response sensor_tier must be an integer 0..3", payload={"error": "collection_plan_sensor_tier_invalid"}, ) if response_sensor_tier != sensor_tier: raise DrSimonError( 502, "collection_plan.sensor_tier does not match response sensor_tier", payload={"error": "collection_plan_sensor_tier_mismatch"}, ) for field in ( "ingest_batch_max_primitives", "ingest_min_interval_seconds", "ingest_max_interval_seconds", "max_cpu_percent", "max_memory_mb", ): value = plan.get(field) if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise DrSimonError( 502, f"collection_plan.{field} must be a positive integer", payload={"error": "collection_plan_budget_invalid"}, ) if int(plan["ingest_min_interval_seconds"]) > int(plan["ingest_max_interval_seconds"]): raise DrSimonError( 502, "collection_plan ingest interval bounds are invalid", payload={"error": "collection_plan_interval_invalid"}, ) for field in ( "enabled_sensor_groups", "disabled_sensor_groups", "reason_codes", "client_actions", ): value = plan.get(field) if not isinstance(value, list) or any( not isinstance(item, str) or not item for item in value ): raise DrSimonError( 502, f"collection_plan.{field} must be a string array", payload={"error": "collection_plan_scope_invalid"}, ) if len(value) > COLLECTION_PLAN_STRING_LIST_MAX_ITEMS or len(set(value)) != len(value): raise DrSimonError( 502, f"collection_plan.{field} must be a bounded unique string array", payload={"error": "collection_plan_scope_invalid"}, ) if field in {"enabled_sensor_groups", "disabled_sensor_groups"}: if set(value) - COLLECTION_PLAN_SENSOR_GROUPS: raise DrSimonError( 502, f"collection_plan.{field} contains an unknown sensor group", payload={"error": "collection_plan_scope_invalid"}, ) elif any(COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is None for item in value): raise DrSimonError( 502, f"collection_plan.{field} must use public identifiers", payload={"error": "collection_plan_scope_invalid"}, ) missing_actions = DrSimonClient._missing_collection_plan_client_actions(plan) if missing_actions: raise DrSimonError( 502, "collection_plan.client_actions is missing required actions: " + ",".join(missing_actions), payload={ "error": "collection_plan_scope_invalid", "missing_client_actions": missing_actions, }, ) if ( "offline_buffer_required" not in plan or "offline_buffer_min_seconds" not in plan ): raise DrSimonError( 502, "collection_plan offline buffer contract is missing", payload={"error": "collection_plan_offline_buffer_invalid"}, ) offline_required = plan["offline_buffer_required"] if not isinstance(offline_required, bool): raise DrSimonError( 502, "collection_plan.offline_buffer_required must be boolean", payload={"error": "collection_plan_offline_buffer_invalid"}, ) offline_min = plan["offline_buffer_min_seconds"] if isinstance(offline_min, bool) or not isinstance(offline_min, int): raise DrSimonError( 502, "collection_plan.offline_buffer_min_seconds must be an integer", payload={"error": "collection_plan_offline_buffer_invalid"}, ) if offline_required and offline_min < 4 * 3600: raise DrSimonError( 502, "collection_plan offline buffer must cover at least 4 hours", payload={"error": "collection_plan_offline_buffer_invalid"}, ) if not offline_required and offline_min != 0: raise DrSimonError( 502, "collection_plan offline buffer must be 0 seconds when disabled", payload={"error": "collection_plan_offline_buffer_invalid"}, ) DrSimonClient._validate_collection_plan_capability_summary(plan) return plan @staticmethod def _validate_analysis_context(payload: JsonObject) -> JsonObject: context_payload = payload.get("analysis_context") if not isinstance(context_payload, dict): raise DrSimonError( 502, "session/start response missing analysis_context", payload={"error": "analysis_context_invalid"}, ) context = cast(JsonObject, context_payload) missing = [ field for field in SESSION_ANALYSIS_CONTEXT_REQUIRED_FIELDS if field not in context ] if missing or context.get("analysis_context_v") != 1: raise DrSimonError( 502, "session/start response analysis_context schema is invalid", payload={"error": "analysis_context_invalid"}, ) for field, allowed_values in ( SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES.items() ): value = context.get(field) if ( not isinstance(value, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(value) is None or value not in allowed_values ): raise DrSimonError( 502, f"analysis_context.{field} must use a public identifier", payload={"error": "analysis_context_invalid"}, ) expected_context_status = ( "resolved" if context["activity_context"] != "unknown" else "partial" if context["assessment_modality"] != "none" else "missing" ) if context["context_status"] != expected_context_status: raise DrSimonError( 502, "analysis_context.context_status is inconsistent", payload={"error": "analysis_context_invalid"}, ) domains_payload = context.get("student_growth_domains") if ( not isinstance(domains_payload, list) or any(not isinstance(item, str) or not item for item in domains_payload) ): raise DrSimonError( 502, "analysis_context.student_growth_domains must be a string array", payload={"error": "analysis_context_invalid"}, ) domains = cast(list[str], domains_payload) if ( len(domains) > COLLECTION_PLAN_STRING_LIST_MAX_ITEMS or len(set(domains)) != len(domains) or any(COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is None for item in domains) or tuple(domains) != STUDENT_GROWTH_DOMAINS ): raise DrSimonError( 502, "analysis_context.student_growth_domains is invalid", payload={"error": "analysis_context_invalid"}, ) return context @staticmethod def _validate_analysis_response_context( payload: JsonObject, *, expected_activity_context: str | None = None, expected_assessment_modality: str | None = None, ) -> JsonObject: context_payload = payload.get("analysisContext") if not isinstance(context_payload, dict): raise DrSimonError( 502, "analysis response missing analysisContext", payload={"error": "analysis_context_invalid"}, ) context = cast(JsonObject, context_payload) missing = [ field for field in ANALYSIS_RESPONSE_CONTEXT_REQUIRED_FIELDS if field not in context ] if missing or context.get("analysis_context_v") != 1: raise DrSimonError( 502, "analysis response analysisContext schema is invalid", payload={ "error": "analysis_context_invalid", "missing_fields": missing, }, ) for field, allowed_values in ( ANALYSIS_RESPONSE_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES.items() ): value = context.get(field) if ( not isinstance(value, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(value) is None or value not in allowed_values ): raise DrSimonError( 502, f"analysisContext.{field} must use a public identifier", payload={"error": "analysis_context_invalid"}, ) expected_context_status = ( "resolved" if context["activity_context"] != "unknown" else "partial" if context["assessment_modality"] != "none" else "missing" ) if context["context_status"] != expected_context_status: raise DrSimonError( 502, "analysisContext.context_status is inconsistent", payload={"error": "analysis_context_invalid"}, ) if context.get("context_required") is not True: raise DrSimonError( 502, "analysisContext.context_required must be true", payload={"error": "analysis_context_invalid"}, ) expected_context_source = ( "client_provided" if expected_activity_context is not None or expected_assessment_modality is not None else None ) if ( expected_context_source is not None and context["context_source"] != expected_context_source ): raise DrSimonError( 502, "analysis response context_source does not match request", payload={ "error": "analysis_context_mismatch", "expected": expected_context_source, "actual": context["context_source"], }, ) if ( expected_activity_context is not None and context["activity_context"] != expected_activity_context ): raise DrSimonError( 502, "analysis response activity_context does not match request", payload={ "error": "analysis_context_mismatch", "expected": expected_activity_context, "actual": context["activity_context"], }, ) derived_assessment_modality = ( expected_assessment_modality if expected_assessment_modality is not None else ASSESSMENT_MODALITY_BY_ACTIVITY_CONTEXT.get( expected_activity_context, "none", ) if expected_activity_context is not None else None ) if ( derived_assessment_modality is not None and context["assessment_modality"] != derived_assessment_modality ): raise DrSimonError( 502, "analysis response assessment_modality does not match request", payload={ "error": "analysis_context_mismatch", "expected": derived_assessment_modality, "actual": context["assessment_modality"], }, ) expected_primary_engine_role = ( "test_behavior_analysis" if expected_activity_context in TEST_BEHAVIOR_ACTIVITY_CONTEXTS else "clinical_signal_analysis" if expected_activity_context == "clinical_research" else "learning_behavior_analysis" if expected_activity_context is not None or expected_assessment_modality is not None else None ) if ( expected_primary_engine_role is not None and context["primary_engine_role"] != expected_primary_engine_role ): raise DrSimonError( 502, "analysis response primary_engine_role does not match request", payload={ "error": "analysis_context_mismatch", "expected": expected_primary_engine_role, "actual": context["primary_engine_role"], }, ) return context @staticmethod def _public_token_list_matches(value: object, expected: tuple[str, ...]) -> bool: return ( isinstance(value, list) and tuple(value) == expected and all( isinstance(item, str) and COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is not None for item in value ) ) @staticmethod def _is_public_token_list(value: object, *, allow_empty: bool = False) -> bool: return ( isinstance(value, list) and (allow_empty or bool(value)) and len(value) <= COLLECTION_PLAN_STRING_LIST_MAX_ITEMS and all( isinstance(item, str) and COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is not None for item in value ) ) @staticmethod def _is_ratio(value: object) -> bool: return ( not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(value) and 0.0 <= float(value) <= 1.0 ) @staticmethod def _validate_analysis_contextual_evidence(payload: JsonObject) -> JsonObject: evidence_payload = payload.get("contextualEvidence") if not isinstance(evidence_payload, dict): raise DrSimonError( 502, "analysis response missing contextualEvidence", payload={"error": "analysis_contextual_evidence_invalid"}, ) evidence = cast(JsonObject, evidence_payload) missing = [ field for field in ANALYSIS_CONTEXTUAL_EVIDENCE_REQUIRED_FIELDS if field not in evidence ] if missing: raise DrSimonError( 502, "analysis response contextualEvidence schema is invalid", payload={ "error": "analysis_contextual_evidence_invalid", "missing_fields": missing, }, ) if evidence.get("evidence_payload_kind") != "derived_behavior_features_only": raise DrSimonError( 502, "contextualEvidence.evidence_payload_kind is invalid", payload={"error": "analysis_contextual_evidence_invalid"}, ) if evidence.get("claim_explanation_policy") != "evidence_bound_behavior_signal": raise DrSimonError( 502, "contextualEvidence.claim_explanation_policy is invalid", payload={"error": "analysis_contextual_evidence_invalid"}, ) if evidence.get("redisclosure_boundary") != "controller_authorized_only": raise DrSimonError( 502, "contextualEvidence.redisclosure_boundary is invalid", payload={"error": "analysis_contextual_evidence_invalid"}, ) for field in ( "no_raw_sensor_payload", "no_clinical_diagnosis", "no_adverse_action", ): if evidence.get(field) is not True: raise DrSimonError( 502, f"contextualEvidence.{field} must be true", payload={"error": "analysis_contextual_evidence_invalid"}, ) dimensions = evidence.get("supported_behavior_dimensions") activity_contexts = evidence.get("supported_activity_contexts") assessment_modalities = evidence.get("supported_assessment_modalities") context_statuses = evidence.get("supported_context_statuses") source_modalities = evidence.get("source_modalities") summary_fields = evidence.get("evidence_summary_fields") allowed_uses = evidence.get("allowed_downstream_uses") positioning_boundaries = evidence.get("public_positioning_boundaries") explanation_components = evidence.get("required_explanation_components") has_cognitive_contract = any( field in evidence for field in COGNITIVE_CONTEXTUAL_EVIDENCE_FIELDS ) cognitive_contract_invalid = False if has_cognitive_contract: cognitive_contract_invalid = ( evidence.get("cognitive_construct_policy") != COGNITIVE_CONSTRUCT_POLICY or evidence.get("cognitive_construct_weighting_policy") != COGNITIVE_CONSTRUCT_WEIGHTING_POLICY or not DrSimonClient._public_token_list_matches( evidence.get("supported_cognitive_constructs"), SUPPORTED_COGNITIVE_CONSTRUCTS, ) or not DrSimonClient._public_token_list_matches( evidence.get("cognitive_construct_evidence_modes"), COGNITIVE_CONSTRUCT_EVIDENCE_MODES, ) or not DrSimonClient._public_token_list_matches( evidence.get("cognitive_construct_product_boundaries"), COGNITIVE_CONSTRUCT_PRODUCT_BOUNDARIES, ) ) if ( not DrSimonClient._public_token_list_matches( activity_contexts, ANALYSIS_SUPPORTED_ACTIVITY_CONTEXTS, ) or not DrSimonClient._public_token_list_matches( assessment_modalities, ANALYSIS_SUPPORTED_ASSESSMENT_MODALITIES, ) or not DrSimonClient._public_token_list_matches( context_statuses, ANALYSIS_SUPPORTED_CONTEXT_STATUSES, ) or not DrSimonClient._public_token_list_matches( source_modalities, ANALYSIS_EVIDENCE_SOURCE_MODALITIES, ) or not DrSimonClient._public_token_list_matches( summary_fields, ANALYSIS_EVIDENCE_SUMMARY_FIELDS, ) or not DrSimonClient._public_token_list_matches( dimensions, SUPPORTED_BEHAVIOR_DIMENSIONS, ) or not isinstance(allowed_uses, list) or tuple(allowed_uses) != ANALYSIS_ALLOWED_DOWNSTREAM_USES or not isinstance(positioning_boundaries, list) or tuple(positioning_boundaries) != REPORT_PUBLIC_POSITIONING_BOUNDARIES or any( not isinstance(item, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is None for item in positioning_boundaries ) or not isinstance(explanation_components, list) or tuple(explanation_components) != EVIDENCE_BOUND_REQUIRED_EXPLANATION_COMPONENTS or any( not isinstance(item, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is None for item in explanation_components ) or cognitive_contract_invalid ): raise DrSimonError( 502, "analysis response contextualEvidence public tokens are invalid", payload={"error": "analysis_contextual_evidence_invalid"}, ) return evidence @staticmethod def _has_evidence_explanation_policy(evidence: JsonObject) -> bool: components = evidence.get("required_explanation_components") return ( evidence.get("claim_explanation_policy") == EVIDENCE_BOUND_CLAIM_EXPLANATION_POLICY and isinstance(components, list) and tuple(components) == EVIDENCE_BOUND_REQUIRED_EXPLANATION_COMPONENTS and all( isinstance(item, str) and COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is not None for item in components ) ) @staticmethod def _has_report_public_positioning_boundaries(evidence: JsonObject) -> bool: boundaries = evidence.get("public_positioning_boundaries") return ( isinstance(boundaries, list) and tuple(boundaries) == REPORT_PUBLIC_POSITIONING_BOUNDARIES and all( isinstance(item, str) and COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is not None for item in boundaries ) ) @staticmethod def _validate_evidence_context_matches( evidence: JsonObject, analysis_context: JsonObject, ) -> None: for field in ("activity_context", "assessment_modality", "context_status"): if evidence.get(field) != analysis_context.get(field): raise DrSimonError( 502, f"analysis evidence {field} does not match analysisContext", payload={ "error": "analysis_evidence_context_mismatch", "field": field, }, ) @staticmethod def _validate_analysis_score_evidence( payload: JsonObject, *, analysis_context: JsonObject, ) -> None: scores = payload.get("scores") if not isinstance(scores, dict): raise DrSimonError( 502, "analysis response scores schema is invalid", payload={"error": "analysis_score_evidence_invalid"}, ) for score_key, score in scores.items(): if ( not isinstance(score_key, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(score_key) is None or not isinstance(score, dict) or not isinstance(score.get("evidence"), dict) ): raise DrSimonError( 502, "analysis response score evidence schema is invalid", payload={"error": "analysis_score_evidence_invalid"}, ) evidence = cast(JsonObject, score["evidence"]) context_status = evidence.get("context_status") if ( evidence.get("evidence_policy") != "contextual_multimodal_evidence_required" or evidence.get("data_quality_reliability") not in DATA_QUALITY_RELIABILITY_VALUES or not isinstance(context_status, str) or context_status not in SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "context_status" ] or not DrSimonClient._has_evidence_explanation_policy(evidence) or not isinstance(evidence.get("context_applied"), bool) or evidence.get("no_raw_sensor_payload") is not True or evidence.get("no_clinical_diagnosis") is not True or evidence.get("no_adverse_action") is not True ): raise DrSimonError( 502, "analysis response score evidence is invalid", payload={"error": "analysis_score_evidence_invalid"}, ) DrSimonClient._validate_evidence_context_matches( evidence, analysis_context, ) @staticmethod def _validate_analysis_item_evidence_context( payload: JsonObject, *, analysis_context: JsonObject, ) -> None: for collection_name in ("alerts", "trends", "anomalies"): collection = payload.get(collection_name) if collection is None: continue if not isinstance(collection, list): raise DrSimonError( 502, f"analysis response {collection_name} schema is invalid", payload={"error": "analysis_evidence_context_invalid"}, ) for item in collection: if not isinstance(item, dict): raise DrSimonError( 502, f"analysis response {collection_name} item is invalid", payload={"error": "analysis_evidence_context_invalid"}, ) evidence = item.get("evidence") if evidence is None: continue if not isinstance(evidence, dict): raise DrSimonError( 502, f"analysis response {collection_name} evidence is invalid", payload={"error": "analysis_evidence_context_invalid"}, ) DrSimonClient._validate_evidence_context_matches( cast(JsonObject, evidence), analysis_context, ) @staticmethod def _validate_analysis_cognitive_construct_evidence( payload: JsonObject, *, analysis_context: JsonObject, ) -> JsonObject: evidence_payload = payload.get("cognitiveConstructEvidence") if not isinstance(evidence_payload, dict): raise DrSimonError( 502, "analysis response missing cognitiveConstructEvidence", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) evidence = cast(JsonObject, evidence_payload) missing = [ field for field in ANALYSIS_COGNITIVE_CONSTRUCT_EVIDENCE_REQUIRED_FIELDS if field not in evidence ] if missing: raise DrSimonError( 502, "analysis response cognitiveConstructEvidence schema is invalid", payload={ "error": "analysis_cognitive_construct_evidence_invalid", "missing_fields": missing, }, ) if ( evidence.get("evidence_v") != 1 or evidence.get("policy_id") != COGNITIVE_CONSTRUCT_POLICY or evidence.get("weighting_policy") != COGNITIVE_CONSTRUCT_WEIGHTING_POLICY or not DrSimonClient._public_token_list_matches( evidence.get("product_boundaries"), COGNITIVE_CONSTRUCT_PRODUCT_BOUNDARIES, ) or not DrSimonClient._is_public_token_list( evidence.get("observed_source_primitives"), allow_empty=True, ) or evidence.get("no_raw_sensor_payload") is not True or evidence.get("no_clinical_diagnosis") is not True or evidence.get("no_adverse_action") is not True ): raise DrSimonError( 502, "analysis response cognitiveConstructEvidence is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) DrSimonClient._validate_evidence_context_matches(evidence, analysis_context) constructs = evidence.get("constructs") if not isinstance(constructs, list): raise DrSimonError( 502, "analysis response cognitiveConstructEvidence.constructs is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) construct_keys: list[str] = [] for item in constructs: if not isinstance(item, dict): raise DrSimonError( 502, "analysis response cognitiveConstructEvidence item is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) construct_key = item.get("construct_key") expected = item.get("expected_source_primitives") observed = item.get("observed_source_primitives") missing_primitives = item.get("missing_source_primitives") science_references = item.get("science_references") warning_codes = item.get("warning_codes") observed_weight = item.get("observed_weight") missing_weight = item.get("missing_weight") if ( not isinstance(construct_key, str) or construct_key not in SUPPORTED_COGNITIVE_CONSTRUCTS or not DrSimonClient._is_public_token_list(expected) or not DrSimonClient._is_public_token_list(observed, allow_empty=True) or not DrSimonClient._is_public_token_list( missing_primitives, allow_empty=True, ) or not DrSimonClient._is_ratio(observed_weight) or not DrSimonClient._is_ratio(missing_weight) or abs( float(cast(float, observed_weight)) + float(cast(float, missing_weight)) - 1.0 ) > 0.0001 or item.get("evidence_mode") not in COGNITIVE_CONSTRUCT_EVIDENCE_MODES or item.get("confidence_cap") not in COGNITIVE_CONSTRUCT_CONFIDENCE_CAPS or item.get("weighting_policy") != COGNITIVE_CONSTRUCT_WEIGHTING_POLICY or not DrSimonClient._is_public_token_list(science_references) or sorted( set(cast(list[str], science_references)) - set(COGNITIVE_CONSTRUCT_SCIENCE_REFERENCES) ) or not DrSimonClient._is_public_token_list( warning_codes, allow_empty=True, ) or sorted(set(cast(list[str], warning_codes)) - set(COGNITIVE_CONSTRUCT_WARNING_CODES)) ): raise DrSimonError( 502, "analysis response cognitiveConstructEvidence item is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) expected_set = set(cast(list[str], expected)) observed_set = set(cast(list[str], observed)) missing_set = set(cast(list[str], missing_primitives)) if observed_set & missing_set or observed_set | missing_set != expected_set: raise DrSimonError( 502, "analysis response cognitiveConstructEvidence partition is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) construct_keys.append(construct_key) if tuple(construct_keys) != SUPPORTED_COGNITIVE_CONSTRUCTS: raise DrSimonError( 502, "analysis response cognitiveConstructEvidence construct order is invalid", payload={"error": "analysis_cognitive_construct_evidence_invalid"}, ) return evidence @staticmethod def _validate_collection_plan_capability_summary(plan: JsonObject) -> None: summary = plan.get("capability_summary") if not isinstance(summary, dict): DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary is required" ) for field in COLLECTION_PLAN_CAPABILITY_SUMMARY_BOOLEAN_FIELDS: if not isinstance(summary.get(field), bool): DrSimonClient._raise_collection_plan_capability_summary_invalid( f"collection_plan.capability_summary.{field} must be boolean" ) tier = summary.get("device_sensor_tier") if tier is not None and ( isinstance(tier, bool) or not isinstance(tier, int) or tier < 0 or tier > 3 ): DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary.device_sensor_tier is invalid" ) for field in COLLECTION_PLAN_CAPABILITY_SUMMARY_LIST_FIELDS: value = summary.get(field) if not isinstance(value, list) or any( not isinstance(item, str) or not item for item in value ): DrSimonClient._raise_collection_plan_capability_summary_invalid( f"collection_plan.capability_summary.{field} must be a string array" ) if ( len(value) > COLLECTION_PLAN_STRING_LIST_MAX_ITEMS or len(set(value)) != len(value) ): DrSimonClient._raise_collection_plan_capability_summary_invalid( f"collection_plan.capability_summary.{field} must be bounded and unique" ) if field.endswith("sensor_groups"): if set(value) - COLLECTION_PLAN_SENSOR_GROUPS: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary has an unknown sensor group" ) elif any( COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(item) is None for item in value ): DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary must use public identifiers" ) enabled_groups = set(cast(list[str], plan["enabled_sensor_groups"])) available_groups = set(cast(list[str], summary["available_sensor_groups"])) unavailable_groups = set( cast(list[str], summary["unavailable_sensor_groups"]) ) if available_groups & unavailable_groups: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary sensor availability overlaps" ) if available_groups | unavailable_groups != COLLECTION_PLAN_SENSOR_GROUPS: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary sensor availability is incomplete" ) if tier is not None and any( COLLECTION_PLAN_SENSOR_GROUP_MIN_TIERS[group] > tier for group in available_groups ): DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary available sensors exceed tier" ) if enabled_groups - available_groups: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary conflicts with enabled sensors" ) runtime_unavailable = ( DrSimonClient._runtime_unavailable_sensor_groups( cast(list[str], plan["reason_codes"]) ) ) if available_groups & runtime_unavailable: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary runtime paused sensors must be unavailable" ) if bool(summary["profile_present"]): if tier != plan["sensor_tier"]: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary tier mismatch" ) elif tier is not None: DrSimonClient._raise_collection_plan_capability_summary_invalid( "collection_plan.capability_summary tier requires a profile" ) expected_flags = { "visual_collection_allowed": bool( enabled_groups & {"camera_visual", "paper_exam_cv"} ), "paper_exam_cv_allowed": "paper_exam_cv" in enabled_groups, "touch_collection_allowed": "touch" in enabled_groups, "stylus_collection_allowed": "stylus" in enabled_groups, "clinical_audio_allowed": "clinical_audio" in enabled_groups, "offline_collection_required": bool(plan["offline_buffer_required"]), "power_saver_required": any( DrSimonClient._collection_plan_reason_requires_power_saver(reason) for reason in cast(list[str], plan["reason_codes"]) ), } for field, expected in expected_flags.items(): if summary[field] != expected: DrSimonClient._raise_collection_plan_capability_summary_invalid( f"collection_plan.capability_summary.{field} is inconsistent" ) @staticmethod def _collection_plan_reason_requires_power_saver(reason: str) -> bool: return reason.endswith("_power_saver") or reason in { "low_memory_power_saver", "low_end_cpu_power_saver", "thermal_serious_power_saver", "thermal_critical_power_saver", } @staticmethod def _runtime_unavailable_sensor_groups(reason_codes: list[str]) -> frozenset[str]: if RUNTIME_SENSOR_PAUSE_REASONS & set(reason_codes): return RUNTIME_PAUSED_SENSOR_GROUPS return frozenset() @staticmethod def _raise_collection_plan_capability_summary_invalid( message: str, ) -> NoReturn: raise DrSimonError( 502, message, payload={"error": "collection_plan_capability_summary_invalid"}, ) @staticmethod def _missing_collection_plan_client_actions( plan: JsonObject, ) -> list[str]: present_actions = set(plan["client_actions"]) missing_actions: list[str] = [] for reason in plan["reason_codes"]: for action in COLLECTION_PLAN_REQUIRED_ACTIONS_BY_REASON.get(reason, ()): if action not in present_actions and action not in missing_actions: missing_actions.append(action) return missing_actions @classmethod def _ingest_artifact(cls, uuid: str, primitives: list[JsonObject]) -> str: times = [str(item["time"]) for item in primitives] moment_digest = cls._hash16_text( "|".join( sorted( f"ingest-moment:{uuid}:{item['time']}:{item['type']}" for item in primitives ) ) ) return ( f"ingest-batch:{uuid}:{len(primitives)}:" f"{min(times)}:{max(times)}:{moment_digest}" ) @staticmethod def _surface_for_product(product: str) -> str: return { "edu": "FOCUSPANG", "home": "MOMENT", "clinic": "PDSS", }.get(product, "CUSTOM") @staticmethod def _sensor_group_for_primitive( primitive_type: str, device_profile: JsonObject | None = None, ) -> str: if primitive_type in CLINICAL_AUDIO_PRIMITIVES: return "clinical_audio" if primitive_type in STYLUS_PRIMITIVES: return "stylus" if primitive_type in TOUCH_PRIMITIVES: return "touch" if primitive_type in PRESSURE_PRIMITIVES: if ( device_profile is not None and device_profile.get("has_stylus") is True and device_profile.get("has_touch") is not True ): return "stylus" return "touch" if primitive_type in PAPER_EXAM_CONTEXT_CV_PRIMITIVES: return "paper_exam_cv" if ( primitive_type in ENHANCED_CAMERA_PRIMITIVES or primitive_type in CAMERA_VISUAL_PRIMITIVES ): return "camera_visual" if "security" in primitive_type or primitive_type.startswith("proctor_bg_"): return "security_events" if ( "key" in primitive_type or "input" in primitive_type or "pointer" in primitive_type ): return "keyboard" if ( "app_" in primitive_type or "tab" in primitive_type or "url" in primitive_type or "scroll" in primitive_type or "focus" in primitive_type or "interaction" in primitive_type ): return "app_interaction" return "timing" def _build_ingest_lineage( self, *, uuid: str, product: str, primitives: list[JsonObject], device_profile: JsonObject | None, batch_id: str, client_clock_at_batch: str, collection_plan: JsonObject | None = None, ) -> dict[str, str]: primitive_types = ",".join( sorted({str(item.get("type", "unknown")) for item in primitives}) ) sensor_group_set = { self._sensor_group_for_primitive( str(item.get("type", "")), device_profile, ) for item in primitives } sensor_groups = ",".join(sorted(sensor_group_set)) times = [str(item.get("time")) for item in primitives if item.get("time")] first_time = min(times, default=client_clock_at_batch) profile_fingerprint = self._hash16_device_profile(device_profile) surface = self._surface_for_product(product) plan_payload = collection_plan or {} plan_id = str(plan_payload.get("inference_plan_id") or "unknown") sensor_tier_value = plan_payload.get("sensor_tier") sensor_tier = str( sensor_tier_value if sensor_tier_value is not None else "unknown" ) cadence = str( plan_payload.get("ingest_min_interval_seconds") or "unknown" ) offline_buffer_required = bool( plan_payload.get("offline_buffer_required", False) ) buffer_mode = "required" if offline_buffer_required else "none" buffer_seconds = str( plan_payload.get("offline_buffer_min_seconds", 0) if offline_buffer_required else 0 ) camera_fps = plan_payload.get("target_camera_fps") camera_fps_field = "" camera_resolution = plan_payload.get("target_camera_resolution") camera_res_field = "" if sensor_group_set & {"camera_visual", "paper_exam_cv"}: camera_fps_field = ( f"camera_fps={int(camera_fps)}:" if isinstance(camera_fps, int) and camera_fps > 0 else "" ) if ( isinstance(camera_resolution, (list, tuple)) and len(camera_resolution) == 2 and all(isinstance(item, int) and item > 0 for item in camera_resolution) ): camera_res_field = ( f"camera_res={int(camera_resolution[0])}x" f"{int(camera_resolution[1])}:" ) base = ( f"{surface}:SDK_CLIENT:PRIMITIVE_UPLOAD:" f"TIER_{sensor_tier}:PRIMITIVE_UPLOAD" ) collection_lineage = ( f"ingest-collection-plan:{base}:local=false:upload=true:" f"plan={plan_id}:cadence={cadence}:buffer={buffer_mode}:" f"buffer_seconds={buffer_seconds}:{camera_fps_field}{camera_res_field}" f"sensors={sensor_groups}:" f"primitives={primitive_types}" ) runtime_lineage = ( f"ingest-runtime-attestation:{base}:observed={first_time}:" f"attested={client_clock_at_batch}:profile={profile_fingerprint}:" f"upload=true:foreground=true:visible=true:raw=false:" f"plan={plan_id}:cadence={cadence}:buffer={buffer_mode}:" f"buffer_seconds={buffer_seconds}:{camera_fps_field}{camera_res_field}" f"sensors={sensor_groups}:" f"primitives={primitive_types}" ) return { "collection_plan_lineage_id": collection_lineage, "runtime_attestation_lineage_id": runtime_lineage, "runtime_attestation_batch_lineage_id": ( "ingest-runtime-attestation-batch:" f"{self._ingest_artifact(uuid, primitives)}:" f"1:{self._hash16_text(runtime_lineage)}" ), } def health(self) -> JsonObject: """GET /health -- no auth required.""" return self._request("GET", "/health") def start_session( self, uuid: str, product: str, *, client_id: str, session_type: str, device_profile: JsonObject, sensor_tier: int, active_primitives: list[str], consent_flags: dict[str, bool], tenant_id: str | None = None, session_id: str | None = None, sdk_v: str = "python-sample/1.0.0", protocol_v: int = PROTOCOL_VERSION, started_at: str | None = None, study_id: str | None = None, ) -> JsonObject: """POST /api/v1/sessions/start and validate the safe collection plan.""" started_at = started_at or self._now_iso_ms() body: dict[str, Any] = { "uuid": uuid, "session_id": session_id or self._make_session_id(uuid, started_at), "client_id": client_id, "session_type": session_type, "product": product, "device_profile": device_profile, "sensor_tier": sensor_tier, "active_primitives": active_primitives, "consent_flags": consent_flags, "sdk_v": sdk_v, "protocol_v": protocol_v, "started_at": started_at, } if tenant_id is not None: body["tenant_id"] = tenant_id if study_id is not None: body["study_id"] = study_id response = self._request("POST", "/api/v1/sessions/start", json=body) collection_plan = self._validate_collection_plan(response) analysis_context = self._validate_analysis_context(response) self._sessions[self._session_key(uuid, product)] = { "session_id": response.get("session_id") or body["session_id"], "tenant_id": response.get("tenant_id") or tenant_id, "device_profile": device_profile, "accepted_primitives": set(response.get("accepted_primitives") or []), "collection_plan": collection_plan, "analysis_context": analysis_context, } return response @staticmethod def _default_ingest_context(session: JsonObject) -> JsonObject: analysis_context = session.get("analysis_context") activity_context = "unknown" if isinstance(analysis_context, dict): candidate = analysis_context.get("activity_context") if isinstance(candidate, str) and candidate: activity_context = candidate return {"context_v": 1, "sessionType": activity_context} @staticmethod def quizpang_response_time_primitive( *, stimulus_display_ts: str, response_ts: str, tier: int = 0, ) -> JsonObject: """Build a Focuspang quizpang `response_time_ms` source primitive. Use the returned primitive in `ingest(..., primitives=[...])` inside a quizpang/quiz session context so the server can materialize response_time_variance and premature_response_rate. """ stimulus_time = DrSimonClient._parse_quizpang_primitive_timestamp( stimulus_display_ts, "stimulus_display_ts", ) response_time = DrSimonClient._parse_quizpang_primitive_timestamp( response_ts, "response_ts", ) response_ms = (response_time - stimulus_time).total_seconds() * 1000.0 if not ( QUIZPANG_RESPONSE_TIME_MIN_MS <= response_ms <= QUIZPANG_RESPONSE_TIME_MAX_MS ): DrSimonClient._raise_quizpang_response_time_invalid( "response_ts", "response_time_ms must be between 100 and 30000 ms", ) if not isinstance(tier, int) or isinstance(tier, bool) or not 0 <= tier <= 3: DrSimonClient._raise_quizpang_response_time_invalid( "tier", "tier must be an integer from 0 to 3", ) return { "time": response_ts, "type": QUIZPANG_RESPONSE_TIME_PRIMITIVE_TYPE, "value": round(response_ms, 4), "tier": tier, } @staticmethod def task_lifecycle_count_primitives( *, observed_at: str, started_count: int, abandoned_count: int, tier: int = 0, ) -> list[JsonObject]: """Build task lifecycle source-count primitives for Focuspang activities. Use these rows in assignment, homework, performance-assessment, or similar task session contexts. The SDK emits only source counts; the server materializes task_avoidance_score when enough evidence is present. """ DrSimonClient._parse_task_lifecycle_primitive_timestamp( observed_at, "observed_at", ) started = DrSimonClient._validate_task_lifecycle_count( "started_count", started_count, min_count=TASK_LIFECYCLE_MIN_STARTED_COUNT, ) abandoned = DrSimonClient._validate_task_lifecycle_count( "abandoned_count", abandoned_count, min_count=0, ) if abandoned > started: DrSimonClient._raise_task_lifecycle_primitive_invalid( "abandoned_count", "abandoned_count must be less than or equal to started_count", ) if not isinstance(tier, int) or isinstance(tier, bool) or not 0 <= tier <= 3: DrSimonClient._raise_task_lifecycle_primitive_invalid( "tier", "tier must be an integer from 0 to 3", ) return [ { "time": observed_at, "type": TASK_LIFECYCLE_STARTED_PRIMITIVE_TYPE, "value": float(started), "tier": tier, }, { "time": observed_at, "type": TASK_LIFECYCLE_ABANDONED_PRIMITIVE_TYPE, "value": float(abandoned), "tier": tier, }, ] @staticmethod def interaction_taxonomy_count_primitives( *, observed_at: str, social_interaction_count: int, total_interaction_count: int, tier: int = 0, ) -> list[JsonObject]: """Build public-safe app interaction taxonomy source-count primitives. Use these rows for classroom, assignment, quizpang, survey, or other Focuspang activity windows where the product can classify social interactions without sending raw chat/comment/share text. """ DrSimonClient._parse_interaction_taxonomy_primitive_timestamp( observed_at, "observed_at", ) social = DrSimonClient._validate_interaction_taxonomy_count( "social_interaction_count", social_interaction_count, min_count=0, max_count=INTERACTION_TAXONOMY_MAX_SOCIAL_COUNT, ) total = DrSimonClient._validate_interaction_taxonomy_count( "total_interaction_count", total_interaction_count, min_count=INTERACTION_TAXONOMY_MIN_TOTAL_COUNT, max_count=INTERACTION_TAXONOMY_MAX_TOTAL_COUNT, ) if social > total: DrSimonClient._raise_interaction_taxonomy_primitive_invalid( "social_interaction_count", "social_interaction_count must be less than or equal to " "total_interaction_count", ) if not isinstance(tier, int) or isinstance(tier, bool) or not 0 <= tier <= 3: DrSimonClient._raise_interaction_taxonomy_primitive_invalid( "tier", "tier must be an integer from 0 to 3", ) return [ { "time": observed_at, "type": INTERACTION_TAXONOMY_SOCIAL_PRIMITIVE_TYPE, "value": float(social), "tier": tier, }, { "time": observed_at, "type": INTERACTION_TAXONOMY_TOTAL_PRIMITIVE_TYPE, "value": float(total), "tier": tier, }, ] @staticmethod def activity_count_primitive( *, observed_at: str, activity_count: int, tier: int = 0, ) -> JsonObject: """Build one public-safe activity-count source primitive. Use this for app-defined activity windows such as foreground activity, typing, touch, scroll, or classroom interaction counts. The server bins accepted rows by UTC day and 4-hour segment before materializing diurnal_activity_flatness. """ DrSimonClient._parse_activity_count_primitive_timestamp( observed_at, "observed_at", ) count = DrSimonClient._validate_activity_count( "activity_count", activity_count, ) if not isinstance(tier, int) or isinstance(tier, bool) or not 0 <= tier <= 3: DrSimonClient._raise_activity_count_primitive_invalid( "tier", "tier must be an integer from 0 to 3", ) return { "time": observed_at, "type": ACTIVITY_COUNT_PRIMITIVE_TYPE, "value": float(count), "tier": tier, } def ingest( self, uuid: str, product: str, primitives: list[JsonObject], **extra: Any, ) -> JsonObject: """POST /api/v1/engine/ingest""" self._reject_server_owned_ingest_fields(primitives, extra) self._reject_raw_media_ingest_fields(primitives, extra) session = self._sessions.get(self._session_key(uuid, product)) if session is not None: extra.setdefault("session_id", session["session_id"]) if session.get("tenant_id"): extra.setdefault("tenant_id", session["tenant_id"]) extra.setdefault("device_profile", session["device_profile"]) if extra.get("context") is None: extra["context"] = self._default_ingest_context(session) self._validate_primitives_against_session_plan(primitives, session) client_clock_at_batch = extra.setdefault( "client_clock_at_batch", self._now_iso_ms(), ) batch_id = str(extra.setdefault("batch_id", self._hash16({ "uuid": uuid, "product": product, "clock": client_clock_at_batch, "primitives": primitives, }).upper().ljust(26, "0")[:26])) device_profile = extra.get("device_profile") collection_plan = ( session.get("collection_plan") if session is not None else None ) if not all( key in extra for key in ( "collection_plan_lineage_id", "runtime_attestation_lineage_id", "runtime_attestation_batch_lineage_id", ) ): extra.update( self._build_ingest_lineage( uuid=uuid, product=product, primitives=primitives, device_profile=device_profile, batch_id=batch_id, client_clock_at_batch=client_clock_at_batch, collection_plan=collection_plan, ) ) body: dict[str, Any] = { "uuid": uuid, "product": product, "primitives": primitives, **extra, } return self._request("POST", "/api/v1/engine/ingest", json=body) def end_session( self, uuid: str, product: str, *, session_id: str | None = None, ended_reason: str = "user_terminated", final_buffer_flushed: bool = True, total_primitives_sent: int = 0, ended_at: str | None = None, client_clock_at_end: str | None = None, ) -> JsonObject: """POST /api/v1/sessions/end and close the cached session plan.""" session_key = self._session_key(uuid, product) session = self._sessions.get(session_key) resolved_session_id = session_id or ( str(session["session_id"]) if session is not None else None ) if not resolved_session_id: raise DrSimonError( 400, "session_id is required when no cached session exists", payload={"error": "session_id_required"}, ) ended_at = ended_at or self._now_iso_ms() client_clock_at_end = client_clock_at_end or ended_at body: JsonObject = { "session_id": resolved_session_id, "product": product, "ended_at": ended_at, "ended_reason": ended_reason, "final_buffer_flushed": final_buffer_flushed, "total_primitives_sent": total_primitives_sent, "client_clock_at_end": client_clock_at_end, } response = self._request("POST", "/api/v1/sessions/end", json=body) self._sessions.pop(session_key, None) return response def start_proctoring_session( self, *, session_id: str, client_id: str, product: str = "edu", started_at: float | None = None, session_type: str = "exam_digital", device_profile: JsonObject | None = None, sensor_tier: int | None = None, consent_flags: dict[str, bool] | None = None, tenant_id: str | None = None, uuid: str | None = None, study_id: str | None = None, seat_layout_positions: JsonObject | None = None, seat_id: str | None = None, session_end_ts: float | None = None, ) -> JsonObject: """POST /api/v1/engine/proctoring/session/start.""" body: dict[str, Any] = { "product": product, "session_id": session_id, "client_id": client_id, "started_at": started_at if started_at is not None else time.time(), "session_type": session_type, } optional_fields = { "device_profile": device_profile, "sensor_tier": sensor_tier, "consent_flags": consent_flags, "tenant_id": tenant_id, "uuid": uuid, "study_id": study_id, "seat_layout_positions": seat_layout_positions, "seat_id": seat_id, "session_end_ts": session_end_ts, } body.update({ key: value for key, value in optional_fields.items() if value is not None }) response = self._request( "POST", "/api/v1/engine/proctoring/session/start", json=body, ) self._validate_collection_plan(response) self._validate_analysis_context(response) return response def proctoring_heartbeat( self, *, session_id: str, client_id: str, timestamp: float, pi: JsonObject, product: str = "edu", ) -> JsonObject: """POST /api/v1/engine/proctoring/heartbeat.""" return self._request( "POST", "/api/v1/engine/proctoring/heartbeat", json={ "product": product, "session_id": session_id, "client_id": client_id, "timestamp": timestamp, "pi": pi, }, ) def end_proctoring_session( self, *, session_id: str, client_id: str, product: str = "edu", ended_at: float | None = None, final_buffer_flushed: bool = True, ) -> JsonObject: """POST /api/v1/engine/proctoring/session/end.""" return self._request( "POST", "/api/v1/engine/proctoring/session/end", json={ "product": product, "session_id": session_id, "client_id": client_id, "ended_at": ended_at if ended_at is not None else time.time(), "final_buffer_flushed": final_buffer_flushed, }, ) def classify_focustime( self, uuid: str, product: str = "edu", *, focus_ratio: float | None = None, pis: JsonObject | None = None, attention_context: JsonObject | None = None, ) -> JsonObject: """POST /api/v1/engine/focustime/classify""" body: JsonObject = { "uuid": uuid, "product": product, "pis": dict(pis or {}), } if focus_ratio is not None: body["focusRatio"] = focus_ratio if attention_context is not None: body["attentionContext"] = dict(attention_context) return self._request( "POST", "/api/v1/engine/focustime/classify", json=body, ) @staticmethod def _validate_primitives_against_session_plan( primitives: list[JsonObject], session: JsonObject, ) -> None: collection_plan = session["collection_plan"] max_rows = int(collection_plan["ingest_batch_max_primitives"]) if len(primitives) > max_rows: raise DrSimonError( 400, f"batch exceeds collection_plan limit of {max_rows} primitives", payload={ "error": "collection_plan_batch_limit_exceeded", "limit": max_rows, }, ) accepted = session.get("accepted_primitives") or set() if not accepted: return rejected = [ str(item.get("type")) for item in primitives if str(item.get("type")) not in accepted ] if rejected: raise DrSimonError( 400, "primitive not accepted by session collection_plan", payload={ "error": "session_plan_primitive_not_accepted", "rejected_primitives": rejected, }, ) @staticmethod def _reject_server_owned_ingest_fields( primitives: list[JsonObject], extra: dict[str, Any], ) -> None: rejected = DrSimonClient._ingest_field_paths( extra, SERVER_OWNED_INGEST_FIELDS, ) for index, primitive in enumerate(primitives): if not isinstance(primitive, dict): continue rejected.extend( DrSimonClient._ingest_field_paths( primitive, SERVER_OWNED_INGEST_FIELDS, root_path=f"primitives[{index}]", ) ) if not rejected: return reasons = [ ( f"field {field} is a server-owned output; SDK must omit it and " "let the engine derive scores from accepted primitives" ) for field in rejected ] message = ( "client-supplied " + "/".join(rejected) + " are not accepted on ingest; omit them and let the server derive " "scores from accepted primitives" ) raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "server_owned_outputs_not_ingest_inputs", "rejected_fields": rejected, "rejection_reasons": reasons, "updatedPrimitives": 0, }, ) @staticmethod def _reject_raw_media_ingest_fields( primitives: list[JsonObject], extra: dict[str, Any], ) -> None: rejected = DrSimonClient._raw_media_ingest_field_paths(extra) for index, primitive in enumerate(primitives): if not isinstance(primitive, dict): continue rejected.extend( DrSimonClient._raw_media_ingest_field_paths( primitive, root_path=f"primitives[{index}]", ) ) if not rejected: return reasons = [ ( f"field {field} contains raw media or raw sensor payload; " "SDK ingest must send derived primitives only" ) for field in rejected ] message = ( "raw media payloads are not accepted on ingest; run local processing " "and send derived primitives only" ) raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "raw_media_payload_not_ingest_input", "rejected_fields": rejected, "rejection_reasons": reasons, "updatedPrimitives": 0, }, ) @staticmethod def _raw_media_ingest_field_paths( value: Any, *, root_path: str = "", ) -> list[str]: return DrSimonClient._ingest_field_paths( value, RAW_MEDIA_INGEST_PRIMITIVE_FIELDS, root_path=root_path, ) @staticmethod def _ingest_field_paths( value: Any, field_names: tuple[str, ...], *, root_path: str = "", ) -> list[str]: field_set = set(field_names) rejected: list[str] = [] seen: set[int] = set() def walk(current: Any, path: str) -> None: if isinstance(current, dict): object_id = id(current) if object_id in seen: return seen.add(object_id) for key, child in current.items(): key_text = str(key) child_path = f"{path}.{key_text}" if path else key_text if key_text in field_set and child is not None: rejected.append(child_path) continue walk(child, child_path) return if isinstance(current, (list, tuple)): object_id = id(current) if object_id in seen: return seen.add(object_id) for index, child in enumerate(current): walk(child, f"{path}[{index}]") walk(value, root_path) return rejected def analysis( self, uuid: str, product: str, *, result_type: str = "realtime", activity_context: str | None = None, assessment_modality: str | None = None, ) -> JsonObject: """GET /api/v1/engine/analysis/{uuid}""" params = {"product": product, "type": result_type} if activity_context is not None: params["activity_context"] = activity_context if assessment_modality is not None: params["assessment_modality"] = assessment_modality response = self._request( "GET", f"/api/v1/engine/analysis/{uuid}", params=params, ) analysis_context = self._validate_analysis_response_context( response, expected_activity_context=activity_context, expected_assessment_modality=assessment_modality, ) self._validate_analysis_contextual_evidence(response) self._validate_analysis_score_evidence( response, analysis_context=analysis_context, ) self._validate_analysis_item_evidence_context( response, analysis_context=analysis_context, ) self._validate_analysis_cognitive_construct_evidence( response, analysis_context=analysis_context, ) return response def params(self, uuid: str, product: str) -> JsonObject: """GET /api/v1/engine/params/{uuid}""" return self._request("GET", f"/api/v1/engine/params/{uuid}", params={"product": product}) def trend( self, uuid: str, product: str, *, activity_context: str | None = None, assessment_modality: str | None = None, ) -> JsonObject: """GET /api/v1/engine/trend/{uuid}""" params = {"product": product} if activity_context is not None: params["activity_context"] = activity_context if assessment_modality is not None: params["assessment_modality"] = assessment_modality response = self._request( "GET", f"/api/v1/engine/trend/{uuid}", params=params, ) analysis_context = self._validate_analysis_response_context( response, expected_activity_context=activity_context, expected_assessment_modality=assessment_modality, ) self._validate_analysis_contextual_evidence(response) self._validate_analysis_item_evidence_context( response, analysis_context=analysis_context, ) return response def anomaly( self, uuid: str, product: str, *, z_warning: float = 2.0, z_critical: float = 3.0, activity_context: str | None = None, assessment_modality: str | None = None, ) -> JsonObject: """GET /api/v1/engine/anomaly/{uuid}""" params: dict[str, str | float] = { "product": product, "z_warning": z_warning, "z_critical": z_critical, } if activity_context is not None: params["activity_context"] = activity_context if assessment_modality is not None: params["assessment_modality"] = assessment_modality response = self._request( "GET", f"/api/v1/engine/anomaly/{uuid}", params=params, ) analysis_context = self._validate_analysis_response_context( response, expected_activity_context=activity_context, expected_assessment_modality=assessment_modality, ) self._validate_analysis_contextual_evidence(response) self._validate_analysis_item_evidence_context( response, analysis_context=analysis_context, ) return response def factor_matrix(self, uuid: str, product: str) -> JsonObject: """GET /api/v1/engine/factor-matrix/{uuid}""" return self._request("GET", f"/api/v1/engine/factor-matrix/{uuid}", params={"product": product}) def factor_profile(self, uuid: str, product: str) -> JsonObject: """GET /api/v1/engine/factor-profile/{uuid}""" return self._request("GET", f"/api/v1/engine/factor-profile/{uuid}", params={"product": product}) def record_attention_scene_transition( self, *, group_id: str, set_by: str, attention_scene: JsonObject, product: str = "edu", tenant_id: str | None = None, session_id: str | None = None, actor_id: str | None = None, source: str = "backend", scene_def_version: str | None = None, server_ts: str | None = None, client_ts: str | None = None, ttl_expires_at: str | None = None, ) -> JsonObject: """POST /api/v1/engine/attention-scene/transitions""" body: JsonObject = { "product": self._validate_scene_transition_product(product), "groupId": self._validate_scene_transition_public_id( group_id, "groupId", ), "setBy": self._validate_scene_transition_public_id(set_by, "setBy"), "source": self._validate_scene_transition_source(source), "attentionScene": self._validate_scene_transition_attention_scene( attention_scene ), "serverTs": self._validate_scene_transition_timestamp( server_ts or self._now_iso_ms(), "serverTs", ), } if tenant_id is not None: body["tenantId"] = self._validate_scene_transition_public_id( tenant_id, "tenantId", max_length=100, ) if session_id is not None: body["sessionId"] = self._validate_scene_transition_public_id( session_id, "sessionId", pattern=SCENE_TRANSITION_SESSION_ID_RE, max_length=256, ) if actor_id is not None: body["actorId"] = self._validate_scene_transition_public_id( actor_id, "actorId", max_length=100, ) if scene_def_version is not None: body["sceneDefVersion"] = self._validate_scene_def_version( scene_def_version, "sceneDefVersion", ) if client_ts is not None: body["clientTs"] = self._validate_scene_transition_timestamp( client_ts, "clientTs", ) if ttl_expires_at is not None: body["ttlExpiresAt"] = self._validate_scene_transition_timestamp( ttl_expires_at, "ttlExpiresAt", ) return self._request( "POST", "/api/v1/engine/attention-scene/transitions", json=body, ) def recompute_session_segments( self, session_id: str, product: str, *, force: bool = False, ) -> JsonObject: """POST /api/v1/engine/sessions/{session_id}/segments/recompute""" return self._request( "POST", f"/api/v1/engine/sessions/{session_id}/segments/recompute", params={"product": product, "force": force}, ) def segment_evidence_report( self, session_id: str, product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, ) -> JsonObject: """GET /api/v1/engine/sessions/{session_id}/segments/evidence-report""" params: dict[str, str | int] = {"product": product} if algo_version is not None: params["algo_version"] = algo_version if activity_context is not None: params["activity_context"] = activity_context if assessment_modality is not None: params["assessment_modality"] = assessment_modality return self._request( "GET", f"/api/v1/engine/sessions/{session_id}/segments/evidence-report", params=params, ) @staticmethod def _segment_consumer_report_params( product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, jurisdiction: str = "kr", audience: str = "teacher", guardian_consent: bool = False, cross_product_consent: bool = False, special_category_basis: bool = False, dpia_completed: bool = False, ferpa_compliant_agreement: bool = False, ) -> dict[str, str | int | bool]: params: dict[str, str | int | bool] = { "product": product, "jurisdiction": jurisdiction, "audience": audience, "guardian_consent": guardian_consent, "cross_product_consent": cross_product_consent, "special_category_basis": special_category_basis, "dpia_completed": dpia_completed, "ferpa_compliant_agreement": ferpa_compliant_agreement, } if algo_version is not None: params["algo_version"] = algo_version if activity_context is not None: params["activity_context"] = activity_context if assessment_modality is not None: params["assessment_modality"] = assessment_modality return params def teacher_segment_report( self, session_id: str, product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, jurisdiction: str = "kr", audience: str = "teacher", guardian_consent: bool = False, cross_product_consent: bool = False, special_category_basis: bool = False, dpia_completed: bool = False, ferpa_compliant_agreement: bool = False, ) -> JsonObject: """GET /api/v1/engine/sessions/{session_id}/segments/teacher-report""" params = self._segment_consumer_report_params( product, algo_version=algo_version, activity_context=activity_context, assessment_modality=assessment_modality, jurisdiction=jurisdiction, audience=audience, guardian_consent=guardian_consent, cross_product_consent=cross_product_consent, special_category_basis=special_category_basis, dpia_completed=dpia_completed, ferpa_compliant_agreement=ferpa_compliant_agreement, ) return self._request( "GET", f"/api/v1/engine/sessions/{session_id}/segments/teacher-report", params=params, ) def consumer_segment_report( self, session_id: str, product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, jurisdiction: str = "kr", audience: str = "student", guardian_consent: bool = False, cross_product_consent: bool = False, special_category_basis: bool = False, dpia_completed: bool = False, ferpa_compliant_agreement: bool = False, ) -> JsonObject: """GET /api/v1/engine/sessions/{session_id}/segments/consumer-report""" params = self._segment_consumer_report_params( product, algo_version=algo_version, activity_context=activity_context, assessment_modality=assessment_modality, jurisdiction=jurisdiction, audience=audience, guardian_consent=guardian_consent, cross_product_consent=cross_product_consent, special_category_basis=special_category_basis, dpia_completed=dpia_completed, ferpa_compliant_agreement=ferpa_compliant_agreement, ) return self._request( "GET", f"/api/v1/engine/sessions/{session_id}/segments/consumer-report", params=params, ) def segment_llm_report_handoff( self, session_id: str, product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, jurisdiction: str = "kr", audience: str = "student", guardian_consent: bool = False, cross_product_consent: bool = False, special_category_basis: bool = False, dpia_completed: bool = False, ferpa_compliant_agreement: bool = False, ) -> JsonObject: """GET /api/v1/engine/sessions/{session_id}/segments/llm-report-handoff""" params = self._segment_consumer_report_params( product, algo_version=algo_version, activity_context=activity_context, assessment_modality=assessment_modality, jurisdiction=jurisdiction, audience=audience, guardian_consent=guardian_consent, cross_product_consent=cross_product_consent, special_category_basis=special_category_basis, dpia_completed=dpia_completed, ferpa_compliant_agreement=ferpa_compliant_agreement, ) return self._request( "GET", f"/api/v1/engine/sessions/{session_id}/segments/llm-report-handoff", params=params, ) def segment_llm_report_draft( self, session_id: str, product: str, *, algo_version: int | None = None, activity_context: str | None = None, assessment_modality: str | None = None, jurisdiction: str = "kr", audience: str = "student", guardian_consent: bool = False, cross_product_consent: bool = False, special_category_basis: bool = False, dpia_completed: bool = False, ferpa_compliant_agreement: bool = False, ) -> JsonObject: """POST /api/v1/engine/sessions/{session_id}/segments/llm-report-draft""" params = self._segment_consumer_report_params( product, algo_version=algo_version, activity_context=activity_context, assessment_modality=assessment_modality, jurisdiction=jurisdiction, audience=audience, guardian_consent=guardian_consent, cross_product_consent=cross_product_consent, special_category_basis=special_category_basis, dpia_completed=dpia_completed, ferpa_compliant_agreement=ferpa_compliant_agreement, ) return self._request( "POST", f"/api/v1/engine/sessions/{session_id}/segments/llm-report-draft", params=params, ) def focuspang_weekly_report( self, uuid: str, week_start: str, **options: Any, ) -> JsonObject: """GET /api/v1/reports/v1/focuspang/student/weekly""" params = self._build_report_request_params( uuid=uuid, date_field="week_start", date_value=week_start, options=options, allowed_fields=REPORT_FOCUSPANG_WEEKLY_OPTION_FIELDS, ) return self._request( "GET", "/api/v1/reports/v1/focuspang/student/weekly", params=params, ) def moment_daily_report( self, uuid: str, report_date: str, **options: Any, ) -> JsonObject: """GET /api/v1/reports/v1/moment/child/daily""" params = self._build_report_request_params( uuid=uuid, date_field="date", date_value=report_date, options=options, allowed_fields=REPORT_MOMENT_DAILY_OPTION_FIELDS, ) return self._request( "GET", "/api/v1/reports/v1/moment/child/daily", params=params, ) @classmethod def _build_report_request_params( cls, *, uuid: str, date_field: str, date_value: str, options: dict[str, Any], allowed_fields: frozenset[str], ) -> dict[str, str | bool]: params: dict[str, str | bool] = { "uuid": uuid, date_field: cls._validate_report_date(date_value, date_field), } for field, value in options.items(): if value is None: continue if field not in allowed_fields: cls._raise_report_request_query_invalid( field, f"{field} is not a supported public report query parameter", ) params[field] = cls._validate_report_option(field, value) return params @classmethod def _validate_report_option(cls, field: str, value: Any) -> str | bool: if field in REPORT_REQUEST_BOOLEAN_FIELDS: if not isinstance(value, bool): cls._raise_report_request_query_invalid( field, f"{field} must be a boolean", ) return value if field == "audience": if not isinstance(value, str) or value not in REPORT_REQUEST_AUDIENCES: cls._raise_report_request_query_invalid( field, "audience must be student, teacher, or parent", ) return value if field == "jurisdiction": return cls._validate_report_public_token( value, field, pattern=REPORT_REQUEST_JURISDICTION_RE, max_length=16, ).lower() if field in {"tenant_id", "policy_assertion_id"}: return cls._validate_report_public_token(value, field) cls._raise_report_request_query_invalid( field, f"{field} is not a supported public report query parameter", ) @staticmethod def _validate_report_date(value: str, field: str) -> str: if not isinstance(value, str) or REPORT_REQUEST_DATE_RE.fullmatch(value) is None: DrSimonClient._raise_report_request_query_invalid( field, f"{field} must be an ISO date", ) return value @staticmethod def _validate_report_public_token( value: Any, field: str, *, pattern: re.Pattern[str] = REPORT_REQUEST_PUBLIC_ID_RE, max_length: int = 128, ) -> str: if not isinstance(value, str): DrSimonClient._raise_report_request_query_invalid( field, f"{field} must be a public identifier", ) token = value.strip() lowered = token.lower() if ( not token or len(token) > max_length or pattern.fullmatch(token) is None or any( lowered.startswith(prefix) for prefix in SCENE_TRANSITION_PRIVATE_ID_PREFIXES ) ): DrSimonClient._raise_report_request_query_invalid( field, f"{field} must be a public identifier", ) return token @staticmethod def _raise_report_request_query_invalid( field: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "report_request_query_invalid", "rejected_fields": [field], "rejection_reasons": [message], }, ) @staticmethod def _parse_quizpang_primitive_timestamp(value: str, field: str) -> datetime: if ( not isinstance(value, str) or SCENE_TRANSITION_ISO_UTC_MS_RE.fullmatch(value) is None ): DrSimonClient._raise_quizpang_response_time_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: DrSimonClient._raise_quizpang_response_time_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) if parsed.tzinfo is None: DrSimonClient._raise_quizpang_response_time_invalid( field, f"{field} must be timezone-aware UTC", ) return parsed.astimezone(timezone.utc) @staticmethod def _raise_quizpang_response_time_invalid( field: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "quizpang_response_time_primitive_invalid", "rejected_fields": [field], "rejection_reasons": [message], }, ) @staticmethod def _parse_task_lifecycle_primitive_timestamp(value: str, field: str) -> datetime: if ( not isinstance(value, str) or SCENE_TRANSITION_ISO_UTC_MS_RE.fullmatch(value) is None ): DrSimonClient._raise_task_lifecycle_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: DrSimonClient._raise_task_lifecycle_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) if parsed.tzinfo is None: DrSimonClient._raise_task_lifecycle_primitive_invalid( field, f"{field} must be timezone-aware UTC", ) return parsed.astimezone(timezone.utc) @staticmethod def _validate_task_lifecycle_count( field: str, value: int, *, min_count: int, ) -> int: if ( not isinstance(value, int) or isinstance(value, bool) or value < min_count or value > TASK_LIFECYCLE_MAX_COUNT ): DrSimonClient._raise_task_lifecycle_primitive_invalid( field, f"{field} must be an integer from {min_count} to " f"{TASK_LIFECYCLE_MAX_COUNT}", ) return value @staticmethod def _raise_task_lifecycle_primitive_invalid( field: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "task_lifecycle_primitive_invalid", "rejected_fields": [field], "rejection_reasons": [message], }, ) @staticmethod def _parse_interaction_taxonomy_primitive_timestamp( value: str, field: str, ) -> datetime: if ( not isinstance(value, str) or SCENE_TRANSITION_ISO_UTC_MS_RE.fullmatch(value) is None ): DrSimonClient._raise_interaction_taxonomy_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: DrSimonClient._raise_interaction_taxonomy_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) if parsed.tzinfo is None: DrSimonClient._raise_interaction_taxonomy_primitive_invalid( field, f"{field} must be timezone-aware UTC", ) return parsed.astimezone(timezone.utc) @staticmethod def _validate_interaction_taxonomy_count( field: str, value: int, *, min_count: int, max_count: int, ) -> int: if ( not isinstance(value, int) or isinstance(value, bool) or value < min_count or value > max_count ): DrSimonClient._raise_interaction_taxonomy_primitive_invalid( field, f"{field} must be an integer from {min_count} to {max_count}", ) return value @staticmethod def _raise_interaction_taxonomy_primitive_invalid( field: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "interaction_taxonomy_primitive_invalid", "rejected_fields": [field], "rejection_reasons": [message], }, ) @staticmethod def _parse_activity_count_primitive_timestamp(value: str, field: str) -> datetime: if ( not isinstance(value, str) or SCENE_TRANSITION_ISO_UTC_MS_RE.fullmatch(value) is None ): DrSimonClient._raise_activity_count_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: DrSimonClient._raise_activity_count_primitive_invalid( field, f"{field} must be an ISO-8601 UTC millisecond timestamp", ) if parsed.tzinfo is None: DrSimonClient._raise_activity_count_primitive_invalid( field, f"{field} must be timezone-aware UTC", ) return parsed.astimezone(timezone.utc) @staticmethod def _validate_activity_count(field: str, value: int) -> int: if ( not isinstance(value, int) or isinstance(value, bool) or value < ACTIVITY_COUNT_MIN_COUNT or value > ACTIVITY_COUNT_MAX_COUNT ): DrSimonClient._raise_activity_count_primitive_invalid( field, f"{field} must be an integer from {ACTIVITY_COUNT_MIN_COUNT} " f"to {ACTIVITY_COUNT_MAX_COUNT}", ) return value @staticmethod def _raise_activity_count_primitive_invalid( field: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "activity_count_primitive_invalid", "rejected_fields": [field], "rejection_reasons": [message], }, ) @staticmethod def factor_quality_for(report: JsonObject, factor: str) -> JsonObject: """Return UI-safe sensor coverage metadata for one report factor.""" factors = report.get("factors") if isinstance(report, dict) else {} factor_payload = ( factors.get(factor) if isinstance(factors, dict) else {} ) if not isinstance(factor_payload, dict): factor_payload = {} data_quality = report.get("data_quality") if isinstance(report, dict) else {} if not isinstance(data_quality, dict): data_quality = {} missing_by_factor = data_quality.get("missing_weighted_primitives_by_factor") missing_from_quality = ( missing_by_factor.get(factor) if isinstance(missing_by_factor, dict) else None ) missing = factor_payload.get("missing_weighted_primitives") if missing is None: missing = missing_from_quality if not isinstance(missing, list): missing = [] reason_codes = factor_payload.get("confidence_reason_codes") or [] return { "factor": factor, "confidence": factor_payload.get("confidence"), "confidence_score": factor_payload.get("factor_quality_confidence_score"), "coverage_score": factor_payload.get("factor_quality_coverage_score"), "confidence_label": factor_payload.get("factor_quality_confidence_label"), "missing_weighted_primitives": missing, "disclosure": factor_payload.get("factor_quality_disclosure") or {}, "capped_by_factor_quality": bool( factor_payload.get("confidence_capped_by_factor_quality") ), "reason_codes": reason_codes if isinstance(reason_codes, list) else [], "has_sensor_coverage_limitation": ( REPORT_FACTOR_QUALITY_REASON in reason_codes if isinstance(reason_codes, list) else False ), "policy_version": data_quality.get("factor_quality_policy_version"), } @staticmethod def report_contextual_evidence(report: JsonObject) -> JsonObject: """Return validated report-level evidence boundaries for UI copy.""" disclosures = report.get("disclosures") if isinstance(report, dict) else {} if not isinstance(disclosures, dict): return {} evidence = disclosures.get("contextual_evidence") if not isinstance(evidence, dict): return {} evidence_object = cast(JsonObject, evidence) allowed_uses = evidence_object.get("allowed_downstream_uses") if ( not DrSimonClient._has_report_public_positioning_boundaries( evidence_object ) or not isinstance(allowed_uses, list) or tuple(allowed_uses) != ANALYSIS_ALLOWED_DOWNSTREAM_USES or evidence_object.get("redisclosure_boundary") != "controller_authorized_only" or evidence_object.get("no_raw_sensor_payload") is not True or evidence_object.get("no_clinical_diagnosis") is not True or evidence_object.get("no_adverse_action") is not True ): return {} return dict(evidence_object) @staticmethod def parent_summary_evidence(report: JsonObject) -> JsonObject: """Return evidence-bound report summary card metadata for UI copy.""" disclosures = report.get("disclosures") if isinstance(report, dict) else {} if not isinstance(disclosures, dict): return {} evidence = disclosures.get("parent_summary_evidence") if not isinstance(evidence, dict): return {} evidence_object = cast(JsonObject, evidence) personalization_n = evidence_object.get("personalization_n_observations") observed_factor_count = evidence_object.get("observed_factor_count") context_status = evidence_object.get("context_status") data_quality_reliability = evidence_object.get("data_quality_reliability") if ( evidence_object.get("evidence_policy") != "contextual_multimodal_evidence_required" or evidence_object.get("summary_kind") != "factor_trend_summary" or evidence_object.get("primary_engine_role") not in SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "primary_engine_role" ] or not isinstance(context_status, str) or context_status not in SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "context_status" ] or not isinstance(data_quality_reliability, str) or data_quality_reliability not in DATA_QUALITY_RELIABILITY_VALUES or not isinstance(personalization_n, int) or isinstance(personalization_n, bool) or personalization_n < 0 or not isinstance(observed_factor_count, int) or isinstance(observed_factor_count, bool) or observed_factor_count < 0 or not isinstance(evidence_object.get("context_applied"), bool) or not DrSimonClient._has_evidence_explanation_policy(evidence_object) or not DrSimonClient._has_report_public_positioning_boundaries( evidence_object ) or any( evidence_object.get(flag) is not True for flag in REPORT_SUMMARY_REQUIRED_EVIDENCE_FLAGS ) ): return {} for field in ( "summary_basis", "source_modalities", "behavior_dimensions", "student_growth_domains", "activity_contexts", "assessment_modalities", "limitation_codes", "interpretation_caveats", ): if not DrSimonClient._is_public_token_list( evidence_object.get(field), allow_empty=(field == "limitation_codes"), ): return {} return dict(evidence_object) @staticmethod def cognitive_construct_evidence_summary(report: JsonObject) -> JsonObject: """Return public-safe cognitive construct coverage for report copy.""" disclosures = report.get("disclosures") if isinstance(report, dict) else {} if not isinstance(disclosures, dict): return {} evidence = disclosures.get("cognitive_construct_evidence_summary") if not isinstance(evidence, dict): return {} evidence_object = cast(JsonObject, evidence) missing = [ field for field in REPORT_COGNITIVE_CONSTRUCT_EVIDENCE_SUMMARY_REQUIRED_FIELDS if field not in evidence_object ] context_status = evidence_object.get("context_status") observed_count = evidence_object.get("observed_construct_count") if ( missing or evidence_object.get("cognitive_construct_evidence_summary_v") != 1 or evidence_object.get("evidence_policy") != "contextual_multimodal_evidence_required" or evidence_object.get("summary_kind") != "report_cognitive_construct_coverage" or evidence_object.get("cognitive_construct_policy") != COGNITIVE_CONSTRUCT_POLICY or evidence_object.get("weighting_policy") != COGNITIVE_CONSTRUCT_WEIGHTING_POLICY or not DrSimonClient._public_token_list_matches( evidence_object.get("product_boundaries"), COGNITIVE_CONSTRUCT_PRODUCT_BOUNDARIES, ) or not isinstance(context_status, str) or context_status not in SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "context_status" ] or not isinstance(observed_count, int) or isinstance(observed_count, bool) or observed_count < 0 or any( evidence_object.get(flag) is not True for flag in ( REPORT_COGNITIVE_CONSTRUCT_EVIDENCE_SUMMARY_REQUIRED_FLAGS ) ) ): return {} if not DrSimonClient._is_public_token_list( evidence_object.get("observed_source_primitives"), allow_empty=True, ): return {} observed_source_primitives = cast( list[str], evidence_object["observed_source_primitives"], ) if any( primitive not in COGNITIVE_CONSTRUCT_SOURCE_PRIMITIVE_SET for primitive in observed_source_primitives ): return {} for field, allowed in ( ( "activity_contexts", SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "activity_context" ], ), ( "assessment_modalities", SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES[ "assessment_modality" ], ), ): values = evidence_object.get(field) if not DrSimonClient._is_public_token_list(values) or any( token not in allowed for token in cast(list[str], values) ): return {} constructs = evidence_object.get("constructs") if not isinstance(constructs, list): return {} construct_keys: list[str] = [] observed_construct_keys: set[str] = set() observed_construct_count = 0 for item in constructs: if not isinstance(item, dict): return {} if not DrSimonClient._is_report_cognitive_construct_summary_item( cast(JsonObject, item) ): return {} construct_key = cast(str, item["construct_key"]) construct_keys.append(construct_key) if float(cast(float, item["observed_weight"])) > 0.0: observed_construct_count += 1 observed_construct_keys.add(construct_key) if ( tuple(construct_keys) != SUPPORTED_COGNITIVE_CONSTRUCTS or observed_construct_count != observed_count or DrSimonClient._report_constructs_for_source_primitives( observed_source_primitives ) != observed_construct_keys ): return {} return dict(evidence_object) @staticmethod def _report_constructs_for_source_primitives( source_primitives: list[str], ) -> set[str]: constructs: set[str] = set() for primitive in source_primitives: for construct_key, construct_primitives in ( COGNITIVE_CONSTRUCT_SOURCE_PRIMITIVES.items() ): if primitive in construct_primitives: constructs.add(construct_key) return constructs @staticmethod def _is_report_cognitive_construct_summary_item(item: JsonObject) -> bool: construct_key = item.get("construct_key") science_references = item.get("science_references") warning_codes = item.get("warning_codes") observed_weight = item.get("observed_weight") missing_weight = item.get("missing_weight") return ( isinstance(construct_key, str) and construct_key in SUPPORTED_COGNITIVE_CONSTRUCTS and DrSimonClient._is_ratio(observed_weight) and DrSimonClient._is_ratio(missing_weight) and abs( float(cast(float, observed_weight)) + float(cast(float, missing_weight)) - 1.0 ) <= 0.0001 and item.get("evidence_mode") in COGNITIVE_CONSTRUCT_EVIDENCE_MODES and item.get("confidence_cap") in COGNITIVE_CONSTRUCT_CONFIDENCE_CAPS and DrSimonClient._is_public_token_list(science_references) and not sorted( set(cast(list[str], science_references)) - set(COGNITIVE_CONSTRUCT_SCIENCE_REFERENCES) ) and DrSimonClient._is_public_token_list( warning_codes, allow_empty=True, ) and not sorted( set(cast(list[str], warning_codes)) - set(COGNITIVE_CONSTRUCT_WARNING_CODES) ) ) @staticmethod def segment_scene_contexts(report: JsonObject) -> list[JsonObject]: """Return UI-safe per-segment scene/activity context evidence rows.""" rows = report.get("segmentSceneContexts") if isinstance(report, dict) else [] if not isinstance(rows, list): return [] safe_rows: list[JsonObject] = [] for row in rows: if not isinstance(row, dict): continue candidate = cast(JsonObject, row) safe = DrSimonClient._segment_scene_context_row(candidate) if safe: safe_rows.append(safe) return safe_rows @staticmethod def _segment_scene_context_row(row: JsonObject) -> JsonObject: if any(key in row for key in SEGMENT_SCENE_CONTEXT_FORBIDDEN_KEYS): return {} seg_index = row.get("segIndex") context_source = row.get("contextSource") context_status = row.get("contextStatus") policy = row.get("sceneContextPolicy") activity_context = row.get("activityContext") overlap_seconds = row.get("overlapSeconds") segment_seconds = row.get("segmentSeconds") coverage_ratio = row.get("coverageRatio") caveats = row.get("contextCaveats") if ( not isinstance(seg_index, int) or isinstance(seg_index, bool) or seg_index < 0 or context_source not in SEGMENT_SCENE_CONTEXT_SOURCES or context_status not in SEGMENT_SCENE_CONTEXT_STATUSES or policy != SEGMENT_SCENE_CONTEXT_POLICY or not isinstance(activity_context, str) or activity_context not in SESSION_ANALYSIS_CONTEXT_PUBLIC_FIELD_ALLOWED_VALUES["activity_context"] or not DrSimonClient._non_negative_finite_number(overlap_seconds) or not DrSimonClient._non_negative_finite_number(segment_seconds) or not DrSimonClient._unit_finite_number(coverage_ratio) or not isinstance(caveats, list) or any( not isinstance(item, str) or item not in SEGMENT_SCENE_CONTEXT_CAVEATS for item in caveats ) ): return {} safe: JsonObject = { "segIndex": seg_index, "contextSource": context_source, "contextStatus": context_status, "sceneContextPolicy": policy, "activityContext": activity_context, "overlapSeconds": overlap_seconds, "segmentSeconds": segment_seconds, "coverageRatio": coverage_ratio, "contextCaveats": list(caveats), } scene_source = row.get("sceneSource") if scene_source is not None: if scene_source not in SEGMENT_SCENE_CONTEXT_SCENE_SOURCES: return {} safe["sceneSource"] = scene_source scene_def_version = row.get("sceneDefVersion") if scene_def_version is not None: if ( not isinstance(scene_def_version, str) or SCENE_DEF_VERSION_RE.fullmatch(scene_def_version) is None ): return {} safe["sceneDefVersion"] = scene_def_version attention_scene = row.get("attentionScene") if attention_scene is not None: if not DrSimonClient._safe_attention_scene_object(attention_scene): return {} safe["attentionScene"] = dict(cast(JsonObject, attention_scene)) return safe @staticmethod def _non_negative_finite_number(value: object) -> bool: return ( isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) and float(value) >= 0.0 ) @staticmethod def _unit_finite_number(value: object) -> bool: return ( isinstance(value, (int, float)) and not isinstance(value, bool) and DrSimonClient._non_negative_finite_number(value) and float(value) <= 1.0 ) @staticmethod def _safe_attention_scene_object(value: object) -> bool: if not isinstance(value, dict): return False if any(key in value for key in SEGMENT_SCENE_CONTEXT_FORBIDDEN_KEYS): return False return all( isinstance(key, str) and (item is None or isinstance(item, str)) for key, item in value.items() ) @staticmethod def _validate_scene_transition_product(value: str) -> str: product = DrSimonClient._validate_scene_transition_public_id( value, "product", pattern=SCENE_TRANSITION_PRODUCT_RE, max_length=20, ).lower() if SCENE_TRANSITION_PRODUCT_RE.fullmatch(product) is None: DrSimonClient._raise_scene_transition_payload_invalid( "product", "product must be a public product identifier", ) return product @staticmethod def _validate_scene_transition_public_id( value: object, field_name: str, *, pattern: re.Pattern[str] = SCENE_TRANSITION_PUBLIC_ID_RE, max_length: int = 128, ) -> str: if not isinstance(value, str): DrSimonClient._raise_scene_transition_payload_invalid( field_name, f"{field_name} must be a public identifier", ) stripped = value.strip() lowered = stripped.lower() if ( not stripped or len(stripped) > max_length or pattern.fullmatch(stripped) is None or lowered.startswith(SCENE_TRANSITION_PRIVATE_ID_PREFIXES) ): DrSimonClient._raise_scene_transition_payload_invalid( field_name, f"{field_name} must be a public identifier", ) return stripped @staticmethod def _validate_scene_transition_source(value: str) -> str: if value not in SCENE_TRANSITION_SOURCES: DrSimonClient._raise_scene_transition_payload_invalid( "source", "source must be one of the public attention scene sources", ) return value @staticmethod def _validate_scene_def_version(value: str, field_name: str) -> str: version = value.strip() if ( not version or SCENE_DEF_VERSION_RE.fullmatch(version) is None or version.lower().startswith(SCENE_TRANSITION_PRIVATE_ID_PREFIXES) ): DrSimonClient._raise_scene_transition_payload_invalid( field_name, f"{field_name} must be a public identifier", ) return version @staticmethod def _validate_scene_transition_timestamp(value: object, field_name: str) -> str: if ( not isinstance(value, str) or SCENE_TRANSITION_ISO_UTC_MS_RE.fullmatch(value) is None ): DrSimonClient._raise_scene_transition_payload_invalid( field_name, f"{field_name} must be an ISO-8601 UTC millisecond timestamp", ) return value @staticmethod def _validate_scene_transition_attention_scene(value: object) -> JsonObject: if not isinstance(value, dict): DrSimonClient._raise_scene_transition_payload_invalid( "attentionScene", "attentionScene must be an object", ) extra_keys = set(value) - SCENE_TRANSITION_SCENE_KEYS if extra_keys: DrSimonClient._raise_scene_transition_payload_invalid( "attentionScene", "attentionScene must contain only group/sub", ) group = value.get("group") if not isinstance(group, str): DrSimonClient._raise_scene_transition_payload_invalid( "attentionScene.group", "attentionScene.group must be a public scene token", ) safe: JsonObject = { "group": DrSimonClient._validate_scene_transition_scene_token( group, "attentionScene.group", ) } sub = value.get("sub") if sub is not None: if not isinstance(sub, str): DrSimonClient._raise_scene_transition_payload_invalid( "attentionScene.sub", "attentionScene.sub must be a public scene token", ) safe["sub"] = DrSimonClient._validate_scene_transition_scene_token( sub, "attentionScene.sub", ) return safe @staticmethod def _validate_scene_transition_scene_token(value: str, field_name: str) -> str: token = value.strip() if ( not token or SCENE_TRANSITION_SCENE_TOKEN_RE.fullmatch(token) is None or token.startswith(SCENE_TRANSITION_PRIVATE_SCENE_TOKEN_PREFIXES) ): DrSimonClient._raise_scene_transition_payload_invalid( field_name, f"{field_name} must be a public scene token", ) return token @staticmethod def _raise_scene_transition_payload_invalid( field_name: str, message: str, ) -> NoReturn: raise DrSimonError( 400, message, payload={ "status": "error", "message": message, "error": "attention_scene_transition_payload_invalid", "rejected_fields": [field_name], "rejection_reasons": [message], }, ) @staticmethod def growth_feedback_points(report: JsonObject) -> list[JsonObject]: """Return evidence-bound Focuspang growth feedback points for UI copy.""" points = ( report.get("growth_feedback_points") if isinstance(report, dict) else [] ) if not isinstance(points, list): return [] safe_points: list[JsonObject] = [] for point in points: if not isinstance(point, dict): continue feedback_kind = point.get("feedback_kind") feedback_token = point.get("feedback_token") evidence = point.get("feedback_evidence") if not isinstance(evidence, dict): continue evidence_object = cast(JsonObject, evidence) data_quality_reliability = evidence_object.get( "data_quality_reliability" ) if ( not isinstance(feedback_kind, str) or feedback_kind not in GROWTH_FEEDBACK_ALLOWED_KINDS or not isinstance(feedback_token, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(feedback_token) is None or not isinstance(data_quality_reliability, str) or data_quality_reliability not in DATA_QUALITY_RELIABILITY_VALUES or evidence_object.get("evidence_policy") != "contextual_multimodal_evidence_required" or not DrSimonClient._has_evidence_explanation_policy( evidence_object ) or not DrSimonClient._has_report_public_positioning_boundaries( evidence_object ) or any( evidence_object.get(flag) is not True for flag in GROWTH_FEEDBACK_REQUIRED_EVIDENCE_FLAGS ) ): continue safe_point = dict(point) safe_point["feedback_evidence"] = dict(evidence_object) safe_points.append(safe_point) return safe_points @staticmethod def recommended_guides(report: JsonObject) -> list[JsonObject]: """Return evidence-bound Moment recommended guides for guardian UI copy.""" guides = ( report.get("recommended_guides") if isinstance(report, dict) else [] ) if not isinstance(guides, list): return [] safe_guides: list[JsonObject] = [] for guide in guides: if not isinstance(guide, dict): continue factor = guide.get("factor") trend_delta = guide.get("trend_delta") reason = guide.get("reason") evidence = guide.get("guidance_evidence") if not isinstance(evidence, dict): continue evidence_object = cast(JsonObject, evidence) data_quality_reliability = evidence_object.get( "data_quality_reliability" ) if ( not isinstance(factor, str) or COLLECTION_PLAN_PUBLIC_TOKEN_RE.fullmatch(factor) is None or not isinstance(trend_delta, (int, float)) or isinstance(trend_delta, bool) or not isinstance(reason, str) or reason not in RECOMMENDED_GUIDE_ALLOWED_REASONS or not isinstance(data_quality_reliability, str) or data_quality_reliability not in DATA_QUALITY_RELIABILITY_VALUES or evidence_object.get("evidence_policy") != "contextual_multimodal_evidence_required" or evidence_object.get("guidance_kind") != "factor_trend_guidance" or evidence_object.get("factor") != factor or not DrSimonClient._has_evidence_explanation_policy( evidence_object ) or not DrSimonClient._has_report_public_positioning_boundaries( evidence_object ) or any( evidence_object.get(flag) is not True for flag in RECOMMENDED_GUIDE_REQUIRED_EVIDENCE_FLAGS ) ): continue safe_guide = dict(guide) safe_guide["guidance_evidence"] = dict(evidence_object) safe_guides.append(safe_guide) return safe_guides def usage( self, partner_code: str, from_date: str, to_date: str, ) -> list[JsonObject]: """GET /api/v1/admin/usage/{partner_code} -- admin key required.""" return cast( list[JsonObject], self._request( "GET", f"/api/v1/admin/usage/{partner_code}", params={"from_date": from_date, "to_date": to_date}, ), ) def usage_summary(self, partner_code: str) -> JsonObject: """GET /api/v1/admin/usage/{partner_code}/summary -- admin key required.""" return self._request("GET", f"/api/v1/admin/usage/{partner_code}/summary") def close(self) -> None: self._client.close() def __enter__(self) -> DrSimonClient: return self def __exit__(self, *args: Any) -> None: self.close() if __name__ == "__main__": with DrSimonClient(api_key="drs_test_key", base_url="http://localhost:8090") as client: print(client.health()) result = client.ingest( uuid="550e8400-e29b-41d4-a716-446655440000", product="edu", primitives=[ {"time": "2026-03-04T10:30:00.000Z", "type": "session_duration", "value": 45.0, "tier": 1}, {"time": "2026-03-04T10:30:00.000Z", "type": "focus_ratio", "value": 0.85, "tier": 1}, ], device_profile={ "device_profile_v": 1, "cpu_class": "pentium_gold", "cpu_score": 1750, "ram_mb": 4096, "has_camera": True, "camera_max_res": [1280, 720], "has_mic": False, "has_touch": False, "has_stylus": False, "thermal_state": "nominal", "power": "plugged", "network_eff": "wifi_strong", "os": "chromeos-126", "device_time_offset_ms": 0, }, ) print("Ingest:", result) analysis = client.analysis("550e8400-e29b-41d4-a716-446655440000", "edu") print("Analysis:", analysis) params = client.params("550e8400-e29b-41d4-a716-446655440000", "edu") print("Params:", params)