diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index 513ae58..cfa2fbb 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -32,4 +32,25 @@ 1. Проверена установленная версия Git: ```text -git version 2.51.1.windows.1 \ No newline at end of file +git version 2.51.1.windows.1 +``` + +--- + +# Запись 002 + +## Дата + +27 июля 2026 года + +## Тема + +Завершение Lab028: пакетирование синхронного BASE + ROI. + +## Выполнено + +- Реализован фиксированный бинарный заголовок размером 32 байта. +- Добавлены packet CRC32 для заголовка и payload и object CRC32 для полного JPEG. +- Исследованы размеры payload 64, 128, 256, 512 и 1024 байта. +- Размер payload 512 байт принят как рабочий кандидат. +- Окончательный выбор размера пакета перенесён в Lab029. diff --git a/data/processed/lab028/lab028_overhead_efficiency.png b/data/processed/lab028/lab028_overhead_efficiency.png new file mode 100644 index 0000000..ee131df Binary files /dev/null and b/data/processed/lab028/lab028_overhead_efficiency.png differ diff --git a/data/processed/lab028/lab028_packet_payload_results.csv b/data/processed/lab028/lab028_packet_payload_results.csv new file mode 100644 index 0000000..c93cca5 --- /dev/null +++ b/data/processed/lab028/lab028_packet_payload_results.csv @@ -0,0 +1,6 @@ +max_payload_bytes,composite_frames,mean_base_packets_per_frame,mean_roi_packets_per_frame,mean_total_packets_per_frame,max_total_packets_per_frame,packets_per_second,jpeg_payload_bytes,jpeg_payload_bitrate_kbps,header_bytes_per_second,header_bitrate_kbps,wire_bytes,wire_bitrate_kbps,service_data_percent,efficiency_percent,mean_wire_packet_bytes,p95_wire_packet_bytes,max_wire_packet_bytes +64,63,36.428571,66.174603,102.603175,117,311.268058,409544,157.769759,9960.577849,79.684623,616392,237.454382,33.557866,66.442134,95.357673,96.000000,96 +128,63,18.428571,33.269841,51.698413,59,156.837881,409544,157.769759,5018.812199,40.150498,513768,197.920257,20.286199,79.713801,157.742708,160.000000,160 +256,63,9.428571,16.920635,26.349206,30,79.935795,409544,157.769759,2557.945425,20.463563,462664,178.233323,11.481334,88.518666,278.713253,288.000000,288 +512,63,5.000000,8.682540,13.682540,15,41.508828,409544,157.769759,1328.282504,10.626260,437128,168.396019,6.310280,93.689720,507.109049,544.000000,544 +1024,63,3.000000,4.587302,7.587302,8,23.017657,409544,157.769759,736.565008,5.892520,424840,163.662279,3.600414,96.399586,888.786611,1056.000000,1056 diff --git a/data/processed/lab028/lab028_packets_wire_bitrate.png b/data/processed/lab028/lab028_packets_wire_bitrate.png new file mode 100644 index 0000000..d5c19ea Binary files /dev/null and b/data/processed/lab028/lab028_packets_wire_bitrate.png differ diff --git a/data/processed/lab028/lab028_report.txt b/data/processed/lab028/lab028_report.txt new file mode 100644 index 0000000..95609ea --- /dev/null +++ b/data/processed/lab028/lab028_report.txt @@ -0,0 +1,52 @@ +Lab028. Пакетирование синхронного BASE + ROI + +Профиль и исходные данные +- Видео: data\raw\lab026_rover_source.mp4 +- Источник: 1280x720, 30.000000 fps, 623 кадров, 20.766667 с. +- ROI: x=0.20...0.80, y=0.42...1.00; пиксели x=256...1024, y=302...720. +- Синхронные составные кадры: 63 при 3.000 fps. +- BASE: 240x135, grayscale JPEG Q23; средний размер 2296.016 B, min/max 2122/2476 B. +- ROI: 320x180, grayscale JPEG Q33; средний размер 4204.683 B, min/max 3457/4938 B. +- BASE и ROI формируются из одного source frame и имеют общий composite_frame_id. + +Бинарный заголовок +- struct format: !4sBBHIHHHHIII +- Размер: 32 байта; network byte order; padding нет. +- Layout: magic[4] @0, version:u8 @4, object_type:u8 @5, header_size:u16 @6, composite_frame_id:u32 @8, fragment_index:u16 @12, fragment_count:u16 @14, payload_length:u16 @16, flags:u16 @18, jpeg_size:u32 @20, object_crc32:u32 @24, packet_crc32:u32 @28. +- object_type: 1=BASE, 2=ROI; flags зарезервирован и равен 0. + +CRC32 +- object_crc32 = zlib.crc32(полный JPEG) & 0xFFFFFFFF; значение помещается во все фрагменты объекта и проверяется после полной сборки. +- packet_crc32: сначала поле packet_crc32 заголовка обнуляется, затем CRC считается как zlib.crc32(header_with_zero_crc + payload) & 0xFFFFFFFF. +- Таким образом packet CRC защищает все поля заголовка и payload. + +Функциональные проверки +- PASS header_serialization_round_trip: fixed 32-byte header round trip is exact +- PASS ordered_lossless_transfer: BASE and ROI match original JPEG bytes +- PASS random_packet_order: arbitrary packet order reconstructed correctly +- PASS exact_duplicate_packets: 2 exact duplicates ignored +- PASS packet_crc_corruption: payload bit flip rejected; composite not emitted +- PASS missing_fragment: ROI fragment 10 reported missing +- PASS object_crc_corruption: valid packet CRCs still failed full-object CRC +- PASS adjacent_frame_isolation: two interleaved frame IDs remained independent +- PASS base_without_roi_atomicity: complete BASE retained while composite stayed unpublished + +Результаты размеров packet payload +payload | BASE pkt/frame | ROI pkt/frame | all pkt/frame | max pkt/frame | pkt/s | JPEG kbit/s | headers B/s | wire kbit/s | service % | efficiency % | packet mean/p95/max +-------:|---------------:|--------------:|--------------:|--------------:|------:|------------:|------------:|------------:|----------:|-------------:|--------------------: +64 | 36.429 | 66.175 | 102.603 | 117 | 311.268 | 157.770 | 9960.578 | 237.454 | 33.558 | 66.442 | 95.4/96.0/96 +128 | 18.429 | 33.270 | 51.698 | 59 | 156.838 | 157.770 | 5018.812 | 197.920 | 20.286 | 79.714 | 157.7/160.0/160 +256 | 9.429 | 16.921 | 26.349 | 30 | 79.936 | 157.770 | 2557.945 | 178.233 | 11.481 | 88.519 | 278.7/288.0/288 +512 | 5.000 | 8.683 | 13.683 | 15 | 41.509 | 157.770 | 1328.283 | 168.396 | 6.310 | 93.690 | 507.1/544.0/544 +1024 | 3.000 | 4.587 | 7.587 | 8 | 23.018 | 157.770 | 736.565 | 163.662 | 3.600 | 96.400 | 888.8/1056.0/1056 + +JPEG payload bitrate одинаков для всех строк, потому что исходные JPEG не меняются при выборе размера фрагмента. +Wire bitrate включает только JPEG payload и 32-байтные заголовки каждого пакета. +Wire bitrate пока НЕ включает FEC, преамбулу, синхронизацию, модуляцию, интервалы, повторные передачи, команды управления и телеметрию. +Лучший размер packet payload автоматически не выбирается: таблица показывает только транспортный компромисс. + +Артефакты +- CSV: data\processed\lab028\lab028_packet_payload_results.csv +- Overhead/efficiency: data\processed\lab028\lab028_overhead_efficiency.png +- Packet count/wire bitrate: data\processed\lab028\lab028_packets_wire_bitrate.png +- Промежуточные JPEG и пакеты сохранялись только в памяти; бинарные дампы не создавались. diff --git a/protocol/video_packet.py b/protocol/video_packet.py new file mode 100644 index 0000000..cf4399f --- /dev/null +++ b/protocol/video_packet.py @@ -0,0 +1,416 @@ +""" +Transport packets for synchronous BASE + ROI JPEG composite frames. + +The wire header has a fixed 32-byte, padding-free network-byte-order layout:: + + !4sBBHIHHHHIII + + Offset Size Field + 0 4 magic (b"SRV1") + 4 1 version + 5 1 object_type (1=BASE, 2=ROI) + 6 2 header_size + 8 4 composite_frame_id + 12 2 fragment_index + 14 2 fragment_count + 16 2 payload_length + 18 2 flags (reserved, must be zero) + 20 4 jpeg_size + 24 4 object_crc32 + 28 4 packet_crc32 + +packet_crc32 is calculated over the complete header with packet_crc32 set to +zero, followed by the packet payload. object_crc32 is calculated over the +complete JPEG before fragmentation and checked after reassembly. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import IntEnum +import struct +import zlib + + +MAGIC = b"SRV1" +VERSION = 1 +HEADER_FORMAT = "!4sBBHIHHHHIII" +HEADER_SIZE = struct.calcsize(HEADER_FORMAT) +MAX_PAYLOAD_LENGTH = 0xFFFF +MAX_FRAGMENT_COUNT = 0xFFFF +MAX_JPEG_SIZE = 0xFFFFFFFF + + +class ObjectType(IntEnum): + """JPEG object carried by a packet.""" + + BASE = 1 + ROI = 2 + + +class VideoPacketError(ValueError): + """Base class for video transport validation errors.""" + + +class PacketCRCError(VideoPacketError): + """The packet header or payload failed its CRC32 check.""" + + +class ObjectCRCError(VideoPacketError): + """A completely reassembled JPEG failed its CRC32 check.""" + + +class ObjectConsistencyError(VideoPacketError): + """Fragments for one object contain inconsistent metadata or data.""" + + +@dataclass(frozen=True) +class VideoPacket: + """Decoded and CRC-checked transport packet.""" + + composite_frame_id: int + object_type: ObjectType + fragment_index: int + fragment_count: int + jpeg_size: int + object_crc32: int + payload: bytes + packet_crc32: int = 0 + + +@dataclass(frozen=True) +class CompositeFrame: + """Atomically reassembled BASE and ROI JPEGs for one source update.""" + + composite_frame_id: int + base_jpeg: bytes + roi_jpeg: bytes + + +def crc32(data: bytes) -> int: + """Return an unsigned IEEE CRC32.""" + + return zlib.crc32(data) & 0xFFFFFFFF + + +def _validate_packet_fields(packet: VideoPacket) -> None: + if not 0 <= packet.composite_frame_id <= 0xFFFFFFFF: + raise VideoPacketError("composite_frame_id is outside uint32") + if packet.object_type not in (ObjectType.BASE, ObjectType.ROI): + raise VideoPacketError("unsupported object_type") + if not 1 <= packet.fragment_count <= MAX_FRAGMENT_COUNT: + raise VideoPacketError("fragment_count is outside uint16") + if not 0 <= packet.fragment_index < packet.fragment_count: + raise VideoPacketError("fragment_index is outside fragment_count") + if not 1 <= len(packet.payload) <= MAX_PAYLOAD_LENGTH: + raise VideoPacketError("payload length is outside uint16") + if not 1 <= packet.jpeg_size <= MAX_JPEG_SIZE: + raise VideoPacketError("jpeg_size is outside uint32") + if len(packet.payload) > packet.jpeg_size: + raise VideoPacketError("payload is larger than the JPEG object") + if not 0 <= packet.object_crc32 <= 0xFFFFFFFF: + raise VideoPacketError("object_crc32 is outside uint32") + + +def _pack_header(packet: VideoPacket, packet_crc32: int) -> bytes: + return struct.pack( + HEADER_FORMAT, + MAGIC, + VERSION, + int(packet.object_type), + HEADER_SIZE, + packet.composite_frame_id, + packet.fragment_index, + packet.fragment_count, + len(packet.payload), + 0, + packet.jpeg_size, + packet.object_crc32, + packet_crc32, + ) + + +def encode_packet(packet: VideoPacket) -> bytes: + """Serialize a packet and calculate its packet CRC32.""" + + if not isinstance(packet, VideoPacket): + raise TypeError("packet must be VideoPacket") + _validate_packet_fields(packet) + header_with_zero_crc = _pack_header(packet, 0) + packet_crc32 = crc32(header_with_zero_crc + packet.payload) + return _pack_header(packet, packet_crc32) + packet.payload + + +def decode_packet(wire_packet: bytes) -> VideoPacket: + """Deserialize a packet and validate its fixed header and packet CRC32.""" + + if not isinstance(wire_packet, (bytes, bytearray)): + raise TypeError("wire_packet must be bytes or bytearray") + wire_packet = bytes(wire_packet) + if len(wire_packet) < HEADER_SIZE + 1: + raise VideoPacketError("packet is shorter than header plus payload") + + ( + magic, + version, + object_type_value, + header_size, + composite_frame_id, + fragment_index, + fragment_count, + payload_length, + flags, + jpeg_size, + object_crc32, + received_packet_crc32, + ) = struct.unpack(HEADER_FORMAT, wire_packet[:HEADER_SIZE]) + + if magic != MAGIC: + raise VideoPacketError("invalid packet magic") + if version != VERSION: + raise VideoPacketError("unsupported packet version") + if header_size != HEADER_SIZE: + raise VideoPacketError("invalid fixed header size") + if flags != 0: + raise VideoPacketError("reserved flags must be zero") + try: + object_type = ObjectType(object_type_value) + except ValueError as error: + raise VideoPacketError("unsupported object_type") from error + + if len(wire_packet) != HEADER_SIZE + payload_length: + raise VideoPacketError("packet length does not match payload_length") + + packet = VideoPacket( + composite_frame_id=composite_frame_id, + object_type=object_type, + fragment_index=fragment_index, + fragment_count=fragment_count, + jpeg_size=jpeg_size, + object_crc32=object_crc32, + payload=wire_packet[HEADER_SIZE:], + packet_crc32=received_packet_crc32, + ) + _validate_packet_fields(packet) + calculated_packet_crc32 = crc32( + _pack_header(packet, 0) + packet.payload + ) + if calculated_packet_crc32 != received_packet_crc32: + raise PacketCRCError( + "packet CRC mismatch: " + f"received 0x{received_packet_crc32:08X}, " + f"calculated 0x{calculated_packet_crc32:08X}" + ) + return packet + + +def packetize_jpeg( + jpeg: bytes, + composite_frame_id: int, + object_type: ObjectType, + max_payload_length: int, +) -> list[bytes]: + """Split one non-empty JPEG object into serialized packets.""" + + if not isinstance(jpeg, (bytes, bytearray)): + raise TypeError("jpeg must be bytes or bytearray") + jpeg = bytes(jpeg) + if not jpeg: + raise ValueError("JPEG object must not be empty") + if len(jpeg) > MAX_JPEG_SIZE: + raise ValueError("JPEG object is larger than uint32") + if not 1 <= max_payload_length <= MAX_PAYLOAD_LENGTH: + raise ValueError("max_payload_length is outside uint16") + if not 0 <= composite_frame_id <= 0xFFFFFFFF: + raise ValueError("composite_frame_id is outside uint32") + try: + object_type = ObjectType(object_type) + except ValueError as error: + raise ValueError("unsupported object_type") from error + + fragment_count = ( + len(jpeg) + max_payload_length - 1 + ) // max_payload_length + if fragment_count > MAX_FRAGMENT_COUNT: + raise ValueError("JPEG needs more than 65535 fragments") + + object_crc32 = crc32(jpeg) + packets = [] + for fragment_index in range(fragment_count): + start = fragment_index * max_payload_length + payload = jpeg[start:start + max_payload_length] + packets.append( + encode_packet( + VideoPacket( + composite_frame_id=composite_frame_id, + object_type=object_type, + fragment_index=fragment_index, + fragment_count=fragment_count, + jpeg_size=len(jpeg), + object_crc32=object_crc32, + payload=payload, + ) + ) + ) + return packets + + +@dataclass +class _ObjectAssembly: + composite_frame_id: int + object_type: ObjectType + fragment_count: int + jpeg_size: int + object_crc32: int + fragments: dict[int, bytes] = field(default_factory=dict) + + @classmethod + def from_packet(cls, packet: VideoPacket) -> "_ObjectAssembly": + return cls( + composite_frame_id=packet.composite_frame_id, + object_type=packet.object_type, + fragment_count=packet.fragment_count, + jpeg_size=packet.jpeg_size, + object_crc32=packet.object_crc32, + ) + + def add(self, packet: VideoPacket) -> bool: + metadata = ( + packet.composite_frame_id, + packet.object_type, + packet.fragment_count, + packet.jpeg_size, + packet.object_crc32, + ) + expected = ( + self.composite_frame_id, + self.object_type, + self.fragment_count, + self.jpeg_size, + self.object_crc32, + ) + if metadata != expected: + raise ObjectConsistencyError( + "fragment metadata conflicts with object assembly" + ) + + existing = self.fragments.get(packet.fragment_index) + if existing is not None: + if existing != packet.payload: + raise ObjectConsistencyError( + "different payload for an existing fragment index" + ) + return False + self.fragments[packet.fragment_index] = packet.payload + return True + + def missing_fragments(self) -> tuple[int, ...]: + return tuple( + index + for index in range(self.fragment_count) + if index not in self.fragments + ) + + def assemble_if_complete(self) -> bytes | None: + if self.missing_fragments(): + return None + jpeg = b"".join( + self.fragments[index] + for index in range(self.fragment_count) + ) + if len(jpeg) != self.jpeg_size: + raise ObjectConsistencyError( + "reassembled JPEG length does not match jpeg_size" + ) + calculated_object_crc32 = crc32(jpeg) + if calculated_object_crc32 != self.object_crc32: + raise ObjectCRCError( + "object CRC mismatch: " + f"received 0x{self.object_crc32:08X}, " + f"calculated 0x{calculated_object_crc32:08X}" + ) + return jpeg + + +class CompositeReassembler: + """Reassemble interleaved BASE/ROI packets and publish atomic frames.""" + + def __init__(self) -> None: + self._objects: dict[ + tuple[int, ObjectType], _ObjectAssembly + ] = {} + self._completed: dict[ + int, dict[ObjectType, bytes] + ] = {} + self._finalized_frame_ids: set[int] = set() + self.duplicate_packets = 0 + + def ingest(self, wire_packet: bytes) -> CompositeFrame | None: + """Accept one packet and return a frame only when BASE and ROI exist.""" + + packet = decode_packet(wire_packet) + if packet.composite_frame_id in self._finalized_frame_ids: + self.duplicate_packets += 1 + return None + + key = (packet.composite_frame_id, packet.object_type) + assembly = self._objects.get(key) + if assembly is None: + assembly = _ObjectAssembly.from_packet(packet) + self._objects[key] = assembly + + if not assembly.add(packet): + self.duplicate_packets += 1 + return None + + jpeg = assembly.assemble_if_complete() + if jpeg is None: + return None + + frame_parts = self._completed.setdefault( + packet.composite_frame_id, {} + ) + frame_parts[packet.object_type] = jpeg + if not all( + object_type in frame_parts + for object_type in (ObjectType.BASE, ObjectType.ROI) + ): + return None + + frame = CompositeFrame( + composite_frame_id=packet.composite_frame_id, + base_jpeg=frame_parts[ObjectType.BASE], + roi_jpeg=frame_parts[ObjectType.ROI], + ) + self._finalized_frame_ids.add(packet.composite_frame_id) + self._completed.pop(packet.composite_frame_id, None) + for object_type in (ObjectType.BASE, ObjectType.ROI): + self._objects.pop( + (packet.composite_frame_id, object_type), None + ) + return frame + + def missing_fragments( + self, + composite_frame_id: int, + object_type: ObjectType, + ) -> tuple[int, ...] | None: + """Return missing indexes, or None if this object has not started.""" + + assembly = self._objects.get( + (composite_frame_id, ObjectType(object_type)) + ) + if assembly is None: + return None + return assembly.missing_fragments() + + def object_is_complete( + self, + composite_frame_id: int, + object_type: ObjectType, + ) -> bool: + """Report whether one CRC-validated object awaits its counterpart.""" + + return ObjectType(object_type) in self._completed.get( + composite_frame_id, {} + ) diff --git a/tests/lab028_video_packetization.py b/tests/lab028_video_packetization.py new file mode 100644 index 0000000..3119b12 --- /dev/null +++ b/tests/lab028_video_packetization.py @@ -0,0 +1,869 @@ +""" +Lab028. Packetization of synchronous BASE + ROI JPEG video objects. + +The laboratory forms the selected Lab027E profile directly from the source +video, keeps every JPEG and packet in memory, verifies the transport layer, +and writes only aggregate CSV/report/plot artifacts. +""" + +from __future__ import annotations + +import csv +from dataclasses import dataclass +from pathlib import Path +import random +from typing import Callable + +import cv2 +import matplotlib +import numpy as np + +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +from protocol.video_packet import ( + CompositeFrame, + CompositeReassembler, + HEADER_FORMAT, + HEADER_SIZE, + ObjectCRCError, + ObjectType, + PacketCRCError, + VideoPacket, + decode_packet, + encode_packet, + packetize_jpeg, +) + + +SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4") +OUTPUT_DIRECTORY = Path("data/processed/lab028") +CSV_PATH = OUTPUT_DIRECTORY / "lab028_packet_payload_results.csv" +REPORT_PATH = OUTPUT_DIRECTORY / "lab028_report.txt" +OVERHEAD_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab028_overhead_efficiency.png" +) +TRAFFIC_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab028_packets_wire_bitrate.png" +) + +COMPOSITE_FPS = 3.0 +BASE_WIDTH = 240 +BASE_HEIGHT = 135 +BASE_QUALITY = 23 +ROI_WIDTH = 320 +ROI_HEIGHT = 180 +ROI_QUALITY = 33 +ROI_X_MIN = 0.20 +ROI_X_MAX = 0.80 +ROI_Y_MIN = 0.42 +ROI_Y_MAX = 1.00 +PAYLOAD_LENGTHS = (64, 128, 256, 512, 1024) +FRAME_TIME_EPSILON_SECONDS = 1e-9 + +CSV_FIELDS = [ + "max_payload_bytes", + "composite_frames", + "mean_base_packets_per_frame", + "mean_roi_packets_per_frame", + "mean_total_packets_per_frame", + "max_total_packets_per_frame", + "packets_per_second", + "jpeg_payload_bytes", + "jpeg_payload_bitrate_kbps", + "header_bytes_per_second", + "header_bitrate_kbps", + "wire_bytes", + "wire_bitrate_kbps", + "service_data_percent", + "efficiency_percent", + "mean_wire_packet_bytes", + "p95_wire_packet_bytes", + "max_wire_packet_bytes", +] + + +@dataclass(frozen=True) +class VideoMetadata: + width: int + height: int + fps: float + frame_count: int + duration_seconds: float + + +@dataclass(frozen=True) +class EncodedComposite: + composite_frame_id: int + source_frame_index: int + base_jpeg: bytes + roi_jpeg: bytes + + +@dataclass(frozen=True) +class PayloadMetrics: + max_payload_bytes: int + composite_frames: int + mean_base_packets_per_frame: float + mean_roi_packets_per_frame: float + mean_total_packets_per_frame: float + max_total_packets_per_frame: int + packets_per_second: float + jpeg_payload_bytes: int + jpeg_payload_bitrate_kbps: float + header_bytes_per_second: float + header_bitrate_kbps: float + wire_bytes: int + wire_bitrate_kbps: float + service_data_percent: float + efficiency_percent: float + mean_wire_packet_bytes: float + p95_wire_packet_bytes: float + max_wire_packet_bytes: int + + +@dataclass(frozen=True) +class TestResult: + name: str + passed: bool + detail: str + + +def normalized_roi_to_pixels( + width: int, + height: int, +) -> tuple[int, int, int, int]: + """Use the normalized ROI coordinates from Lab027 through Lab027E.""" + + coordinates = ( + int(round(width * ROI_X_MIN)), + int(round(height * ROI_Y_MIN)), + int(round(width * ROI_X_MAX)), + int(round(height * ROI_Y_MAX)), + ) + x_min, y_min, x_max, y_max = coordinates + if not ( + 0 <= x_min < x_max <= width + and 0 <= y_min < y_max <= height + ): + raise RuntimeError("calculated ROI is outside the source frame") + return coordinates + + +def encode_grayscale_jpeg(image: np.ndarray, quality: int) -> bytes: + """Encode one grayscale image to an in-memory JPEG.""" + + encoded, buffer = cv2.imencode( + ".jpg", + image, + [int(cv2.IMWRITE_JPEG_QUALITY), int(quality)], + ) + if not encoded: + raise RuntimeError("OpenCV could not encode JPEG") + jpeg = buffer.tobytes() + if not jpeg: + raise RuntimeError("OpenCV produced an empty JPEG") + return jpeg + + +def encode_composite( + source_frame: np.ndarray, + source_roi: tuple[int, int, int, int], + composite_frame_id: int, + source_frame_index: int, +) -> EncodedComposite: + """Form synchronous BASE and ROI JPEGs from exactly one source frame.""" + + grayscale = cv2.cvtColor(source_frame, cv2.COLOR_BGR2GRAY) + base = cv2.resize( + grayscale, + (BASE_WIDTH, BASE_HEIGHT), + interpolation=cv2.INTER_AREA, + ) + x_min, y_min, x_max, y_max = source_roi + roi = grayscale[y_min:y_max, x_min:x_max] + if roi.size == 0: + raise RuntimeError("source ROI is empty") + roi = cv2.resize( + roi, + (ROI_WIDTH, ROI_HEIGHT), + interpolation=cv2.INTER_AREA, + ) + return EncodedComposite( + composite_frame_id=composite_frame_id, + source_frame_index=source_frame_index, + base_jpeg=encode_grayscale_jpeg(base, BASE_QUALITY), + roi_jpeg=encode_grayscale_jpeg(roi, ROI_QUALITY), + ) + + +def load_video_profile( + source_path: Path, +) -> tuple[VideoMetadata, list[EncodedComposite]]: + """Read the video sequentially and select synchronous updates at 3 fps.""" + + if not source_path.exists(): + raise FileNotFoundError(f"source video is missing: {source_path}") + capture = cv2.VideoCapture(str(source_path)) + if not capture.isOpened(): + raise RuntimeError(f"OpenCV could not open {source_path}") + + width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT)) + fps = float(capture.get(cv2.CAP_PROP_FPS)) + declared_frame_count = int( + capture.get(cv2.CAP_PROP_FRAME_COUNT) + ) + if width <= 0 or height <= 0 or fps <= 0.0: + capture.release() + raise RuntimeError("invalid source video metadata") + + source_roi = normalized_roi_to_pixels(width, height) + selected: list[EncodedComposite] = [] + source_frame_index = 0 + next_composite_time = 0.0 + try: + while True: + frame_read, source_frame = capture.read() + if not frame_read or source_frame is None: + break + source_time = source_frame_index / fps + if ( + source_time + FRAME_TIME_EPSILON_SECONDS + >= next_composite_time + ): + selected.append( + encode_composite( + source_frame, + source_roi, + len(selected), + source_frame_index, + ) + ) + next_composite_time += 1.0 / COMPOSITE_FPS + source_frame_index += 1 + finally: + capture.release() + + if source_frame_index <= 0 or not selected: + raise RuntimeError("source video did not yield frames") + if ( + declared_frame_count > 0 + and source_frame_index != declared_frame_count + ): + raise RuntimeError( + "decoded frame count differs from video metadata: " + f"{source_frame_index} != {declared_frame_count}" + ) + metadata = VideoMetadata( + width=width, + height=height, + fps=fps, + frame_count=source_frame_index, + duration_seconds=source_frame_index / fps, + ) + return metadata, selected + + +def packets_for_composite( + composite: EncodedComposite, + max_payload_bytes: int, +) -> tuple[list[bytes], list[bytes]]: + """Packetize BASE and ROI separately with one composite frame ID.""" + + base_packets = packetize_jpeg( + composite.base_jpeg, + composite.composite_frame_id, + ObjectType.BASE, + max_payload_bytes, + ) + roi_packets = packetize_jpeg( + composite.roi_jpeg, + composite.composite_frame_id, + ObjectType.ROI, + max_payload_bytes, + ) + return base_packets, roi_packets + + +def calculate_payload_metrics( + composites: list[EncodedComposite], + duration_seconds: float, + max_payload_bytes: int, +) -> PayloadMetrics: + """Calculate actual packet and bitrate statistics for one payload limit.""" + + base_counts: list[int] = [] + roi_counts: list[int] = [] + total_counts: list[int] = [] + wire_packet_sizes: list[int] = [] + for composite in composites: + base_packets, roi_packets = packets_for_composite( + composite, max_payload_bytes + ) + base_counts.append(len(base_packets)) + roi_counts.append(len(roi_packets)) + total_counts.append(len(base_packets) + len(roi_packets)) + wire_packet_sizes.extend( + len(packet) for packet in base_packets + roi_packets + ) + + jpeg_payload_bytes = sum( + len(composite.base_jpeg) + len(composite.roi_jpeg) + for composite in composites + ) + packet_count = len(wire_packet_sizes) + header_bytes = packet_count * HEADER_SIZE + wire_bytes = jpeg_payload_bytes + header_bytes + return PayloadMetrics( + max_payload_bytes=max_payload_bytes, + composite_frames=len(composites), + mean_base_packets_per_frame=float(np.mean(base_counts)), + mean_roi_packets_per_frame=float(np.mean(roi_counts)), + mean_total_packets_per_frame=float(np.mean(total_counts)), + max_total_packets_per_frame=max(total_counts), + packets_per_second=packet_count / duration_seconds, + jpeg_payload_bytes=jpeg_payload_bytes, + jpeg_payload_bitrate_kbps=( + jpeg_payload_bytes * 8.0 / duration_seconds / 1000.0 + ), + header_bytes_per_second=header_bytes / duration_seconds, + header_bitrate_kbps=( + header_bytes * 8.0 / duration_seconds / 1000.0 + ), + wire_bytes=wire_bytes, + wire_bitrate_kbps=( + wire_bytes * 8.0 / duration_seconds / 1000.0 + ), + service_data_percent=header_bytes / wire_bytes * 100.0, + efficiency_percent=jpeg_payload_bytes / wire_bytes * 100.0, + mean_wire_packet_bytes=float(np.mean(wire_packet_sizes)), + p95_wire_packet_bytes=float( + np.percentile(wire_packet_sizes, 95) + ), + max_wire_packet_bytes=max(wire_packet_sizes), + ) + + +def feed_packets( + packets: list[bytes], + reassembler: CompositeReassembler | None = None, +) -> tuple[list[CompositeFrame], CompositeReassembler]: + """Feed packets and collect every atomically published frame.""" + + receiver = reassembler or CompositeReassembler() + completed = [] + for packet in packets: + frame = receiver.ingest(packet) + if frame is not None: + completed.append(frame) + return completed, receiver + + +def assert_frame_matches( + frame: CompositeFrame, + expected: EncodedComposite, +) -> None: + if frame.composite_frame_id != expected.composite_frame_id: + raise AssertionError("composite frame ID differs") + if frame.base_jpeg != expected.base_jpeg: + raise AssertionError("BASE JPEG differs byte-for-byte") + if frame.roi_jpeg != expected.roi_jpeg: + raise AssertionError("ROI JPEG differs byte-for-byte") + + +def run_functional_tests( + composites: list[EncodedComposite], +) -> list[TestResult]: + """Run header and all mandatory Lab028 transport checks.""" + + if len(composites) < 2: + raise RuntimeError("functional checks need two video frames") + first = composites[0] + second = composites[1] + base_packets, roi_packets = packets_for_composite(first, 256) + all_packets = base_packets + roi_packets + tests: list[tuple[str, Callable[[], str]]] = [] + + def header_round_trip() -> str: + parsed = decode_packet(all_packets[0]) + rebuilt = encode_packet( + VideoPacket( + composite_frame_id=parsed.composite_frame_id, + object_type=parsed.object_type, + fragment_index=parsed.fragment_index, + fragment_count=parsed.fragment_count, + jpeg_size=parsed.jpeg_size, + object_crc32=parsed.object_crc32, + payload=parsed.payload, + ) + ) + if rebuilt != all_packets[0]: + raise AssertionError("serialized bytes changed after round trip") + return f"fixed {HEADER_SIZE}-byte header round trip is exact" + + def ordered_lossless() -> str: + frames, _ = feed_packets(all_packets) + if len(frames) != 1: + raise AssertionError("ordered transfer did not emit one frame") + assert_frame_matches(frames[0], first) + return "BASE and ROI match original JPEG bytes" + + def shuffled_packets() -> str: + shuffled = list(all_packets) + random.Random(28001).shuffle(shuffled) + frames, _ = feed_packets(shuffled) + if len(frames) != 1: + raise AssertionError("shuffled transfer did not emit one frame") + assert_frame_matches(frames[0], first) + return "arbitrary packet order reconstructed correctly" + + def duplicate_packets() -> str: + duplicated = list(all_packets) + duplicated.extend( + [all_packets[0], all_packets[len(base_packets)]] + ) + random.Random(28002).shuffle(duplicated) + frames, receiver = feed_packets(duplicated) + if len(frames) != 1: + raise AssertionError("duplicates changed publication count") + assert_frame_matches(frames[0], first) + if receiver.duplicate_packets < 2: + raise AssertionError("exact duplicates were not counted") + return f"{receiver.duplicate_packets} exact duplicates ignored" + + def packet_crc_corruption() -> str: + corrupted = bytearray(all_packets[0]) + corrupted[-1] ^= 0x01 + receiver = CompositeReassembler() + try: + receiver.ingest(bytes(corrupted)) + except PacketCRCError: + pass + else: + raise AssertionError("corrupted packet passed packet CRC") + frames, _ = feed_packets(all_packets[1:], receiver) + if frames: + raise AssertionError("frame emitted despite rejected packet") + return "payload bit flip rejected; composite not emitted" + + def missing_fragment() -> str: + missing_packet = roi_packets[len(roi_packets) // 2] + missing_index = decode_packet(missing_packet).fragment_index + remaining = [ + packet + for packet in all_packets + if packet is not missing_packet + ] + frames, receiver = feed_packets(remaining) + if frames: + raise AssertionError("frame emitted with a missing fragment") + missing = receiver.missing_fragments( + first.composite_frame_id, ObjectType.ROI + ) + if missing is None or missing_index not in missing: + raise AssertionError("missing fragment was not reported") + return f"ROI fragment {missing_index} reported missing" + + def object_crc_corruption() -> str: + target_position = len(base_packets) + parsed = decode_packet(all_packets[target_position]) + changed_payload = bytearray(parsed.payload) + changed_payload[0] ^= 0x01 + altered = encode_packet( + VideoPacket( + composite_frame_id=parsed.composite_frame_id, + object_type=parsed.object_type, + fragment_index=parsed.fragment_index, + fragment_count=parsed.fragment_count, + jpeg_size=parsed.jpeg_size, + object_crc32=parsed.object_crc32, + payload=bytes(changed_payload), + ) + ) + formally_valid = list(all_packets) + formally_valid[target_position] = altered + receiver = CompositeReassembler() + emitted = [] + object_error_seen = False + for packet in formally_valid: + try: + frame = receiver.ingest(packet) + except ObjectCRCError: + object_error_seen = True + continue + if frame is not None: + emitted.append(frame) + if not object_error_seen: + raise AssertionError("object CRC did not detect changed JPEG") + if emitted: + raise AssertionError("frame emitted after object CRC failure") + return "valid packet CRCs still failed full-object CRC" + + def adjacent_frames_do_not_mix() -> str: + first_packets = sum(packets_for_composite(first, 256), []) + second_packets = sum(packets_for_composite(second, 256), []) + interleaved = first_packets + second_packets + random.Random(28003).shuffle(interleaved) + frames, _ = feed_packets(interleaved) + by_id = {frame.composite_frame_id: frame for frame in frames} + if set(by_id) != { + first.composite_frame_id, + second.composite_frame_id, + }: + raise AssertionError("adjacent frames were lost or mixed") + assert_frame_matches(by_id[first.composite_frame_id], first) + assert_frame_matches(by_id[second.composite_frame_id], second) + return "two interleaved frame IDs remained independent" + + def base_only_is_not_atomic() -> str: + frames, receiver = feed_packets(base_packets) + if frames: + raise AssertionError("BASE-only input emitted a composite") + if not receiver.object_is_complete( + first.composite_frame_id, ObjectType.BASE + ): + raise AssertionError("complete BASE object was not retained") + return "complete BASE retained while composite stayed unpublished" + + tests.extend( + [ + ("header_serialization_round_trip", header_round_trip), + ("ordered_lossless_transfer", ordered_lossless), + ("random_packet_order", shuffled_packets), + ("exact_duplicate_packets", duplicate_packets), + ("packet_crc_corruption", packet_crc_corruption), + ("missing_fragment", missing_fragment), + ("object_crc_corruption", object_crc_corruption), + ("adjacent_frame_isolation", adjacent_frames_do_not_mix), + ("base_without_roi_atomicity", base_only_is_not_atomic), + ] + ) + + results = [] + for name, test in tests: + try: + detail = test() + except Exception as error: + results.append(TestResult(name, False, str(error))) + else: + results.append(TestResult(name, True, detail)) + failed = [result for result in results if not result.passed] + if failed: + details = "; ".join( + f"{result.name}: {result.detail}" for result in failed + ) + raise RuntimeError(f"functional transport tests failed: {details}") + return results + + +def save_csv(metrics: list[PayloadMetrics]) -> None: + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + with CSV_PATH.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=CSV_FIELDS) + writer.writeheader() + for item in metrics: + row = {} + for field_name in CSV_FIELDS: + value = getattr(item, field_name) + row[field_name] = ( + f"{value:.6f}" + if isinstance(value, float) + else value + ) + writer.writerow(row) + + +def save_plots(metrics: list[PayloadMetrics]) -> None: + payloads = [item.max_payload_bytes for item in metrics] + + figure, axis = plt.subplots(figsize=(9, 5.5)) + axis.plot( + payloads, + [item.service_data_percent for item in metrics], + marker="o", + linewidth=2, + label="Service data (header / wire)", + ) + axis.plot( + payloads, + [item.efficiency_percent for item in metrics], + marker="s", + linewidth=2, + label="Efficiency (JPEG / wire)", + ) + axis.set_xscale("log", base=2) + axis.set_xticks(payloads) + axis.set_xticklabels([str(value) for value in payloads]) + axis.set_xlabel("Maximum packet payload, bytes") + axis.set_ylabel("Share, %") + axis.set_title("Lab028 packet overhead and efficiency") + axis.grid(True, alpha=0.3) + axis.legend() + figure.tight_layout() + figure.savefig(OVERHEAD_PLOT_PATH, dpi=160) + plt.close(figure) + + figure, packet_axis = plt.subplots(figsize=(9, 5.5)) + bitrate_axis = packet_axis.twinx() + packet_line = packet_axis.plot( + payloads, + [item.packets_per_second for item in metrics], + color="tab:blue", + marker="o", + linewidth=2, + label="Packets/s", + ) + bitrate_lines = bitrate_axis.plot( + payloads, + [item.wire_bitrate_kbps for item in metrics], + color="tab:red", + marker="s", + linewidth=2, + label="Wire bitrate", + ) + bitrate_axis.plot( + payloads, + [item.jpeg_payload_bitrate_kbps for item in metrics], + color="tab:green", + linestyle="--", + linewidth=2, + label="JPEG payload bitrate", + ) + packet_axis.set_xscale("log", base=2) + packet_axis.set_xticks(payloads) + packet_axis.set_xticklabels([str(value) for value in payloads]) + packet_axis.set_xlabel("Maximum packet payload, bytes") + packet_axis.set_ylabel("Packets per second", color="tab:blue") + bitrate_axis.set_ylabel("Bitrate, kbit/s", color="tab:red") + packet_axis.set_title("Lab028 packet rate and wire bitrate") + packet_axis.grid(True, alpha=0.3) + lines = packet_line + bitrate_lines + bitrate_axis.lines[1:] + packet_axis.legend( + lines, + [line.get_label() for line in lines], + loc="best", + ) + figure.tight_layout() + figure.savefig(TRAFFIC_PLOT_PATH, dpi=160) + plt.close(figure) + + +def metrics_table(metrics: list[PayloadMetrics]) -> list[str]: + lines = [ + ( + "payload | BASE pkt/frame | ROI pkt/frame | all pkt/frame | " + "max pkt/frame | pkt/s | JPEG kbit/s | headers B/s | " + "wire kbit/s | service % | efficiency % | packet mean/p95/max" + ), + ( + "-------:|---------------:|--------------:|--------------:|" + "--------------:|------:|------------:|------------:|" + "------------:|----------:|-------------:|--------------------:" + ), + ] + for item in metrics: + lines.append( + f"{item.max_payload_bytes} | " + f"{item.mean_base_packets_per_frame:.3f} | " + f"{item.mean_roi_packets_per_frame:.3f} | " + f"{item.mean_total_packets_per_frame:.3f} | " + f"{item.max_total_packets_per_frame} | " + f"{item.packets_per_second:.3f} | " + f"{item.jpeg_payload_bitrate_kbps:.3f} | " + f"{item.header_bytes_per_second:.3f} | " + f"{item.wire_bitrate_kbps:.3f} | " + f"{item.service_data_percent:.3f} | " + f"{item.efficiency_percent:.3f} | " + f"{item.mean_wire_packet_bytes:.1f}/" + f"{item.p95_wire_packet_bytes:.1f}/" + f"{item.max_wire_packet_bytes}" + ) + return lines + + +def write_report( + metadata: VideoMetadata, + composites: list[EncodedComposite], + metrics: list[PayloadMetrics], + test_results: list[TestResult], +) -> None: + source_roi = normalized_roi_to_pixels( + metadata.width, metadata.height + ) + base_sizes = [len(item.base_jpeg) for item in composites] + roi_sizes = [len(item.roi_jpeg) for item in composites] + lines = [ + "Lab028. Пакетирование синхронного BASE + ROI", + "", + "Профиль и исходные данные", + f"- Видео: {SOURCE_VIDEO_PATH}", + ( + f"- Источник: {metadata.width}x{metadata.height}, " + f"{metadata.fps:.6f} fps, {metadata.frame_count} кадров, " + f"{metadata.duration_seconds:.6f} с." + ), + ( + f"- ROI: x={ROI_X_MIN:.2f}...{ROI_X_MAX:.2f}, " + f"y={ROI_Y_MIN:.2f}...{ROI_Y_MAX:.2f}; " + f"пиксели x={source_roi[0]}...{source_roi[2]}, " + f"y={source_roi[1]}...{source_roi[3]}." + ), + ( + f"- Синхронные составные кадры: {len(composites)} " + f"при {COMPOSITE_FPS:.3f} fps." + ), + ( + f"- BASE: {BASE_WIDTH}x{BASE_HEIGHT}, grayscale JPEG Q" + f"{BASE_QUALITY}; средний размер {np.mean(base_sizes):.3f} B, " + f"min/max {min(base_sizes)}/{max(base_sizes)} B." + ), + ( + f"- ROI: {ROI_WIDTH}x{ROI_HEIGHT}, grayscale JPEG Q" + f"{ROI_QUALITY}; средний размер {np.mean(roi_sizes):.3f} B, " + f"min/max {min(roi_sizes)}/{max(roi_sizes)} B." + ), + ( + "- BASE и ROI формируются из одного source frame и имеют " + "общий composite_frame_id." + ), + "", + "Бинарный заголовок", + f"- struct format: {HEADER_FORMAT}", + f"- Размер: {HEADER_SIZE} байта; network byte order; padding нет.", + ( + "- Layout: magic[4] @0, version:u8 @4, object_type:u8 @5, " + "header_size:u16 @6, composite_frame_id:u32 @8, " + "fragment_index:u16 @12, fragment_count:u16 @14, " + "payload_length:u16 @16, flags:u16 @18, jpeg_size:u32 @20, " + "object_crc32:u32 @24, packet_crc32:u32 @28." + ), + "- object_type: 1=BASE, 2=ROI; flags зарезервирован и равен 0.", + "", + "CRC32", + ( + "- object_crc32 = zlib.crc32(полный JPEG) & 0xFFFFFFFF; " + "значение помещается во все фрагменты объекта и проверяется " + "после полной сборки." + ), + ( + "- packet_crc32: сначала поле packet_crc32 заголовка " + "обнуляется, затем CRC считается как " + "zlib.crc32(header_with_zero_crc + payload) & 0xFFFFFFFF." + ), + "- Таким образом packet CRC защищает все поля заголовка и payload.", + "", + "Функциональные проверки", + ] + lines.extend( + f"- {'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}" + for item in test_results + ) + lines.extend( + [ + "", + "Результаты размеров packet payload", + *metrics_table(metrics), + "", + ( + "JPEG payload bitrate одинаков для всех строк, потому что " + "исходные JPEG не меняются при выборе размера фрагмента." + ), + ( + "Wire bitrate включает только JPEG payload и 32-байтные " + "заголовки каждого пакета." + ), + ( + "Wire bitrate пока НЕ включает FEC, преамбулу, " + "синхронизацию, модуляцию, интервалы, повторные передачи, " + "команды управления и телеметрию." + ), + ( + "Лучший размер packet payload автоматически не выбирается: " + "таблица показывает только транспортный компромисс." + ), + "", + "Артефакты", + f"- CSV: {CSV_PATH}", + f"- Overhead/efficiency: {OVERHEAD_PLOT_PATH}", + f"- Packet count/wire bitrate: {TRAFFIC_PLOT_PATH}", + ( + "- Промежуточные JPEG и пакеты сохранялись только в памяти; " + "бинарные дампы не создавались." + ), + "", + ] + ) + REPORT_PATH.write_text("\n".join(lines), encoding="utf-8") + + +def validate_outputs(metrics: list[PayloadMetrics]) -> None: + if len(metrics) != len(PAYLOAD_LENGTHS): + raise RuntimeError("not all payload sizes were measured") + if [item.max_payload_bytes for item in metrics] != list( + PAYLOAD_LENGTHS + ): + raise RuntimeError("payload result order differs") + if any( + abs( + item.service_data_percent + + item.efficiency_percent + - 100.0 + ) > 1e-9 + for item in metrics + ): + raise RuntimeError("overhead and efficiency do not sum to 100%") + for path in ( + CSV_PATH, + REPORT_PATH, + OVERHEAD_PLOT_PATH, + TRAFFIC_PLOT_PATH, + ): + if not path.exists() or path.stat().st_size <= 0: + raise RuntimeError(f"missing or empty output: {path}") + + +def main() -> None: + print("Lab028: loading and JPEG-encoding the Lab027E profile...") + metadata, composites = load_video_profile(SOURCE_VIDEO_PATH) + print( + f" source={metadata.width}x{metadata.height}, " + f"{metadata.fps:.3f} fps, frames={metadata.frame_count}" + ) + print(f" synchronous composite frames={len(composites)}") + + print("Running functional transport checks...") + test_results = run_functional_tests(composites) + for result in test_results: + print(f" PASS {result.name}: {result.detail}") + + print("Measuring packet payload sizes...") + metrics = [ + calculate_payload_metrics( + composites, + metadata.duration_seconds, + payload_length, + ) + for payload_length in PAYLOAD_LENGTHS + ] + for item in metrics: + print( + f" payload={item.max_payload_bytes:4d} B: " + f"{item.packets_per_second:.3f} packet/s, " + f"wire={item.wire_bitrate_kbps:.3f} kbit/s, " + f"service={item.service_data_percent:.3f}%" + ) + + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + save_csv(metrics) + save_plots(metrics) + write_report(metadata, composites, metrics, test_results) + validate_outputs(metrics) + print(f"CSV: {CSV_PATH}") + print(f"Report: {REPORT_PATH}") + print(f"Plots: {OVERHEAD_PLOT_PATH}, {TRAFFIC_PLOT_PATH}") + print("Lab028 completed successfully.") + + +if __name__ == "__main__": + main()