Lab050: model fresh-frame SDR video stream

protocol/video_frame_scheduler.py

@@ -4,7 +4,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from enum import Enum
-from typing import Iterable
+import math
+from typing import Callable, Iterable
 
 from protocol.link_packet import TrafficClass, encode_link_packet
 from protocol.priority_scheduler import transmission_duration_seconds
@@ -35,6 +36,12 @@ class VideoFrameGroup:
     composite_frame_id: int
     generation_time_us: int
     packets: tuple[AgePolicyPacket, ...]
+    available_time_us: int | None = None
+
+    @property
+    def available_time_seconds(self) -> float:
+        """Готовность закодированного изображения отдельно от момента съёмки."""
+        return (self.generation_time_us if self.available_time_us is None else self.available_time_us) / 1_000_000.0
 
     @property
     def generation_time_seconds(self) -> float:
@@ -90,6 +97,7 @@ class FrameScheduleResult:
     started_frame_ids: tuple[int, ...]
     completed_frame_ids: tuple[int, ...]
     admissions: tuple[FrameAdmission, ...]
+    maximum_waiting_frames: int = 0
 
 
 def _replace_state(
@@ -195,13 +203,23 @@ def schedule_video_frames(
     high_priority_packets: Iterable[AgePolicyPacket],
     policy: FramePolicy,
     channel_bitrate_bps: float,
+    *,
+    max_video_age_seconds: float | None = None,
+    completion_latency_seconds: float | Callable[[VideoFrameGroup], float] = 0.0,
 ) -> FrameScheduleResult:
     """Run strict-priority service while keeping video frames contiguous."""
 
     policy = FramePolicy(policy)
     if channel_bitrate_bps <= 0.0:
         raise ValueError("channel bitrate must be positive")
-    frame_arrivals = tuple(sorted(frames, key=lambda frame: (frame.generation_time_seconds, frame.composite_frame_id)))
+    if max_video_age_seconds is not None and (not math.isfinite(max_video_age_seconds) or max_video_age_seconds <= 0):
+        raise ValueError("Возраст видео должен быть конечным положительным числом")
+    if not callable(completion_latency_seconds) and (not math.isfinite(completion_latency_seconds) or completion_latency_seconds < 0):
+        raise ValueError("Задержка после RF должна быть конечной и неотрицательной")
+    frame_arrivals = tuple(sorted(frames, key=lambda frame: (frame.available_time_seconds, frame.composite_frame_id)))
+    for frame in frame_arrivals:
+        if not frame.packets or frame.available_time_seconds < frame.generation_time_seconds:
+            raise ValueError("Нужны непустые изображения, готовые не раньше съёмки")
     high_arrivals = tuple(sorted(high_priority_packets, key=lambda item: (item.available_time_seconds, item.arrival_order)))
     ready_high: list[AgePolicyPacket] = []
     pending_frames: list[VideoFrameGroup] = []
@@ -214,15 +232,17 @@ def schedule_video_frames(
     actual_by_frame: dict[int, float] = {}
     active: VideoFrameGroup | None = None
     active_index = 0
+    maximum_waiting = 0
+    newest_capture_key: tuple[int, int] | None = None
     cursor = 0.0
     frame_index = high_index = 0
     blocker_by_order: dict[int, tuple[AgePolicyPacket, float]] = {}
 
-    def drop_frame(frame: VideoFrameGroup, reason: str) -> None:
-        dropped.append(FrameDrop(frame, cursor, reason))
+    def drop_frame(frame: VideoFrameGroup, reason: str, when: float | None = None) -> None:
+        dropped.append(FrameDrop(frame, cursor if when is None else when, reason))
 
     def admit(now: float, active_packet: AgePolicyPacket | None = None, active_end: float = 0.0) -> None:
-        nonlocal frame_index, high_index
+        nonlocal frame_index, high_index, maximum_waiting, newest_capture_key
         while high_index < len(high_arrivals) and high_arrivals[high_index].available_time_seconds <= now + TIME_EPSILON_SECONDS:
             item = high_arrivals[high_index]
             high_index += 1
@@ -232,12 +252,19 @@ def schedule_video_frames(
                     max(0.0, active_end - item.available_time_seconds),
                 )
             _replace_state(ready_high, item, replacements)
-        while frame_index < len(frame_arrivals) and frame_arrivals[frame_index].generation_time_seconds <= now + TIME_EPSILON_SECONDS:
+        while frame_index < len(frame_arrivals) and frame_arrivals[frame_index].available_time_seconds <= now + TIME_EPSILON_SECONDS:
             frame = frame_arrivals[frame_index]
             frame_index += 1
             if policy is FramePolicy.LATEST_ONLY:
+                capture_key = (frame.generation_time_us, frame.composite_frame_id)
+                # Позднее завершение кодирования не делает старый снимок новым.
+                # Помним свежесть и когда очередь уже опустела/новый кадр активен.
+                if newest_capture_key is not None and capture_key < newest_capture_key:
+                    drop_frame(frame, "older_capture_arrived_late", frame.available_time_seconds)
+                    continue
+                newest_capture_key = capture_key
                 for old in pending_frames:
-                    drop_frame(old, "replaced_by_newest")
+                    drop_frame(old, "replaced_by_newest", frame.available_time_seconds)
                 pending_frames[:] = [frame]
             elif policy is FramePolicy.TWO_WAITING:
                 pending_frames.append(frame)
@@ -245,6 +272,7 @@ def schedule_video_frames(
                     drop_frame(pending_frames.pop(0), "waiting_limit")
             else:
                 pending_frames.append(frame)
+            maximum_waiting = max(maximum_waiting, len(pending_frames))
 
     while (
         frame_index < len(frame_arrivals)
@@ -256,7 +284,7 @@ def schedule_video_frames(
         if not ready_high and not pending_frames and active is None:
             next_times = []
             if frame_index < len(frame_arrivals):
-                next_times.append(frame_arrivals[frame_index].generation_time_seconds)
+                next_times.append(frame_arrivals[frame_index].available_time_seconds)
             if high_index < len(high_arrivals):
                 next_times.append(high_arrivals[high_index].available_time_seconds)
             cursor = max(cursor, min(next_times))
@@ -273,7 +301,7 @@ def schedule_video_frames(
             if active is None:
                 while pending_frames and active is None:
                     candidate = pending_frames.pop(0)
-                    deadline = policy.deadline_seconds
+                    deadline = max_video_age_seconds if max_video_age_seconds is not None else policy.deadline_seconds
                     if deadline is not None:
                         future_high = high_arrivals[high_index:]
                         predicted = predict_frame_completion(
@@ -283,7 +311,11 @@ def schedule_video_frames(
                             tuple(ready_high),
                             future_high,
                         )
-                        if predicted - candidate.generation_time_seconds > deadline + TIME_EPSILON_SECONDS:
+                        post_latency = (completion_latency_seconds(candidate) if callable(completion_latency_seconds)
+                                        else completion_latency_seconds)
+                        if not math.isfinite(post_latency) or post_latency < 0:
+                            raise ValueError("Прогноз задержки должен быть конечным и неотрицательным")
+                        if predicted + post_latency - candidate.generation_time_seconds > deadline + TIME_EPSILON_SECONDS:
                             drop_frame(candidate, "prediction_reject")
                             continue
                         predicted_by_frame[candidate.composite_frame_id] = predicted
@@ -327,4 +359,5 @@ def schedule_video_frames(
         started_frame_ids=tuple(started),
         completed_frame_ids=tuple(completed),
         admissions=admissions,
+        maximum_waiting_frames=maximum_waiting,
     )