Spaces:
Sleeping
Sleeping
File size: 4,952 Bytes
1b30630 | 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 | """
Wrong-Side Driving Detector.
Detects vehicles travelling against the expected flow of traffic in their assigned lane.
"""
import logging
from typing import List, Optional
import numpy as np
from config.settings import ViolationType, VIOLATION_DISPLAY_NAMES, TWO_WHEELER_CLASSES, FOUR_WHEELER_CLASSES
from config.settings import EntityClass
from core.violation_engine import ViolationRecord
from core.scene_graph import SceneGraph
from violations.base import BaseViolationDetector
logger = logging.getLogger(__name__)
class WrongSideViolationDetector(BaseViolationDetector):
"""
Detector for wrong-side driving violations.
"""
violation_type = ViolationType.WRONG_SIDE_DRIVING
def detect(
self,
scene_graph: SceneGraph,
frame: Optional[np.ndarray] = None,
) -> List[ViolationRecord]:
"""
Scan the scene graph for vehicles driving in the wrong direction.
Args:
scene_graph: Structured representation of the current frame.
frame: Optional raw BGR image.
Returns:
List of Wrong-Side driving violations.
"""
violations: List[ViolationRecord] = []
import datetime
# Expected directions by lane half (left / right of frame).
# In India, traffic flows left (Left-Hand Driving). We assume:
# - Left lanes (indices < 2) should move NORTH (away from camera).
# - Right lanes (indices >= 2) should move SOUTH (oncoming/towards camera).
expected_left_half = {"NORTH", "NORTHWEST", "NORTHEAST"}
expected_right_half = {"SOUTH", "SOUTHWEST", "SOUTHEAST"}
all_vehicle_classes = (
TWO_WHEELER_CLASSES | FOUR_WHEELER_CLASSES | {EntityClass.AUTO_RICKSHAW}
)
for node in scene_graph.nodes.values():
if node.entity_class not in all_vehicle_classes:
continue
direction = scene_graph.get_direction(node.node_id)
if direction is None or direction == "STATIONARY":
continue
# Skip nodes with negligible movement to avoid false flags from tracking jitter
import math
vector = node.attributes.get("direction_vector")
if vector is not None:
dx, dy = vector
magnitude = math.hypot(dx, dy)
if magnitude < 2.0:
continue
lane_index = node.attributes.get("lane_index")
if lane_index is None:
continue
is_left_half = lane_index < 2
wrong_side = False
if is_left_half and direction in expected_right_half:
wrong_side = True
elif not is_left_half and direction in expected_left_half:
wrong_side = True
if not wrong_side:
continue
involved = [node.node_id]
plate_node = scene_graph.get_plate_of(node.node_id)
plate_text = ""
if plate_node is not None:
involved.append(plate_node.node_id)
plate_text = plate_node.attributes.get("plate_text", "")
display_name = VIOLATION_DISPLAY_NAMES.get(
self.violation_type, "Wrong-Side Driving"
)
timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
unique_suffix = f"{node.node_id[-3:]}-{direction}"
temp_id = f"VIO-WRONGSIDE-{unique_suffix}"
expected_dir = "SOUTH-bound" if is_left_half else "NORTH-bound"
chain = [
f"Detected {node.entity_class.value}_{node.node_id} (conf={node.confidence:.2f}) with trajectory data",
f"Computed direction: {direction} in lane {lane_index}",
f"Lane {lane_index} expected {expected_dir} flow, vehicle moving {direction} (against flow)",
f"VIOLATION: {display_name} — {node.entity_class.value}_{node.node_id}",
]
violations.append(
ViolationRecord(
violation_id=temp_id,
violation_type=self.violation_type,
confidence=node.confidence * 0.8, # Heuristic penalty
involved_nodes=involved,
description=(
f"{display_name}: {node.entity_class.value} moving "
f"{direction} in lane {lane_index} (expected opposite flow)."
),
bbox=node.bbox,
timestamp=timestamp,
metadata={
"direction": direction,
"lane_index": lane_index,
"plate_text": plate_text,
"vehicle_type": node.entity_class.value,
},
reasoning_chain=chain,
)
)
return violations
|