68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""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
|