diff --git a/tests/unit/test_wildlife_queries.py b/tests/unit/test_wildlife_queries.py new file mode 100644 index 0000000..2631624 --- /dev/null +++ b/tests/unit/test_wildlife_queries.py @@ -0,0 +1,38 @@ +import time +import pytest +from vigilar.storage.queries import ( + get_wildlife_stats, get_wildlife_frequency, get_wildlife_sightings_paginated, + insert_wildlife_sighting, +) + +@pytest.fixture +def seeded_wildlife(test_db): + for i in range(5): + insert_wildlife_sighting(test_db, species="deer", threat_level="PASSIVE", + camera_id="front", confidence=0.9, event_id=i + 1) + for i in range(3): + insert_wildlife_sighting(test_db, species="raccoon", threat_level="NUISANCE", + camera_id="back", confidence=0.8, event_id=i + 10) + insert_wildlife_sighting(test_db, species="bear", threat_level="PREDATOR", + camera_id="front", confidence=0.95, event_id=20) + return test_db + +def test_get_wildlife_stats(seeded_wildlife): + stats = get_wildlife_stats(seeded_wildlife) + assert stats["total"] == 9 + assert stats["per_species"]["deer"] == 5 + +def test_get_wildlife_frequency(seeded_wildlife): + freq = get_wildlife_frequency(seeded_wildlife) + assert len(freq) == 6 + +def test_get_wildlife_sightings_paginated(seeded_wildlife): + page1 = get_wildlife_sightings_paginated(seeded_wildlife, limit=5, offset=0) + assert len(page1) == 5 + page2 = get_wildlife_sightings_paginated(seeded_wildlife, limit=5, offset=5) + assert len(page2) == 4 + +def test_get_wildlife_sightings_filter_species(seeded_wildlife): + result = get_wildlife_sightings_paginated(seeded_wildlife, species="bear", limit=50, offset=0) + assert len(result) == 1 + assert result[0]["species"] == "bear" diff --git a/vigilar/storage/queries.py b/vigilar/storage/queries.py index 91b295b..973d647 100644 --- a/vigilar/storage/queries.py +++ b/vigilar/storage/queries.py @@ -426,6 +426,65 @@ def get_wildlife_sightings( return [dict(r._mapping) for r in rows] +# --- Wildlife Journal Queries --- + +def get_wildlife_sightings_paginated( + engine: Engine, + species: str | None = None, + threat_level: str | None = None, + camera_id: str | None = None, + since_ts: float | None = None, + until_ts: float | None = None, + limit: int = 50, + offset: int = 0, +) -> list[dict[str, Any]]: + query = ( + select(wildlife_sightings).order_by(desc(wildlife_sightings.c.ts)) + .limit(limit).offset(offset) + ) + if species: + query = query.where(wildlife_sightings.c.species == species) + if threat_level: + query = query.where(wildlife_sightings.c.threat_level == threat_level) + if camera_id: + query = query.where(wildlife_sightings.c.camera_id == camera_id) + if since_ts: + query = query.where(wildlife_sightings.c.ts >= since_ts) + if until_ts: + query = query.where(wildlife_sightings.c.ts < until_ts) + with engine.connect() as conn: + return [dict(r) for r in conn.execute(query).mappings().all()] + + +def get_wildlife_stats(engine: Engine) -> dict[str, Any]: + from sqlalchemy import func + with engine.connect() as conn: + total = conn.execute( + select(func.count()).select_from(wildlife_sightings) + ).scalar() or 0 + species_rows = conn.execute( + select(wildlife_sightings.c.species, func.count().label("cnt")) + .group_by(wildlife_sightings.c.species) + ).mappings().all() + per_species = {r["species"]: r["cnt"] for r in species_rows} + return {"total": total, "species_count": len(per_species), "per_species": per_species} + + +def get_wildlife_frequency(engine: Engine) -> dict[str, dict[str, int]]: + import datetime + buckets: dict[str, dict[str, int]] = { + "00-04": {}, "04-08": {}, "08-12": {}, "12-16": {}, "16-20": {}, "20-24": {} + } + with engine.connect() as conn: + rows = conn.execute(select(wildlife_sightings)).mappings().all() + for r in rows: + dt = datetime.datetime.fromtimestamp(r["ts"]) + bucket_key = list(buckets.keys())[dt.hour // 4] + species = r["species"] + buckets[bucket_key][species] = buckets[bucket_key].get(species, 0) + 1 + return buckets + + # --- Training Images --- def insert_training_image(