frigate/frigate/test/test_recording_coverage.py

"""Tests for the unified recording coverage resolver."""

import asyncio
import json
import unittest
from types import SimpleNamespace

from peewee import SqliteDatabase

from frigate.api.media import _vod_response
from frigate.const import MAX_SEGMENT_DURATION
from frigate.models import Recordings
from frigate.util.media import parse_playback_index
from frigate.util.recording_coverage import (
    _rows_query,
    coverage_spans,
    null_audio_glitches,
    plan_clip,
    realized_timeline,
    recording_overlap,
    resolve_coverage,
    stream_media_summary,
)


class CoverageDbTestCase(unittest.TestCase):
    def setUp(self):
        # in-memory database keeps these tests isolated from the shared
        # on-disk test.db used by the http api tests
        self.db = SqliteDatabase(":memory:")
        models = [Recordings]
        self.db.bind(models)
        self.db.create_tables(models)

    def tearDown(self):
        self.db.close()

    def _insert(
        self,
        id,
        start,
        end,
        stream_type,
        camera="front_door",
        video_codec=None,
        audio_rate=None,
        audio_codec=None,
        has_audio=None,
        segment_size=0,
        keyframes=None,
        audio_duration=None,
    ):
        Recordings.create(
            id=id,
            camera=camera,
            path=f"/tmp/{id}.mp4",
            start_time=start,
            end_time=end,
            duration=end - start,
            stream_type=stream_type,
            video_codec=video_codec,
            audio_rate=audio_rate,
            audio_codec=audio_codec,
            has_audio=has_audio,
            segment_size=segment_size,
            keyframes=keyframes,
            audio_duration=audio_duration,
        )


