Spaces:
Sleeping
Sleeping
| """ | |
| Chronos-Graph: 4D Forensic Timeline Prototype | |
| Time-reveal interface: all evidence starts dim, lights up as slider scrubs through time. | |
| """ | |
| import gradio as gr | |
| import plotly.graph_objects as go | |
| import pandas as pd | |
| import re | |
| from datetime import datetime | |
| import numpy as np | |
| import traceback | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # MOCK DATASET: The Riverside Park Homicide | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| MOCK_AUTOPSY = """ | |
| AUTOPSY REPORT | |
| Case No: 2024-01-15-RP | |
| Deceased: Victor Hale, Male, 34 y.o. | |
| Recovery: 2024-01-16 06:00 AM | |
| Location: Riverside Park, approximate coordinates 40.7829 N, 73.9654 W | |
| POSTMORTEM INTERVAL (PMI): | |
| - Rectal temperature at 06:00 AM: 24.2C | |
| - Ambient temperature: 4C | |
| - Using Henssge nomogram (1.5C/hr cooling rate): | |
| (37.0 - 24.2) / 1.5 = 8.53 hours | |
| - Estimated Time of Death: 21:28 (9:28 PM) on 2024-01-15 | |
| - Confidence interval: 21:00 - 22:00 | |
| PHYSICAL FINDINGS: | |
| - Fully established rigor mortis (consistent with 9+ hours) | |
| - Fixed lividity posterior | |
| - No evidence of body repositioning post-mortem | |
| - Fatal exsanguination from single stab wound to chest | |
| """ | |
| MOCK_PHONE_GPS = """ | |
| timestamp,latitude,longitude,event | |
| 2024-01-15 19:30:00,40.7484,-73.9847,Home - Brooklyn Heights | |
| 2024-01-15 20:00:00,40.7614,-73.9776,Blue Note Jazz Club - Manhattan | |
| 2024-01-15 20:45:00,40.7681,-73.9819,Subway Station - 59th St | |
| 2024-01-15 21:00:00,40.7829,-73.9654,Riverside Park - Entrance | |
| 2024-01-15 21:30:00,40.7829,-73.9654,Riverside Park - Deep Trail | |
| 2024-01-15 22:00:00,40.7282,-74.0776,Jersey City - Moving rapidly | |
| 2024-01-15 22:15:00,40.7282,-74.0776,Jersey City - STOP at gas station | |
| 2024-01-15 22:30:00,40.7282,-74.0776,Jersey City - Still stationary | |
| 2024-01-15 23:00:00,40.7282,-74.0776,Jersey City - Stationary | |
| """ | |
| MOCK_CCTV = """ | |
| [CCTV-LOG-v1.2] | |
| 2024-01-15 19:28:00 | CAM-BH-001 | 40.7485, -73.9846 | SUBJECT exits residence | |
| 2024-01-15 20:05:00 | CAM-MN-042 | 40.7613, -73.9777 | SUBJECT enters Blue Note Jazz Club | |
| 2024-01-15 20:42:00 | CAM-MN-042 | 40.7613, -73.9777 | SUBJECT exits club, walks north | |
| 2024-01-15 21:03:00 | CAM-RP-007 | 40.7828, -73.9655 | SUBJECT seen entering Riverside Park | |
| 2024-01-15 21:35:00 | CAM-RP-007 | 40.7828, -73.9655 | NO ACTIVITY - camera clear | |
| 2024-01-15 22:08:00 | CAM-HW-112 | 40.7260, -74.0340 | VICTIMS VEHICLE (plate NY-HA-8842) detected on Holland Tunnel eastbound | |
| 2024-01-15 22:12:00 | CAM-HW-113 | 40.7280, -74.0760 | Same vehicle exits tunnel Jersey side | |
| """ | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # PARSERS | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def parse_autopsy(text): | |
| events = [] | |
| coord_match = re.search(r'(\d+\.\d+)[\s°]*[NS]?,?\s*(\d+\.\d+)[\s°]*[EW]?', text) | |
| lat, lon = 40.7829, -73.9654 | |
| if coord_match: | |
| lat, lon = float(coord_match.group(1)), float(coord_match.group(2)) | |
| if 'S' in text[coord_match.start():coord_match.end()]: | |
| lat = -lat | |
| if 'W' in text[coord_match.start():coord_match.end()]: | |
| lon = -lon | |
| tod_match = re.search(r'Estimated Time of Death[:\s]+(\d{2}):(\d{2})', text) | |
| if tod_match: | |
| hour, minute = int(tod_match.group(1)), int(tod_match.group(2)) | |
| tod = datetime(2024, 1, 15, hour, minute) | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": tod, | |
| "event": f"Estimated Time of Death: {hour:02d}:{minute:02d}", | |
| "source": "Autopsy", "icon": "skull" | |
| }) | |
| rec_match = re.search(r'Recovery[:\s]+(\d{4}-\d{2}-\d{2})\s+(\d{2}):(\d{2})', text) | |
| if rec_match: | |
| year, month, day = map(int, rec_match.group(1).split('-')) | |
| hour, minute = int(rec_match.group(2)), int(rec_match.group(3)) | |
| rec_time = datetime(year, month, day, hour, minute) | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": rec_time, | |
| "event": "Body Recovered", | |
| "source": "Autopsy", "icon": "ambulance" | |
| }) | |
| ci_match = re.search(r'Confidence interval[:\s]+(\d{2}):(\d{2})\s+-\s+(\d{2}):(\d{2})', text) | |
| if ci_match: | |
| h1, m1 = int(ci_match.group(1)), int(ci_match.group(2)) | |
| h2, m2 = int(ci_match.group(3)), int(ci_match.group(4)) | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": datetime(2024, 1, 15, h1, m1), | |
| "event": "Earliest TOD (confidence bound)", | |
| "source": "Autopsy", "icon": "bound" | |
| }) | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": datetime(2024, 1, 15, h2, m2), | |
| "event": "Latest TOD (confidence bound)", | |
| "source": "Autopsy", "icon": "bound" | |
| }) | |
| return events | |
| def parse_gps(text): | |
| events = [] | |
| lines = text.strip().split('\n') | |
| if len(lines) < 2: | |
| return events | |
| for line in lines[1:]: | |
| parts = line.split(',') | |
| if len(parts) >= 4: | |
| try: | |
| ts = datetime.strptime(parts[0].strip(), "%Y-%m-%d %H:%M:%S") | |
| lat, lon = float(parts[1]), float(parts[2]) | |
| event = parts[3].strip() | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": ts, | |
| "event": event, "source": "Phone GPS", "icon": "phone" | |
| }) | |
| except Exception: | |
| continue | |
| return events | |
| def parse_cctv(text): | |
| events = [] | |
| pattern = r'(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\|\s+([\w-]+)\s+\|\s+([\d.-]+),?\s+([\d.-]+)\s+\|\s+(.+)' | |
| for match in re.finditer(pattern, text): | |
| ts_str = match.group(1) | |
| cam_id = match.group(2) | |
| lat = float(match.group(3)) | |
| lon = float(match.group(4)) | |
| event = match.group(5).strip() | |
| ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") | |
| events.append({ | |
| "lat": lat, "lon": lon, "time": ts, | |
| "event": f"[{cam_id}] {event}", | |
| "source": "CCTV", "icon": "camera" | |
| }) | |
| return events | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # ANOMALY DETECTION ENGINE | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def detect_anomalies(events): | |
| anomalies = [] | |
| autopsy_events = [e for e in events if e["source"] == "Autopsy" and "Death" in e["event"]] | |
| if not autopsy_events: | |
| return anomalies | |
| tod = autopsy_events[0]["time"] | |
| tod_lat, tod_lon = autopsy_events[0]["lat"], autopsy_events[0]["lon"] | |
| gps_events = sorted([e for e in events if e["source"] == "Phone GPS"], key=lambda x: x["time"]) | |
| for ev in gps_events: | |
| if ev["time"] > tod: | |
| dist = np.sqrt((ev["lat"] - tod_lat)**2 + (ev["lon"] - tod_lon)**2) * 111 | |
| if dist > 1.0: | |
| anomalies.append({ | |
| "severity": "CRITICAL", | |
| "type": "Impossible Intersection", | |
| "message": f"Phone GPS shows device at ({ev['lat']:.4f}, {ev['lon']:.4f}) at {ev['time'].strftime('%H:%M')} — {int((ev['time']-tod).total_seconds()/60)} min AFTER estimated death ({tod.strftime('%H:%M')}). Distance from body: {dist:.1f} km.", | |
| "time": ev["time"], | |
| "lat": ev["lat"], "lon": ev["lon"] | |
| }) | |
| cctv_events = sorted([e for e in events if e["source"] == "CCTV"], key=lambda x: x["time"]) | |
| for ev in cctv_events: | |
| if ev["time"] > tod and "vehicle" in ev["event"].lower(): | |
| dist = np.sqrt((ev["lat"] - tod_lat)**2 + (ev["lon"] - tod_lon)**2) * 111 | |
| if dist > 1.0: | |
| anomalies.append({ | |
| "severity": "CRITICAL", | |
| "type": "Post-Mortem Vehicle Movement", | |
| "message": f"CCTV ({ev['event']}) at {ev['time'].strftime('%H:%M')} — {int((ev['time']-tod).total_seconds()/60)} min AFTER death. {dist:.1f} km from body.", | |
| "time": ev["time"], | |
| "lat": ev["lat"], "lon": ev["lon"] | |
| }) | |
| for ev in cctv_events: | |
| if ev["time"] > tod and "SUBJECT" in ev["event"]: | |
| anomalies.append({ | |
| "severity": "CRITICAL", | |
| "type": "Post-Mortem Sighting", | |
| "message": f"CCTV shows SUBJECT at {ev['time'].strftime('%H:%M')} — AFTER estimated death. Possible identity theft or recording tampering.", | |
| "time": ev["time"], | |
| "lat": ev["lat"], "lon": ev["lon"] | |
| }) | |
| return anomalies | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # VISUALIZATION: TIME-REVEAL EFFECT | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| SOURCE_COLORS = { | |
| "Autopsy": "#FF4444", | |
| "Phone GPS": "#44AA44", | |
| "CCTV": "#4488FF" | |
| } | |
| DIM_COLOR = "#555555" | |
| def hex_to_rgba(hex_color, alpha): | |
| """Convert hex color to rgba string for Plotly.""" | |
| hex_color = hex_color.lstrip('#') | |
| r = int(hex_color[0:2], 16) | |
| g = int(hex_color[2:4], 16) | |
| b = int(hex_color[4:6], 16) | |
| return f"rgba({r},{g},{b},{alpha})" | |
| def time_to_float(t): | |
| return t.hour + t.minute / 60.0 + t.second / 3600.0 | |
| def build_visualization(events, slider_time_float): | |
| if not events: | |
| return go.Figure(), go.Figure() | |
| df = pd.DataFrame(events) | |
| df["time_f"] = df["time"].apply(time_to_float) | |
| df["time_str"] = df["time"].apply(lambda t: t.strftime("%H:%M")) | |
| slider_time = datetime(2024, 1, 15, int(slider_time_float), int((slider_time_float % 1) * 60)) | |
| nearest_idx = (df["time_f"] - slider_time_float).abs().idxmin() | |
| nearest = df.loc[nearest_idx] | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # 2D GEO MAP | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| fig_map = go.Figure() | |
| def reveal_state(t_f): | |
| diff = abs(t_f - slider_time_float) | |
| if diff < 0.5: | |
| return {"opacity": 1.0, "size": 18, "color": None, "show_text": True} | |
| elif diff < 1.5: | |
| return {"opacity": 0.35, "size": 10, "color": DIM_COLOR, "show_text": False} | |
| else: | |
| return {"opacity": 0.08, "size": 5, "color": DIM_COLOR, "show_text": False} | |
| for source in df["source"].unique(): | |
| sub = df[df["source"] == source] | |
| states = [reveal_state(row["time_f"]) for _, row in sub.iterrows()] | |
| sizes = [s["size"] for s in states] | |
| opacities = [s["opacity"] for s in states] | |
| colors = [SOURCE_COLORS.get(source, "#888") if s["color"] is None else s["color"] for s in states] | |
| texts = [row["event"] if states[i]["show_text"] else "" for i, (_, row) in enumerate(sub.iterrows())] | |
| fig_map.add_trace(go.Scattergeo( | |
| lon=sub["lon"].tolist(), | |
| lat=sub["lat"].tolist(), | |
| mode='markers+text', | |
| marker=dict(size=sizes, color=colors, opacity=opacities, line=dict(width=1, color='black')), | |
| text=texts, | |
| textposition="top center", | |
| textfont=dict(size=9, color=SOURCE_COLORS.get(source, "#888")), | |
| name=source, | |
| hoverinfo='text', | |
| hovertext=[f"{row['event']}<br>Time: {row['time_str']}<br>Lat: {row['lat']:.4f}, Lon: {row['lon']:.4f}" for _, row in sub.iterrows()] | |
| )) | |
| # GPS trajectory | |
| gps_df = df[df["source"] == "Phone GPS"].sort_values("time_f") | |
| for i in range(len(gps_df) - 1): | |
| r1 = gps_df.iloc[i] | |
| r2 = gps_df.iloc[i + 1] | |
| seg_mid = (r1["time_f"] + r2["time_f"]) / 2.0 | |
| diff = abs(seg_mid - slider_time_float) | |
| if diff < 0.5: | |
| lw, lc, lo = 4, SOURCE_COLORS["Phone GPS"], 1.0 | |
| elif diff < 1.5: | |
| lw, lc, lo = 2, DIM_COLOR, 0.3 | |
| else: | |
| lw, lc, lo = 1, DIM_COLOR, 0.05 | |
| fig_map.add_trace(go.Scattergeo( | |
| lon=[r1["lon"], r2["lon"]], lat=[r1["lat"], r2["lat"]], mode='lines', | |
| line=dict(color=lc, width=lw), opacity=lo, hoverinfo='skip', showlegend=False | |
| )) | |
| # CCTV dashed connections | |
| cctv_df = df[df["source"] == "CCTV"].sort_values("time_f") | |
| for i in range(len(cctv_df) - 1): | |
| r1 = cctv_df.iloc[i] | |
| r2 = cctv_df.iloc[i + 1] | |
| seg_mid = (r1["time_f"] + r2["time_f"]) / 2.0 | |
| diff = abs(seg_mid - slider_time_float) | |
| if diff < 0.5: | |
| lw, lc, lo = 3, SOURCE_COLORS["CCTV"], 0.8 | |
| elif diff < 1.5: | |
| lw, lc, lo = 1, DIM_COLOR, 0.2 | |
| else: | |
| lw, lc, lo = 1, DIM_COLOR, 0.03 | |
| fig_map.add_trace(go.Scattergeo( | |
| lon=[r1["lon"], r2["lon"]], lat=[r1["lat"], r2["lat"]], mode='lines', | |
| line=dict(color=lc, width=lw, dash='dot'), opacity=lo, hoverinfo='skip', showlegend=False | |
| )) | |
| # Gold highlight ring | |
| fig_map.add_trace(go.Scattergeo( | |
| lon=[nearest["lon"]], lat=[nearest["lat"]], mode='markers', | |
| marker=dict(size=30, color='rgba(0,0,0,0)', line=dict(width=3, color='gold')), | |
| name='Current Time Focus', hoverinfo='skip' | |
| )) | |
| fig_map.update_layout( | |
| title=dict(text=f"Chronos-Graph 2D Map — Time: {slider_time.strftime('%H:%M')}", font=dict(size=14, color='white')), | |
| geo=dict( | |
| scope='usa', center=dict(lat=40.75, lon=-73.98), projection_scale=150, | |
| showland=True, landcolor='rgb(30,30,35)', subunitcolor='rgb(60,60,70)', countrycolor='rgb(60,60,70)', | |
| showsubunits=True, showcountries=True, resolution=50, | |
| lonaxis=dict(range=[-74.3, -73.6]), lataxis=dict(range=[40.5, 41.0]), | |
| ), | |
| paper_bgcolor='rgb(15,15,20)', plot_bgcolor='rgb(15,15,20)', | |
| font=dict(color='white'), height=550, margin=dict(l=0, r=0, t=50, b=0), | |
| legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01, bgcolor='rgba(0,0,0,0.5)', font=dict(color='white')) | |
| ) | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # 3D SPACE-TIME GRAPH — FIXED: no per-point opacity lists | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| fig_3d = go.Figure() | |
| for i in range(len(gps_df) - 1): | |
| r1 = gps_df.iloc[i] | |
| r2 = gps_df.iloc[i + 1] | |
| seg_mid = (r1["time_f"] + r2["time_f"]) / 2.0 | |
| diff = abs(seg_mid - slider_time_float) | |
| if diff < 0.5: | |
| lw, lc, lo = 5, SOURCE_COLORS["Phone GPS"], 1.0 | |
| elif diff < 1.5: | |
| lw, lc, lo = 2, DIM_COLOR, 0.3 | |
| else: | |
| lw, lc, lo = 1, DIM_COLOR, 0.05 | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[r1["lon"], r2["lon"]], y=[r1["lat"], r2["lat"]], z=[r1["time_f"], r2["time_f"]], | |
| mode='lines', line=dict(color=lc, width=lw), opacity=lo, hoverinfo='skip', showlegend=False | |
| )) | |
| for i in range(len(cctv_df) - 1): | |
| r1 = cctv_df.iloc[i] | |
| r2 = cctv_df.iloc[i + 1] | |
| seg_mid = (r1["time_f"] + r2["time_f"]) / 2.0 | |
| diff = abs(seg_mid - slider_time_float) | |
| if diff < 0.5: | |
| lw, lc, lo = 4, SOURCE_COLORS["CCTV"], 0.8 | |
| elif diff < 1.5: | |
| lw, lc, lo = 2, DIM_COLOR, 0.25 | |
| else: | |
| lw, lc, lo = 1, DIM_COLOR, 0.04 | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[r1["lon"], r2["lon"]], y=[r1["lat"], r2["lat"]], z=[r1["time_f"], r2["time_f"]], | |
| mode='lines', line=dict(color=lc, width=lw, dash='dot'), opacity=lo, hoverinfo='skip', showlegend=False | |
| )) | |
| # 3D points: per-source, per-opacity-band (bright / fade / dim) to avoid per-point opacity lists | |
| for source in df["source"].unique(): | |
| sub = df[df["source"] == source].sort_values("time_f") | |
| base_color = SOURCE_COLORS.get(source, "#888888") | |
| # Split into 3 opacity bands and emit separate traces | |
| bright_pts = [] | |
| fade_pts = [] | |
| dim_pts = [] | |
| for _, row in sub.iterrows(): | |
| diff = abs(row["time_f"] - slider_time_float) | |
| pt = { | |
| "x": row["lon"], "y": row["lat"], "z": row["time_f"], | |
| "text": f"{row['event']}<br>Time: {row['time_str']}<br>Lat: {row['lat']:.4f}, Lon: {row['lon']:.4f}" | |
| } | |
| if diff < 0.5: | |
| bright_pts.append(pt) | |
| elif diff < 1.5: | |
| fade_pts.append(pt) | |
| else: | |
| dim_pts.append(pt) | |
| # Bright trace | |
| if bright_pts: | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[p["x"] for p in bright_pts], | |
| y=[p["y"] for p in bright_pts], | |
| z=[p["z"] for p in bright_pts], | |
| mode='markers', | |
| marker=dict(size=12, color=base_color, opacity=1.0, line=dict(width=2, color='black')), | |
| name=f"{source} (active)", | |
| hoverinfo='text', | |
| hovertext=[p["text"] for p in bright_pts] | |
| )) | |
| # Fade trace | |
| if fade_pts: | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[p["x"] for p in fade_pts], | |
| y=[p["y"] for p in fade_pts], | |
| z=[p["z"] for p in fade_pts], | |
| mode='markers', | |
| marker=dict(size=7, color=DIM_COLOR, opacity=0.35, line=dict(width=1, color='black')), | |
| name=f"{source} (fading)", | |
| hoverinfo='text', | |
| hovertext=[p["text"] for p in fade_pts] | |
| )) | |
| # Dim trace | |
| if dim_pts: | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[p["x"] for p in dim_pts], | |
| y=[p["y"] for p in dim_pts], | |
| z=[p["z"] for p in dim_pts], | |
| mode='markers', | |
| marker=dict(size=4, color=DIM_COLOR, opacity=0.08, line=dict(width=1, color='black')), | |
| name=f"{source} (inactive)", | |
| hoverinfo='text', | |
| hovertext=[p["text"] for p in dim_pts] | |
| )) | |
| # Gold highlight ring in 3D | |
| fig_3d.add_trace(go.Scatter3d( | |
| x=[nearest["lon"]], y=[nearest["lat"]], z=[nearest["time_f"]], | |
| mode='markers', | |
| marker=dict(size=20, color='rgba(0,0,0,0)', line=dict(width=3, color='gold')), | |
| name='Current Time Focus', hoverinfo='skip' | |
| )) | |
| # Cyan time plane | |
| fig_3d.add_trace(go.Mesh3d( | |
| x=[-74.3, -73.6, -73.6, -74.3], y=[40.5, 40.5, 41.0, 41.0], z=[slider_time_float]*4, | |
| color='cyan', opacity=0.2, name='Current Time Plane', hoverinfo='skip' | |
| )) | |
| tick_vals = list(range(18, 25)) | |
| tick_text = [f"{h:02d}:00" for h in tick_vals] | |
| fig_3d.update_layout( | |
| title=dict(text=f"4D Chronos-Graph — Time: {slider_time.strftime('%H:%M')}", font=dict(size=14, color='white')), | |
| scene=dict( | |
| xaxis_title="Longitude", yaxis_title="Latitude", zaxis_title="Time", | |
| xaxis=dict(range=[-74.3, -73.6], dtick=0.1), | |
| yaxis=dict(range=[40.5, 41.0], dtick=0.1), | |
| zaxis=dict(range=[18, 24], dtick=0.5, ticktext=tick_text, tickvals=tick_vals), | |
| aspectmode='manual', aspectratio=dict(x=2, y=2, z=1.2), | |
| camera=dict(eye=dict(x=1.5, y=1.5, z=0.8)), | |
| ), | |
| paper_bgcolor='rgb(15,15,20)', plot_bgcolor='rgb(15,15,20)', | |
| font=dict(color='white'), height=650, margin=dict(l=0, r=0, t=50, b=0), | |
| legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01, bgcolor='rgba(0,0,0,0.5)', font=dict(color='white')) | |
| ) | |
| return fig_map, fig_3d | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # GRADIO UI | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def process_all(autopsy_text, gps_text, cctv_text, slider_hour): | |
| try: | |
| events = [] | |
| events.extend(parse_autopsy(autopsy_text)) | |
| events.extend(parse_gps(gps_text)) | |
| events.extend(parse_cctv(cctv_text)) | |
| anomalies = detect_anomalies(events) | |
| if anomalies: | |
| anomaly_html = "<div style='background:#ffebee; border-left:5px solid #f44336; padding:15px; margin:10px 0; border-radius:4px;'>" | |
| anomaly_html += "<h3 style='color:#b71c1c;margin-top:0;'>ANOMALIES DETECTED</h3>" | |
| for a in anomalies: | |
| anomaly_html += f"<p style='margin:8px 0;'><b style='color:#d32f2f;'>[{a['severity']}] {a['type']}</b><br>{a['message']}</p>" | |
| anomaly_html += "</div>" | |
| else: | |
| anomaly_html = "<div style='background:#e8f5e9; border-left:5px solid #4caf50; padding:15px; margin:10px 0; border-radius:4px;'>" | |
| anomaly_html += "<h3 style='color:#1b5e20;margin-top:0;'>No temporal impossibilities detected.</h3>" | |
| anomaly_html += "</div>" | |
| fig_map, fig_3d = build_visualization(events, slider_hour) | |
| table_html = "<table style='width:100%; border-collapse:collapse; font-size:12px; font-family:monospace;'>" | |
| table_html += "<tr style='background:#222; color:#fff;'><th>Source</th><th>Time</th><th>Lat</th><th>Lon</th><th>Event</th></tr>" | |
| for e in sorted(events, key=lambda x: x["time"]): | |
| color = SOURCE_COLORS.get(e["source"], "#888") | |
| table_html += f"<tr style='border-bottom:1px solid #333; color:#ddd;'>" | |
| table_html += f"<td style='color:{color}; font-weight:bold;'>{e['source']}</td>" | |
| table_html += f"<td>{e['time'].strftime('%H:%M')}</td>" | |
| table_html += f"<td>{e['lat']:.4f}</td><td>{e['lon']:.4f}</td>" | |
| table_html += f"<td>{e['event']}</td></tr>" | |
| table_html += "</table>" | |
| return fig_map, fig_3d, anomaly_html, table_html, len(events) | |
| except Exception as e: | |
| err_msg = f"<pre style='color:red; background:#fee; padding:10px; border-radius:4px;'>{traceback.format_exc()}</pre>" | |
| return go.Figure(), go.Figure(), err_msg, "", 0 | |
| with gr.Blocks(title="Chronos-Graph: 4D Forensic Timeline") as demo: | |
| gr.Markdown(""" | |
| <div style="text-align:center; padding:10px 0 20px 0;"> | |
| <h1 style="font-size:2.2em; margin-bottom:5px;">⏳ Chronos-Graph</h1> | |
| <p style="font-size:1.1em; color:#888; margin:0;"> | |
| The 4D Unified Forensic Timeline — <b>Drag the Time Slider to reveal evidence</b> | |
| </p> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("<h3 style='color:#555; margin-top:0;'>📥 Ingestion Engine</h3>") | |
| autopsy_input = gr.TextArea(label="Autopsy Report (unstructured text)", value=MOCK_AUTOPSY, lines=12) | |
| gps_input = gr.TextArea(label="Phone GPS Metadata (CSV)", value=MOCK_PHONE_GPS, lines=8) | |
| cctv_input = gr.TextArea(label="CCTV Logs", value=MOCK_CCTV, lines=8) | |
| slider_time = gr.Slider(minimum=18.0, maximum=24.0, value=21.5, step=0.25, label="🕐 Time Slider") | |
| process_btn = gr.Button("🔍 PARSE & VISUALIZE", variant="primary") | |
| events_count = gr.Number(label="Parsed Events", interactive=False) | |
| with gr.Column(scale=3): | |
| with gr.Row(): | |
| anomaly_output = gr.HTML(label="Anomaly Detection") | |
| with gr.Tabs(): | |
| with gr.Tab("🗺️ 2D Map View"): | |
| map_plot = gr.Plot(label="Geographic Evidence Plot") | |
| with gr.Tab("⏳ 3D Space-Time Graph"): | |
| graph_3d = gr.Plot(label="4D Chronos-Graph (X=Lon, Y=Lat, Z=Time)") | |
| with gr.Tab("📊 Parsed Events Table"): | |
| table_output = gr.HTML(label="Structured Events") | |
| slider_time.release( | |
| fn=process_all, | |
| inputs=[autopsy_input, gps_input, cctv_input, slider_time], | |
| outputs=[map_plot, graph_3d, anomaly_output, table_output, events_count] | |
| ) | |
| slider_time.change( | |
| fn=process_all, | |
| inputs=[autopsy_input, gps_input, cctv_input, slider_time], | |
| outputs=[map_plot, graph_3d, anomaly_output, table_output, events_count] | |
| ) | |
| process_btn.click( | |
| fn=process_all, | |
| inputs=[autopsy_input, gps_input, cctv_input, slider_time], | |
| outputs=[map_plot, graph_3d, anomaly_output, table_output, events_count] | |
| ) | |
| demo.load( | |
| fn=process_all, | |
| inputs=[autopsy_input, gps_input, cctv_input, slider_time], | |
| outputs=[map_plot, graph_3d, anomaly_output, table_output, events_count] | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() | |