""" ForensicAI Analysis API — Hugging Face Space backend Scientific forensic tools: fingerprints, DNA STR, TOD, ballistics, toxicology, digital evidence, chain of custody, and case classification. """ from __future__ import annotations import math from typing import Optional from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field app = FastAPI(title="ForensicAI Analysis API", version="1.0.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ── Fingerprint data ────────────────────────────────────────────────────────── PATTERN_DATA = { "whorl": { "subtypes": ["plain_whorl", "central_pocket_loop", "double_loop", "accidental"], "population_freq": 0.30, "delta_count": 2, "henry_code_range": "1–32", }, "loop": { "subtypes": ["ulnar_loop", "radial_loop"], "population_freq": 0.60, "delta_count": 1, "henry_code_range": "1–16", }, "arch": { "subtypes": ["plain_arch", "tented_arch"], "population_freq": 0.05, "delta_count": 0, "henry_code_range": "A/T", }, } # ── CODIS 13-core loci allele frequencies (simplified Caucasian population) ─── CODIS_FREQ: dict[str, dict[str, float]] = { "CSF1PO": {"10": .212, "11": .248, "12": .335, "13": .152, "14": .040, "other": .013}, "FGA": {"18": .062, "19": .139, "20": .147, "21": .176, "22": .153, "23": .118, "24": .089, "other": .116}, "TH01": {"6": .220, "7": .218, "8": .099, "9": .138, "9.3": .237, "10": .066, "other": .022}, "TPOX": {"8": .509, "9": .089, "10": .059, "11": .255, "12": .063, "other": .025}, "vWA": {"14": .090, "15": .119, "16": .183, "17": .229, "18": .186, "19": .122, "20": .047, "other": .024}, "D3S1358": {"14": .059, "15": .215, "16": .263, "17": .218, "18": .163, "other": .082}, "D5S818": {"7": .010, "9": .057, "10": .088, "11": .360, "12": .379, "13": .093, "other": .013}, "D7S820": {"8": .112, "9": .115, "10": .272, "11": .247, "12": .183, "13": .052, "other": .019}, "D8S1179": {"10": .050, "11": .062, "12": .136, "13": .337, "14": .195, "15": .138, "16": .069, "other": .013}, "D13S317": {"8": .189, "9": .098, "10": .070, "11": .312, "12": .259, "13": .052, "other": .020}, "D16S539": {"9": .111, "10": .078, "11": .282, "12": .332, "13": .161, "14": .025, "other": .011}, "D18S51": {"12": .066, "13": .130, "14": .151, "15": .148, "16": .128, "17": .115, "18": .092, "other": .170}, "D21S11": {"27": .052, "28": .167, "29": .211, "30": .261, "31": .101, "32": .082, "other": .126}, } # ── Toxicology symptom map ──────────────────────────────────────────────────── SYMPTOM_MAP: dict[str, list[str]] = { "pinpoint_pupils": ["opioids", "organophosphates", "clonidine"], "dilated_pupils": ["stimulants", "anticholinergics", "hallucinogens", "withdrawal"], "bradycardia": ["opioids", "beta_blockers", "organophosphates", "digoxin"], "tachycardia": ["stimulants", "anticholinergics", "ethanol_withdrawal", "cocaine"], "hyperthermia": ["stimulants", "anticholinergics", "serotonin_syndrome", "NMS"], "hypothermia": ["opioids", "ethanol", "sedatives", "antipsychotics"], "seizures": ["stimulants", "organophosphates", "GHB_withdrawal", "tricyclics"], "excessive_salivation": ["organophosphates", "cholinergics", "ketamine"], "dry_mouth": ["anticholinergics", "antihistamines", "stimulants"], "cherry_red_skin": ["carbon_monoxide"], "cyanosis": ["methemoglobin_formers", "asphyxiants", "nitrites"], "garlic_odor": ["organophosphates", "arsenic", "selenium"], "almond_odor": ["cyanide"], "fruity_odor": ["ethanol", "diabetic_ketoacidosis", "isopropanol"], "muscle_rigidity": ["serotonin_syndrome", "NMS", "strychnine", "tetanus"], "nausea_vomiting": ["heavy_metals", "opioids", "GHB", "ethanol", "cyanide"], "confusion_altered_ms": ["ethanol", "GHB", "benzodiazepines", "heavy_metals", "CO"], "respiratory_depression": ["opioids", "barbiturates", "benzodiazepines", "GHB"], "sweating_diaphoresis": ["organophosphates", "stimulants", "serotonin_syndrome", "opioid_withdrawal"], "urinary_retention": ["anticholinergics", "opioids"], "excessive_urination": ["ethanol", "lithium", "caffeine"], "bruising_bleeding": ["anticoagulants", "thrombocytopenia_agents", "salicylates"], } CONFIRMATORY_TESTS: dict[str, list[str]] = { "opioids": ["urine_immunoassay", "GC-MS_confirmation", "serum_opioid_panel"], "organophosphates": ["plasma_cholinesterase", "RBC_cholinesterase", "urine_alkyl_phosphates"], "stimulants": ["urine_amphetamine_screen", "serum_cocaine_metabolites", "GC-MS"], "anticholinergics": ["serum_anticholinergic_assay", "urine_toxicology_extended"], "carbon_monoxide": ["carboxyhemoglobin_co-oximetry", "blood_CO_level"], "cyanide": ["whole_blood_cyanide", "plasma_thiocyanate"], "heavy_metals": ["ICP-MS_blood_metals", "urine_24h_metals", "hair_metal_analysis"], "ethanol": ["serum_ethanol_quantitative", "breath_alcohol_correlation"], "benzodiazepines": ["urine_benzo_screen", "serum_specific_benzo_levels"], "GHB": ["serum_GHB_within_6h", "urine_GHB_within_12h"], "serotonin_syndrome": ["clinical_diagnosis_Hunter_criteria", "serum_serotonin", "CK_level"], } # ── Ballistics caliber table ────────────────────────────────────────────────── CALIBER_TABLE = [ {"min_mm": 0, "max_mm": 6.5, "calibers": [".22 LR", ".22 WMR", ".25 ACP"], "weapon_types": ["pistol", "revolver", "rifle"]}, {"min_mm": 6.5, "max_mm": 8.5, "calibers": [".32 ACP", ".32 S&W", "7.62x25mm"], "weapon_types": ["pistol", "revolver"]}, {"min_mm": 8.5, "max_mm": 10.0, "calibers": [".380 ACP", "9mm Luger", ".38 Special"], "weapon_types": ["pistol", "revolver", "submachine_gun"]}, {"min_mm": 10.0, "max_mm": 11.5, "calibers": [".40 S&W", "10mm Auto", ".357 Magnum"], "weapon_types": ["pistol", "revolver"]}, {"min_mm": 11.5, "max_mm": 13.5, "calibers": [".44 Magnum", ".45 ACP", ".45 Colt"], "weapon_types": ["pistol", "revolver"]}, {"min_mm": 13.5, "max_mm": 20.0, "calibers": [".50 AE", ".500 S&W", "12 ga slug"], "weapon_types": ["pistol", "revolver", "shotgun"]}, {"min_mm": 20.0, "max_mm": 999, "calibers": ["12 ga shot", "20 ga shot", "rifled slug"], "weapon_types": ["shotgun"]}, ] # ── Utilities ───────────────────────────────────────────────────────────────── def _allele_freq(locus: str, allele: str) -> float: lf = CODIS_FREQ.get(locus, {}) return lf.get(str(allele), lf.get("other", 0.01)) def _score_symptoms(symptoms: list[str]) -> dict[str, float]: scores: dict[str, float] = {} for sym in symptoms: key = sym.lower().replace(" ", "_") for substance in SYMPTOM_MAP.get(key, []): scores[substance] = scores.get(substance, 0) + 1 total = len(symptoms) or 1 return {k: round(v / total, 3) for k, v in sorted(scores.items(), key=lambda x: -x[1])} def _ke_joules(grains: float, fps: float) -> float: kg = grains * 6.479891e-5 ms = fps * 0.3048 return round(0.5 * kg * ms ** 2, 2) def _caliber_from_diameter(mm: float) -> dict: for row in CALIBER_TABLE: if row["min_mm"] <= mm < row["max_mm"]: return row return CALIBER_TABLE[-1] # ── Request models ──────────────────────────────────────────────────────────── class FingerprintRequest(BaseModel): pattern_type: str = Field(..., description="whorl | loop | arch | unknown") minutiae_count: int = Field(0, ge=0, le=200) ridge_count: int = Field(0, ge=0, le=50) finger: str = Field("unknown", description="thumb | index | middle | ring | little") hand: str = Field("unknown", description="left | right | unknown") scar_present: bool = Field(False) latent_quality: str = Field("medium", description="high | medium | low | insufficient") class TODRequest(BaseModel): body_temp_c: float = Field(..., ge=0, le=42) ambient_temp_c: float = Field(..., ge=-30, le=50) body_weight_kg: float = Field(70.0, ge=20, le=200) lividity_stage: str = Field("unfixed", description="none | unfixed | fixed | blanching") rigor_stage: str = Field("early", description="none | early | complete | passing | absent") location: str = Field("indoor", description="indoor | outdoor | water | buried") clothing: str = Field("clothed", description="clothed | partially_clothed | nude") class DNASTRRequest(BaseModel): loci: dict[str, list[str]] = Field(..., description="Dict of locus → [allele1, allele2]") population: str = Field("caucasian", description="caucasian | african_american | hispanic | asian") class DigitalEvidenceRequest(BaseModel): file_hash: str = Field(..., description="MD5/SHA1/SHA256 hex string") hash_algorithm: str = Field("sha256", description="md5 | sha1 | sha256 | sha512") file_size_bytes: int = Field(0, ge=0) creation_timestamp: Optional[str] = None modified_timestamp: Optional[str] = None metadata_intact: bool = Field(True) source_device: str = Field("unknown") acquisition_method: str = Field("unknown", description="dd | FTK | EnCase | manual | unknown") class ToxicologyRequest(BaseModel): symptoms: list[str] = Field(..., description="List of observed symptoms") biological_sample: str = Field("blood", description="blood | urine | hair | vitreous | liver") time_since_exposure_h: float = Field(0.0, ge=0) circumstances: str = Field("", description="Optional scene/context description") class BallisticsRequest(BaseModel): bullet_weight_grains: float = Field(..., ge=10, le=800) velocity_fps: float = Field(..., ge=100, le=5000) wound_diameter_mm: Optional[float] = None wound_depth_mm: Optional[float] = None wound_type: str = Field("penetrating", description="penetrating | perforating | tangential | graze") distance_markers: list[str] = Field(default_factory=list, description="stippling | soot | singeing | none") class EvidenceIntegrityRequest(BaseModel): evidence_type: str = Field(..., description="biological | digital | physical | documentary | trace") collection_method: str = Field(..., description="swab | bagging | photography | seizure | other") storage_conditions: str = Field("refrigerated", description="refrigerated | frozen | room_temp | unknown") elapsed_hours: float = Field(0.0, ge=0) packaging_sealed: bool = Field(True) documentation_complete: bool = Field(True) witness_count: int = Field(1, ge=0) contamination_risk: str = Field("low", description="low | medium | high") class CaseClassificationRequest(BaseModel): evidence_types: list[str] = Field(..., description="List of evidence types found") scene_description: str = Field("") victim_condition: str = Field("", description="alive | deceased | unknown") location_type: str = Field("", description="residential | commercial | outdoor | vehicle | other") # ── Endpoints ───────────────────────────────────────────────────────────────── @app.get("/health") def health(): return {"status": "ok", "service": "ForensicAI Analysis API"} @app.post("/fingerprint_analysis") def fingerprint_analysis(body: FingerprintRequest): pt = body.pattern_type.lower() if pt not in PATTERN_DATA and pt != "unknown": raise HTTPException(422, detail={"error": f"Unknown pattern type. Use: {list(PATTERN_DATA.keys())}"}) data = PATTERN_DATA.get(pt, {"subtypes": [], "population_freq": 0.05, "delta_count": "?", "henry_code_range": "?"}) # Quality-adjusted match threshold quality_thresholds = {"high": 12, "medium": 8, "low": 6, "insufficient": 0} threshold = quality_thresholds.get(body.latent_quality, 8) meets_threshold = body.minutiae_count >= threshold # ACE-V suitability if body.latent_quality == "insufficient" or body.minutiae_count < 6: suitability = "insufficient_for_comparison" elif body.minutiae_count >= 12 and body.latent_quality in ("high", "medium"): suitability = "suitable_for_ACE-V" else: suitability = "limited_value" individualisation_potential = "high" if meets_threshold and body.latent_quality == "high" else \ "moderate" if meets_threshold else "low" return { "pattern_type": pt, "subtypes": data["subtypes"], "population_frequency": data["population_freq"], "delta_count": data["delta_count"], "henry_code_range": data["henry_code_range"], "finger": body.finger, "hand": body.hand, "minutiae_count": body.minutiae_count, "ridge_count": body.ridge_count, "latent_quality": body.latent_quality, "scar_present": body.scar_present, "acevo_suitability": suitability, "individualisation_potential": individualisation_potential, "min_minutiae_threshold": threshold, "meets_threshold": meets_threshold, "notes": "Conclusions require ACE-V examination by a certified latent print examiner.", } @app.post("/time_of_death") def time_of_death(body: TODRequest): # Henssge nomogram cooling formula (simplified) # Corrected body temp constant: 37.2°C (rectal) T0, Ta = 37.2, body.ambient_temp_c Tb = body.body_temp_c W = body.body_weight_kg # Location/clothing correction factor correction = {"indoor_clothed": 1.0, "outdoor_clothed": 1.1, "indoor_nude": 0.75, "outdoor_nude": 0.9}.get( f"{body.location}_{body.clothing}", 1.0) # Compute k (cooling rate constant) k = 0.0347 * ((70 / W) ** 0.625) * correction if Tb <= Ta: cooling_pmi_h = None cooling_note = "Body temp at or below ambient — cooling PMI unreliable beyond ~24–36h" elif Tb >= T0: cooling_pmi_h = 0 cooling_note = "Body temp at or above 37.2°C — death very recent (<1–2h)" else: ratio = (Tb - Ta) / (T0 - Ta) cooling_pmi_h = round(-math.log(ratio) / k, 1) if ratio > 0 else None cooling_note = "Henssge formula applied" # Livor mortis PMI ranges lividity_ranges = { "none": (0, 2), "unfixed": (1, 8), "blanching": (4, 12), "fixed": (8, 36), } livor_range = lividity_ranges.get(body.lividity_stage, (0, 36)) # Rigor mortis PMI ranges rigor_ranges = { "none": (0, 3), "early": (2, 8), "complete": (6, 24), "passing": (18, 48), "absent": (36, 96), } rigor_range = rigor_ranges.get(body.rigor_stage, (0, 96)) # Consensus interval (intersection of all three) low = max(livor_range[0], rigor_range[0]) high = min(livor_range[1], rigor_range[1]) if cooling_pmi_h is not None: low = max(low, cooling_pmi_h * 0.7) high = min(high, cooling_pmi_h * 1.3) low = round(max(low, 0), 1) high = round(max(high, low + 1), 1) return { "estimated_pmi_hours": {"low": low, "high": high}, "cooling_pmi_hours": cooling_pmi_h, "lividity_pmi_range_hours": livor_range, "rigor_pmi_range_hours": rigor_range, "body_temp_c": body.body_temp_c, "ambient_temp_c": body.ambient_temp_c, "cooling_constant_k": round(k, 5), "correction_factor": correction, "cooling_formula_note": cooling_note, "confidence": "moderate" if cooling_pmi_h else "low", "caveat": "PMI estimates are probabilistic. Multiple confounding factors apply. " "Confirm with scene investigators and forensic pathologist.", } @app.post("/dna_str_match") def dna_str_match(body: DNASTRRequest): if not body.loci: raise HTTPException(422, detail={"error": "At least one CODIS locus required"}) codis_core = set(CODIS_FREQ.keys()) loci_analyzed = [] rmp = 1.0 # random match probability profile_complete = 0 for locus, alleles in body.loci.items(): if len(alleles) != 2: continue a1, a2 = str(alleles[0]), str(alleles[1]) f1 = _allele_freq(locus, a1) f2 = _allele_freq(locus, a2) # Hardy-Weinberg: 2pq (hetero) or p² (homo) if a1 == a2: locus_prob = f1 ** 2 genotype_type = "homozygous" else: locus_prob = 2 * f1 * f2 genotype_type = "heterozygous" rmp *= locus_prob if locus in codis_core: profile_complete += 1 loci_analyzed.append({ "locus": locus, "alleles": [a1, a2], "genotype_type": genotype_type, "allele_freqs": [round(f1, 5), round(f2, 5)], "locus_prob": round(locus_prob, 8), "in_codis_core": locus in codis_core, }) rmp = round(rmp, 15) if loci_analyzed else None rmp_display = f"1 in {round(1/rmp):,}" if rmp and rmp > 0 else "N/A" completeness_pct = round((profile_complete / 13) * 100, 1) strength = "inconclusive" if rmp and rmp < 1e-10: strength = "extremely_strong" elif rmp and rmp < 1e-7: strength = "very_strong" elif rmp and rmp < 1e-4: strength = "strong" elif rmp and rmp < 0.01: strength = "moderate" return { "population": body.population, "loci_analyzed": loci_analyzed, "loci_count": len(loci_analyzed), "codis_core_loci_present": profile_complete, "profile_completeness_pct": completeness_pct, "random_match_probability": rmp, "rmp_display": rmp_display, "statistical_strength": strength, "codis_uploadable": profile_complete >= 10, "notes": "Frequencies based on published Caucasian population data (Butler 2006). " "Population-specific databases should be consulted for court use.", } @app.post("/digital_evidence_integrity") def digital_evidence_integrity(body: DigitalEvidenceRequest): # Hash format validation expected_lengths = {"md5": 32, "sha1": 40, "sha256": 64, "sha512": 128} expected_len = expected_lengths.get(body.hash_algorithm.lower(), 64) hash_clean = body.file_hash.strip().lower().replace(":", "") hash_valid = len(hash_clean) == expected_len and all(c in "0123456789abcdef" for c in hash_clean) # Acquisition method trust score method_trust = {"dd": 0.95, "FTK": 0.95, "EnCase": 0.95, "X-Ways": 0.90, "manual": 0.40, "unknown": 0.20} trust = method_trust.get(body.acquisition_method, 0.20) # Integrity scoring score = 100 issues = [] if not hash_valid: score -= 30 issues.append("Hash format invalid for selected algorithm") if not body.metadata_intact: score -= 20 issues.append("Metadata integrity compromised") if body.acquisition_method == "manual": score -= 25 issues.append("Manual acquisition lacks write-blocker verification") if body.acquisition_method == "unknown": score -= 35 issues.append("Acquisition method unknown — chain of custody break risk") if body.creation_timestamp and body.modified_timestamp: if body.modified_timestamp < body.creation_timestamp: score -= 20 issues.append("Modified timestamp precedes creation — potential anti-forensic manipulation") score = max(0, score) admissibility_risk = "low" if score >= 80 else "medium" if score >= 55 else "high" return { "hash_algorithm": body.hash_algorithm, "hash_provided": body.file_hash, "hash_format_valid": hash_valid, "file_size_bytes": body.file_size_bytes, "acquisition_method": body.acquisition_method, "acquisition_trust_score": trust, "metadata_intact": body.metadata_intact, "integrity_score": score, "admissibility_risk": admissibility_risk, "issues_identified": issues, "recommended_actions": [ "Verify hash with original acquisition log", "Confirm write-blocker use during acquisition", "Document chain of custody for every transfer", "Use Cellebrite / EnCase / FTK for re-acquisition if needed", ] if issues else ["Evidence integrity appears sound — maintain chain of custody"], } @app.post("/toxicology_panel") def toxicology_panel(body: ToxicologyRequest): if not body.symptoms: raise HTTPException(422, detail={"error": "At least one symptom required"}) scores = _score_symptoms(body.symptoms) if not scores: return {"candidates": [], "message": "No substance matches found for provided symptoms"} # Sample window by biological matrix window = { "blood": "6–12 hours", "urine": "1–4 days (some drugs up to 30 days)", "hair": "90+ days", "vitreous": "mirrors blood at time of death", "liver": "extended (weeks)", }.get(body.biological_sample, "variable") top = list(scores.items())[:6] candidates = [ { "substance_class": subst, "match_score": sc, "confidence": "high" if sc > 0.5 else "moderate" if sc > 0.25 else "low", "confirmatory_tests": CONFIRMATORY_TESTS.get(subst, ["extended_toxicology_screen"]), } for subst, sc in top ] return { "symptoms_evaluated": body.symptoms, "biological_sample": body.biological_sample, "detection_window": window, "time_since_exposure_h": body.time_since_exposure_h, "substance_candidates": candidates, "priority_tests": candidates[0]["confirmatory_tests"] if candidates else [], "caveat": "Symptom-based screening only. GC-MS or LC-MS/MS confirmation required for legal proceedings.", } @app.post("/ballistics_profile") def ballistics_profile(body: BallisticsRequest): ke = _ke_joules(body.bullet_weight_grains, body.velocity_fps) mass_g = round(body.bullet_weight_grains * 0.0647989, 2) vel_ms = round(body.velocity_fps * 0.3048, 1) # Energy classification if ke < 100: energy_class = "low_energy" elif ke < 500: energy_class = "medium_energy" elif ke < 2000: energy_class = "high_energy" else: energy_class = "very_high_energy" caliber_info = _caliber_from_diameter(body.wound_diameter_mm) if body.wound_diameter_mm else None # Range indicators from distance markers range_estimate = "contact_or_close" if "soot" in body.distance_markers or "singeing" in body.distance_markers \ else "intermediate" if "stippling" in body.distance_markers \ else "distant" if body.distance_markers == ["none"] \ else "undetermined" return { "bullet_weight_grains": body.bullet_weight_grains, "bullet_weight_grams": mass_g, "velocity_fps": body.velocity_fps, "velocity_ms": vel_ms, "kinetic_energy_joules": ke, "energy_classification": energy_class, "wound_type": body.wound_type, "wound_diameter_mm": body.wound_diameter_mm, "probable_calibers": caliber_info["calibers"] if caliber_info else ["undetermined"], "probable_weapon_types": caliber_info["weapon_types"] if caliber_info else ["undetermined"], "firing_distance_class": range_estimate, "distance_markers_present": body.distance_markers, "notes": "Wound morphology is influenced by intermediate targets, clothing, and tissue type. " "Firearm examiner verification required.", } @app.post("/evidence_integrity") def evidence_integrity(body: EvidenceIntegrityRequest): score = 100 issues = [] recommendations = [] # Time-based degradation (biological evidence degrades fastest) degradation_rates = {"biological": 2.5, "digital": 0.5, "physical": 0.3, "documentary": 0.2, "trace": 1.5} rate = degradation_rates.get(body.evidence_type, 1.0) time_penalty = min(40, body.elapsed_hours * rate * 0.5) score -= time_penalty if not body.packaging_sealed: score -= 20 issues.append("Packaging not sealed — contamination risk") if not body.documentation_complete: score -= 15 issues.append("Documentation incomplete — chain of custody gap") if body.witness_count == 0: score -= 10 issues.append("No witnesses to collection — procedural weakness") if body.contamination_risk == "high": score -= 20 issues.append("High contamination risk flagged at collection") if body.storage_conditions == "room_temp" and body.evidence_type == "biological": score -= 15 issues.append("Biological evidence stored at room temperature — DNA degradation risk") if body.storage_conditions == "unknown": score -= 10 issues.append("Storage conditions unknown") score = max(0, round(score)) admissibility = "acceptable" if score >= 75 else "challenged" if score >= 50 else "compromised" if body.evidence_type == "biological": recommendations.append("Maintain cold chain (4°C); freeze for long-term storage") if not body.packaging_sealed: recommendations.append("Re-seal in tamper-evident packaging immediately") if time_penalty > 20: recommendations.append("Expedite laboratory analysis due to elapsed time") return { "evidence_type": body.evidence_type, "collection_method": body.collection_method, "storage_conditions": body.storage_conditions, "elapsed_hours": body.elapsed_hours, "integrity_score": score, "admissibility_status": admissibility, "issues": issues, "recommendations": recommendations, "witness_count": body.witness_count, "contamination_risk": body.contamination_risk, } @app.post("/case_classification") def case_classification(body: CaseClassificationRequest): ev = [e.lower() for e in body.evidence_types] desc = body.scene_description.lower() loc = body.location_type.lower() scores: dict[str, float] = {} type_signals = { "homicide": ["blood", "weapon", "victim", "ligature", "blunt_force", "gunshot", "stabbing"], "sexual_assault": ["biological_fluid", "dna_swab", "clothing_fiber", "bruising", "victim_clothing"], "burglary": ["tool_mark", "glass", "footprint", "fingerprint", "forced_entry"], "fraud": ["document", "financial_record", "digital", "forgery", "counterfeit"], "drug_offense": ["controlled_substance", "paraphernalia", "scale", "currency", "packaging"], "digital_crime": ["digital", "device", "log_file", "network_trace", "metadata", "hash"], "arson": ["accelerant", "char_pattern", "fire_debris", "heat_damage"], "hit_and_run": ["paint_transfer", "glass_shatter", "tyre_mark", "blood", "vehicle_part"], } for case_type, signals in type_signals.items(): match = sum(1 for s in signals if any(s in e for e in ev) or s in desc) if match: scores[case_type] = match / len(signals) # Victim condition strongly shifts toward homicide if body.victim_condition == "deceased": scores["homicide"] = scores.get("homicide", 0) + 0.50 scores["hit_and_run"] = scores.get("hit_and_run", 0) + 0.15 elif body.victim_condition == "alive": scores["sexual_assault"] = scores.get("sexual_assault", 0) + 0.10 scores["burglary"] = scores.get("burglary", 0) + 0.05 # Residential location with deceased victim is strong homicide signal if loc == "residential" and body.victim_condition == "deceased": scores["homicide"] = scores.get("homicide", 0) + 0.20 if not scores: primary = "undetermined" confidence = "low" else: primary = max(scores, key=lambda k: scores[k]) top_score = scores[primary] confidence = "high" if top_score > 0.6 else "moderate" if top_score > 0.35 else "low" priority_evidence = { "homicide": ["DNA swabs", "firearms/weapons", "CCTV footage", "mobile phone records"], "sexual_assault": ["rape kit", "clothing", "DNA profile", "toxicology screen"], "burglary": ["fingerprints", "tool marks", "footwear impressions", "CCTV"], "fraud": ["financial records", "digital devices", "handwriting samples", "metadata"], "drug_offense": ["substance weight/purity", "fingerprints on packaging", "cell tower data"], "digital_crime": ["device imaging", "hash verification", "network logs", "cloud warrant"], "arson": ["fire debris GC-MS", "accelerant swabs", "origin point photography"], "hit_and_run": ["paint chip analysis", "glass refraction index", "tyre impression cast"], }.get(primary, ["scene photography", "witness statements", "physical evidence inventory"]) return { "primary_classification": primary, "confidence": confidence, "all_scores": {k: round(v, 3) for k, v in sorted(scores.items(), key=lambda x: -x[1])}, "evidence_types_input": body.evidence_types, "victim_condition": body.victim_condition, "location_type": loc, "priority_evidence": priority_evidence, "recommended_specialists": { "homicide": ["forensic pathologist", "bloodstain pattern analyst", "firearms examiner"], "sexual_assault": ["sexual assault nurse examiner (SANE)", "DNA analyst"], "digital_crime": ["digital forensics examiner", "network forensics specialist"], "arson": ["fire investigator", "accelerant detection canine"], }.get(primary, ["forensic scientist", "crime scene investigator"]), }