class TestRecordingCoverage(CoverageDbTestCase):
    def test_both_streams_offset_boundaries(self):
        # main: [1000,1010) [1010,1020)   sub: [1003,1013) [1013,1023)
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("m2", 1010.0, 1020.0, "main")
        self._insert("s1", 1003.0, 1013.0, "sub")
        self._insert("s2", 1013.0, 1023.0, "sub")
        intervals = resolve_coverage("front_door", 1000.0, 1020.0)
        # boundaries: 1000,1003,1010,1013,1020
        assert [round(i.start_time) for i in intervals] == [1000, 1003, 1010, 1013]
        assert intervals[0].main is not None and intervals[0].sub is None
        assert intervals[1].main is not None and intervals[1].sub is not None

    def test_sub_only_when_main_expired(self):
        self._insert("s1", 1000.0, 1010.0, "sub")
        intervals = resolve_coverage("front_door", 1000.0, 1010.0)
        assert len(intervals) == 1
        assert intervals[0].main is None and intervals[0].sub is not None

    def test_gap_when_neither_exists(self):
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("m2", 1030.0, 1040.0, "main")
        intervals = resolve_coverage("front_door", 1000.0, 1040.0)
        assert len(intervals) == 2  # the [1010,1030) gap produces no interval

    def test_spans_merge_same_availability(self):
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("m2", 1010.0, 1020.0, "main")
        spans = coverage_spans(resolve_coverage("front_door", 1000.0, 1020.0))
        assert spans == [
            {"start_time": 1000.0, "end_time": 1020.0, "streams": ["main"]}
        ]

    def test_requested_range_clamps_intervals(self):
        # query range partially overlaps the rows; the first and last
        # intervals must be clamped to the requested [after, before]
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("m2", 1010.0, 1020.0, "main")
        intervals = resolve_coverage("front_door", 1005.0, 1015.0)
        assert intervals[0].start_time == 1005.0
        assert intervals[-1].end_time == 1015.0
        spans = coverage_spans(intervals)
        assert spans == [
            {"start_time": 1005.0, "end_time": 1015.0, "streams": ["main"]}
        ]

    def test_stream_media_summary_newest_known_wins(self):
        # the newer main row's NULL codecs must not mask the older row's
        # known codecs, while its known audio fields take precedence
        self._insert(
            "m1",
            1000.0,
            1010.0,
            "main",
            video_codec="hevc",
            audio_rate=16000,
            audio_codec="aac",
        )
        self._insert("m2", 1010.0, 1020.0, "main", audio_rate=8000, has_audio=True)
        self._insert("s1", 1000.0, 1010.0, "sub", has_audio=False)
        summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1020.0))
        assert summary == {
            "main": {
                "video_codec": "hevc",
                "audio_rate": 8000,
                "audio_codec": "aac",
                "has_audio": True,
                "bitrate": None,
            },
            "sub": {
                "video_codec": None,
                "audio_rate": None,
                "audio_codec": None,
                "has_audio": False,
                "bitrate": None,
            },
        }

    def test_stream_media_summary_omits_absent_stream(self):
        self._insert("m1", 1000.0, 1010.0, "main", video_codec="h264")
        summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1010.0))
        assert list(summary.keys()) == ["main"]
        assert stream_media_summary([]) == {}

    def test_stream_media_summary_bitrate_weighted_by_duration(self):
        # main: 10 MiB over 10s + 30 MiB over 20s -> 40 MiB / 30s
        self._insert("m1", 1000.0, 1010.0, "main", segment_size=10)
        self._insert("m2", 1010.0, 1030.0, "main", segment_size=30)
        # sub rows without a stored size (legacy default 0) report None
        self._insert("s1", 1000.0, 1030.0, "sub")
        summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1030.0))
        expected = int(40 * 1024 * 1024 * 8 / 30)
        assert summary["main"]["bitrate"] == expected
        assert summary["sub"]["bitrate"] is None

    def test_stream_media_summary_bitrate_skips_zero_size_rows(self):
        # the zero-size glitch row contributes neither bytes nor seconds
        self._insert("m1", 1000.0, 1010.0, "main", segment_size=5)
        self._insert("m2", 1010.0, 1020.0, "main", segment_size=0)
        summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1020.0))
        expected = int(5 * 1024 * 1024 * 8 / 10)
        assert summary["main"]["bitrate"] == expected

    def test_unknown_audio_row_keeps_video_only_stream(self):
        self._insert("s1", 1000.0, 1010.0, "sub", has_audio=False)
        self._insert("s2", 1010.0, 1020.0, "sub", has_audio=None)
        self._insert("s3", 1020.0, 1030.0, "sub", has_audio=False)
        kept = null_audio_glitches(resolve_coverage("front_door", 1000.0, 1030.0))
        self.assertEqual(
            [i.sub.path for i in kept], [f"/tmp/s{n}.mp4" for n in (1, 2, 3)]
        )

    def test_video_only_glitch_dropped_on_audio_stream(self):
        self._insert("s1", 1000.0, 1010.0, "sub", has_audio=True)
        self._insert("s2", 1010.0, 1020.0, "sub", has_audio=False)
        self._insert("s3", 1020.0, 1030.0, "sub", has_audio=None)
        kept = null_audio_glitches(resolve_coverage("front_door", 1000.0, 1030.0))
        self.assertEqual([i.sub.path for i in kept], ["/tmp/s1.mp4", "/tmp/s3.mp4"])

    def test_other_camera_rows_excluded(self):
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("o1", 1000.0, 1010.0, "main", camera="back_yard")
        self._insert("o2", 1000.0, 1010.0, "sub", camera="back_yard")
        intervals = resolve_coverage("front_door", 1000.0, 1010.0)
        assert len(intervals) == 1
        assert intervals[0].main is not None and intervals[0].sub is None
        assert resolve_coverage("side_gate", 1000.0, 1010.0) == []

    def test_realized_timeline_snap_lead_in(self):
        """A mid-file resume's back-snap lead-in appears in the realized duration.

        Sub resumes at 1010 with a 7000ms inpoint; the stored keyframe
        index snaps back to 6000ms, so the clip serves 1000ms of pre-span
        content on top of the 3000ms span.
        """
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("s1", 1003.0, 1013.0, "sub", keyframes=[0, 2000, 4000, 6000, 8000])
        timeline = realized_timeline(
            resolve_coverage("front_door", 1000.0, 1013.0), None
        )
        assert timeline == [
            {"start_time": 1000.0, "end_time": 1010.0, "duration": 10000},
            {"start_time": 1010.0, "end_time": 1013.0, "duration": 4000},
        ]

    def test_realized_timeline_no_keyframe_index_serves_whole_file(self):
        """A row without a stored keyframe index serves the whole file.

        NULL keyframes mean the record-time probe failed, so a mid-file
        entry cannot be snapped to a decodable frame.
        """
        self._insert("m1", 1000.0, 1010.0, "main")
        self._insert("s1", 1003.0, 1013.0, "sub")
        timeline = realized_timeline(
            resolve_coverage("front_door", 1000.0, 1013.0), None
        )
        # the resume clip serves all 10000ms of the sub file for a 3000ms span
        assert timeline[1] == {
            "start_time": 1010.0,
            "end_time": 1013.0,
            "duration": 10000,
        }

    def test_realized_timeline_pinned_stream_no_mid_range_lead_in(self):
        """A pinned stream's spans start at its own file starts: no lead-in."""
        self._insert("s1", 1000.0, 1010.0, "sub", keyframes=[0, 5000])
        self._insert("s2", 1010.0, 1020.0, "sub", keyframes=[0, 5000])
        timeline = realized_timeline(
            resolve_coverage("front_door", 1000.0, 1020.0), "sub"
        )
        assert [t["duration"] for t in timeline] == [10000, 10000]

    def test_rows_query_scan_is_bounded_below(self):
        """The row query must give SQLite a lower bound on start_time.

        Without one the best index range is (camera=? AND start_time<?),
        which walks the camera's entire history on every coverage call:
        ~20s on a cold page cache, blocking the API event loop.
        """
        # the index the planner picks in production (migration 020)
        self.db.execute_sql(
            "CREATE INDEX recordings_api_recordings_summary ON recordings "
            '("camera", "start_time" DESC, "duration", "motion", "objects")'
        )
        sql, params = _rows_query("front_door", 2000.0, 3000.0, "main").sql()
        plan = " ".join(
            row[-1] for row in self.db.execute_sql("EXPLAIN QUERY PLAN " + sql, params)
        )
        assert "start_time>?" in plan and "start_time<?" in plan, plan

    def test_longest_segment_spanning_window_start_included(self):
        # a maximum-length segment overlapping `after` sits right at the
        # start_time lower bound the query is allowed to apply
        self._insert("m1", 1000.0, 1000.0 + MAX_SEGMENT_DURATION - 0.5, "main")
        intervals = resolve_coverage("front_door", 1599.0, 1700.0)
        assert len(intervals) == 1
        assert intervals[0].main is not None

    def test_plan_clip_skips_too_short(self):
        """A clip left with under 100ms of media is skipped (duration 0 in timelines)."""
        row = SimpleNamespace(
            path="/tmp/x.mp4",
            start_time=1000.0,
            end_time=1000.15,
            duration=0.15,
            keyframes=[0, 100],
        )
        plan = plan_clip(row, 1000.1, 1000.15)
        assert plan.skipped
        assert plan.duration_ms == 0


