Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
__pycache__/
*.pyc
eagle_surveillance.egg-info/
2 changes: 2 additions & 0 deletions services/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
from services import tracking
__all__ = ['tracking']
176 changes: 92 additions & 84 deletions services/detection/zones.py
Original file line number Diff line number Diff line change
@@ -1,84 +1,92 @@
"""
services/detection/zones.py

Zone definitions are now loaded from config/zones.yaml via ZoneConfigLoader.
Set ZONES_CONFIG_PATH env var to override the default config location.

Previously hardcoded DEFAULT_ZONES have been removed.
"""

import logging
from libs.config.zone_loader import ZoneConfigLoader
import numpy as np
from typing import List, Tuple


class Zone:
"""Lightweight wrapper around zone dicts from ZoneConfigLoader."""

def __init__(self, data: dict) -> None:
self.name: str = data.get("name")
self.polygon: List[Tuple[float, float]] = data.get("polygon", [])
self.alert_on_entry: bool = data.get("alert_on_entry", False)
self.color_hex: str = data.get("color_hex", "#FF0000")

def as_array(self) -> np.ndarray:
return np.array(self.polygon, dtype=np.int32)

@property
def color_bgr(self) -> Tuple[int, int, int]:
# hex #RRGGBB -> BGR tuple for OpenCV
h = self.color_hex.lstrip("#")
r = int(h[0:2], 16)
g = int(h[2:4], 16)
b = int(h[4:6], 16)
return (b, g, r)

def contains_point(self, x: float, y: float) -> bool:
# Ray casting algorithm for point-in-polygon
pts = self.polygon
inside = False
n = len(pts)
j = n - 1
for i in range(n):
xi, yi = pts[i]
xj, yj = pts[j]
intersect = ((yi > y) != (yj > y)) and (
x < (xj - xi) * (y - yi) / (yj - yi + 1e-12) + xi
)
if intersect:
inside = not inside
j = i
return inside

logger = logging.getLogger(__name__)

# Module-level singleton loader — starts hot-reload background thread
_loader = ZoneConfigLoader()
_loader.start()


def get_zones() -> list[dict]:
"""
Return the current list of zone dicts loaded from YAML.
Each zone has: name, polygon, alert_on_entry, color_hex.
"""
return _loader.get_zones()


def get_camera_id() -> str | None:
"""Return the camera_id from the active zone config."""
return _loader.get_camera_id()


# Convenience alias for code that previously referenced DEFAULT_ZONES directly
DEFAULT_ZONES = [Zone(z) for z in get_zones()]


def get_zones_for_point(x: float, y: float) -> List[Zone]:
"""Return list of Zone objects that contain the point (x, y).

Coordinates are in image pixel space (x horizontal, y vertical).
"""
zones = [Zone(z) for z in _loader.get_zones()]
return [z for z in zones if z.contains_point(x, y)]
"""
services/detection/zones.py
Zone definitions are now loaded from config/zones.yaml via ZoneConfigLoader.
Set ZONES_CONFIG_PATH env var to override the default config location.
Previously hardcoded DEFAULT_ZONES have been removed.
"""
import logging
from libs.config.zone_loader import ZoneConfigLoader

logger = logging.getLogger(__name__)

# Module-level singleton loader — starts hot-reload background thread
_loader = ZoneConfigLoader()
_loader.start()


def get_zones() -> list[dict]:
"""
Return the current list of zone dicts loaded from YAML.
Each zone has: name, polygon, alert_on_entry, color_hex.
"""
return _loader.get_zones()


def get_camera_id() -> str | None:
"""Return the camera_id from the active zone config."""
return _loader.get_camera_id()


def get_zones_for_point(x: float, y: float) -> list:
"""
Return all zones whose polygon contains the point (x, y).

Used by tracker.py to determine zone membership of a tracked object
given its centre-point coordinates.

Each returned object exposes a `.name` attribute so callers can do:
zones = [z.name for z in get_zones_for_point(cx, cy)]

Falls back gracefully (returns empty list) when:
- No zones are configured yet
- A zone polygon is malformed
- shapely is not installed (point-in-polygon skipped)
"""
zones = _loader.get_zones()
if not zones:
return []

matched = []

try:
from shapely.geometry import Point, Polygon

point = Point(x, y)
for zone in zones:
try:
poly = Polygon(zone["polygon"])
if poly.contains(point):
# Return a lightweight object with .name attribute
matched.append(_Zone(zone["name"]))
except Exception as e:
logger.warning(
"Skipping malformed polygon for zone '%s': %s",
zone.get("name", "unknown"),
e,
)

except ImportError:
# shapely not installed — skip spatial check, return empty
logger.debug(
"shapely not available; get_zones_for_point returning [] for (%.1f, %.1f)",
x, y,
)

return matched


class _Zone:
"""Lightweight zone result object exposing just the .name attribute."""

__slots__ = ("name",)

def __init__(self, name: str) -> None:
"""Store the zone name."""
self.name = name

def __repr__(self) -> str:
"""Return a readable string representation of the zone."""
return f"Zone(name={self.name!r})"

# Convenience alias for code that previously referenced DEFAULT_ZONES directly
DEFAULT_ZONES = get_zones()
Loading
Loading