108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
from app.detector import CycleDetector, PowerSample
|
|
from app.storage import Storage, to_db_time
|
|
|
|
|
|
def at(second: int) -> datetime:
|
|
return datetime(2026, 9, 5, 10, 0, second, tzinfo=UTC)
|
|
|
|
|
|
def make_storage(tmp_path) -> Storage:
|
|
storage = Storage(str(tmp_path / "monitor.sqlite3"))
|
|
storage.open()
|
|
return storage
|
|
|
|
|
|
def feed(storage: Storage, detector: CycleDetector, second: int, watts: float):
|
|
sampled_at = at(second)
|
|
storage.record_sample(sampled_at, watts, 230.0, 0.0, True)
|
|
return detector.process(PowerSample(sampled_at, watts))
|
|
|
|
|
|
def test_records_confirmed_cycle_using_first_threshold_crossings(tmp_path):
|
|
storage = make_storage(tmp_path)
|
|
detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2)
|
|
|
|
feed(storage, detector, 0, 2)
|
|
assert feed(storage, detector, 1, 120) == []
|
|
started = feed(storage, detector, 2, 710)
|
|
assert started[0]["type"] == "started"
|
|
assert detector.running
|
|
|
|
feed(storage, detector, 3, 700)
|
|
assert feed(storage, detector, 4, 10) == []
|
|
stopped = feed(storage, detector, 5, 3)
|
|
|
|
assert stopped[0]["type"] == "stopped"
|
|
cycle = stopped[0]["cycle"]
|
|
assert cycle["started_at"] == "2026-09-05T10:00:01.000Z"
|
|
assert cycle["ended_at"] == "2026-09-05T10:00:04.000Z"
|
|
assert cycle["duration_seconds"] == 3.0
|
|
assert cycle["maximum_watts"] == 710.0
|
|
assert cycle["incomplete_start"] == 0
|
|
assert cycle["incomplete_end"] == 0
|
|
|
|
|
|
def test_hysteresis_ignores_noise_between_thresholds(tmp_path):
|
|
storage = make_storage(tmp_path)
|
|
detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2)
|
|
|
|
feed(storage, detector, 0, 0)
|
|
feed(storage, detector, 1, 105)
|
|
feed(storage, detector, 2, 80)
|
|
assert not detector.running
|
|
|
|
feed(storage, detector, 3, 150)
|
|
feed(storage, detector, 4, 160)
|
|
assert detector.running
|
|
feed(storage, detector, 5, 45)
|
|
assert detector.running
|
|
feed(storage, detector, 6, 25)
|
|
feed(storage, detector, 7, 40)
|
|
assert detector.running
|
|
|
|
|
|
def test_initial_running_state_and_interruption_are_marked_partial(tmp_path):
|
|
storage = make_storage(tmp_path)
|
|
detector = CycleDetector(storage, start_watts=100, stop_watts=30, confirm_samples=2)
|
|
|
|
feed(storage, detector, 0, 600)
|
|
feed(storage, detector, 1, 610)
|
|
event = detector.interrupt(at(2))
|
|
|
|
assert event is not None
|
|
cycle = event["cycle"]
|
|
assert cycle["incomplete_start"] == 1
|
|
assert cycle["incomplete_end"] == 1
|
|
|
|
|
|
def test_open_cycle_summary_uses_current_time_not_end_of_day(tmp_path, monkeypatch):
|
|
storage = make_storage(tmp_path)
|
|
start = datetime.now(UTC) - timedelta(seconds=12)
|
|
storage.start_cycle(start)
|
|
summary = storage.summary_between(start - timedelta(minutes=1), start + timedelta(days=1))
|
|
|
|
assert 11 <= summary["runtime_seconds"] <= 14
|
|
|
|
|
|
def test_history_downsampling_preserves_peak_and_cycle_samples(tmp_path):
|
|
storage = make_storage(tmp_path)
|
|
start = at(0)
|
|
for index in range(120):
|
|
sampled_at = start + timedelta(seconds=index)
|
|
watts = 1700.0 if index == 61 else (1100.0 if 60 <= index <= 64 else 0.0)
|
|
storage.record_sample(sampled_at, watts, 230.0, 0.0, True)
|
|
|
|
history = storage.sample_history(
|
|
start,
|
|
start + timedelta(seconds=119),
|
|
max_points=20,
|
|
preserve_ranges=[(start + timedelta(seconds=60), start + timedelta(seconds=64))],
|
|
)
|
|
|
|
retained = {row["sampled_at"]: row["watts"] for row in history}
|
|
assert max(retained.values()) == 1700.0
|
|
for second in range(60, 65):
|
|
assert to_db_time(start + timedelta(seconds=second)) in retained
|