feat: add flexible multi-coop discovery, virtual floor synthesis, and stationary camera capture
This commit is contained in:
andrew committed
1 parent
c3b9c70221
commit
29384e5871
34 files changed
+1803
-516
No files matched your search
@@ -0,0 +1,184 @@
|
||||
"""Unit tests for dynamic coop, floor, and mortality discovery."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from chicken_counter.config import load_or_synthesize_floor_config
|
||||
from chicken_counter.discovery import (
|
||||
discover_all_mortality_dirs,
|
||||
discover_farm_structure,
|
||||
get_coop_for_location,
|
||||
resolve_floor_video_dir,
|
||||
sync_floor_configs,
|
||||
)
|
||||
|
||||
|
||||
class TestFlexibleDiscovery(unittest.TestCase):
|
||||
|
||||
def test_get_coop_for_location(self):
|
||||
self.assertEqual(get_coop_for_location("K1-L1"), "K1")
|
||||
self.assertEqual(get_coop_for_location("K2-L3"), "K2")
|
||||
self.assertEqual(get_coop_for_location("k1-l2"), "K1")
|
||||
self.assertEqual(get_coop_for_location("K3/L1"), "K3")
|
||||
self.assertEqual(get_coop_for_location("K4_L2"), "K4")
|
||||
self.assertEqual(get_coop_for_location("kandang-atas"), "kandang-atas")
|
||||
self.assertEqual(get_coop_for_location("kandang-bawah"), "kandang-bawah")
|
||||
self.assertEqual(get_coop_for_location("K5"), "K5")
|
||||
|
||||
def test_discovery_on_current_workspace(self):
|
||||
structure = discover_farm_structure()
|
||||
self.assertIsNotNone(structure)
|
||||
|
||||
# Check detected coops in workspace
|
||||
self.assertIn("K1", structure.coops)
|
||||
self.assertIn("K2", structure.coops)
|
||||
self.assertIn("K3", structure.coops)
|
||||
self.assertIn("K4", structure.coops)
|
||||
self.assertIn("K5", structure.coops)
|
||||
|
||||
# Verify covered floors for K1 (3 floors) and K4 (2 floors)
|
||||
k1_floors = structure.get_covered_floors("K1")
|
||||
self.assertIn("K1-L1", k1_floors)
|
||||
self.assertIn("K1-L2", k1_floors)
|
||||
self.assertIn("K1-L3", k1_floors)
|
||||
|
||||
k4_floors = structure.get_covered_floors("K4")
|
||||
self.assertIn("K4-L1", k4_floors)
|
||||
self.assertIn("K4-L2", k4_floors)
|
||||
|
||||
# Verify mortality directories
|
||||
mort_dirs = structure.all_mortality_dirs()
|
||||
self.assertIn("K1", mort_dirs)
|
||||
self.assertIn("K2", mort_dirs)
|
||||
self.assertIn("K3", mort_dirs)
|
||||
self.assertIn("K4", mort_dirs)
|
||||
self.assertIn("K5", mort_dirs)
|
||||
self.assertEqual(mort_dirs["K1"].name, "mortality")
|
||||
self.assertEqual(mort_dirs["K1"].parent.name, "K1")
|
||||
|
||||
def test_flexible_discovery_arbitrary_site(self):
|
||||
"""Test that discovery dynamically adapts to arbitrary site layouts with varying coops and floors."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp = Path(tmpdir)
|
||||
# Create arbitrary site layout:
|
||||
# Coop A (Kandang-Utama) has 4 floors (L1..L4) + mortality
|
||||
# Coop B (K9) has 1 floor (L1) + mortality
|
||||
# Coop C (K10) has 2 floors (L1, L2) with no mortality folder yet
|
||||
(tmp / "cycle1" / "Kandang-Utama" / "L1").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "Kandang-Utama" / "L2").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "Kandang-Utama" / "L3").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "Kandang-Utama" / "L4").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "Kandang-Utama" / "mortality").mkdir(parents=True)
|
||||
|
||||
(tmp / "cycle1" / "K9" / "L1").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "K9" / "mortality").mkdir(parents=True)
|
||||
|
||||
(tmp / "cycle1" / "K10" / "L1").mkdir(parents=True)
|
||||
(tmp / "cycle1" / "K10" / "L2").mkdir(parents=True)
|
||||
|
||||
structure = discover_farm_structure(videos_root=tmp)
|
||||
|
||||
self.assertTrue("KANDANG-UTAMA" in structure.coops or "Kandang-Utama" in structure.coops)
|
||||
coop_u = structure.get_coop("Kandang-Utama")
|
||||
self.assertIsNotNone(coop_u)
|
||||
self.assertEqual(len(coop_u.covered_floors), 4)
|
||||
self.assertIsNotNone(coop_u.mortality_dir)
|
||||
|
||||
coop_k9 = structure.get_coop("K9")
|
||||
self.assertIsNotNone(coop_k9)
|
||||
self.assertEqual(len(coop_k9.covered_floors), 1)
|
||||
self.assertIsNotNone(coop_k9.mortality_dir)
|
||||
|
||||
coop_k10 = structure.get_coop("K10")
|
||||
self.assertIsNotNone(coop_k10)
|
||||
self.assertEqual(len(coop_k10.covered_floors), 2)
|
||||
self.assertIsNone(coop_k10.mortality_dir)
|
||||
|
||||
m_dirs = structure.all_mortality_dirs()
|
||||
self.assertIn("K9", m_dirs)
|
||||
self.assertTrue("Kandang-Utama" in m_dirs or "KANDANG-UTAMA" in m_dirs)
|
||||
self.assertNotIn("K10", m_dirs)
|
||||
|
||||
def test_resolve_floor_video_dir(self):
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp = Path(tmpdir)
|
||||
# Create hierarchical directory
|
||||
hier_dir = tmp / "cycle7" / "K1" / "L1"
|
||||
hier_dir.mkdir(parents=True)
|
||||
|
||||
# Query using old flat path
|
||||
flat_query = tmp / "cycle7" / "K1-L1"
|
||||
resolved = resolve_floor_video_dir(flat_query, "K1-L1")
|
||||
self.assertEqual(resolved, hier_dir)
|
||||
|
||||
def test_sync_floor_configs(self):
|
||||
"""Test auto-generating floor YAML configs for newly discovered coops/floors."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp = Path(tmpdir)
|
||||
vid_dir = tmp / "VIDEOS" / "cycle7"
|
||||
cfg_dir = tmp / "configs" / "floor_config"
|
||||
|
||||
# Create new coop K8 with 3 floors
|
||||
(vid_dir / "K8" / "L1").mkdir(parents=True)
|
||||
(vid_dir / "K8" / "L2").mkdir(parents=True)
|
||||
(vid_dir / "K8" / "L3").mkdir(parents=True)
|
||||
|
||||
created = sync_floor_configs(videos_root=vid_dir.parent, configs_dir=cfg_dir)
|
||||
self.assertEqual(len(created), 3)
|
||||
|
||||
# Check that files were written
|
||||
self.assertTrue((cfg_dir / "K8-L1.yaml").exists())
|
||||
self.assertTrue((cfg_dir / "K8-L2.yaml").exists())
|
||||
self.assertTrue((cfg_dir / "K8-L3.yaml").exists())
|
||||
|
||||
content = (cfg_dir / "K8-L1.yaml").read_text()
|
||||
self.assertIn('location: "K8-L1"', content)
|
||||
self.assertIn('root_dir: "../VIDEOS/cycle7/K8/L1"', content)
|
||||
|
||||
# Running sync again should create 0 files
|
||||
created_again = sync_floor_configs(videos_root=vid_dir.parent, configs_dir=cfg_dir)
|
||||
self.assertEqual(len(created_again), 0)
|
||||
|
||||
def test_load_or_synthesize_floor_config(self):
|
||||
"""Test virtual in-memory config synthesis without any YAML file."""
|
||||
# Load virtual floor K9-L4 (no YAML exists)
|
||||
settings = load_or_synthesize_floor_config("K9-L4")
|
||||
self.assertEqual(settings.batch.location, "K9-L4")
|
||||
self.assertTrue(settings.batch.root_dir.endswith("VIDEOS/cycle7/K9/L4"))
|
||||
self.assertEqual(len(settings.cameras), 4)
|
||||
self.assertIn("CC1", settings.cameras)
|
||||
|
||||
def test_dated_mortality_discovery(self):
|
||||
"""Test that dated mortality subfolders (e.g. K1/mortality/2026-08-20/) are discovered."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp = Path(tmpdir)
|
||||
mort_day = tmp / "cycle7" / "K1" / "mortality" / "2026-08-20"
|
||||
mort_day.mkdir(parents=True)
|
||||
report = mort_day / "mortality_report.json"
|
||||
report.write_text('{"date":"2026-08-20","total_mortality_count":15,"results":[]}')
|
||||
|
||||
structure = discover_farm_structure(videos_root=tmp)
|
||||
self.assertIn("K1", structure.coops)
|
||||
self.assertIsNotNone(structure.coops["K1"].mortality_dir)
|
||||
|
||||
def test_stationary_camera_config(self):
|
||||
"""Test retrieving stationary camera sources for farm coops."""
|
||||
from chicken_counter.mortality import capture_camera_snapshot, get_camera_source_for_coop
|
||||
|
||||
cam_k1 = get_camera_source_for_coop("K1")
|
||||
self.assertIsNotNone(cam_k1)
|
||||
self.assertTrue("192.168.1.101" in str(cam_k1) or "rtsp" in str(cam_k1))
|
||||
|
||||
cam_k5 = get_camera_source_for_coop("K5")
|
||||
self.assertIsNotNone(cam_k5)
|
||||
|
||||
# Non-existent camera stream capture should return None gracefully without raising
|
||||
invalid_cap = capture_camera_snapshot("rtsp://invalid-non-existent-host.local:554/live", timeout_sec=0.5, warmup_frames=1)
|
||||
self.assertIsNone(invalid_cap)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user