class TestRecordingOverlap(CoverageDbTestCase):
    def test_matches_unbounded_overlap(self):
        spans = [(1000, 1010), (1010, 1020), (1100, 1100 + MAX_SEGMENT_DURATION - 0.5)]
        for i, (start, end) in enumerate(spans):
            self._insert(f"r{i}", start, end, "main")

        probes = [995, 1000, 1005, 1010, 1015, 1020, 1050, 1100, 1650, 1699.5, 1700]
        for after in probes:
            for before in probes:
                if before < after:
                    continue

                with self.subTest(after=after, before=before):
                    unbounded = Recordings.select(Recordings.id).where(
                        Recordings.start_time.between(after, before)
                        | Recordings.end_time.between(after, before)
                        | (
                            (after > Recordings.start_time)
                            & (before < Recordings.end_time)
                        )
                    )
                    bounded = Recordings.select(Recordings.id).where(
                        recording_overlap(after, before)
                    )
                    self.assertEqual({r.id for r in bounded}, {r.id for r in unbounded})


class TestVodManifestPolicy(CoverageDbTestCase):
    """Discontinuity policy of the vod mapping builder."""

    def _mapping(self, start, end, stream=None):
        response = asyncio.run(
            _vod_response("front_door", start, end, stream_preference=stream)
        )
        return json.loads(response.body)

    def _insert_stream(self, prefix, stream_type, spans):
        for idx, (start, end) in enumerate(spans):
            self._insert(
                f"{prefix}{idx}",
                start,
                end,
                stream_type,
                video_codec="h264",
                audio_codec="aac",
                audio_rate=16000,
                has_audio=True,
            )

    def test_cross_stream_handoff_forces_discontinuity(self):
        # codec and audio params match on both streams, so only the stream
        # mix can force per-clip init segments here
        self._insert_stream("m", "main", [(1000.0, 1010.0)])
        self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
        mapping = self._mapping(1000.0, 1020.0)
        assert mapping["discontinuity"] is True
        assert mapping["initialClipIndex"] == 1
        assert len(mapping["sequences"][0]["clips"]) == 2

    def test_single_stream_range_stays_consistent(self):
        self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
        mapping = self._mapping(1000.0, 1020.0)
        assert mapping["discontinuity"] is False
        assert "initialClipIndex" not in mapping

    def test_pinned_stream_over_mixed_coverage_stays_consistent(self):
        self._insert_stream("m", "main", [(1000.0, 1010.0), (1010.0, 1020.0)])
        self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
        mapping = self._mapping(1000.0, 1020.0, stream="sub")
        assert mapping["discontinuity"] is False
        assert "initialClipIndex" not in mapping


