File size: 7,733 Bytes
1c0c94d | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """Deterministic, pure Rule Engine. One function per violation; emits *candidates*.
No ML, no I/O, no randomness here — that is what makes it unit-testable in isolation.
Tiers (see docs/DESIGN.md §3):
A appearance — helmet, triple riding (may auto-confirm downstream)
C spatial — stop-line, red-light, illegal parking (need camera calibration;
never auto-confirm; routed to VLM/human)
D temporal — wrong-side (a still image can't prove it; stays inert by design)
"""
from __future__ import annotations
from collections.abc import Callable
from core.calibration import point_in_polygon
from core.schemas import Candidate, EvidenceGraph, LightState, Tier, Vehicle
Rule = Callable[[EvidenceGraph], list[Candidate]]
def _ground_point(v: Vehicle) -> tuple[float, float]:
"""Bottom-centre of the box ~ where the vehicle meets the road."""
return (v.bbox.x1 + v.bbox.x2) / 2, v.bbox.y2
# --------------------------------------------------------------------------- #
# Tier A — appearance
# --------------------------------------------------------------------------- #
def helmet_rule(graph: EvidenceGraph) -> list[Candidate]:
out: list[Candidate] = []
for v in graph.vehicles:
if v.type not in {"motorcycle", "bicycle"}:
continue
for r in graph.riders_of(v.id):
# helmet None => the classifier could not tell; not a reportable violation.
if r.helmet is False:
out.append(
Candidate(
type="HELMET_NON_COMPLIANCE",
tier=Tier.A,
subjects=[v.id, r.id],
rule_score=1.0,
# use the classifier's real confidence when we have it
attribute_score=r.helmet_score
if r.helmet_score is not None
else 0.9,
detection_score=v.confidence,
pre_verified=True, # set by the classifier; skip the 2nd VLM
reason="A rider on a motorcycle is not wearing a helmet.",
)
)
return out
# --------------------------------------------------------------------------- #
# Tier B — hard appearance (seatbelt: genuinely unreliable from traffic cams)
# --------------------------------------------------------------------------- #
def seatbelt_rule(graph: EvidenceGraph) -> list[Candidate]:
"""Flag car/truck/bus drivers for a seatbelt check. Tier B: a still image rarely
proves it, so these are ``speculative`` candidates routed to the VLM/human — never
auto-confirmed, and dropped if the VLM can't substantiate them.
"""
out: list[Candidate] = []
for v in graph.vehicles:
if v.type not in {"car", "truck", "bus"}:
continue
for d in graph.drivers_of(v.id):
if d.seatbelt is True:
continue # visibly belted -> no violation
speculative = (
d.seatbelt is None
) # None = a guess; False = classifier said so
out.append(
Candidate(
type="SEATBELT_NON_COMPLIANCE",
tier=Tier.B,
subjects=[v.id, d.id],
rule_score=0.5 if speculative else 0.8,
detection_score=v.confidence,
speculative=speculative,
reason=(
"A driver is visible in a car; a seatbelt check is needed."
if speculative
else "A car driver appears not to be wearing a seatbelt."
),
)
)
return out
def triple_riding_rule(graph: EvidenceGraph) -> list[Candidate]:
out: list[Candidate] = []
for v in graph.vehicles:
if v.type != "motorcycle":
continue
riders = graph.riders_of(v.id)
if len(riders) >= 3:
out.append(
Candidate(
type="TRIPLE_RIDING",
tier=Tier.A,
subjects=[v.id, *[r.id for r in riders]],
rule_score=min(1.0, len(riders) / 3.0),
detection_score=v.confidence,
reason=f"{len(riders)} riders are linked to a single motorcycle.",
)
)
return out
# --------------------------------------------------------------------------- #
# Tier C — spatial (require per-camera calibration zones; never auto-confirm)
# --------------------------------------------------------------------------- #
def stop_line_rule(graph: EvidenceGraph) -> list[Candidate]:
out: list[Candidate] = []
for z in graph.zones_by_kind("stop_line"):
for v in graph.vehicles:
if point_in_polygon(_ground_point(v), z.polygon):
out.append(
Candidate(
type="STOP_LINE_VIOLATION",
tier=Tier.C,
subjects=[v.id],
rule_score=0.8,
detection_score=v.confidence,
reason="Vehicle is over the calibrated stop-line zone.",
)
)
return out
def red_light_rule(graph: EvidenceGraph) -> list[Candidate]:
if not any(light.state == LightState.red for light in graph.lights):
return []
out: list[Candidate] = []
for z in graph.zones_by_kind("stop_line"):
for v in graph.vehicles:
if point_in_polygon(_ground_point(v), z.polygon):
out.append(
Candidate(
type="RED_LIGHT_VIOLATION",
tier=Tier.C,
subjects=[v.id],
rule_score=0.7,
detection_score=v.confidence,
reason="Light is red and vehicle is past the stop line; "
"confirm with an image sequence.",
)
)
return out
def illegal_parking_rule(graph: EvidenceGraph) -> list[Candidate]:
out: list[Candidate] = []
for z in graph.zones_by_kind("no_parking"):
for v in graph.vehicles:
if point_in_polygon(_ground_point(v), z.polygon):
out.append(
Candidate(
type="ILLEGAL_PARKING",
tier=Tier.C,
subjects=[v.id],
rule_score=0.6,
detection_score=v.confidence,
reason="Vehicle is inside a no-parking zone; "
"needs dwell-time confirmation.",
)
)
return out
# --------------------------------------------------------------------------- #
# Tier D — temporal (a single frame cannot prove it; inert by design)
# --------------------------------------------------------------------------- #
def wrong_side_rule(graph: EvidenceGraph) -> list[Candidate]:
"""Direction of travel needs motion; a still image can't establish it. We abstain
rather than emit false positives — fill in once orientation/tracking exists."""
return []
RULES: list[Rule] = [
helmet_rule,
triple_riding_rule,
seatbelt_rule,
stop_line_rule,
red_light_rule,
illegal_parking_rule,
wrong_side_rule,
]
def run_rules(graph: EvidenceGraph) -> list[Candidate]:
candidates: list[Candidate] = []
for rule in RULES:
candidates.extend(rule(graph))
return candidates
|