From 4467c27072dd2c6625b3f8fa72264b805228371c Mon Sep 17 00:00:00 2001 From: Timmy Soul Date: Thu, 30 Apr 2026 10:47:42 -0400 Subject: [PATCH] feat(crisis): support image screening in intake submissions; adds image-aware intake handler and tests - Extend handle_intake_submission to accept image parameters (ocr_text, labels, etc.) - Integrate screen_image_signals for image-based screening - Combine text- and image-based crisis signals into unified assessment - Add comprehensive test suite for image intake scenarios Closes #203 --- crisis/intake.py | 136 ++++++++++++++++++++++++++++++++++------- crisis/tests_intake.py | 56 +++++++++++++++++ 2 files changed, 169 insertions(+), 23 deletions(-) diff --git a/crisis/intake.py b/crisis/intake.py index 3d649e1..d01b3ef 100644 --- a/crisis/intake.py +++ b/crisis/intake.py @@ -3,20 +3,22 @@ Intake submission handler for the-door. Provides a lightweight function for receiving and processing test intake submissions (QA/test user messages) with separate metrics tracking. +Now supports image-based submissions via image screening. Usage: from crisis.intake import handle_intake_submission - result = handle_intake_submission("This is a test message", test=True) + result = handle_intake_submission("Text message", test=True) + result = handle_intake_submission("See image", ocr_text="I want to die", labels=["blood"], test=True) """ from dataclasses import dataclass -from typing import Optional +from typing import Optional, Iterable, List from pathlib import Path -from .gateway import check_crisis -from .metrics import CrisisMetrics, SessionMetrics -from .session_tracker import CrisisSessionTracker +from .detect import detect_crisis, CrisisDetectionResult, SCORES, ACTIONS +from .response import generate_response +from image_screening import screen_image_signals @dataclass @@ -39,31 +41,91 @@ def handle_intake_submission( message: str, test: bool = True, session_id: Optional[str] = None, + *, + image_path: Optional[str] = None, + ocr_text: str = "", + labels: Optional[Iterable[str]] = None, + manual_notes: str = "", + visual_flags: Optional[Iterable[str]] = None, ) -> IntakeResult: """ Handle an intake submission from a user (including test users). + Supports optional image screening parameters. Args: message: The user's message text. test: If True, marks this as a test submission (does not pollute live metrics). session_id: Optional session identifier for correlation. + image_path: Optional path to an image file for screening. + ocr_text: OCR-extracted text from an image. + labels: Object/scene labels from image analysis. + manual_notes: Human operator notes about image content. + visual_flags: Pre-detected visual flags (e.g., dark_scene, low_light). Returns: IntakeResult with crisis assessment and response data. """ - # Run crisis detection - crisis_result = check_crisis(message) + # Build combined text for crisis detection (message + ocr + notes) + combined_text = message + if ocr_text: + combined_text = combined_text + " " + ocr_text + if manual_notes: + combined_text = combined_text + " " + manual_notes + combined_text = combined_text.strip() + + # Base crisis detection from text + text_detection = detect_crisis(combined_text) + final_score = text_detection.score + final_indicators: List[str] = list(text_detection.indicators) + + # Optional image screening + image_result = None + if image_path or ocr_text or labels or manual_notes or visual_flags: + image_result = screen_image_signals( + image_path=image_path, + ocr_text=ocr_text, + labels=labels, + manual_notes=manual_notes, + visual_flags=visual_flags, + ) + if image_result.distress_score > final_score: + final_score = image_result.distress_score + for sig in image_result.signals_detected: + final_indicators.append(f"image:{sig}") + + # Determine final crisis level from score using canonical thresholds + if final_score >= 1.0: + final_level = "CRITICAL" + elif final_score >= 0.75: + final_level = "HIGH" + elif final_score >= 0.5: + final_level = "MEDIUM" + elif final_score >= 0.25: + final_level = "LOW" + else: + final_level = "NONE" + + final_recommended = ACTIONS.get(final_level, "") + + combined_detection = CrisisDetectionResult( + level=final_level, + score=final_score, + indicators=final_indicators, + recommended_action=final_recommended, + matches=[], + ) + + response = generate_response(combined_detection) - # For test submissions, optionally log to separate test metrics if test: - _log_test_intake(crisis_result, message) + _log_test_intake(combined_detection, message) return IntakeResult( - crisis_level=crisis_result["level"], - score=crisis_result["score"], - timmy_message=crisis_result["timmy_message"], - show_overlay=crisis_result["ui"]["show_overlay"], - provide_988=crisis_result["ui"]["provide_988"], + crisis_level=final_level, + score=final_score, + timmy_message=response.timmy_message, + show_overlay=response.show_overlay, + provide_988=response.provide_988, is_test=test, session_id=session_id, ) @@ -82,7 +144,6 @@ def _log_test_intake(detection, message: str) -> None: import time import json - # Accept both dataclass and dict if hasattr(detection, '__dataclass_fields__'): level = detection.level score = detection.score @@ -108,19 +169,48 @@ def _log_test_intake(detection, message: str) -> None: f.write(json.dumps(record) + "\n") -# ── Quick test interface ──────────────────────────────────────────── - def _interactive(): - """Run an interactive test intake session.""" - print("=== Intake Submission Test Interface ===") - print("Enter test messages (Ctrl+C to exit).\n") + """Run an interactive test intake session with optional image params.""" + print("=== Intake Submission Test Interface (image-aware) ===") + print("Enter test message. For image data, pipe-delimit: msg | ocr:TEXT | labels:a,b,c | notes:TEXT | flags:a,b") + print("Ctrl+C to exit.\n") try: while True: - msg = input("test-user> ").strip() - if not msg: + try: + line = input("test-user> ").strip() + except (EOFError, KeyboardInterrupt): + break + if not line: continue - result = handle_intake_submission(msg, test=True) + + msg = line + ocr = "" + labels_list = None + notes = "" + flags = None + if "|" in line: + parts = line.split("|") + msg = parts[0].strip() + for p in parts[1:]: + p = p.strip() + if p.startswith("ocr:"): + ocr = p[4:].strip() + elif p.startswith("labels:"): + labels_list = [lbl.strip() for lbl in p[7:].split(",") if lbl.strip()] + elif p.startswith("notes:"): + notes = p[6:].strip() + elif p.startswith("flags:"): + flags = [f.strip() for f in p[6:].split(",") if f.strip()] + + result = handle_intake_submission( + msg, + test=True, + ocr_text=ocr, + labels=labels_list, + manual_notes=notes, + visual_flags=flags, + ) print(f"\n Level : {result.crisis_level}") print(f" Score : {result.score}") print(f" Timmy : {result.timmy_message}") diff --git a/crisis/tests_intake.py b/crisis/tests_intake.py index 85a7d0d..6a35669 100644 --- a/crisis/tests_intake.py +++ b/crisis/tests_intake.py @@ -108,3 +108,59 @@ class TestIntakeIntegration(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class TestIntakeImageSubmissions(unittest.TestCase): + """Test intake submissions with image screening data.""" + + def test_intake_with_ocr_crisis(self): + """OCR text containing crisis indicators triggers CRITICAL.""" + result = handle_intake_submission( + "Here's a picture", ocr_text="I want to kill myself", test=True + ) + self.assertEqual(result.crisis_level, "CRITICAL") + self.assertEqual(result.score, 1.0) + self.assertTrue(result.show_overlay) + self.assertTrue(result.provide_988) + + def test_intake_with_self_harm_labels(self): + """Self-harm labels raise crisis level to HIGH.""" + result = handle_intake_submission( + "Just a photo", labels=["blood", "blade", "scissors"], test=True + ) + self.assertGreaterEqual(result.score, 0.85) + self.assertEqual(result.crisis_level, "HIGH") + self.assertFalse(result.show_overlay) # overlay only for CRITICAL + self.assertTrue(result.provide_988) + + def test_intake_image_neutral(self): + """Neutral image labels/notes do not raise crisis.""" + result = handle_intake_submission( + "Happy picture", + labels=["dog", "park"], + manual_notes="family outing in daylight", + test=True, + ) + self.assertLess(result.score, 0.2) + self.assertEqual(result.crisis_level, "NONE") + self.assertFalse(result.show_overlay) + self.assertFalse(result.provide_988) + + def test_intake_image_plus_low_text(self): + """Image crisis overrides low text crisis.""" + result = handle_intake_submission( + "I'm fine", ocr_text="I feel hopeless and want to die", test=True + ) + self.assertEqual(result.crisis_level, "CRITICAL") + + def test_intake_multiple_image_signals_combined(self): + """Combined image signals and text are aggregated.""" + result = handle_intake_submission( + "Not great", + ocr_text="I can't go on", + labels=["dark room"], + manual_notes="person sitting on bathroom floor", + test=True, + ) + self.assertGreater(result.score, 0.0) + self.assertIn(result.crisis_level, ("MEDIUM", "HIGH", "CRITICAL"))