core/violation_engine.py

"""
Violation Engine — Layer 4 of the HSUP Pipeline.

Queries the :class:`SceneGraph` to detect traffic violations by reasoning
over the relational structure of detected entities.  Each violation type
has a dedicated detection method that encodes Indian traffic-law logic.

Supported Violations
────────────────────
- **Helmet non-compliance** — rider on two-wheeler without helmet.
- **Triple riding** — three or more persons on a single two-wheeler.
- **Seatbelt non-compliance** — driver/passenger without seatbelt.
- **Wrong-side driving** — vehicle moving against expected traffic flow.
- **Stop-line violation** — vehicle crossing stop line while signal is red.
- **Red-light violation** — vehicle proceeding through red signal.
- **Illegal parking** — vehicle stationary in a no-parking zone.

Architecture
────────────
    SceneGraph (Layer 3)
          │
          ▼
    ViolationEngine.analyze()
      ├── _check_helmet_violations()
      ├── _check_triple_riding()
      ├── _check_seatbelt_violations()
      ├── _check_wrong_side()
      ├── _check_stop_line()
      ├── _check_red_light()
      └── _check_illegal_parking()
          │
          ▼
    List[ViolationRecord]  →  Evidence Generator (Layer 5)
"""

from __future__ import annotations

import datetime
import logging
import threading
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

import numpy as np

from config.settings import (
    EntityClass,
    Settings,
    SETTINGS,
    TWO_WHEELER_CLASSES,
    FOUR_WHEELER_CLASSES,
    ViolationType,
    VIOLATION_DISPLAY_NAMES,
)
from core.scene_graph import (
    DRIVES,
    HAS_PLATE,
    NOT_WEARS,
    RIDES,
    WEARS,
    NEAR_SIGNAL,
    CROSSED_LINE,
    SceneEdge,
    SceneGraph,
    SceneNode,
)

logger = logging.getLogger(__name__)


# ══════════════════════════════════════════════════════════════
# VIOLATION RECORD
# ══════════════════════════════════════════════════════════════

@dataclass
class ViolationRecord:
    """
    Immutable record of a single detected traffic violation.

    Produced by :meth:`ViolationEngine.analyze` and consumed downstream by
    the evidence generator (Layer 5) and the dashboard.

    Attributes:
        violation_id:   Unique identifier (e.g. ``"VIO-20260616-001"``).
        violation_type: Semantic type from :class:`ViolationType`.
        confidence:     Aggregate confidence ``[0, 1]``.
        involved_nodes: ``node_id`` values of all participating entities.
        description:    Human-readable violation description.
        bbox:           Union bounding box ``(x1, y1, x2, y2)`` that
                        encompasses all involved entities.
        timestamp:      ISO-8601 timestamp of detection.
        metadata:       Free-form extras — plate text, fine amount, section
                        of law, etc.
    """

    violation_id: str
    violation_type: ViolationType
    confidence: float
    involved_nodes: List[str]
    description: str
    bbox: Tuple[int, int, int, int]
    timestamp: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    reasoning_chain: List[str] = field(default_factory=list)


# ══════════════════════════════════════════════════════════════
# VIOLATION ENGINE
# ══════════════════════════════════════════════════════════════