class TestClipKeyFrames(unittest.TestCase):
    """Keyframe data emitted for nginx-vod segmentation."""

    def _row(self, keyframes):
        return SimpleNamespace(
            path="/tmp/kf.mp4",
            start_time=1000.0,
            end_time=1010.0,
            duration=10.0,
            keyframes=keyframes,
        )

    def test_whole_file_clip_emits_keyframe_gaps(self):
        keyframes = [0, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000]
        plan = plan_clip(self._row(keyframes), 1000.0, 1010.0)
        assert plan.clip_from_ms is None
        assert plan.key_frame_durations == [1000] * 9
        assert plan.first_key_frame_offset_ms == 0

    def test_mid_file_clip_offsets_are_clip_relative(self):
        # start 3.5s in: clipFrom snaps back to the 3000ms keyframe, so
        # the emitted offsets must be relative to that snapped start
        keyframes = [0, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000]
        plan = plan_clip(self._row(keyframes), 1003.5, 1010.0)
        assert plan.clip_from_ms == 3000
        assert plan.duration_ms == 7000
        assert plan.key_frame_durations == [1000] * 6
        assert plan.first_key_frame_offset_ms == 0

    def test_no_index_reports_none(self):
        plan = plan_clip(self._row(None), 1000.0, 1010.0)
        assert plan.key_frame_durations is None

    def test_single_keyframe_in_range_reports_none(self):
        # one cut point cannot split anything; the manifest falls back
        # to a whole-clip declaration
        plan = plan_clip(self._row([0]), 1000.0, 1010.0)
        assert plan.key_frame_durations is None

    def test_keyframe_on_clip_end_excluded(self):
        # a keyframe landing exactly on the clip end would only declare
        # a zero-length tail segment
        plan = plan_clip(self._row([0, 5000, 10000]), 1000.0, 1010.0)
        assert plan.key_frame_durations == [5000]


class TestVodManifestKeyFrames(CoverageDbTestCase):
    """Mapping emission of real vs placeholder keyFrameDurations."""

    def _clips(self, start, end):
        response = asyncio.run(_vod_response("front_door", start, end))
        return json.loads(response.body)["sequences"][0]["clips"]

    def test_indexed_row_emits_real_gaps(self):
        self._insert(
            "m1",
            1000.0,
            1010.0,
            "main",
            keyframes=[0, 2000, 4000, 6000, 8000],
        )
        clips = self._clips(1000.0, 1010.0)
        assert clips[0]["keyFrameDurations"] == [2000, 2000, 2000, 2000]
        assert "firstKeyFrameOffset" not in clips[0]

    def test_unindexed_row_keeps_whole_clip_placeholder(self):
        self._insert("m1", 1000.0, 1010.0, "main")
        clips = self._clips(1000.0, 1010.0)
        assert clips[0]["keyFrameDurations"] == [10000]


