feat: batch clip download from motionEye (trim+concat, history Klip button)
This commit is contained in:
andrew committed
1 parent
d3357fccef
commit
110d557dbf
8 files changed
+694
-6
No files matched your search
@@ -0,0 +1,275 @@
|
||||
"""Tests for src/clip.py — stdlib only, no network, no ffmpeg."""
|
||||
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
import src.clip as clip
|
||||
from src.clip import ClipError, find_clips, _path_start
|
||||
|
||||
|
||||
def dt(value: str) -> datetime:
|
||||
return datetime.fromisoformat(value)
|
||||
|
||||
|
||||
def fake_probe(durations):
|
||||
"""probe(path) -> duration; records calls on probe.calls."""
|
||||
calls = []
|
||||
|
||||
def probe(path):
|
||||
calls.append(path)
|
||||
return durations.get(path, 0.0)
|
||||
|
||||
probe.calls = calls
|
||||
return probe
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- path parsing
|
||||
|
||||
def test_path_start_valid_path():
|
||||
assert _path_start("2026-09-26/09-51-40.mp4") == datetime(2026, 9, 26, 9, 51, 40)
|
||||
assert _path_start("/2026-09-26/09-51-40.mp4") == datetime(2026, 9, 26, 9, 51, 40)
|
||||
|
||||
|
||||
def test_path_start_skips_unparseable_paths():
|
||||
assert _path_start("2026-09-26/09-51-40.avi") is None
|
||||
assert _path_start("09-51-40.mp4") is None
|
||||
assert _path_start("2026-9-26/09-51-40.mp4") is None
|
||||
assert _path_start("") is None
|
||||
assert _path_start(None) is None
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ find_clips
|
||||
|
||||
def test_find_clips_only_overlapping_clip():
|
||||
movies = [
|
||||
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
||||
{"path": "2026-09-26/09-20-00.mp4", "start": dt("2026-09-26 09:20:00")},
|
||||
]
|
||||
probe = fake_probe({
|
||||
"2026-09-26/09-00-00.mp4": 300.0,
|
||||
"2026-09-26/09-20-00.mp4": 60.0,
|
||||
})
|
||||
# B(09:20, 60s): window starts 10s into B and ends 50s into B.
|
||||
got = find_clips(movies, dt("2026-09-26 09:20:10"), dt("2026-09-26 09:20:50"), probe)
|
||||
assert [c["path"] for c in got] == ["2026-09-26/09-20-00.mp4"]
|
||||
assert got[0]["off0"] == 10.0
|
||||
assert got[0]["off1"] == 50.0
|
||||
assert got[0]["duration"] == 60.0
|
||||
|
||||
|
||||
def test_find_clips_window_touching_clip_end_excluded():
|
||||
# B(09:20, 60s) ends exactly 09:21; window 09:21→09:25 overlaps by 0s.
|
||||
movies = [
|
||||
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
||||
{"path": "2026-09-26/09-20-00.mp4", "start": dt("2026-09-26 09:20:00")},
|
||||
]
|
||||
probe = fake_probe({
|
||||
"2026-09-26/09-00-00.mp4": 300.0,
|
||||
"2026-09-26/09-20-00.mp4": 60.0,
|
||||
})
|
||||
got = find_clips(movies, dt("2026-09-26 09:21:00"), dt("2026-09-26 09:25:00"), probe)
|
||||
assert got == []
|
||||
|
||||
|
||||
def test_find_clips_window_inside_one_clip():
|
||||
movies = [{"path": "2026-09-26/10-00-00.mp4", "start": dt("2026-09-26 10:00:00")}]
|
||||
probe = fake_probe({"2026-09-26/10-00-00.mp4": 600.0})
|
||||
got = find_clips(movies, dt("2026-09-26 10:02:00"), dt("2026-09-26 10:04:00"), probe)
|
||||
assert len(got) == 1
|
||||
assert got[0]["off0"] == 120.0
|
||||
assert got[0]["off1"] == 240.0
|
||||
|
||||
|
||||
def test_find_clips_window_spanning_two_clips_in_order():
|
||||
movies = [
|
||||
{"path": "2026-09-26/09-04-00.mp4", "start": dt("2026-09-26 09:04:00")},
|
||||
{"path": "2026-09-26/09-00-00.mp4", "start": dt("2026-09-26 09:00:00")},
|
||||
]
|
||||
probe = fake_probe({
|
||||
"2026-09-26/09-00-00.mp4": 300.0,
|
||||
"2026-09-26/09-04-00.mp4": 600.0,
|
||||
})
|
||||
got = find_clips(movies, dt("2026-09-26 09:03:00"), dt("2026-09-26 09:05:30"), probe)
|
||||
assert [c["path"] for c in got] == [
|
||||
"2026-09-26/09-00-00.mp4",
|
||||
"2026-09-26/09-04-00.mp4",
|
||||
]
|
||||
assert (got[0]["off0"], got[0]["off1"]) == (180.0, 300.0)
|
||||
assert (got[1]["off0"], got[1]["off1"]) == (0.0, 90.0)
|
||||
|
||||
|
||||
def test_find_clips_no_overlap_empty():
|
||||
movies = [{"path": "2026-09-26/10-35-00.mp4", "start": dt("2026-09-26 10:35:00")}]
|
||||
probe = fake_probe({"2026-09-26/10-35-00.mp4": 60.0})
|
||||
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"), probe)
|
||||
assert got == []
|
||||
|
||||
|
||||
def test_find_clips_probe_zero_duration_skipped():
|
||||
movies = [{"path": "2026-09-26/11-59-00.mp4", "start": dt("2026-09-26 11:59:00")}]
|
||||
probe = fake_probe({})
|
||||
assert probe("2026-09-26/11-59-00.mp4") == 0.0
|
||||
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"),
|
||||
fake_probe({"2026-09-26/11-59-00.mp4": 0.0}))
|
||||
assert got == []
|
||||
|
||||
|
||||
def test_find_clips_probe_error_treated_as_zero():
|
||||
def broken_probe(path):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
movies = [{"path": "2026-09-26/11-59-00.mp4", "start": dt("2026-09-26 11:59:00")}]
|
||||
got = find_clips(movies, dt("2026-09-26 12:00:00"), dt("2026-09-26 12:01:00"),
|
||||
broken_probe)
|
||||
assert got == []
|
||||
|
||||
|
||||
def test_find_clips_midnight_crossing_window():
|
||||
movies = [
|
||||
{"path": "/2026-09-26/23-58-00.mp4", "start": dt("2026-09-26 23:58:00")},
|
||||
{"path": "/2026-09-27/00-00-30.mp4", "start": dt("2026-09-27 00:00:30")},
|
||||
]
|
||||
probe = fake_probe({
|
||||
"/2026-09-26/23-58-00.mp4": 300.0,
|
||||
"/2026-09-27/00-00-30.mp4": 300.0,
|
||||
})
|
||||
got = find_clips(movies, dt("2026-09-26 23:59:00"), dt("2026-09-27 00:01:00"), probe)
|
||||
assert [c["path"] for c in got] == [
|
||||
"/2026-09-26/23-58-00.mp4",
|
||||
"/2026-09-27/00-00-30.mp4",
|
||||
]
|
||||
assert (got[1]["off0"], got[1]["off1"]) == (0.0, 30.0)
|
||||
|
||||
|
||||
def test_find_clips_old_clip_not_probed():
|
||||
start, end = dt("2026-09-26 12:00:00"), dt("2026-09-26 12:10:00")
|
||||
old = {"path": "old.mp4", "start": start - clip._CANDIDATE_BACK - clip.timedelta(seconds=1)}
|
||||
edge = {"path": "edge.mp4", "start": start - clip._CANDIDATE_BACK}
|
||||
inside = {"path": "inside.mp4", "start": dt("2026-09-26 11:59:00")}
|
||||
probe = fake_probe({
|
||||
"old.mp4": 3600.0,
|
||||
"edge.mp4": 3600.0,
|
||||
"inside.mp4": 60.0,
|
||||
})
|
||||
find_clips([old, edge, inside], start, end, probe)
|
||||
assert "old.mp4" not in probe.calls
|
||||
assert "edge.mp4" in probe.calls
|
||||
assert "inside.mp4" in probe.calls
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- clip_batch
|
||||
|
||||
BASE = "http://192.168.192.253:8765"
|
||||
CLIP_A = "/2026-09-26/09-00-00.mp4"
|
||||
CLIP_B = "/2026-09-26/09-04-00.mp4"
|
||||
START_ISO = "2026-09-26T09:03:00"
|
||||
END_ISO = "2026-09-26T09:05:30"
|
||||
|
||||
|
||||
def test_clip_batch_no_recording_raises(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(clip, "list_movies", lambda *a, **k: [])
|
||||
monkeypatch.setattr(clip, "find_clips", lambda *a, **k: [])
|
||||
with pytest.raises(ClipError, match="Tidak ada rekaman motionEye"):
|
||||
clip.clip_batch(BASE, 1, START_ISO, END_ISO, str(tmp_path / "out.mp4"))
|
||||
|
||||
|
||||
def fake_run_factory(calls, captured):
|
||||
def fake_run(args, timeout):
|
||||
args = list(args)
|
||||
calls.append((args, timeout))
|
||||
if "concat" in args:
|
||||
list_file = args[args.index("-i") + 1]
|
||||
with open(list_file) as handle:
|
||||
captured["concat_list"] = handle.read()
|
||||
target = args[-1]
|
||||
os.makedirs(os.path.dirname(target), exist_ok=True)
|
||||
with open(target, "wb") as handle:
|
||||
handle.write(b"piece-bytes")
|
||||
return fake_run
|
||||
|
||||
|
||||
def patch_batch_env(monkeypatch, movies, duration=600.0):
|
||||
monkeypatch.setattr(clip, "list_movies", lambda *a, **k: list(movies))
|
||||
monkeypatch.setattr(clip, "probe_duration", lambda url, timeout=30: duration)
|
||||
|
||||
|
||||
def test_clip_batch_single_piece_argv(monkeypatch, tmp_path):
|
||||
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
||||
patch_batch_env(monkeypatch, movies, duration=600.0)
|
||||
calls, captured = [], {}
|
||||
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
||||
out = tmp_path / "sub" / "batch.mp4"
|
||||
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50", str(out))
|
||||
assert len(calls) == 1
|
||||
args, timeout = calls[0]
|
||||
assert timeout == 180
|
||||
piece = args[-1]
|
||||
assert args[:-1] == [
|
||||
"ffmpeg", "-y",
|
||||
"-ss", "10.000",
|
||||
"-to", "50.000",
|
||||
"-i", BASE + "/movie/1/download/2026-09-26/09-00-00.mp4",
|
||||
"-c", "copy",
|
||||
]
|
||||
assert piece.endswith(".mp4")
|
||||
# temp dir created next to out_path's dir
|
||||
assert os.path.dirname(piece).startswith(str(tmp_path / "sub") + os.sep)
|
||||
assert out.exists() and out.read_bytes() == b"piece-bytes"
|
||||
assert not os.path.exists(os.path.dirname(piece)) # temp dir cleaned up
|
||||
|
||||
|
||||
def test_clip_batch_two_pieces_trim_and_concat(monkeypatch, tmp_path):
|
||||
movies = [
|
||||
{"path": CLIP_B, "start": dt("2026-09-26 09:04:00")},
|
||||
{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")},
|
||||
]
|
||||
patch_batch_env(monkeypatch, movies, duration=600.0)
|
||||
calls, captured = [], {}
|
||||
monkeypatch.setattr(clip, "_run_ffmpeg", fake_run_factory(calls, captured))
|
||||
out = tmp_path / "batch.mp4"
|
||||
clip.clip_batch(BASE, 1, START_ISO, END_ISO, str(out), timeout=60)
|
||||
assert len(calls) == 3
|
||||
(piece_a, timeout_a), (piece_b, _), (concat_args, timeout_b) = calls
|
||||
assert timeout_a == timeout_b == 60
|
||||
assert piece_a[:8] == [
|
||||
"ffmpeg", "-y", "-ss", "180.000", "-to", "330.000", "-i",
|
||||
BASE + "/movie/1/download/2026-09-26/09-00-00.mp4",
|
||||
]
|
||||
assert piece_a[8:10] == ["-c", "copy"]
|
||||
assert piece_b[:8] == [
|
||||
"ffmpeg", "-y", "-ss", "0.000", "-to", "90.000", "-i",
|
||||
BASE + "/movie/1/download/2026-09-26/09-04-00.mp4",
|
||||
]
|
||||
assert piece_b[8:10] == ["-c", "copy"]
|
||||
assert concat_args == [
|
||||
"ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i",
|
||||
concat_args[concat_args.index("-i") + 1],
|
||||
"-c", "copy", str(out),
|
||||
]
|
||||
lines = captured["concat_list"].splitlines()
|
||||
assert len(lines) == 2
|
||||
assert all(line.startswith("file '") and line.endswith(".mp4'") for line in lines)
|
||||
assert lines[0].endswith("piece_000.mp4'")
|
||||
assert lines[1].endswith("piece_001.mp4'")
|
||||
assert out.exists()
|
||||
|
||||
|
||||
def test_clip_batch_ffmpeg_failure_raises_cliperror(monkeypatch, tmp_path):
|
||||
movies = [{"path": CLIP_A, "start": dt("2026-09-26 09:00:00")}]
|
||||
patch_batch_env(monkeypatch, movies, duration=600.0)
|
||||
|
||||
def failing_run(args, timeout):
|
||||
raise ClipError("ffmpeg gagal: ...Connection refused")
|
||||
|
||||
monkeypatch.setattr(clip, "_run_ffmpeg", failing_run)
|
||||
with pytest.raises(ClipError, match="Connection refused"):
|
||||
clip.clip_batch(BASE, 1, "2026-09-26T09:00:10", "2026-09-26T09:00:50",
|
||||
str(tmp_path / "out.mp4"))
|
||||
|
||||
|
||||
def test_clip_batch_bad_iso_raises_cliperror(tmp_path):
|
||||
with pytest.raises(ClipError):
|
||||
clip.clip_batch(BASE, 1, "bukan-tanggal", "2026-09-26T10:00:00",
|
||||
str(tmp_path / "out.mp4"))
|
||||
Reference in new issue
Block a user