Spaces:
Paused
Paused
| """Failure Evidence Graph. | |
| Structured, typed nodes (Object / Environment / Sensors / Prediction / | |
| GroundTruth / Tracking / Planner / HistoricalSimilarity / Frequency / | |
| SafetyConsequence) each carrying an explicit evidence status: | |
| OBSERVED directly measured from the campaign data / replay | |
| DERIVED computed deterministically from observed data | |
| HYPOTHESIS produced by an advisory agent; requires confirmation | |
| UNAVAILABLE the evidence does not exist yet; the graph says so instead | |
| of letting anything downstream pretend it does | |
| The graph starts from deterministic scene data and is enriched as analysis | |
| stages complete. Nothing here consults an LLM. | |
| """ | |
| from __future__ import annotations | |
| from typing import Optional | |
| from sensorflow.agentic import data as data_mod | |
| from sensorflow.agentic.models import (EvidenceEdge, EvidenceGraph, EvidenceNode, | |
| FailureEvent, new_id) | |
| def _node(node_type: str, status: str, summary: str, fields: dict, | |
| source: str, caveats: Optional[list] = None) -> EvidenceNode: | |
| return EvidenceNode(node_id=new_id("ev"), node_type=node_type, status=status, | |
| summary=summary, fields=fields, source=source, | |
| caveats=caveats or []) | |
| def build_graph(failure: FailureEvent, campaign: data_mod.Campaign) -> EvidenceGraph: | |
| """Initial graph from deterministic campaign evidence for the failure's | |
| representative instance (first instance).""" | |
| graph = EvidenceGraph(failure_id=failure.failure_id) | |
| inst = failure.instances[0] if failure.instances else None | |
| if inst is None: | |
| graph.nodes.append(_node( | |
| "Object", "UNAVAILABLE", | |
| "No concrete failure instance is attached to this pattern", | |
| {}, "pipeline")) | |
| return graph | |
| history = data_mod.observations_for_instance( | |
| campaign, inst.sequence_id, inst.object_instance_id) | |
| obs = next((o for o in history if o.frame_id == inst.frame_id), None) | |
| graph.nodes.append(_node( | |
| "Object", "OBSERVED", | |
| f"{inst.gt_class} instance {inst.object_instance_id} at " | |
| f"{inst.distance_m:.1f} m", | |
| {"instance_id": inst.object_instance_id, "gt_class": inst.gt_class, | |
| "distance_m": inst.distance_m, | |
| "bbox_3d": obs.gt_bbox_3d if obs else None}, | |
| "bevfusion synthetic scene ground truth")) | |
| graph.nodes.append(_node( | |
| "Environment", "OBSERVED", | |
| f"{inst.time_of_day}/{inst.weather}, construction_zone=" | |
| f"{inst.construction_zone}, geo={inst.geo_bucket}", | |
| {"time_of_day": inst.time_of_day, "weather": inst.weather, | |
| "construction_zone": inst.construction_zone, | |
| "geo_bucket": inst.geo_bucket, "occluded": inst.occluded}, | |
| "sequence context (campaign metadata)")) | |
| graph.nodes.append(_node( | |
| "Sensors", "DERIVED", | |
| "Camera/LiDAR coherence not yet verified — pending fusion agent", | |
| {"modalities": ["camera", "lidar"], "verified": False}, | |
| "sensorflow.bevfusion simulated sensors", | |
| caveats=["fusion verification runs in FAILURE_ANALYSIS"])) | |
| if obs is not None: | |
| graph.nodes.append(_node( | |
| "Prediction", "OBSERVED", | |
| f"candidate predicted {obs.candidate.predicted_class} " | |
| f"(conf {obs.candidate.confidence}); baseline predicted " | |
| f"{obs.baseline.predicted_class} (conf {obs.baseline.confidence})", | |
| {"candidate": obs.candidate.model_dump(), | |
| "baseline": obs.baseline.model_dump(), | |
| "candidate_model": failure.candidate_model, | |
| "baseline_model": failure.baseline_model}, | |
| "campaign prediction log")) | |
| graph.nodes.append(_node( | |
| "GroundTruth", "OBSERVED", | |
| f"GT class {obs.gt_class} (synthetic vendor-style GT)", | |
| {"gt_class": obs.gt_class, "gt_bbox_3d": obs.gt_bbox_3d, | |
| "gt_source": "synthetic_vendor_gt"}, | |
| "bevfusion synthetic scene ground truth")) | |
| graph.nodes.append(_node( | |
| "Tracking", "OBSERVED", | |
| f"candidate track {obs.candidate.track_id}; " | |
| f"{len(history)} frames observed for this instance", | |
| {"candidate_track_id": obs.candidate.track_id, | |
| "baseline_track_id": obs.baseline.track_id, | |
| "frames_observed": len(history)}, | |
| "campaign prediction log")) | |
| else: | |
| graph.nodes.append(_node("Prediction", "UNAVAILABLE", | |
| "prediction record missing", {}, "campaign")) | |
| graph.nodes.append(_node("GroundTruth", "UNAVAILABLE", | |
| "ground truth record missing", {}, "campaign")) | |
| graph.nodes.append(_node("Tracking", "UNAVAILABLE", | |
| "tracking record missing", {}, "campaign")) | |
| if inst.has_planner_trace: | |
| graph.nodes.append(_node( | |
| "Planner", "OBSERVED", | |
| "Replay planner-response trace exists for this instance", | |
| {"replay_available": True, "analyzed": False}, | |
| "synthetic planner replay trace", | |
| caveats=["safety-impact agent quantifies behavior in FAILURE_ANALYSIS"])) | |
| else: | |
| graph.nodes.append(_node( | |
| "Planner", "UNAVAILABLE", | |
| "No behavioral replay evidence for this instance — downstream " | |
| "consequence cannot be observed, only hypothesized", | |
| {"replay_available": False}, | |
| "synthetic planner replay trace")) | |
| graph.nodes.append(_node( | |
| "HistoricalSimilarity", "DERIVED", | |
| "Similar-failure retrieval pending (scenario mining agent)", | |
| {"retrieved": False}, "scenario mining (embedding + attribute retrieval)")) | |
| basis = failure.detection_basis | |
| graph.nodes.append(_node( | |
| "Frequency", "DERIVED", | |
| f"candidate {basis.candidate_events}/{basis.denominator} " | |
| f"({basis.candidate_rate:.6f}) vs baseline {basis.baseline_events}" | |
| f"/{basis.denominator} ({basis.baseline_rate:.6f}) — CI/significance " | |
| "pending statistical agent", | |
| {"candidate_events": basis.candidate_events, | |
| "baseline_events": basis.baseline_events, | |
| "denominator": basis.denominator, | |
| "candidate_rate": basis.candidate_rate, | |
| "baseline_rate": basis.baseline_rate}, | |
| "deterministic detection scan")) | |
| graph.nodes.append(_node( | |
| "SafetyConsequence", "UNAVAILABLE", | |
| "Not yet assessed — requires safety-impact analysis over replay " | |
| "evidence; hypotheses are never recorded as consequences", | |
| {}, "safety impact agent (FAILURE_ANALYSIS)")) | |
| ids = {n.node_type: n.node_id for n in graph.nodes} | |
| for src, dst, rel in [ | |
| ("Object", "Prediction", "was_predicted_as"), | |
| ("Object", "GroundTruth", "labeled_as"), | |
| ("Object", "Environment", "situated_in"), | |
| ("Prediction", "Tracking", "feeds"), | |
| ("Tracking", "Planner", "feeds"), | |
| ("Prediction", "Frequency", "aggregated_into"), | |
| ("Planner", "SafetyConsequence", "determines"), | |
| ("Object", "HistoricalSimilarity", "compared_against"), | |
| ("Sensors", "Prediction", "produced"), | |
| ]: | |
| if src in ids and dst in ids: | |
| graph.edges.append(EvidenceEdge(src=ids[src], dst=ids[dst], relation=rel)) | |
| return graph | |
| def set_node(graph: EvidenceGraph, node_type: str, status: str, summary: str, | |
| fields: dict, source: str, caveats: Optional[list] = None) -> None: | |
| """Replace (or add) a node after an analysis stage produced its evidence.""" | |
| for i, n in enumerate(graph.nodes): | |
| if n.node_type == node_type: | |
| graph.nodes[i] = EvidenceNode( | |
| node_id=n.node_id, node_type=node_type, status=status, | |
| summary=summary, fields=fields, source=source, | |
| caveats=caveats or []) | |
| return | |
| graph.nodes.append(_node(node_type, status, summary, fields, source, caveats)) | |