class ViolationEngine:
    """
    Rule-based violation reasoning engine.

    Inspects a :class:`SceneGraph` and emits :class:`ViolationRecord`
    instances for every detected infraction.  Designed to be called
    once per frame.

    Usage::

        engine = ViolationEngine()
        violations = engine.analyze(scene_graph, frame=current_frame)
        for v in violations:
            print(v.violation_type, v.description)
    """

    _global_counter: int = -1
    _global_counter_lock = threading.Lock()
    _global_date_str: str = ""

    def __init__(self, settings: Settings | None = None) -> None:
        """
        Initialise the violation engine.

        Args:
            settings: Global settings instance.  Falls back to
                      :data:`SETTINGS` if ``None``.
        """
        self._settings: Settings = settings or SETTINGS
        today = datetime.date.today().strftime("%Y%m%d")

        # Check if running within a Streamlit session
        self._use_session_state = False
        try:
            import streamlit as st
            # Confirm we are in a Streamlit thread/execution context
            if hasattr(st, "session_state"):
                self._use_session_state = True
        except Exception:
            pass

        if self._use_session_state:
            import streamlit as st
            if "global_violation_counter" not in st.session_state or st.session_state.get("global_violation_date") != today:
                max_seq = 0
                try:
                    import re
                    evidence_dir = self._settings.evidence_dir
                    if evidence_dir.exists():
                        pattern = re.compile(rf"^VIO-{today}-(\d+)$")
                        for path in evidence_dir.iterdir():
                            if path.is_dir():
                                match = pattern.match(path.name)
                                if match:
                                    seq_val = int(match.group(1))
                                    if seq_val > max_seq:
                                        max_seq = seq_val
                except Exception as e:
                    logger.warning("Failed to scan evidence directory: %s", e)
                st.session_state["global_violation_counter"] = max_seq
                st.session_state["global_violation_date"] = today
        else:
            with ViolationEngine._global_counter_lock:
                # If the class-level counter is not initialized or the date changed:
                if ViolationEngine._global_counter == -1 or ViolationEngine._global_date_str != today:
                    ViolationEngine._global_date_str = today
                    max_seq = 0
                    try:
                        import re
                        evidence_dir = self._settings.evidence_dir
                        if evidence_dir.exists():
                            pattern = re.compile(rf"^VIO-{today}-(\d+)$")
                            for path in evidence_dir.iterdir():
                                if path.is_dir():
                                    match = pattern.match(path.name)
                                    if match:
                                        seq_val = int(match.group(1))
                                        if seq_val > max_seq:
                                            max_seq = seq_val
                    except Exception as e:
                        logger.warning("Failed to scan evidence directory: %s", e)
                    ViolationEngine._global_counter = max_seq

        # Load all registered violation detectors explicitly to handle Streamlit caching/reloading issues
        from violations.helmet import HelmetViolationDetector
        from violations.seatbelt import SeatbeltViolationDetector
        from violations.triple_riding import TripleRidingViolationDetector
        from violations.wrong_side import WrongSideViolationDetector
        from violations.stop_line import StopLineViolationDetector
        from violations.red_light import RedLightViolationDetector
        from violations.illegal_parking import IllegalParkingViolationDetector
        from violations.missing_plate import MissingPlateViolationDetector

        detector_map = {
            ViolationType.HELMET_NON_COMPLIANCE: HelmetViolationDetector,
            ViolationType.SEATBELT_NON_COMPLIANCE: SeatbeltViolationDetector,
            ViolationType.TRIPLE_RIDING: TripleRidingViolationDetector,
            ViolationType.WRONG_SIDE_DRIVING: WrongSideViolationDetector,
            ViolationType.STOP_LINE_VIOLATION: StopLineViolationDetector,
            ViolationType.RED_LIGHT_VIOLATION: RedLightViolationDetector,
            ViolationType.ILLEGAL_PARKING: IllegalParkingViolationDetector,
            ViolationType.MISSING_LICENSE_PLATE: MissingPlateViolationDetector,
        }

        self._detectors = {}
        for vtype, detector_cls in detector_map.items():
            self._detectors[vtype] = detector_cls(
                enabled=True,
                min_confidence=self._settings.min_violation_confidence
            )

        logger.info(
            "ViolationEngine initialised with %d detectors: %s",
            len(self._detectors),
            [d.__class__.__name__ for d in self._detectors.values()]
        )

    # ──────────────────────────────────────────────────────────
    # PUBLIC API
    # ──────────────────────────────────────────────────────────

    def analyze(
        self,
        scene_graph: SceneGraph,
        frame: np.ndarray | None = None,
    ) -> List[ViolationRecord]:
        """
        Run all registered violation detectors on a scene graph.

        Args:
            scene_graph: A fully-built :class:`SceneGraph` for the current
                         frame.
            frame:       Optional raw frame.

        Returns:
            List of :class:`ViolationRecord` instances.
        """
        violations: List[ViolationRecord] = []

        for vtype, detector in self._detectors.items():
            try:
                results = detector.run(scene_graph, frame)
                for vr in results:
                    # Standardise violation ID
                    vr.violation_id = self._generate_violation_id()
                    violations.append(vr)
            except Exception:
                logger.error(
                    "Detector %s failed during run().",
                    detector.__class__.__name__,
                    exc_info=True
                )

        # De-duplicate: remove duplicate violations for the same vehicle
        seen = set()
        unique_violations = []
        for v in violations:
            # Create a key from violation type + primary vehicle node
            vehicle_nodes = [n for n in v.involved_nodes if any(vt in n for vt in ['car', 'motorcycle', 'auto_rickshaw', 'bus', 'truck', 'tempo'])]
            key = (v.violation_type.value, tuple(sorted(vehicle_nodes)))
            if key not in seen:
                # Filter out background/small vehicle violations
                img_h, img_w = None, None
                if frame is not None:
                    img_h, img_w = frame.shape[:2]
                elif getattr(scene_graph, "frame_shape", None) is not None:
                    img_h, img_w = scene_graph.frame_shape[:2]

                if img_h is not None and img_w is not None:
                    total_area = img_h * img_w
                    
                    # Find the primary vehicle involved in this violation
                    primary_veh_node = None
                    for node_id in v.involved_nodes:
                        node = scene_graph.nodes.get(node_id)
                        if node and node.entity_class.value in ['car', 'motorcycle', 'auto_rickshaw', 'bus', 'truck', 'tempo']:
                            primary_veh_node = node
                            break
                    
                    if primary_veh_node is not None:
                        vx1, vy1, vx2, vy2 = primary_veh_node.bbox
                        v_area = (vx2 - vx1) * (vy2 - vy1)
                        # Filter if vehicle area is less than configured threshold (focus on near objects)
                        if (v_area / total_area) < self._settings.min_vehicle_area_ratio:
                            logger.info(
                                "Filtering out background vehicle violation: %s (area ratio: %.4f < %.4f)",
                                primary_veh_node.node_id, v_area / total_area, self._settings.min_vehicle_area_ratio
                            )
                            continue
                seen.add(key)
                unique_violations.append(v)
            else:
                logger.debug("De-duplicated violation: %s for %s", v.violation_type.value, vehicle_nodes)
        violations = unique_violations

        logger.info(
            "ViolationEngine found %d violation(s) in current frame.",
            len(violations),
        )
        return violations


    # ──────────────────────────────────────────────────────────
    # VIOLATION CHECKS
    # ──────────────────────────────────────────────────────────

    def _check_helmet_violations(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect helmet non-compliance on two-wheelers.

        Logic:
        1. Find all ``RIDES`` edges linking a person to a two-wheeler.
        2. For each rider, query ``get_helmet_status()``.
        3. If status is ``"no_helmet"`` → emit a violation.
        4. If status is ``None`` (no helmet detection at all) → skip
           (ambiguous — the detector may not have picked it up).

        Returns:
            List of helmet-violation records.
        """
        violations: List[ViolationRecord] = []

        # Query all person → two-wheeler relationships
        for ec in TWO_WHEELER_CLASSES:
            triplets = sg.query(
                source_class=None, relation=RIDES, target_class=ec
            )
            for rider_node, edge, vehicle_node in triplets:
                status = sg.get_helmet_status(rider_node.node_id)
                if status != "no_helmet":
                    continue

                # Build violation record
                involved = [rider_node.node_id, vehicle_node.node_id]
                plate_node = sg.get_plate_of(vehicle_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(
                    ViolationType.HELMET_NON_COMPLIANCE,
                    "Helmet Non-Compliance",
                )

                violations.append(
                    ViolationRecord(
                        violation_id=self._generate_violation_id(),
                        violation_type=ViolationType.HELMET_NON_COMPLIANCE,
                        confidence=min(
                            rider_node.confidence, vehicle_node.confidence
                        ),
                        involved_nodes=involved,
                        description=(
                            f"{display_name}: Rider on {vehicle_node.entity_class.value} "
                            f"without helmet."
                        ),
                        bbox=self._merge_bbox(
                            [rider_node, vehicle_node]
                        ),
                        timestamp=self._now_iso(),
                        metadata={
                            "plate_text": plate_text,
                            "vehicle_type": vehicle_node.entity_class.value,
                            "rider_node": rider_node.node_id,
                        },
                    )
                )

        return violations

    def _check_triple_riding(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect three or more persons on a single two-wheeler.

        Logic:
        1. For each two-wheeler node, call ``get_passenger_count()``.
        2. If count ≥ ``triple_riding_min_persons`` → violation.

        Returns:
            List of triple-riding violation records.
        """
        violations: List[ViolationRecord] = []
        min_persons = self._settings.triple_riding_min_persons

        all_two_wheelers = [
            n
            for n in sg.nodes.values()
            if n.entity_class in TWO_WHEELER_CLASSES
        ]

        for vehicle in all_two_wheelers:
            count = sg.get_passenger_count(vehicle.node_id)
            if count < min_persons:
                continue

            riders = sg.get_riders_of(vehicle.node_id)
            involved = [vehicle.node_id] + [r.node_id for r in riders]

            plate_node = sg.get_plate_of(vehicle.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(
                ViolationType.TRIPLE_RIDING, "Triple Riding"
            )

            all_involved_nodes = [vehicle] + riders
            violations.append(
                ViolationRecord(
                    violation_id=self._generate_violation_id(),
                    violation_type=ViolationType.TRIPLE_RIDING,
                    confidence=min(
                        n.confidence for n in all_involved_nodes
                    ),
                    involved_nodes=involved,
                    description=(
                        f"{display_name}: {count} persons detected on "
                        f"{vehicle.entity_class.value}."
                    ),
                    bbox=self._merge_bbox(all_involved_nodes),
                    timestamp=self._now_iso(),
                    metadata={
                        "passenger_count": count,
                        "plate_text": plate_text,
                        "vehicle_type": vehicle.entity_class.value,
                    },
                )
            )

        return violations

    def _check_seatbelt_violations(
        self,
        sg: SceneGraph,
        frame: np.ndarray | None = None,
    ) -> List[ViolationRecord]:
        """
        Detect seatbelt non-compliance in four-wheelers.

        Logic:
        1. Find all ``DRIVES`` edges linking a person to a four-wheeler.
        2. Check for ``NOT_WEARS`` edges targeting ``NO_SEATBELT`` on that
           person.
        3. If the frame is available, a future enhancement can crop the
           windshield region for CNN-based seatbelt verification.

        Returns:
            List of seatbelt-violation records.
        """
        violations: List[ViolationRecord] = []

        for ec in FOUR_WHEELER_CLASSES:
            triplets = sg.query(
                source_class=None, relation=DRIVES, target_class=ec
            )
            for driver_node, edge, vehicle_node in triplets:
                # Check for explicit no-seatbelt detection
                has_no_seatbelt = any(
                    e.source_id == driver_node.node_id
                    and e.relation == NOT_WEARS
                    and sg.nodes.get(e.target_id, SceneNode(
                        node_id="", entity_class=EntityClass.PERSON,
                        bbox=(0, 0, 0, 0), confidence=0
                    )).entity_class == EntityClass.NO_SEATBELT
                    for e in sg.edges
                )

                if not has_no_seatbelt:
                    continue

                involved = [driver_node.node_id, vehicle_node.node_id]
                plate_node = sg.get_plate_of(vehicle_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(
                    ViolationType.SEATBELT_NON_COMPLIANCE,
                    "Seatbelt Non-Compliance",
                )

                violations.append(
                    ViolationRecord(
                        violation_id=self._generate_violation_id(),
                        violation_type=ViolationType.SEATBELT_NON_COMPLIANCE,
                        confidence=min(
                            driver_node.confidence, vehicle_node.confidence
                        ),
                        involved_nodes=involved,
                        description=(
                            f"{display_name}: Driver in {vehicle_node.entity_class.value} "
                            f"without seatbelt."
                        ),
                        bbox=self._merge_bbox([driver_node, vehicle_node]),
                        timestamp=self._now_iso(),
                        metadata={
                            "plate_text": plate_text,
                            "vehicle_type": vehicle_node.entity_class.value,
                            "frame_available": frame is not None,
                        },
                    )
                )

        return violations

    def _check_wrong_side(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect wrong-side driving.

        Logic:
        1. For each vehicle with a direction attribute, check if the
           direction matches the expected traffic flow for its lane.
        2. Indian traffic drives on the LEFT.  In a typical camera view:
           - Vehicles in left lanes should move in one direction (e.g. SOUTH).
           - Vehicles in right lanes should move the opposite (e.g. NORTH).
        3. If a vehicle moves in the *opposite* expected direction for its
           lane → violation.

        This is a heuristic — production deployment would calibrate the
        expected direction per camera installation.

        Returns:
            List of wrong-side violation records.
        """
        violations: List[ViolationRecord] = []

        # Default expected directions by lane half (left / right of frame)
        # This should be configured per-camera in production.
        expected_left_half = {"SOUTH", "SOUTHWEST", "SOUTHEAST"}
        expected_right_half = {"NORTH", "NORTHWEST", "NORTHEAST"}

        all_vehicle_classes = (
            TWO_WHEELER_CLASSES | FOUR_WHEELER_CLASSES | {EntityClass.AUTO_RICKSHAW}
        )

        for node in sg.nodes.values():
            if node.entity_class not in all_vehicle_classes:
                continue

            direction = sg.get_direction(node.node_id)
            if direction is None or direction == "STATIONARY":
                continue

            lane_index = node.attributes.get("lane_index")
            if lane_index is None:
                continue

            # Determine which half of the road
            is_left_half = lane_index < 2  # lanes 0, 1 = left half

            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
            else:
                wrong_side = False

            if not wrong_side:
                continue

            involved = [node.node_id]
            plate_node = sg.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(
                ViolationType.WRONG_SIDE_DRIVING, "Wrong-Side Driving"
            )

            violations.append(
                ViolationRecord(
                    violation_id=self._generate_violation_id(),
                    violation_type=ViolationType.WRONG_SIDE_DRIVING,
                    confidence=node.confidence * 0.8,  # penalise heuristic
                    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=self._now_iso(),
                    metadata={
                        "direction": direction,
                        "lane_index": lane_index,
                        "plate_text": plate_text,
                        "vehicle_type": node.entity_class.value,
                    },
                )
            )

        return violations

    def _check_stop_line(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect vehicles crossing a stop line while the signal is red.

        Logic:
        1. Find all ``STOP_LINE`` nodes in the graph.
        2. Find all ``TRAFFIC_LIGHT_RED`` nodes in the graph.
        3. For each vehicle, check if it overlaps with / has crossed past
           the stop-line bbox while a red signal is visible.

        Returns:
            List of stop-line violation records.
        """
        violations: List[ViolationRecord] = []

        stop_lines = [
            n for n in sg.nodes.values()
            if n.entity_class == EntityClass.STOP_LINE
        ]
        red_lights = [
            n for n in sg.nodes.values()
            if n.entity_class == EntityClass.TRAFFIC_LIGHT_RED
        ]

        # No stop line or no red light → no violation possible
        if not stop_lines or not red_lights:
            return violations

        all_vehicle_classes = (
            TWO_WHEELER_CLASSES | FOUR_WHEELER_CLASSES | {EntityClass.AUTO_RICKSHAW}
        )

        for vehicle in sg.nodes.values():
            if vehicle.entity_class not in all_vehicle_classes:
                continue

            for stop_line in stop_lines:
                # Vehicle has crossed stop line if its bottom edge is
                # below the stop line's y-centre (vehicle went past it)
                vehicle_bottom = vehicle.bbox[3]
                stop_line_y = (stop_line.bbox[1] + stop_line.bbox[3]) / 2.0

                # Check horizontal overlap (vehicle is in the same lane)
                v_cx = (vehicle.bbox[0] + vehicle.bbox[2]) / 2.0
                sl_x1 = stop_line.bbox[0]
                sl_x2 = stop_line.bbox[2]

                if not (sl_x1 <= v_cx <= sl_x2):
                    continue

                if vehicle_bottom <= stop_line_y:
                    continue  # vehicle has NOT crossed

                # Vehicle has crossed the stop line while signal is red
                involved = [vehicle.node_id, stop_line.node_id]
                plate_node = sg.get_plate_of(vehicle.node_id)
                plate_text = ""
                if plate_node is not None:
                    involved.append(plate_node.node_id)
                    plate_text = plate_node.attributes.get("plate_text", "")

                # Include the nearest red light
                if red_lights:
                    involved.append(red_lights[0].node_id)

                display_name = VIOLATION_DISPLAY_NAMES.get(
                    ViolationType.STOP_LINE_VIOLATION,
                    "Stop-Line Violation",
                )

                violations.append(
                    ViolationRecord(
                        violation_id=self._generate_violation_id(),
                        violation_type=ViolationType.STOP_LINE_VIOLATION,
                        confidence=min(
                            vehicle.confidence, stop_line.confidence
                        ),
                        involved_nodes=involved,
                        description=(
                            f"{display_name}: {vehicle.entity_class.value} "
                            f"crossed stop line during red signal."
                        ),
                        bbox=self._merge_bbox(
                            [vehicle, stop_line]
                        ),
                        timestamp=self._now_iso(),
                        metadata={
                            "plate_text": plate_text,
                            "vehicle_type": vehicle.entity_class.value,
                            "signal_state": "red",
                        },
                    )
                )
                break  # one violation per vehicle per frame

        return violations

    def _check_red_light(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect vehicles running a red light.

        This differs from stop-line violation in that it focuses on vehicles
        that are *actively proceeding through* an intersection during a red
        signal (i.e. moving past the stop line area).

        Logic:
        1. Find red-light signals.
        2. Find vehicles with direction data that are **beyond** the stop-line
           zone and still moving.
        3. A vehicle that was already past the stop line before the light
           turned red is not a violator (would need temporal state —
           simplified here).

        Returns:
            List of red-light violation records.
        """
        violations: List[ViolationRecord] = []

        red_lights = [
            n for n in sg.nodes.values()
            if n.entity_class == EntityClass.TRAFFIC_LIGHT_RED
        ]
        stop_lines = [
            n for n in sg.nodes.values()
            if n.entity_class == EntityClass.STOP_LINE
        ]

        if not red_lights:
            return violations

        all_vehicle_classes = (
            TWO_WHEELER_CLASSES | FOUR_WHEELER_CLASSES | {EntityClass.AUTO_RICKSHAW}
        )

        for vehicle in sg.nodes.values():
            if vehicle.entity_class not in all_vehicle_classes:
                continue

            direction = sg.get_direction(vehicle.node_id)
            if direction is None or direction == "STATIONARY":
                continue  # not moving → not running a red light

            # Check if vehicle is in the intersection zone (past stop lines)
            crossed_stop = False
            for sl in stop_lines:
                vehicle_bottom = vehicle.bbox[3]
                sl_y = (sl.bbox[1] + sl.bbox[3]) / 2.0

                v_cx = (vehicle.bbox[0] + vehicle.bbox[2]) / 2.0
                if sl.bbox[0] <= v_cx <= sl.bbox[2] and vehicle_bottom > sl_y:
                    crossed_stop = True
                    break

            if not crossed_stop and stop_lines:
                continue  # vehicle hasn't crossed into intersection

            # If no stop lines are detected, use proximity to red light
            if not stop_lines:
                from core.scene_graph import _bbox_center_distance

                min_dist = min(
                    _bbox_center_distance(vehicle.bbox, rl.bbox)
                    for rl in red_lights
                )
                # Only flag if vehicle is reasonably close to the signal
                frame_diag = max(vehicle.bbox[2], vehicle.bbox[3]) * 5
                if min_dist > frame_diag:
                    continue

            involved = [vehicle.node_id, red_lights[0].node_id]
            plate_node = sg.get_plate_of(vehicle.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(
                ViolationType.RED_LIGHT_VIOLATION, "Red-Light Violation"
            )

            violations.append(
                ViolationRecord(
                    violation_id=self._generate_violation_id(),
                    violation_type=ViolationType.RED_LIGHT_VIOLATION,
                    confidence=min(
                        vehicle.confidence, red_lights[0].confidence
                    ) * 0.9,  # slight discount for heuristic
                    involved_nodes=involved,
                    description=(
                        f"{display_name}: {vehicle.entity_class.value} "
                        f"proceeding through red signal (direction: {direction})."
                    ),
                    bbox=self._merge_bbox(
                        [vehicle, red_lights[0]]
                    ),
                    timestamp=self._now_iso(),
                    metadata={
                        "direction": direction,
                        "plate_text": plate_text,
                        "vehicle_type": vehicle.entity_class.value,
                        "signal_state": "red",
                    },
                )
            )

        return violations

    def _check_illegal_parking(
        self, sg: SceneGraph
    ) -> List[ViolationRecord]:
        """
        Detect illegally parked vehicles.

        Logic:
        1. Find vehicles with direction ``"STATIONARY"`` or no direction
           data and no ``track_id`` (untracked ≈ stationary for a while).
        2. In production this would also consult a no-parking zone map.
           For now, we flag stationary vehicles that have been detected
           consistently (placeholder for temporal accumulation).

        Note:
            Full illegal-parking detection requires multi-frame temporal
            analysis (duration threshold from ``Settings.illegal_parking_duration_sec``).
            This single-frame check only flags *candidates*.

        Returns:
            List of illegal-parking violation records.
        """
        violations: List[ViolationRecord] = []

        all_vehicle_classes = (
            TWO_WHEELER_CLASSES | FOUR_WHEELER_CLASSES | {EntityClass.AUTO_RICKSHAW}
        )

        for vehicle in sg.nodes.values():
            if vehicle.entity_class not in all_vehicle_classes:
                continue

            direction = vehicle.attributes.get("direction")
            # Only consider stationary vehicles
            if direction is not None and direction != "STATIONARY":
                continue

            # Require parking flag from temporal analysis (set externally)
            is_parked_flag = vehicle.attributes.get("is_parked", False)
            if not is_parked_flag:
                continue

            involved = [vehicle.node_id]
            plate_node = sg.get_plate_of(vehicle.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(
                ViolationType.ILLEGAL_PARKING, "Illegal Parking"
            )

            violations.append(
                ViolationRecord(
                    violation_id=self._generate_violation_id(),
                    violation_type=ViolationType.ILLEGAL_PARKING,
                    confidence=vehicle.confidence * 0.7,  # low conf for single-frame
                    involved_nodes=involved,
                    description=(
                        f"{display_name}: {vehicle.entity_class.value} appears "
                        f"to be illegally parked."
                    ),
                    bbox=vehicle.bbox,
                    timestamp=self._now_iso(),
                    metadata={
                        "plate_text": plate_text,
                        "vehicle_type": vehicle.entity_class.value,
                        "stationary": True,
                    },
                )
            )

        return violations

    # ──────────────────────────────────────────────────────────
    # INTERNAL HELPERS
    # ──────────────────────────────────────────────────────────

    def _generate_violation_id(self) -> str:
        """
        Generate a unique, human-readable violation ID.

        Format: ``VIO-YYYYMMDD-NNN`` where NNN is a zero-padded sequence
        number that resets daily.

        Returns:
            Unique violation ID string.
        """
        today = datetime.date.today().strftime("%Y%m%d")

        if self._use_session_state:
            import streamlit as st
            if "global_violation_date" not in st.session_state or st.session_state.get("global_violation_date") != today:
                st.session_state["global_violation_date"] = today
                st.session_state["global_violation_counter"] = 0
            st.session_state["global_violation_counter"] += 1
            seq = st.session_state["global_violation_counter"]
        else:
            with ViolationEngine._global_counter_lock:
                if today != ViolationEngine._global_date_str:
                    ViolationEngine._global_date_str = today
                    ViolationEngine._global_counter = 0

                ViolationEngine._global_counter += 1
                seq = ViolationEngine._global_counter

        return f"VIO-{today}-{seq:03d}"

    def _merge_bbox(
        self, nodes: List[SceneNode]
    ) -> Tuple[int, int, int, int]:
        """
        Compute the union bounding box that encompasses all *nodes*.

        Args:
            nodes: List of :class:`SceneNode` instances whose bounding
                   boxes should be merged.

        Returns:
            ``(x1, y1, x2, y2)`` — the smallest axis-aligned box that
            contains all input boxes.  Returns ``(0, 0, 0, 0)`` if the
            list is empty.
        """
        if not nodes:
            return (0, 0, 0, 0)

        x1 = min(n.bbox[0] for n in nodes)
        y1 = min(n.bbox[1] for n in nodes)
        x2 = max(n.bbox[2] for n in nodes)
        y2 = max(n.bbox[3] for n in nodes)
        return (x1, y1, x2, y2)

    @staticmethod
    def _now_iso() -> str:
        """Return the current UTC time as an ISO-8601 string."""
        return datetime.datetime.now(datetime.timezone.utc).isoformat()

    def __repr__(self) -> str:  # pragma: no cover
        return f"ViolationEngine(counter={ViolationEngine._global_counter})"