docs: new repo URL (andrew/...), requirements.txt, pytest smoke tests + CI

tests/test_batch.py

@@ -0,0 +1,67 @@
+"""Smoke tests for BatchLifecycleManager (src/batch.py). Stdlib only."""
+
+from src.batch import BatchLifecycleManager
+
+
+def _mgr(**kw):
+    args = dict(
+        stabilize_seconds=5.0,
+        stabilize_threshold_px=15.0,
+        sack_idle_timeout=10.0,
+        min_batch_duration=0.0,
+        truck_gone_tolerance=3.0,
+    )
+    args.update(kw)
+    return BatchLifecycleManager(**args)
+
+
+def test_idle_to_counting_after_stable_truck():
+    m = _mgr()
+    started = []
+    m.on_batch_start(lambda bid, ts: started.append(bid))
+    m.update_truck(True, (500.0, 300.0), timestamp=1000.0)
+    assert m.state == "TRUCK_STABILIZING"
+    assert not m.is_active
+    m.update_truck(True, (501.0, 301.0), timestamp=1006.0)  # stable 6s
+    assert m.state == "COUNTING_SACKS"
+    assert m.is_active and m.is_counting
+    assert started == [1]
+
+
+def test_truck_leaving_resets_to_idle():
+    m = _mgr(stabilize_seconds=0.0)  # instant start
+    m.update_truck(True, (500.0, 300.0), timestamp=1000.0)
+    assert m.is_active
+    ended = []
+    m.on_batch_end(ended.append)
+    m.update_sacks(False, 1, timestamp=1001.0, loading_count=3, unloading_count=1)
+    m.update_sacks(False, 0, timestamp=1012.0, loading_count=3, unloading_count=1)  # idle -> waiting
+    assert m.state == "WAITING_FOR_ACTIVITY"
+    m.update_truck(False, None, timestamp=1020.0)  # truck gone past tolerance
+    assert m.state == "IDLE"
+    assert len(ended) == 1
+    assert (ended[0].loading_count, ended[0].unloading_count) == (3, 1)
+    assert ended[0].net_count == 2
+
+
+def test_activity_resumes_same_batch():
+    m = _mgr(stabilize_seconds=0.0)
+    m.update_truck(True, (500.0, 300.0), timestamp=1000.0)
+    bid = m.current_batch_id
+    m.update_sacks(False, 0, timestamp=1012.0)  # -> waiting
+    assert m.is_waiting
+    m.update_sacks(True, 2, timestamp=1015.0)  # sacks resume
+    assert m.state == "COUNTING_SACKS"
+    assert m.current_batch_id == bid  # same batch, not a new one
+
+
+def test_legacy_update_shim():
+    m = _mgr(stabilize_seconds=0.0)
+    # shim passes centroid=None, so IDLE never leaves without a real centroid
+    m.update(truck_detected=True, timestamp=1000.0)
+    assert m.state == "IDLE"
+    # ...but once counting via the real API, the shim feeds update_sacks fine
+    m.update_truck(True, (500.0, 300.0), timestamp=1000.0)
+    assert m.is_active
+    m.update(truck_detected=True, timestamp=1001.0, loading_count=2)
+    assert m.is_active

tests/test_config.py

@@ -0,0 +1,28 @@
+"""Smoke tests for config loader (src/config.py). Needs python-dotenv only."""
+
+from src.config import load_config
+
+_KEYS = [
+    "LOCAL_RTSP", "JETSON_RTSP", "MODEL_SACK_PATH", "MODEL_TRUCK_PATH",
+    "COUNTING_LINE_Y", "COUNTING_LINE_X_START", "COUNTING_LINE_X_END",
+    "SACK_CONF_THRESHOLD", "TRUCK_CONF_THRESHOLD", "BATCH_TIMEOUT_SECONDS",
+    "CSV_OUTPUT_DIR", "DATA_SEED",
+]
+
+
+def test_load_from_env_file(tmp_path, monkeypatch):
+    for k in _KEYS:
+        monkeypatch.delenv(k, raising=False)
+    env = tmp_path / "test.env"
+    env.write_text(
+        "LOCAL_RTSP=rtsp://cam/1\n"
+        "MODEL_SACK_PATH=/m/sack.pt\n"
+        "MODEL_TRUCK_PATH=/m/truck.pt\n"
+        "SACK_CONF_THRESHOLD=0.55\n"
+    )
+    cfg = load_config(str(env))
+    assert cfg.local_rtsp == "rtsp://cam/1"
+    assert cfg.sack_model_path == "/m/sack.pt"
+    assert cfg.truck_model_path == "/m/truck.pt"
+    assert cfg.sack_conf == 0.55
+    assert cfg.truck_conf == 0.50  # default preserved

tests/test_counting.py

@@ -0,0 +1,60 @@
+"""Smoke tests for LineCrossCounter (src/counting.py). Pure-Python: needs only numpy."""
+
+from src.counting import LineCrossCounter
+from src.interfaces import Detection
+
+
+def _det(tid, y1, cx=500.0):
+    return Detection(
+        bbox=(cx - 20, y1, cx + 20, y1 + 60),
+        confidence=0.9,
+        class_id=0,
+        class_name="sack",
+        track_id=tid,
+    )
+
+
+def _counter():
+    return LineCrossCounter(line_y=100, line_x_start=0, line_x_end=1000, margin=20)
+
+
+def test_loading_above_then_below():
+    c = _counter()
+    assert c.update([_det(1, y1=10)]) == []          # above line
+    events = c.update([_det(1, y1=150)])             # below line
+    assert len(events) == 1
+    assert events[0]["direction"] == "loading"
+    assert c.loading_count == 1
+
+
+def test_below_first_counts_as_unloading_not_loading():
+    c = _counter()
+    assert c.update([_det(2, y1=150)]) == []         # appeared below first
+    assert c.update([_det(2, y1=10)]) != []          # moved above => unloading
+    assert c.unloading_count == 1
+    assert c.loading_count == 0
+
+
+def test_track_counted_once_per_direction():
+    c = _counter()
+    c.update([_det(3, y1=10)])
+    c.update([_det(3, y1=150)])
+    c.update([_det(3, y1=10)])
+    c.update([_det(3, y1=150)])
+    assert c.loading_count == 1  # track_id counted once for loading
+
+
+def test_outside_x_bounds_skipped():
+    c = _counter()
+    c.update([_det(4, y1=10, cx=500.0)])
+    assert c.update([_det(4, y1=150, cx=5000.0)]) == []
+    assert c.loading_count == 0
+
+
+def test_net_and_reset():
+    c = _counter()
+    c.update([_det(5, y1=10)])
+    c.update([_det(5, y1=150)])
+    assert c.net_count == 1
+    c.reset()
+    assert (c.loading_count, c.unloading_count, c.net_count) == (0, 0, 0)