332 lines
15 KiB
Python
332 lines
15 KiB
Python
"""Ограниченный транспорт свежих H.264 кадров поверх уже проверенных пакетов.
|
||
|
||
Модуль получает только полезную нагрузку после строгого PHY/FEC/CRC. Он не
|
||
запускает радио и не заменяет существующие заголовки или исправляющий код.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections import deque
|
||
from dataclasses import dataclass, field
|
||
from enum import IntEnum
|
||
import math
|
||
import struct
|
||
|
||
|
||
VERSION = 1
|
||
HEADER_FORMAT = "!BIIHHHH"
|
||
HEADER_SIZE = struct.calcsize(HEADER_FORMAT) # 17 байт
|
||
MAX_FRAME_BYTES = 131_072
|
||
MAX_FRAGMENTS = 256
|
||
|
||
|
||
class VideoTransportError(ValueError):
|
||
"""Некорректный фрагмент или состояние видеотранспорта."""
|
||
|
||
|
||
class FrameType(IntEnum):
|
||
KEYFRAME = 1
|
||
DELTA_FRAME = 2
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class VideoFrame:
|
||
stream_id: int
|
||
frame_id: int
|
||
frame_type: FrameType
|
||
capture_ms: int
|
||
data: bytes
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class VideoFragment:
|
||
stream_id: int
|
||
frame_id: int
|
||
fragment_index: int
|
||
fragment_count: int
|
||
frame_type: FrameType
|
||
age_ms: int
|
||
data: bytes
|
||
|
||
|
||
def _validate_uint(value: int, maximum: int, name: str) -> None:
|
||
if not isinstance(value, int) or not 0 <= value <= maximum:
|
||
raise VideoTransportError(f"{name} вне допустимого диапазона")
|
||
|
||
|
||
def is_newer(frame_id: int, reference: int) -> bool:
|
||
"""Сравнить 32-битные номера с защитой от старых пакетов при переходе через ноль."""
|
||
difference = (frame_id - reference) & 0xFFFFFFFF
|
||
return 0 < difference < 0x80000000
|
||
|
||
|
||
def encode_fragment(fragment: VideoFragment, max_phy_payload: int) -> bytes:
|
||
"""Упаковать фрагмент в полезную нагрузку проверенного пакета."""
|
||
_validate_uint(fragment.stream_id, 0xFFFFFFFF, "stream_id")
|
||
_validate_uint(fragment.frame_id, 0xFFFFFFFF, "frame_id")
|
||
_validate_uint(fragment.fragment_count, MAX_FRAGMENTS, "fragment_count")
|
||
if fragment.fragment_count == 0:
|
||
raise VideoTransportError("fragment_count равен нулю")
|
||
_validate_uint(fragment.fragment_index, fragment.fragment_count - 1, "fragment_index")
|
||
_validate_uint(fragment.age_ms, 0xFFFF, "age_ms")
|
||
if fragment.frame_type not in (FrameType.KEYFRAME, FrameType.DELTA_FRAME):
|
||
raise VideoTransportError("неизвестный тип кадра")
|
||
if not isinstance(fragment.data, bytes) or not fragment.data:
|
||
raise VideoTransportError("пустые или некорректные данные")
|
||
if len(fragment.data) > min(0xFFFF, max_phy_payload - HEADER_SIZE):
|
||
raise VideoTransportError("фрагмент не помещается в пакет")
|
||
flags = (VERSION << 4) | int(fragment.frame_type)
|
||
header = struct.pack(HEADER_FORMAT, flags, fragment.stream_id,
|
||
fragment.frame_id, fragment.fragment_index,
|
||
fragment.fragment_count, fragment.age_ms,
|
||
len(fragment.data))
|
||
return header + fragment.data
|
||
|
||
|
||
def decode_fragment(wire: bytes, max_phy_payload: int) -> VideoFragment:
|
||
"""Проверить длины и поля после внешнего CRC, не восстанавливая данные."""
|
||
if not isinstance(wire, bytes) or not HEADER_SIZE < len(wire) <= max_phy_payload:
|
||
raise VideoTransportError("длина видеопакета вне допустимого диапазона")
|
||
flags, stream_id, frame_id, index, count, age_ms, length = struct.unpack(
|
||
HEADER_FORMAT, wire[:HEADER_SIZE])
|
||
if flags >> 4 != VERSION or (flags & 0x0F) not in (1, 2):
|
||
raise VideoTransportError("версия или тип кадра не поддерживается")
|
||
if not 1 <= count <= MAX_FRAGMENTS or index >= count:
|
||
raise VideoTransportError("неверная нумерация фрагментов")
|
||
if length == 0 or length != len(wire) - HEADER_SIZE:
|
||
raise VideoTransportError("payload_length не совпадает с размером пакета")
|
||
return VideoFragment(stream_id, frame_id, index, count,
|
||
FrameType(flags & 0x0F), age_ms, wire[HEADER_SIZE:])
|
||
|
||
|
||
def split_frame(frame: VideoFrame, max_phy_payload: int,
|
||
age_ms: int = 0) -> tuple[VideoFragment, ...]:
|
||
"""Разбить одно непустое изображение H.264 без потери байтов."""
|
||
if not isinstance(frame.data, bytes) or not 1 <= len(frame.data) <= MAX_FRAME_BYTES:
|
||
raise VideoTransportError("размер кадра вне допустимого диапазона")
|
||
capacity = max_phy_payload - HEADER_SIZE
|
||
if capacity <= 0:
|
||
raise VideoTransportError("пакет меньше заголовка")
|
||
count = math.ceil(len(frame.data) / capacity)
|
||
if count > MAX_FRAGMENTS:
|
||
raise VideoTransportError("слишком много фрагментов")
|
||
return tuple(VideoFragment(frame.stream_id, frame.frame_id, index, count,
|
||
frame.frame_type, age_ms,
|
||
frame.data[index * capacity:(index + 1) * capacity])
|
||
for index in range(count))
|
||
|
||
|
||
@dataclass
|
||
class _QueuedFrame:
|
||
frame: VideoFrame
|
||
parts: tuple[VideoFragment, ...]
|
||
next_index: int = 0
|
||
|
||
|
||
class BoundedFrameQueue:
|
||
"""Очередь ограничивает число кадров и срок каждого фрагмента."""
|
||
|
||
def __init__(self, max_phy_payload: int, max_frames: int = 2,
|
||
max_age_ms: int = 300) -> None:
|
||
if max_frames < 1 or not 1 <= max_age_ms <= 0xFFFF:
|
||
raise ValueError("недопустимые ограничения очереди")
|
||
self.max_phy_payload = max_phy_payload
|
||
self.max_frames = max_frames
|
||
self.max_age_ms = max_age_ms
|
||
self._frames: deque[_QueuedFrame] = deque()
|
||
self.max_depth = 0
|
||
self.max_oldest_age_ms = 0
|
||
self.stale_frame_drops = 0
|
||
self.capacity_frame_drops = 0
|
||
self.sent_frames = 0
|
||
self._last_capture_ms = -1
|
||
|
||
@property
|
||
def depth(self) -> int:
|
||
return len(self._frames)
|
||
|
||
def _observe(self, now_ms: int) -> None:
|
||
self.max_depth = max(self.max_depth, self.depth)
|
||
if self._frames:
|
||
self.max_oldest_age_ms = max(
|
||
self.max_oldest_age_ms, now_ms - self._frames[0].frame.capture_ms)
|
||
|
||
def enqueue(self, frame: VideoFrame, now_ms: int) -> None:
|
||
if frame.capture_ms <= self._last_capture_ms or frame.capture_ms > now_ms:
|
||
raise VideoTransportError("время создания кадров должно возрастать")
|
||
self._last_capture_ms = frame.capture_ms
|
||
parts = split_frame(frame, self.max_phy_payload)
|
||
self.expire(now_ms)
|
||
if now_ms - frame.capture_ms >= self.max_age_ms:
|
||
self.stale_frame_drops += 1
|
||
return
|
||
if len(self._frames) >= self.max_frames:
|
||
self._frames.popleft()
|
||
self.capacity_frame_drops += 1
|
||
self._frames.append(_QueuedFrame(frame, parts))
|
||
self._observe(now_ms)
|
||
|
||
def expire(self, now_ms: int, packet_end_ms: float | None = None) -> None:
|
||
retained: deque[_QueuedFrame] = deque()
|
||
for item in self._frames:
|
||
deadline = item.frame.capture_ms + self.max_age_ms
|
||
if (now_ms >= deadline or (packet_end_ms is not None
|
||
and math.ceil(packet_end_ms) >= deadline)):
|
||
self.stale_frame_drops += 1
|
||
else:
|
||
retained.append(item)
|
||
self._frames = retained
|
||
self._observe(now_ms)
|
||
|
||
def next_fragment(self, now_ms: int, packet_end_ms: float) -> VideoFragment | None:
|
||
self.expire(now_ms, packet_end_ms)
|
||
if not self._frames:
|
||
return None
|
||
item = self._frames[0]
|
||
part = item.parts[item.next_index]
|
||
age_ms = math.ceil(packet_end_ms - item.frame.capture_ms)
|
||
if age_ms > 0xFFFF:
|
||
self._frames.popleft()
|
||
self.stale_frame_drops += 1
|
||
return None
|
||
self._observe(now_ms)
|
||
return VideoFragment(part.stream_id, part.frame_id, part.fragment_index,
|
||
part.fragment_count, part.frame_type, age_ms, part.data)
|
||
|
||
def mark_sent(self) -> None:
|
||
if not self._frames:
|
||
raise VideoTransportError("нет активного кадра")
|
||
self._frames[0].next_index += 1
|
||
if self._frames[0].next_index == len(self._frames[0].parts):
|
||
self._frames.popleft()
|
||
self.sent_frames += 1
|
||
|
||
|
||
@dataclass
|
||
class _Assembly:
|
||
frame_type: FrameType
|
||
count: int
|
||
deadline_ms: int
|
||
first_arrival_ms: int
|
||
parts: dict[int, bytes] = field(default_factory=dict)
|
||
size: int = 0
|
||
|
||
|
||
class FrameReassembler:
|
||
"""Собирать только свежие кадры выбранного сеанса, без ожидания старых."""
|
||
|
||
def __init__(self, stream_id: int, max_phy_payload: int,
|
||
max_pending_frames: int = 2, max_age_ms: int = 300) -> None:
|
||
_validate_uint(stream_id, 0xFFFFFFFF, "stream_id")
|
||
if max_pending_frames < 1 or not 1 <= max_age_ms <= 0xFFFF:
|
||
raise ValueError("недопустимые ограничения сборщика")
|
||
self.stream_id = stream_id
|
||
self.max_phy_payload = max_phy_payload
|
||
self.max_pending_frames = max_pending_frames
|
||
self.max_age_ms = max_age_ms
|
||
self._pending: dict[int, _Assembly] = {}
|
||
self._closed: deque[int] = deque(maxlen=1024)
|
||
self.last_completed_id: int | None = None
|
||
self.latest_admitted_id: int | None = None
|
||
self.max_depth = 0
|
||
self.expired_frames = 0
|
||
self.capacity_drops = 0
|
||
self.rejected_fragments = 0
|
||
|
||
@property
|
||
def depth(self) -> int:
|
||
return len(self._pending)
|
||
|
||
def expire(self, now_ms: int) -> tuple[int, ...]:
|
||
expired = tuple(frame_id for frame_id, item in self._pending.items()
|
||
if now_ms >= item.deadline_ms)
|
||
for frame_id in expired:
|
||
del self._pending[frame_id]
|
||
self._closed.append(frame_id)
|
||
self.expired_frames += 1
|
||
return expired
|
||
|
||
def offer(self, wire: bytes, now_ms: int) -> VideoFrame | None:
|
||
self.expire(now_ms)
|
||
fragment = decode_fragment(wire, self.max_phy_payload)
|
||
return self._offer_fragment(fragment, now_ms)
|
||
|
||
def offer_fragment(self, fragment: VideoFragment, now_ms: int) -> VideoFrame | None:
|
||
"""Принять уже проверенный фрагмент нового видеоформата без сериализации."""
|
||
self.expire(now_ms)
|
||
if (not isinstance(fragment, VideoFragment)
|
||
or not isinstance(fragment.data, bytes) or not fragment.data
|
||
or len(fragment.data) > self.max_phy_payload - HEADER_SIZE
|
||
or not 1 <= fragment.fragment_count <= MAX_FRAGMENTS
|
||
or not 0 <= fragment.fragment_index < fragment.fragment_count
|
||
or fragment.frame_type not in (FrameType.KEYFRAME, FrameType.DELTA_FRAME)
|
||
or not 0 <= fragment.age_ms <= 0xFFFF
|
||
or not 0 <= fragment.stream_id <= 0xFFFFFFFF
|
||
or not 0 <= fragment.frame_id <= 0xFFFFFFFF):
|
||
raise VideoTransportError("некорректный проверенный видеофрагмент")
|
||
return self._offer_fragment(fragment, now_ms)
|
||
|
||
def _offer_fragment(self, fragment: VideoFragment, now_ms: int) -> VideoFrame | None:
|
||
if (fragment.stream_id != self.stream_id or fragment.frame_id in self._closed
|
||
or (self.last_completed_id is not None
|
||
and not is_newer(fragment.frame_id, self.last_completed_id))):
|
||
self.rejected_fragments += 1
|
||
return None
|
||
if fragment.age_ms >= self.max_age_ms:
|
||
self.rejected_fragments += 1
|
||
return None
|
||
deadline = now_ms + self.max_age_ms - fragment.age_ms
|
||
item = self._pending.get(fragment.frame_id)
|
||
if item is None:
|
||
if (self.latest_admitted_id is not None
|
||
and not is_newer(fragment.frame_id, self.latest_admitted_id)
|
||
and not 0 < ((self.latest_admitted_id - fragment.frame_id) & 0xFFFFFFFF)
|
||
< self.max_pending_frames):
|
||
self.rejected_fragments += 1
|
||
return None
|
||
if len(self._pending) >= self.max_pending_frames:
|
||
oldest_id = min(self._pending,
|
||
key=lambda frame_id: self._pending[frame_id].first_arrival_ms)
|
||
if not is_newer(fragment.frame_id, oldest_id):
|
||
self.rejected_fragments += 1
|
||
return None
|
||
del self._pending[oldest_id]
|
||
self._closed.append(oldest_id)
|
||
self.capacity_drops += 1
|
||
item = _Assembly(fragment.frame_type, fragment.fragment_count,
|
||
deadline, now_ms)
|
||
self._pending[fragment.frame_id] = item
|
||
if (self.latest_admitted_id is None
|
||
or is_newer(fragment.frame_id, self.latest_admitted_id)):
|
||
self.latest_admitted_id = fragment.frame_id
|
||
self.max_depth = max(self.max_depth, len(self._pending))
|
||
elif item.frame_type != fragment.frame_type or item.count != fragment.fragment_count:
|
||
del self._pending[fragment.frame_id]
|
||
self._closed.append(fragment.frame_id)
|
||
self.rejected_fragments += 1
|
||
raise VideoTransportError("противоречивые метаданные кадра")
|
||
item.deadline_ms = min(item.deadline_ms, deadline)
|
||
previous = item.parts.get(fragment.fragment_index)
|
||
if previous is not None:
|
||
if previous != fragment.data:
|
||
del self._pending[fragment.frame_id]
|
||
self._closed.append(fragment.frame_id)
|
||
self.rejected_fragments += 1
|
||
raise VideoTransportError("противоречивые данные фрагмента")
|
||
return None
|
||
item.parts[fragment.fragment_index] = fragment.data
|
||
item.size += len(fragment.data)
|
||
if item.size > MAX_FRAME_BYTES:
|
||
del self._pending[fragment.frame_id]
|
||
self._closed.append(fragment.frame_id)
|
||
self.rejected_fragments += 1
|
||
raise VideoTransportError("кадр превышает ограничение памяти")
|
||
if len(item.parts) != item.count:
|
||
return None
|
||
data = b"".join(item.parts[index] for index in range(item.count))
|
||
del self._pending[fragment.frame_id]
|
||
self._closed.append(fragment.frame_id)
|
||
self.last_completed_id = fragment.frame_id
|
||
return VideoFrame(fragment.stream_id, fragment.frame_id,
|
||
fragment.frame_type, now_ms - fragment.age_ms, data)
|