""" Report Analysis Service. Coordinates report parsing, LLM analysis, and structured output generation. """ from __future__ import annotations from typing import Dict, Any, List from loguru import logger from multimodal.report_parser import parse_report_text from models.report_llm import ReportLLM class ReportService: """Service for analyzing medical reports.""" def __init__(self, llm: ReportLLM): self.llm = llm logger.info("[ReportService] Initialized") def _build_sources(self, extracted_values: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Build source citations for report analysis.""" sources = [] if extracted_values: abnormal_items = [x for x in extracted_values if (x.get("flag") or "").lower() in {"high", "low"}] if abnormal_items: sources.append({ "title": "Lab Value Interpretation Reference", "score": 0.91, "category": "clinical_reference", "preview": f"Detected abnormal values: {', '.join(x['name'] for x in abnormal_items[:4])}" }) if not sources: sources.append({ "title": "Uploaded Medical Report", "score": 0.84, "category": "uploaded_document", "preview": "Analysis generated from the uploaded report content." }) return sources def _compute_confidence(self, extracted_values: List[Dict[str, Any]], raw_text: str) -> float: """Compute confidence score for report analysis.""" signal = 0.55 if len(raw_text) > 250: signal += 0.10 if len(extracted_values) >= 3: signal += 0.15 if len(extracted_values) >= 6: signal += 0.08 abnormal_count = sum( 1 for x in extracted_values if (x.get("flag") or "").lower() in {"high", "low"} ) if abnormal_count > 0: signal += 0.05 return round(min(signal, 0.95), 2) def analyze(self, raw_text: str) -> Dict[str, Any]: """ Analyze medical report text. Args: raw_text: Raw report text Returns: Structured analysis with summary, explanation, values, sources """ logger.info(f"[ReportService] Analyzing report ({len(raw_text)} chars)") # Extract structured values extracted_values = parse_report_text(raw_text) logger.info(f"[ReportService] Extracted {len(extracted_values)} values") # Generate LLM analysis try: llm_output = self.llm.analyze_report( raw_text=raw_text, extracted_values=extracted_values, ) except Exception as e: logger.error(f"[ReportService] LLM analysis failed: {e}") llm_output = { "summary": "Analysis failed due to processing error.", "simple_explanation": "Unable to generate explanation.", "potential_concerns": [], "next_steps": ["Please try again or consult a healthcare professional."], "safety_note": "This analysis is informational only." } # Build sources and compute confidence sources = self._build_sources(extracted_values) confidence = self._compute_confidence(extracted_values, raw_text) result = { "summary": llm_output.get("summary", "No summary available."), "simple_explanation": llm_output.get("simple_explanation", "No explanation available."), "potential_concerns": llm_output.get("potential_concerns", []), "next_steps": llm_output.get("next_steps", []), "confidence": confidence, "extracted_values": extracted_values, "sources": sources, "safety_note": llm_output.get( "safety_note", "This analysis is informational and should not replace professional medical advice." ), "report_type": llm_output.get("report_type", "Medical Report"), } logger.success(f"[ReportService] Analysis complete (confidence: {confidence})") return result