Spaces:
Paused
Paused
File size: 3,593 Bytes
729628a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Launch gate evaluator: block export until quality thresholds pass."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Dict, List, Optional
DEFAULT_THRESHOLDS_PATH = Path("runs/pipeline/gate_thresholds.json")
DEFAULT_THRESHOLDS = {
"map_3d": 0.65,
"orientation_error_deg": 5.0,
"id_swap_rate": 0.02,
"track_fragmentation_rate": 0.05,
"position_error_m": 2.0,
}
class LaunchGateEvaluator:
"""Validate benchmark metrics against safety thresholds before export."""
def __init__(self, thresholds_path: Optional[Path] = None):
self.thresholds_path = thresholds_path or DEFAULT_THRESHOLDS_PATH
self.thresholds = self._load_thresholds()
def _load_thresholds(self) -> Dict[str, float]:
if self.thresholds_path.exists():
with open(self.thresholds_path) as f:
return json.load(f)
self.thresholds_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.thresholds_path, "w") as f:
json.dump(DEFAULT_THRESHOLDS, f, indent=2)
return DEFAULT_THRESHOLDS.copy()
def evaluate(self, sequence_id: str) -> Dict[str, Any]:
metric_path = Path("runs/pipeline") / sequence_id / "benchmark" / "metric_card.json"
if not metric_path.exists():
return {
"passed": False,
"failures": [{"metric": "benchmark", "message": "No benchmark results found. Run quality gate first."}],
"blocked_stages": ["export"],
}
with open(metric_path) as f:
metrics = json.load(f)
failures: List[Dict] = []
checks = [
("map_3d", metrics.get("map_3d", 0), self.thresholds["map_3d"], ">="),
("orientation_error_deg", metrics.get("orientation_error_deg", 999), self.thresholds["orientation_error_deg"], "<="),
("id_swap_rate", metrics.get("id_swap_rate", 1), self.thresholds["id_swap_rate"], "<="),
("track_fragmentation_rate", metrics.get("track_fragmentation_rate", 1), self.thresholds["track_fragmentation_rate"], "<="),
]
for name, value, threshold, op in checks:
failed = (value < threshold) if op == ">=" else (value > threshold)
if failed:
failures.append({
"metric": name,
"value": value,
"threshold": threshold,
"operator": op,
})
passed = len(failures) == 0
result = {
"passed": passed,
"failures": failures,
"blocked_stages": [] if passed else ["export"],
"metrics": metrics,
}
gate_path = Path("runs/pipeline") / sequence_id / "launch_gate.json"
gate_path.parent.mkdir(parents=True, exist_ok=True)
with open(gate_path, "w") as f:
json.dump(result, f, indent=2)
state_path = Path("runs/pipeline/state.json")
state = {}
if state_path.exists():
with open(state_path) as f:
state = json.load(f)
state.setdefault(sequence_id, {})["launch_gate_passed"] = passed
with open(state_path, "w") as f:
json.dump(state, f, indent=2)
return result
def is_export_allowed(self, sequence_id: str) -> bool:
gate_path = Path("runs/pipeline") / sequence_id / "launch_gate.json"
if gate_path.exists():
with open(gate_path) as f:
return json.load(f).get("passed", False)
return False
|