51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
|
"""1-minute downsampled tracks for DVR playback."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
from tracks import (
|
||
|
|
TRACK_BUCKET,
|
||
|
|
downsample_tracks,
|
||
|
|
minute_bucket,
|
||
|
|
positions_at_timestamp,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _ts(h, m, s=0):
|
||
|
|
return datetime(2026, 8, 28, h, m, s, tzinfo=timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
def test_minute_bucket_floors_seconds():
|
||
|
|
assert minute_bucket(_ts(12, 4, 47)) == _ts(12, 4, 0)
|
||
|
|
assert TRACK_BUCKET == "1 minute"
|
||
|
|
|
||
|
|
|
||
|
|
def test_downsample_keeps_last_sample_per_id_per_minute():
|
||
|
|
rows = [
|
||
|
|
{"id": "a1", "ts": _ts(12, 4, 10), "lat": 35.0, "lon": -79.0},
|
||
|
|
{"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1},
|
||
|
|
{"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2},
|
||
|
|
{"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0},
|
||
|
|
]
|
||
|
|
out = downsample_tracks(rows)
|
||
|
|
by = {(r["id"], r["bucket"]): r for r in out}
|
||
|
|
assert by[("a1", _ts(12, 4))]["lat"] == 35.1
|
||
|
|
assert by[("a1", _ts(12, 5))]["lat"] == 35.2
|
||
|
|
assert by[("b2", _ts(12, 4))]["lat"] == 36.0
|
||
|
|
assert len(out) == 3
|
||
|
|
|
||
|
|
|
||
|
|
def test_positions_at_timestamp_uses_that_minute_window():
|
||
|
|
rows = [
|
||
|
|
{"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1, "heading": 90, "speed": 10, "label": "A1"},
|
||
|
|
{"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2, "heading": 91, "speed": 11, "label": "A1"},
|
||
|
|
{"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0, "heading": 0, "speed": 0, "label": "B2"},
|
||
|
|
]
|
||
|
|
at = positions_at_timestamp(rows, _ts(12, 4, 59))
|
||
|
|
ids = {p["id"]: p for p in at}
|
||
|
|
assert ids["a1"]["lat"] == 35.1
|
||
|
|
assert ids["b2"]["lat"] == 36.0
|
||
|
|
later = positions_at_timestamp(rows, _ts(12, 5, 30))
|
||
|
|
assert {p["id"]: p["lat"] for p in later} == {"a1": 35.2}
|