Buckets:
| """ | |
| Traffix — Synapse read-only demo on Hugging Face Spaces. | |
| What this demo shows: | |
| • The Synapse explainability service answers questions about recorded traffic decisions. | |
| • No traffic simulator is running. No signals are actuated. | |
| • Evidence events are synthetic fixtures that match the schema of a real traffix run. | |
| Built by ESPRIT 4DS students (azizmessaoud/traffix). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from pathlib import Path | |
| # spaces MUST come before any CUDA-touching import (ZeroGPU rule). | |
| # This app is CPU-only; we include a no-op @spaces.GPU so the Space | |
| # can be hosted on zero-a10g without needing a paid plan. | |
| import spaces # noqa: E402 (keep first) | |
| sys.path.insert(0, str(Path(__file__).parent / "src")) | |
| from coflow5.synapse.service import EvidenceEvent, SynapseService # noqa: E402 | |
| import gradio as gr # noqa: E402 | |
| # --------------------------------------------------------------------------- | |
| # Demo evidence: realistic synthetic events that match the traffix schema | |
| # --------------------------------------------------------------------------- | |
| DEMO_EVENTS: dict[str, EvidenceEvent] = { | |
| "evt-j1-morning-peak": EvidenceEvent( | |
| run_id="demo-readonly-0001", | |
| scenario_hash="sha256:demo", | |
| event_id="evt-j1-morning-peak", | |
| message_id="msg-001", | |
| bundle_digest="sha256:demo-bundle-abc123", | |
| facts={ | |
| "junction": "J1", | |
| "simulation_time": "08:07:23", | |
| "selected_action": "PHASE_GREEN_NS (north-south green, 34 s)", | |
| "reason_code": ( | |
| "max-pressure algorithm detected 14 queued vehicles on the " | |
| "north approach vs 2 on east-west; pressure ratio 7:1" | |
| ), | |
| "controller": "traffix-hierarchy (A1)", | |
| "emergency_override": False, | |
| }, | |
| ), | |
| "evt-j3-ambulance": EvidenceEvent( | |
| run_id="demo-readonly-0001", | |
| scenario_hash="sha256:demo", | |
| event_id="evt-j3-ambulance", | |
| message_id="msg-042", | |
| bundle_digest="sha256:demo-bundle-abc123", | |
| facts={ | |
| "junction": "J3", | |
| "simulation_time": "08:15:44", | |
| "selected_action": "EMERGENCY_PREEMPT (all-red 4.2 s → green on ambulance corridor)", | |
| "reason_code": ( | |
| "ambulance1 advisory received from A4 adviser; safety mask " | |
| "triggered preemption with 4.2 s clearance time; " | |
| "conflicting phases held red" | |
| ), | |
| "controller": "traffix-hierarchy (A1 + A4 adviser)", | |
| "emergency_override": True, | |
| }, | |
| ), | |
| "evt-j2-bus-priority": EvidenceEvent( | |
| run_id="demo-readonly-0001", | |
| scenario_hash="sha256:demo", | |
| event_id="evt-j2-bus-priority", | |
| message_id="msg-019", | |
| bundle_digest="sha256:demo-bundle-abc123", | |
| facts={ | |
| "junction": "J2", | |
| "simulation_time": "08:11:05", | |
| "selected_action": "PHASE_GREEN_EW extended +12 s (east-west green)", | |
| "reason_code": ( | |
| "morning_peak calendar scenario active; A3 bus-priority adviser " | |
| "promoted bus line 12 (34 passengers, 3 min late); " | |
| "extension capped at schedule-recovery maximum" | |
| ), | |
| "controller": "traffix-hierarchy (A1 + A3 adviser)", | |
| "emergency_override": False, | |
| }, | |
| ), | |
| "evt-j5-fallback": EvidenceEvent( | |
| run_id="demo-readonly-0001", | |
| scenario_hash="sha256:demo", | |
| event_id="evt-j5-fallback", | |
| message_id="msg-088", | |
| bundle_digest="sha256:demo-bundle-abc123", | |
| facts={ | |
| "junction": "J5", | |
| "simulation_time": "08:33:17", | |
| "selected_action": "PHASE_FIXED_CYCLE (fixed 60 s cycle)", | |
| "reason_code": ( | |
| "fallback probe detected A1 heartbeat gap > 5 s; " | |
| "recovery module engaged fixed-cycle baseline; " | |
| "A1 resumed after 12 s" | |
| ), | |
| "controller": "recovery-fallback", | |
| "emergency_override": False, | |
| }, | |
| ), | |
| } | |
| DEMO_POLICIES: dict[str, str] = { | |
| "policy-max-pressure": ( | |
| "Max-pressure: always serve the phase whose product of upstream queue " | |
| "and downstream vacancy is highest. Proven throughput-optimal under " | |
| "independent queues." | |
| ), | |
| "policy-emergency-preemption": ( | |
| "Emergency preemption: on receiving an ambulance advisory, hold all " | |
| "conflicting phases red for a clearance interval, then give green on " | |
| "the emergency corridor until the vehicle clears." | |
| ), | |
| "policy-bus-priority": ( | |
| "Bus priority: when A3 detects a late bus with passengers, request a " | |
| "green extension on its approach, bounded by the schedule-recovery cap." | |
| ), | |
| "policy-fallback": ( | |
| "Fallback: if A1 misses its heartbeat for > 5 s, the recovery module " | |
| "switches to a pre-configured fixed cycle and alerts the operator." | |
| ), | |
| } | |
| synapse = SynapseService( | |
| events=DEMO_EVENTS, | |
| policies=DEMO_POLICIES, | |
| corpus_hash="sha256:demo-corpus-v1", | |
| ) | |
| HEALTH_RESPONSE = { | |
| "status": "ok", | |
| "actuates": False, | |
| "sumo": False, | |
| "space": "huggingface-demo", | |
| "claim_flags": [ | |
| "simulation is not deployment evidence", | |
| "no lives saved — simulated emergency times only", | |
| "no measured air quality — HBEFA proxies only", | |
| "evidence events in this demo are synthetic fixtures", | |
| ], | |
| "note": ( | |
| "This Space runs the Synapse read-only explainer. " | |
| "No traffic simulator is active." | |
| ), | |
| } | |
| # --------------------------------------------------------------------------- | |
| # ZeroGPU no-op — required to host on zero-a10g without a paid plan. | |
| # This app is CPU-only; the decorator is never triggered during normal use. | |
| # --------------------------------------------------------------------------- | |
| def _gpu_noop() -> None: # pragma: no cover | |
| """Placeholder so HF accepts this Space on zero-a10g hardware.""" | |
| pass | |
| # --------------------------------------------------------------------------- | |
| # Business logic (pure CPU) | |
| # --------------------------------------------------------------------------- | |
| def check_health() -> str: | |
| """Return the API health status as formatted JSON.""" | |
| return json.dumps(HEALTH_RESPONSE, indent=2) | |
| def list_runs() -> str: | |
| """List available evidence bundles.""" | |
| runs = [ | |
| { | |
| "run_id": "demo-readonly-0001", | |
| "status": "fixture", | |
| "controller": "traffix-hierarchy", | |
| "scenario": "morning_peak (synthetic)", | |
| "junctions": ["J1", "J2", "J3", "J4", "J5", "J6"], | |
| "events_available": list(DEMO_EVENTS.keys()), | |
| "note": ( | |
| "Demo fixture. Replace with a real evidence bundle " | |
| "produced by scripts/run_experiments.py." | |
| ), | |
| } | |
| ] | |
| return json.dumps(runs, indent=2) | |
| def list_events() -> list[list]: | |
| """Return a table of available demo events.""" | |
| rows = [] | |
| for eid, ev in DEMO_EVENTS.items(): | |
| rows.append( | |
| [ | |
| eid, | |
| ev.facts.get("junction", ""), | |
| ev.facts.get("simulation_time", ""), | |
| str(ev.facts.get("emergency_override", False)), | |
| str(ev.facts.get("selected_action", ""))[:60], | |
| ] | |
| ) | |
| return rows | |
| def ask_synapse(event_id: str, question: str, policy_id: str) -> tuple[str, str]: | |
| """ | |
| Ask Synapse to explain a recorded traffic decision. | |
| Args: | |
| event_id: The decision event to query (from the dropdown). | |
| question: A natural-language question about the decision. | |
| policy_id: Optional policy citation to attach to the answer. | |
| Returns: | |
| Tuple of (formatted answer markdown, provenance JSON). | |
| """ | |
| if not event_id or not question.strip(): | |
| return "⚠️ Please select an event and enter a question.", "{}" | |
| pid = policy_id if policy_id and policy_id != "none" else None | |
| result = synapse.explain(question.strip(), event_id=event_id, policy_id=pid) | |
| provenance_json = json.dumps(result.provenance.as_dict(), indent=2) | |
| fallback_label = { | |
| "verified": "✅ verified (policy cited)", | |
| "template": "📋 template (no policy)", | |
| "abstain": "🚫 abstained (event not found)", | |
| }.get(result.fallback, result.fallback) | |
| citations_text = ( | |
| ", ".join(c for c in result.citations if c) or "none" | |
| ) | |
| answer_md = f"""### Answer | |
| {result.answer} | |
| --- | |
| **Fallback mode:** {fallback_label} | |
| **Citations:** `{citations_text}` | |
| """ | |
| return answer_md, provenance_json | |
| def get_metrics() -> str: | |
| """Return Synapse service metrics.""" | |
| return json.dumps(synapse.metrics(), indent=2) | |
| # --------------------------------------------------------------------------- | |
| # Gradio UI | |
| # --------------------------------------------------------------------------- | |
| EVENTS_LIST = list(DEMO_EVENTS.keys()) | |
| POLICIES_LIST = ["none"] + list(DEMO_POLICIES.keys()) | |
| with gr.Blocks( | |
| title="Traffix — Synapse Demo", | |
| theme=gr.themes.Soft(), | |
| ) as demo: | |
| gr.Markdown( | |
| """ | |
| # 🚦 Traffix · Synapse Decision Explainer | |
| **Multi-agent AI traffic signal control · Tunis city centre (J1–J6)** | |
| *ESPRIT 4DS capstone project · [azizmessaoud/traffix](https://github.com/azizmessaoud/traffix)* | |
| > ⚠️ **Simulation only.** This Space runs the Synapse read-only API on pre-recorded | |
| > decision logs. No traffic simulator is active. No signals are actuated. | |
| > Events shown are synthetic fixtures that match the real schema. | |
| """ | |
| ) | |
| with gr.Tab("🏥 Health"): | |
| gr.Markdown( | |
| "Check what the API reports about itself — including the honesty flags " | |
| "the project requires every hosted endpoint to expose." | |
| ) | |
| health_btn = gr.Button("Check API health", variant="primary") | |
| health_out = gr.Code(language="json", label="Health response") | |
| health_btn.click(fn=check_health, outputs=health_out) | |
| with gr.Tab("📁 Runs"): | |
| gr.Markdown( | |
| "Evidence bundles produced by `scripts/run_experiments.py`. " | |
| "This demo ships with one synthetic fixture." | |
| ) | |
| runs_btn = gr.Button("List evidence runs") | |
| runs_out = gr.Code(language="json", label="Available runs") | |
| runs_btn.click(fn=list_runs, outputs=runs_out) | |
| gr.Markdown("### Events in this run") | |
| events_table = gr.Dataframe( | |
| headers=["Event ID", "Junction", "Time", "Emergency", "Action (truncated)"], | |
| value=list_events(), | |
| interactive=False, | |
| ) | |
| with gr.Tab("🔍 Ask Synapse"): | |
| gr.Markdown( | |
| """ | |
| Ask the Synapse service *why* an agent made a specific signal decision. | |
| Synapse reads from the decision log — it never touches the simulator. | |
| **How it works:** | |
| 1. Pick a recorded decision event | |
| 2. Optionally cite a policy the decision should be grounded in | |
| 3. Ask your question in plain English | |
| """ | |
| ) | |
| with gr.Row(): | |
| event_dd = gr.Dropdown( | |
| choices=EVENTS_LIST, | |
| label="Event ID", | |
| value=EVENTS_LIST[0], | |
| ) | |
| policy_dd = gr.Dropdown( | |
| choices=POLICIES_LIST, | |
| label="Policy citation (optional)", | |
| value="none", | |
| ) | |
| question_box = gr.Textbox( | |
| label="Question", | |
| placeholder="Why did the agent make this decision?", | |
| lines=2, | |
| ) | |
| ask_btn = gr.Button("Ask Synapse", variant="primary") | |
| with gr.Row(): | |
| answer_out = gr.Markdown(label="Answer") | |
| prov_out = gr.Code(language="json", label="Provenance (audit trail)") | |
| ask_btn.click( | |
| fn=ask_synapse, | |
| inputs=[event_dd, question_box, policy_dd], | |
| outputs=[answer_out, prov_out], | |
| ) | |
| gr.Examples( | |
| examples=[ | |
| [ | |
| "evt-j1-morning-peak", | |
| "Why did the agent give green to north-south traffic at J1?", | |
| "policy-max-pressure", | |
| ], | |
| [ | |
| "evt-j3-ambulance", | |
| "Why was an emergency preemption triggered at J3?", | |
| "policy-emergency-preemption", | |
| ], | |
| [ | |
| "evt-j2-bus-priority", | |
| "Why was the bus given priority over cars at J2?", | |
| "policy-bus-priority", | |
| ], | |
| [ | |
| "evt-j5-fallback", | |
| "Why did J5 fall back to a fixed cycle?", | |
| "policy-fallback", | |
| ], | |
| ], | |
| inputs=[event_dd, question_box, policy_dd], | |
| ) | |
| with gr.Tab("📊 Metrics"): | |
| gr.Markdown( | |
| "Live Synapse service metrics: request count, fallback rates, latency." | |
| ) | |
| metrics_btn = gr.Button("Get metrics") | |
| metrics_out = gr.Code(language="json", label="Synapse metrics") | |
| metrics_btn.click(fn=get_metrics, outputs=metrics_out) | |
| with gr.Tab("🏗️ Architecture"): | |
| gr.Markdown( | |
| """ | |
| ## Two-tier agent hierarchy | |
| ``` | |
| ╔══════════════════════════════ NETWORK LEVEL ══════════════════════════════╗ | |
| ║ ║ | |
| ║ 8 Network Agents ┌─────────────────────┐ ║ | |
| ║ ───────────────── │ Supermaster │ ║ | |
| ║ • Demand forecasting ──outputs──► │ Network Coordinator │ ║ | |
| ║ • Incident detection │ │ ║ | |
| ║ • Emergency routing │ • Aggregates network │ directives ║ | |
| ║ • Corridor monitoring │ info across all J │ ──────────► ║ | |
| ║ • Transit monitoring │ • Coordinates │ ║ | |
| ║ • Weather analysis │ network strategy │ ◄────────── ║ | |
| ║ • … (2 others) │ • ML models for │ status / ║ | |
| ║ │ demand & incidents │ ack ║ | |
| ║ └─────────────────────┘ ║ | |
| ╚══════════════════════════════════════════════════════════════════════════╝ | |
| │ directives (JSON) | |
| ▼ | |
| ╔═══════════════════════ JUNCTION LEVEL (×6) ═══════════════════════════════╗ | |
| ║ ║ | |
| ║ Real-time sensors (loops, cameras, detectors) ║ | |
| ║ │ ║ | |
| ║ ▼ state data (JSON / Python objects) ║ | |
| ║ ┌──────────────────────────────────────────────────────────┐ ║ | |
| ║ │ 5 Local Agents (per junction) │ ║ | |
| ║ │ │ ║ | |
| ║ │ 🚦 Traffic-Light 👟 Pedestrian 🚌 Transit-Priority │ ║ | |
| ║ │ checks queues checks wait checks bus ETA │ ║ | |
| ║ │ proposes phase requests requests priority │ ║ | |
| ║ │ (Max-Pressure) crossing │ ║ | |
| ║ │ │ ║ | |
| ║ │ 🚑 Emergency-Priority 📡 Sensor-Health │ ║ | |
| ║ │ detects ambulances checks data freshness │ ║ | |
| ║ │ requests preemption flags DATA_STALE issues │ ║ | |
| ║ └───────────────────┬──────────────────────────────────────┘ ║ | |
| ║ │ requests / proposals / status (JSON) ║ | |
| ║ ▼ ║ | |
| ║ ┌──────────────────────────────────────────────────────────┐ ║ | |
| ║ │ Junction Master (Decision Coordinator) │ ║ | |
| ║ │ │ ║ | |
| ║ │ Priority rules: Safety ▸ Emergency ▸ Supermaster │ ║ | |
| ║ │ ▸ Transit ▸ Pedestrian ▸ Normal │ ║ | |
| ║ │ │ ║ | |
| ║ │ Evaluates agent proposals using: │ ║ | |
| ║ │ • Rule-based logic • Traffic control strategies │ ║ | |
| ║ │ • Current state (queues, phase, directives) │ ║ | |
| ║ └───────────────────┬──────────────────────────────────────┘ ║ | |
| ║ │ selected action (JSON) ║ | |
| ║ ▼ ║ | |
| ║ ┌──────────┐ permitted ┌──────────┐ command ┌──────────┐ ║ | |
| ║ │ Safety │ ──────────────► │ Executor │ ────TraCI───► │ SUMO │ ║ | |
| ║ │ Check │ action │ │ │ (sim) │ ║ | |
| ║ └──────────┘ └──────────┘ └──────────┘ ║ | |
| ║ ║ | |
| ║ Safety Check validates: signal-phase constraints · clearance times ║ | |
| ║ pedestrian safety rules · weather conditions ║ | |
| ║ ║ | |
| ║ Every decision ─── JSONL log ──────────────────────────────────────────► ║ | |
| ╚═══════════════════════════════════════════════════════════════════════════╝ | |
| │ decision logs / events (JSONL) | |
| ▼ | |
| ╔════════════════════════ SYNAPSE (this Space) ═════════════════════════════╗ | |
| ║ Read-only · Never actuates signals · Never runs SUMO ║ | |
| ║ Answers natural-language questions from the decision log ║ | |
| ║ Falls back to templates when no LLM key is set ║ | |
| ║ └── feeds ──► Dashboard / API (visualisation + XAI) ║ | |
| ╚═══════════════════════════════════════════════════════════════════════════╝ | |
| ``` | |
| ## Local agent roles (5 per junction) | |
| | Agent | Inputs | Action | | |
| |-------|--------|--------| | |
| | **Traffic-Light Agent** | Queue lengths, downstream vacancies | Proposes phase via Max-Pressure algorithm | | |
| | **Pedestrian Agent** | Pedestrian count, waiting time, thresholds | Requests crossing green when threshold exceeded | | |
| | **Transit-Priority Agent** | Bus lateness, headway, ETA, passenger count | Requests green extension on bus approach | | |
| | **Emergency-Priority Agent** | Vehicle location, distance, ETA | Requests all-red clearance + corridor green | | |
| | **Sensor-Health Agent** | Data freshness, health indicators | Flags DATA_STALE; triggers fallback probe | | |
| ## Network agents (8 at Supermaster level) | |
| Demand forecasting · Incident detection · Emergency route planning · | |
| Corridor monitoring · Transit monitoring · Weather analysis + 2 others. | |
| Each publishes to the Supermaster via JSON messages; the Supermaster | |
| pushes directives back to all Junction Masters. | |
| ## Architecture invariants (checked by `scripts/check_architecture.py`) | |
| - ✅ **Only one Junction Master** per light — no race conditions on signal write | |
| - ✅ **Safety Check** runs on every proposed action before TraCI command is issued | |
| - ✅ **Reason code logged before action applied** — full audit trail in JSONL | |
| - ✅ **Synapse is read-only** — it never writes to the simulation or TraCI | |
| - ✅ **Emergency clearance time** (4.2 s) held on conflicting phases before corridor green | |
| - ✅ **Fallback probe** restores fixed cycle if Junction Master heartbeat is missed > 5 s | |
| ## Message format (JSON) | |
| ```json | |
| { | |
| "type": "Request", | |
| "source": "pedestrian_agent", | |
| "junction_id": "J2", | |
| "priority": "pedestrian", | |
| "data": { | |
| "waiting_count": 8, | |
| "crossing_time_est": 20 | |
| }, | |
| "timestamp": "2025-05-15T10:24:08Z" | |
| } | |
| ``` | |
| ## Honesty flags | |
| - `simulation is not deployment evidence` — SUMO metrics ≠ real-world proof | |
| - `no lives saved` — simulated emergency times only | |
| - `no measured air quality` — HBEFA emission proxies only | |
| - `evidence events in this demo are synthetic fixtures` | |
| """ | |
| ) | |
| demo.launch(mcp_server=True) | |
Xet Storage Details
- Size:
- 22.7 kB
- Xet hash:
- 030a37f7b57c0bd5aabbd2abc8b448e51fc61e7b5009d6f33b571ca76e890276
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.