class TestVodAudioCutShort(CoverageDbTestCase):
    """A clip whose audio ends before its clipFrom has no audio track to nginx-vod."""

    KEYFRAMES = [0, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000]

    def _insert_pair(self, first_audio_duration):
        # the first row's audio dropped out; the range starts 5s into it
        for id, start, audio_duration in (
            ("m1", 1000.0, first_audio_duration),
            ("m2", 1010.0, 10.0),
        ):
            self._insert(
                id,
                start,
                start + 10.0,
                "main",
                has_audio=True,
                audio_codec="aac",
                audio_rate=16000,
                keyframes=self.KEYFRAMES,
                audio_duration=audio_duration,
            )

    def _clips(self, start, end):
        response = asyncio.run(_vod_response("front_door", start, end))
        return json.loads(response.body)["sequences"][0]["clips"]

    def test_clip_past_audio_end_is_skipped(self):
        self._insert_pair(2.112)
        clips = self._clips(1005.0, 1020.0)
        assert [c["path"] for c in clips] == ["/tmp/m2.mp4"]
        timeline = realized_timeline(
            resolve_coverage("front_door", 1005.0, 1020.0), None
        )
        assert [t["duration"] for t in timeline] == [0, 10000]

    def test_clip_with_audio_left_is_kept(self):
        # nginx-vod keeps the track while any sample lies past clipFrom
        self._insert_pair(5.001)
        clips = self._clips(1005.0, 1020.0)
        assert clips[0]["clipFrom"] == 5000
        assert len(clips) == 2

    def test_clip_starting_on_audio_end_is_skipped(self):
        self._insert_pair(5.0)
        clips = self._clips(1005.0, 1020.0)
        assert len(clips) == 1

    def test_legacy_row_is_kept(self):
        self._insert_pair(None)
        clips = self._clips(1005.0, 1020.0)
        assert len(clips) == 2

    def test_video_only_range_keeps_clip(self):
        # a video-only sub row serves the whole range without audio, so
        # nginx-vod never compares audio tracks
        self._insert(
            "m1",
            1000.0,
            1010.0,
            "main",
            has_audio=True,
            audio_codec="aac",
            audio_rate=16000,
            keyframes=self.KEYFRAMES,
            audio_duration=2.112,
        )
        self._insert(
            "s1", 1010.0, 1020.0, "sub", has_audio=False, keyframes=self.KEYFRAMES
        )
        clips = self._clips(1005.0, 1020.0)
        assert [c["path"] for c in clips] == ["/tmp/m1.mp4", "/tmp/s1.mp4"]
        assert all(c["tracks"] == "v" for c in clips)
        timeline = realized_timeline(
            resolve_coverage("front_door", 1005.0, 1020.0), None
        )
        assert timeline[0]["duration"] > 0

    def test_sole_clip_is_kept(self):
        # alone in its sequence the clip plays video-only, so keep it
        self._insert_pair(2.112)
        clips = self._clips(1005.0, 1009.0)
        assert clips[0]["path"] == "/tmp/m1.mp4"
        timeline = realized_timeline(
            resolve_coverage("front_door", 1005.0, 1009.0), None
        )
        assert timeline[0]["duration"] > 0


class TestParsePlaybackIndex(unittest.TestCase):
    def test_splits_video_keyframes_from_audio_duration(self):
        output = "\n".join(
            [
                "video,0,0.000000,0.040000,K__",
                "audio,1,0.000000,0.064000,K__",
                "video,0,0.040000,0.040000,___",
                "audio,1,0.064000,0.064000,K__",
                "video,0,1.000000,0.040000,K__",
                "audio,1,0.128000,N/A,K__",
            ]
        )
        keyframes, audio_duration = parse_playback_index(output)
        assert keyframes == [0, 1000]
        assert audio_duration == 0.128

    def test_no_audio_packets_reports_zero(self):
        keyframes, audio_duration = parse_playback_index(
            "video,0,0.000000,0.040000,K__"
        )
        assert keyframes == [0]
        assert audio_duration == 0.0

    def test_only_first_video_and_audio_streams_count(self):
        output = "\n".join(
            [
                "video,1,0.000000,0.100000,K__",
                "video,0,0.000000,0.040000,K__",
                "audio,3,0.000000,0.064000,K__",
                "audio,2,0.000000,0.064000,K__",
                "video,1,2.500000,0.100000,K__",
                "video,0,1.000000,0.040000,K__",
                "audio,3,0.064000,0.064000,K__",
            ]
        )
        keyframes, audio_duration = parse_playback_index(output)
        assert keyframes == [0, 1000]
        assert audio_duration == 0.064