ATA / src /integration /safety_gate.py
vishanth11
Deploy AI Traffic Adaptation System with Git LFS binaries to Hugging Face Spaces
f234112
Raw History Blame Contribute Delete
2.42 kB
class SafetyGate:
"""
Validates a proposed signal phase before execution.
The Safety Gate does not choose the phase.
It only decides whether the proposed phase is safe to execute.
"""
def __init__(
self,
min_green_seconds=10,
yellow_seconds=3,
all_red_seconds=1,
confidence_threshold=0.50,
risk_threshold=0.75,
accident_probability_threshold=0.75
):
self.min_green_seconds = min_green_seconds
self.yellow_seconds = yellow_seconds
self.all_red_seconds = all_red_seconds
self.confidence_threshold = confidence_threshold
self.risk_threshold = risk_threshold
self.accident_probability_threshold = accident_probability_threshold
def validate(
self,
phase,
topology_valid,
conflict_free,
timing_valid,
pedestrian_clear,
downstream_available,
emergency_safe,
confidence,
safety_risk=0.0,
accident_detected=False,
safety_confidence=None
):
"""
Validate a proposed signal phase.
Returns:
{
"approved": bool,
"phase": phase,
"checks": {...},
"reasons": [...]
}
"""
checks = {
"topology_valid": bool(topology_valid),
"conflict_free": bool(conflict_free),
"timing_valid": bool(timing_valid),
"pedestrian_clear": bool(pedestrian_clear),
"downstream_available": bool(downstream_available),
"emergency_safe": bool(emergency_safe),
"confidence_valid": (
confidence >= self.confidence_threshold
),
"risk_valid": (
0.0 <= safety_risk <= self.risk_threshold
),
"accident_safe": not bool(accident_detected),
}
if safety_confidence is not None:
checks["safety_confidence_valid"] = (
safety_confidence >= self.confidence_threshold
)
reasons = []
for name, passed in checks.items():
if not passed:
reasons.append(name)
approved = len(reasons) == 0
return {
"approved": approved,
"phase": phase,
"checks": checks,
"reasons": reasons
}