""" Lab030. Systematic packet-erasure FEC for the synchronous video stream. Real Lab028 packets with 512-byte payload are grouped consecutively in blocks of k=8. Modes without FEC and with r=1,2,4 parity packets are compared over the time-based Lab029B channel at 300 kbit/s. Blocks may cross composite frame boundaries. Systematic packets precede parity packets; no interleaving, ARQ, or retransmission is used. """ from __future__ import annotations from bisect import bisect_right import csv from dataclasses import asdict, dataclass from itertools import combinations from pathlib import Path from typing import Callable import cv2 import matplotlib import numpy as np matplotlib.use("Agg") import matplotlib.pyplot as plt from protocol.packet_erasure_fec import ( GF_PRIMITIVE_POLYNOMIAL, OUTER_HEADER_FORMAT, OUTER_HEADER_SIZE, DecodedFECBlock, InsufficientSymbolsError, OuterPacketCRCError, decode_fec_block, decode_outer_symbol, encode_fec_block, gf_add, gf_div, gf_inverse, gf_mul, ) from protocol.video_packet import ( CompositeReassembler, ObjectType, decode_packet as decode_inner_packet, ) from tests.lab028_video_packetization import ( COMPOSITE_FPS, SOURCE_VIDEO_PATH, EncodedComposite, VideoMetadata, load_video_profile, ) from tests.lab029_packet_channel_simulation import ( PreparedProfile, prepare_profiles, ) from tests.lab029b_time_based_burst_simulation import ( BAD_TIME_FRACTION, CONTROL_STREAM_BITRATE_BPS, CONTROL_STREAM_BITRATE_KBPS, MEAN_BAD_DURATIONS_SECONDS, TimeInterval, generate_bad_intervals, percentile, positive_runs, ) OUTPUT_DIRECTORY = Path("data/processed/lab030") CSV_PATH = OUTPUT_DIRECTORY / "lab030_results.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab030_report.txt" COMPOSITE_SUCCESS_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab030_composite_success.png" ) NO_IMAGE_PLOT_PATH = OUTPUT_DIRECTORY / "lab030_no_image_duration.png" STREAM_RATE_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab030_stream_rate_overhead.png" ) DELAY_QUEUE_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab030_delay_queue.png" ) MODE_COMPARISON_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab030_mode_comparison.png" ) INNER_PAYLOAD_SIZE = 512 SOURCE_BLOCK_SIZE = 8 MONTE_CARLO_REPETITIONS = 200 MASTER_SEED = 300_300 SEED_BASE = MASTER_SEED TIME_EPSILON_SECONDS = 1e-12 @dataclass(frozen=True) class FECMode: name: str parity_count: int label: str FEC_MODES = ( FECMode("none", 0, "Без FEC"), FECMode("8+1", 1, "8+1"), FECMode("8+2", 2, "8+2"), FECMode("8+4", 4, "8+4"), ) CSV_FIELDS = [ "mode", "source_block_size", "nominal_parity_count", "mean_bad_duration_ms", "mean_good_duration_ms", "target_bad_time_fraction", "actual_bad_time_fraction", "monte_carlo_repetitions", "seed", "source_jpeg_bitrate_kbps", "inner_packet_stream_bitrate_kbps", "outer_fec_stream_bitrate_kbps", "service_and_parity_percent", "control_stream_bitrate_kbps", "schedule_duration_seconds", "mean_queue_length_packets", "max_queue_length_packets", "mean_publication_delay_seconds", "p95_publication_delay_seconds", "max_publication_delay_seconds", "transmitted_source_packets", "transmitted_parity_packets", "lost_source_packets", "lost_parity_packets", "recovered_source_packets", "fec_recovered_blocks", "fec_unrecoverable_blocks", "fec_affected_block_recovery_rate", "base_objects_completed", "roi_objects_completed", "atomic_composite_frames_completed", "composite_success_rate", "base_only_frames", "roi_only_frames", "incomplete_frames", "mean_no_new_image_duration_seconds", "p95_no_new_image_duration_seconds", "max_no_new_image_duration_seconds", "mean_consecutive_incomplete_frames", "p95_consecutive_incomplete_frames", "max_consecutive_incomplete_frames", "effective_delivered_video_bitrate_kbps", ] @dataclass(frozen=True) class SourcePacket: global_index: int composite_frame_id: int generation_time_seconds: float inner_packet: bytes @dataclass(frozen=True) class FECBlockPlan: block_id: int source_global_indices: tuple[int, ...] source_count: int parity_count: int symbol_size: int @dataclass(frozen=True) class TransmissionUnit: sequence_index: int generation_time_seconds: float wire_packet: bytes is_parity: bool block_id: int symbol_index: int source_global_index: int | None @dataclass(frozen=True) class ScheduledUnit: unit: TransmissionUnit start_seconds: float end_seconds: float @dataclass(frozen=True) class ModeSchedule: mode: FECMode source_packets: tuple[SourcePacket, ...] blocks: tuple[FECBlockPlan, ...] units: tuple[ScheduledUnit, ...] source_duration_seconds: float duration_seconds: float total_jpeg_bytes: int total_inner_bytes: int total_transmitted_bytes: int mean_queue_length_packets: float max_queue_length_packets: int @dataclass class BlockReceiveState: plan: FECBlockPlan received_outer_packets: list[bytes] delivered_source_indices: set[int] lost_source_indices: set[int] decoded: bool = False recovery_failed: bool = False recovered_source_packets: int = 0 @dataclass(frozen=True) class RepetitionResult: lost_source_packets: int lost_parity_packets: int recovered_source_packets: int fec_recovered_blocks: int fec_unrecoverable_blocks: int base_objects_completed: int roi_objects_completed: int atomic_composite_frames_completed: int base_only_frames: int roi_only_frames: int incomplete_frames: int delivered_jpeg_bytes: int publication_delays: tuple[float, ...] no_new_image_durations: tuple[float, ...] incomplete_frame_runs: tuple[int, ...] bad_time_seconds: float @dataclass(frozen=True) class SimulationResult: mode: str source_block_size: int nominal_parity_count: int mean_bad_duration_ms: float mean_good_duration_ms: float target_bad_time_fraction: float actual_bad_time_fraction: float monte_carlo_repetitions: int seed: int source_jpeg_bitrate_kbps: float inner_packet_stream_bitrate_kbps: float outer_fec_stream_bitrate_kbps: float service_and_parity_percent: float control_stream_bitrate_kbps: float schedule_duration_seconds: float mean_queue_length_packets: float max_queue_length_packets: int mean_publication_delay_seconds: float p95_publication_delay_seconds: float max_publication_delay_seconds: float transmitted_source_packets: int transmitted_parity_packets: int lost_source_packets: int lost_parity_packets: int recovered_source_packets: int fec_recovered_blocks: int fec_unrecoverable_blocks: int fec_affected_block_recovery_rate: float base_objects_completed: int roi_objects_completed: int atomic_composite_frames_completed: int composite_success_rate: float base_only_frames: int roi_only_frames: int incomplete_frames: int mean_no_new_image_duration_seconds: float p95_no_new_image_duration_seconds: float max_no_new_image_duration_seconds: float mean_consecutive_incomplete_frames: float p95_consecutive_incomplete_frames: float max_consecutive_incomplete_frames: int effective_delivered_video_bitrate_kbps: float @dataclass(frozen=True) class FunctionalTestResult: name: str passed: bool detail: str def parity_for_last_block( source_count: int, nominal_parity_count: int, ) -> int: """Preserve the nominal parity ratio in a final partial block.""" if nominal_parity_count == 0: return 0 return max( 1, int( np.ceil( source_count * nominal_parity_count / SOURCE_BLOCK_SIZE ) ), ) def prepare_source_packets( profile: PreparedProfile, ) -> tuple[SourcePacket, ...]: return tuple( SourcePacket( global_index=index, composite_frame_id=prepared.composite_frame_id, generation_time_seconds=( prepared.composite_frame_id / COMPOSITE_FPS ), inner_packet=prepared.wire_packet, ) for index, prepared in enumerate(profile.packets) ) def build_mode_units( source_packets: tuple[SourcePacket, ...], mode: FECMode, ) -> tuple[ tuple[TransmissionUnit, ...], tuple[FECBlockPlan, ...], ]: """Build sequential systematic/parity transmission units.""" if mode.parity_count == 0: units = tuple( TransmissionUnit( sequence_index=index, generation_time_seconds=packet.generation_time_seconds, wire_packet=packet.inner_packet, is_parity=False, block_id=index // SOURCE_BLOCK_SIZE, symbol_index=index % SOURCE_BLOCK_SIZE, source_global_index=index, ) for index, packet in enumerate(source_packets) ) return units, () units = [] blocks = [] sequence_index = 0 for block_id, start in enumerate( range(0, len(source_packets), SOURCE_BLOCK_SIZE) ): block_sources = source_packets[ start:start + SOURCE_BLOCK_SIZE ] source_count = len(block_sources) parity_count = ( mode.parity_count if source_count == SOURCE_BLOCK_SIZE else parity_for_last_block( source_count, mode.parity_count ) ) inner_packets = tuple( packet.inner_packet for packet in block_sources ) outer_packets = encode_fec_block( inner_packets, block_id, parity_count ) parsed_outer = [ decode_outer_symbol(packet) for packet in outer_packets ] generation_complete = max( packet.generation_time_seconds for packet in block_sources ) blocks.append( FECBlockPlan( block_id=block_id, source_global_indices=tuple( packet.global_index for packet in block_sources ), source_count=source_count, parity_count=parity_count, symbol_size=parsed_outer[0].symbol_size, ) ) for local_index, (wire_packet, parsed) in enumerate( zip(outer_packets, parsed_outer) ): is_parity = parsed.is_parity units.append( TransmissionUnit( sequence_index=sequence_index, generation_time_seconds=( generation_complete if is_parity else block_sources[ local_index ].generation_time_seconds ), wire_packet=wire_packet, is_parity=is_parity, block_id=block_id, symbol_index=parsed.symbol_index, source_global_index=( None if is_parity else block_sources[ local_index ].global_index ), ) ) sequence_index += 1 return tuple(units), tuple(blocks) def schedule_units( units: tuple[TransmissionUnit, ...], ) -> tuple[ tuple[ScheduledUnit, ...], float, float, int, ]: """Run the transmitter as one FIFO queue at 300 kbit/s.""" scheduled = [] cursor = 0.0 for unit in units: start = max(cursor, unit.generation_time_seconds) end = ( start + len(unit.wire_packet) * 8.0 / CONTROL_STREAM_BITRATE_BPS ) scheduled.append(ScheduledUnit(unit, start, end)) cursor = end queue_depth_samples = [] for arriving in units: generation_time = arriving.generation_time_seconds queue_depth_samples.append( sum( 1 for scheduled_unit in scheduled if ( scheduled_unit.unit.generation_time_seconds <= generation_time + TIME_EPSILON_SECONDS and scheduled_unit.end_seconds > generation_time + TIME_EPSILON_SECONDS ) ) ) return ( tuple(scheduled), scheduled[-1].end_seconds, float(np.mean(queue_depth_samples)), max(queue_depth_samples), ) def build_mode_schedules( metadata: VideoMetadata, composites: list[EncodedComposite], profile: PreparedProfile, ) -> dict[str, ModeSchedule]: source_packets = prepare_source_packets(profile) total_jpeg_bytes = sum( len(composite.base_jpeg) + len(composite.roi_jpeg) for composite in composites ) total_inner_bytes = sum( len(packet.inner_packet) for packet in source_packets ) schedules = {} for mode in FEC_MODES: units, blocks = build_mode_units(source_packets, mode) scheduled, duration, mean_queue, max_queue = schedule_units( units ) schedules[mode.name] = ModeSchedule( mode=mode, source_packets=source_packets, blocks=blocks, units=scheduled, source_duration_seconds=metadata.duration_seconds, duration_seconds=duration, total_jpeg_bytes=total_jpeg_bytes, total_inner_bytes=total_inner_bytes, total_transmitted_bytes=sum( len(unit.unit.wire_packet) for unit in scheduled ), mean_queue_length_packets=mean_queue, max_queue_length_packets=max_queue, ) return schedules def overlap_loss_flags( schedule: ModeSchedule, intervals: tuple[TimeInterval, ...], ) -> np.ndarray: flags = np.zeros(len(schedule.units), dtype=np.bool_) interval_index = 0 for unit_index, scheduled in enumerate(schedule.units): while ( interval_index < len(intervals) and intervals[interval_index].end_seconds <= scheduled.start_seconds + TIME_EPSILON_SECONDS ): interval_index += 1 if interval_index >= len(intervals): break interval = intervals[interval_index] if ( interval.start_seconds < scheduled.end_seconds - TIME_EPSILON_SECONDS and interval.end_seconds > scheduled.start_seconds + TIME_EPSILON_SECONDS ): flags[unit_index] = True return flags def frame_outage_metrics( frame_count: int, completion_times: dict[int, float], schedule_end: float, ) -> tuple[ tuple[float, ...], tuple[int, ...], tuple[float, ...], ]: complete = [ frame_id in completion_times for frame_id in range(frame_count) ] incomplete_runs = positive_runs([not flag for flag in complete]) no_image = [] index = 0 while index < frame_count: if complete[index]: index += 1 continue run_start = index while index < frame_count and not complete[index]: index += 1 start_time = ( completion_times[run_start - 1] if run_start > 0 else 0.0 ) end_time = ( completion_times[index] if index < frame_count else schedule_end ) no_image.append(max(0.0, end_time - start_time)) publication_delays = tuple( completion_time - frame_id / COMPOSITE_FPS for frame_id, completion_time in completion_times.items() ) return tuple(no_image), incomplete_runs, publication_delays def simulate_baseline( schedule: ModeSchedule, loss_flags: np.ndarray, frame_count: int, ) -> RepetitionResult: receiver = CompositeReassembler() completion_times = {} lost_source = 0 delivered_jpeg_bytes = 0 for lost, scheduled in zip(loss_flags, schedule.units): if bool(lost): lost_source += 1 continue completed = receiver.ingest(scheduled.unit.wire_packet) if completed is not None: completion_times[completed.composite_frame_id] = ( scheduled.end_seconds ) delivered_jpeg_bytes += ( len(completed.base_jpeg) + len(completed.roi_jpeg) ) return finish_repetition( schedule, receiver, completion_times, frame_count, lost_source, 0, 0, 0, 0, delivered_jpeg_bytes, ) def deliver_inner_packet( receiver: CompositeReassembler, inner_packet: bytes, delivery_time: float, completion_times: dict[int, float], ) -> int: completed = receiver.ingest(inner_packet) if completed is None: return 0 completion_times[completed.composite_frame_id] = delivery_time return len(completed.base_jpeg) + len(completed.roi_jpeg) def simulate_fec( schedule: ModeSchedule, loss_flags: np.ndarray, frame_count: int, ) -> RepetitionResult: receiver = CompositeReassembler() completion_times = {} plans = {plan.block_id: plan for plan in schedule.blocks} states = { block_id: BlockReceiveState( plan=plan, received_outer_packets=[], delivered_source_indices=set(), lost_source_indices=set(), ) for block_id, plan in plans.items() } lost_source = 0 lost_parity = 0 delivered_jpeg_bytes = 0 for lost, scheduled in zip(loss_flags, schedule.units): unit = scheduled.unit state = states[unit.block_id] if bool(lost): if unit.is_parity: lost_parity += 1 else: lost_source += 1 assert unit.source_global_index is not None state.lost_source_indices.add( unit.source_global_index ) continue state.received_outer_packets.append(unit.wire_packet) parsed = decode_outer_symbol(unit.wire_packet) if not parsed.is_parity: global_index = state.plan.source_global_indices[ parsed.symbol_index ] if global_index not in state.delivered_source_indices: state.delivered_source_indices.add(global_index) delivered_jpeg_bytes += deliver_inner_packet( receiver, parsed.data, scheduled.end_seconds, completion_times, ) if ( not state.decoded and len(state.received_outer_packets) >= state.plan.source_count ): decoded = decode_fec_block( tuple(state.received_outer_packets) ) state.decoded = True for local_index in decoded.recovered_indices: global_index = state.plan.source_global_indices[ local_index ] if global_index in state.delivered_source_indices: continue state.delivered_source_indices.add(global_index) state.recovered_source_packets += 1 delivered_jpeg_bytes += deliver_inner_packet( receiver, decoded.source_packets[local_index], scheduled.end_seconds, completion_times, ) recovered_packets = sum( state.recovered_source_packets for state in states.values() ) recovered_blocks = sum( 1 for state in states.values() if ( state.lost_source_indices and state.lost_source_indices <= state.delivered_source_indices ) ) failed_blocks = sum( 1 for state in states.values() if ( state.lost_source_indices and not ( state.lost_source_indices <= state.delivered_source_indices ) ) ) return finish_repetition( schedule, receiver, completion_times, frame_count, lost_source, lost_parity, recovered_packets, recovered_blocks, failed_blocks, delivered_jpeg_bytes, ) def finish_repetition( schedule: ModeSchedule, receiver: CompositeReassembler, completion_times: dict[int, float], frame_count: int, lost_source: int, lost_parity: int, recovered_packets: int, recovered_blocks: int, failed_blocks: int, delivered_jpeg_bytes: int, ) -> RepetitionResult: base_completed = 0 roi_completed = 0 base_only = 0 roi_only = 0 for frame_id in range(frame_count): atomic = frame_id in completion_times base = ( atomic or receiver.object_is_complete(frame_id, ObjectType.BASE) ) roi = ( atomic or receiver.object_is_complete(frame_id, ObjectType.ROI) ) base_completed += int(base) roi_completed += int(roi) base_only += int(base and not roi) roi_only += int(roi and not base) no_image, incomplete_runs, delays = frame_outage_metrics( frame_count, completion_times, schedule.duration_seconds ) return RepetitionResult( lost_source_packets=lost_source, lost_parity_packets=lost_parity, recovered_source_packets=recovered_packets, fec_recovered_blocks=recovered_blocks, fec_unrecoverable_blocks=failed_blocks, base_objects_completed=base_completed, roi_objects_completed=roi_completed, atomic_composite_frames_completed=len(completion_times), base_only_frames=base_only, roi_only_frames=roi_only, incomplete_frames=frame_count - len(completion_times), delivered_jpeg_bytes=delivered_jpeg_bytes, publication_delays=delays, no_new_image_durations=no_image, incomplete_frame_runs=incomplete_runs, bad_time_seconds=0.0, ) def simulate_condition( schedule: ModeSchedule, frame_count: int, mean_bad_duration_seconds: float, seed: int, repetitions: int, ) -> SimulationResult: rng = np.random.default_rng(seed) repetition_results = [] bad_time_seconds = 0.0 for _ in range(repetitions): intervals = generate_bad_intervals( schedule.duration_seconds, mean_bad_duration_seconds, rng, ) bad_time_seconds += sum( interval.duration_seconds for interval in intervals ) loss_flags = overlap_loss_flags(schedule, intervals) repetition_results.append( simulate_baseline(schedule, loss_flags, frame_count) if schedule.mode.parity_count == 0 else simulate_fec(schedule, loss_flags, frame_count) ) def total(field: str) -> int: return sum( int(getattr(result, field)) for result in repetition_results ) def flattened(field: str) -> list[float]: return [ float(value) for result in repetition_results for value in getattr(result, field) ] publication_delays = flattened("publication_delays") no_image = flattened("no_new_image_durations") incomplete_runs = flattened("incomplete_frame_runs") recovered_blocks = total("fec_recovered_blocks") failed_blocks = total("fec_unrecoverable_blocks") affected_blocks = recovered_blocks + failed_blocks total_frames = frame_count * repetitions source_packets_per_pass = len(schedule.source_packets) parity_packets_per_pass = sum( 1 for unit in schedule.units if unit.unit.is_parity ) source_jpeg_rate = ( schedule.total_jpeg_bytes * 8.0 / schedule.source_duration_seconds / 1000.0 ) inner_rate = ( schedule.total_inner_bytes * 8.0 / schedule.source_duration_seconds / 1000.0 ) outer_rate = ( schedule.total_transmitted_bytes * 8.0 / schedule.source_duration_seconds / 1000.0 ) delivered_bytes = total("delivered_jpeg_bytes") return SimulationResult( mode=schedule.mode.name, source_block_size=SOURCE_BLOCK_SIZE, nominal_parity_count=schedule.mode.parity_count, mean_bad_duration_ms=mean_bad_duration_seconds * 1000.0, mean_good_duration_ms=( mean_bad_duration_seconds * (1.0 - BAD_TIME_FRACTION) / BAD_TIME_FRACTION * 1000.0 ), target_bad_time_fraction=BAD_TIME_FRACTION, actual_bad_time_fraction=( bad_time_seconds / (schedule.duration_seconds * repetitions) ), monte_carlo_repetitions=repetitions, seed=seed, source_jpeg_bitrate_kbps=source_jpeg_rate, inner_packet_stream_bitrate_kbps=inner_rate, outer_fec_stream_bitrate_kbps=outer_rate, service_and_parity_percent=( ( schedule.total_transmitted_bytes - schedule.total_jpeg_bytes ) / schedule.total_transmitted_bytes * 100.0 ), control_stream_bitrate_kbps=CONTROL_STREAM_BITRATE_KBPS, schedule_duration_seconds=schedule.duration_seconds, mean_queue_length_packets=( schedule.mean_queue_length_packets ), max_queue_length_packets=schedule.max_queue_length_packets, mean_publication_delay_seconds=( float(np.mean(publication_delays)) if publication_delays else 0.0 ), p95_publication_delay_seconds=percentile( publication_delays, 95 ), max_publication_delay_seconds=( max(publication_delays) if publication_delays else 0.0 ), transmitted_source_packets=( source_packets_per_pass * repetitions ), transmitted_parity_packets=( parity_packets_per_pass * repetitions ), lost_source_packets=total("lost_source_packets"), lost_parity_packets=total("lost_parity_packets"), recovered_source_packets=total("recovered_source_packets"), fec_recovered_blocks=recovered_blocks, fec_unrecoverable_blocks=failed_blocks, fec_affected_block_recovery_rate=( recovered_blocks / affected_blocks if affected_blocks else 0.0 ), base_objects_completed=total("base_objects_completed"), roi_objects_completed=total("roi_objects_completed"), atomic_composite_frames_completed=total( "atomic_composite_frames_completed" ), composite_success_rate=( total("atomic_composite_frames_completed") / total_frames ), base_only_frames=total("base_only_frames"), roi_only_frames=total("roi_only_frames"), incomplete_frames=total("incomplete_frames"), mean_no_new_image_duration_seconds=( float(np.mean(no_image)) if no_image else 0.0 ), p95_no_new_image_duration_seconds=percentile(no_image, 95), max_no_new_image_duration_seconds=( max(no_image) if no_image else 0.0 ), mean_consecutive_incomplete_frames=( float(np.mean(incomplete_runs)) if incomplete_runs else 0.0 ), p95_consecutive_incomplete_frames=percentile( incomplete_runs, 95 ), max_consecutive_incomplete_frames=( int(max(incomplete_runs)) if incomplete_runs else 0 ), effective_delivered_video_bitrate_kbps=( delivered_bytes * 8.0 / (schedule.duration_seconds * repetitions) / 1000.0 ), ) def run_monte_carlo( schedules: dict[str, ModeSchedule], frame_count: int, ) -> list[SimulationResult]: results = [] for duration_index, duration in enumerate( MEAN_BAD_DURATIONS_SECONDS ): seed = SEED_BASE + duration_index for mode in FEC_MODES: results.append( simulate_condition( schedules[mode.name], frame_count, duration, seed, MONTE_CARLO_REPETITIONS, ) ) return results def result_lookup( results: list[SimulationResult], mean_bad_duration_ms: float, ) -> dict[str, SimulationResult]: return { result.mode: result for result in results if result.mean_bad_duration_ms == mean_bad_duration_ms } def run_functional_tests( composites: list[EncodedComposite], schedules: dict[str, ModeSchedule], results: list[SimulationResult], ) -> list[FunctionalTestResult]: tests: list[tuple[str, Callable[[], str]]] = [] sample_inner = tuple( packet.inner_packet for packet in schedules["none"].source_packets[:8] ) def gf_arithmetic() -> str: for value in range(1, 256): inverse = gf_inverse(value) if gf_mul(value, inverse) != 1: raise AssertionError(f"inverse failed for {value}") if gf_div(value, value) != 1: raise AssertionError(f"division failed for {value}") for left in range(0, 256, 17): for right in range(0, 256, 19): for third in range(0, 256, 31): if gf_mul(left, gf_add(right, third)) != gf_add( gf_mul(left, right), gf_mul(left, third) ): raise AssertionError("distributivity failed") return "inverse, division, and distributivity passed" def encode_decode_without_loss() -> str: outer = encode_fec_block(sample_inner, 1, 4) decoded = decode_fec_block(outer) if decoded.source_packets != sample_inner: raise AssertionError("lossless FEC round trip changed bytes") return "systematic 8+4 block is byte-exact without loss" def recover_any_r_losses() -> str: checked = 0 for parity_count in (1, 2, 4): outer = encode_fec_block( sample_inner, parity_count, parity_count ) for missing in combinations( range(len(outer)), parity_count ): available = tuple( packet for index, packet in enumerate(outer) if index not in missing ) decoded = decode_fec_block(available) if decoded.source_packets != sample_inner: raise AssertionError( f"failed r={parity_count}, missing={missing}" ) checked += 1 return f"all {checked} exact-r erasure combinations recovered" def r_plus_one_is_not_guaranteed() -> str: outer = encode_fec_block(sample_inner, 7, 2) available = outer[3:] try: decode_fec_block(available) except InsufficientSymbolsError: return "8+2 correctly rejected seven available symbols" raise AssertionError("r+1 erasures unexpectedly guaranteed") def recovered_inner_is_exact_and_crc_valid() -> str: outer = encode_fec_block(sample_inner, 9, 4) available = tuple( packet for index, packet in enumerate(outer) if index not in {0, 2, 5, 9} ) decoded = decode_fec_block(available) for original, restored in zip( sample_inner, decoded.source_packets ): if restored != original: raise AssertionError("restored inner bytes differ") decode_inner_packet(restored) return "restored Lab028 packets are byte-exact and CRC-valid" def outer_crc_detects_corruption() -> str: outer = bytearray( encode_fec_block(sample_inner, 10, 2)[0] ) outer[-1] ^= 0x01 try: decode_outer_symbol(bytes(outer)) except OuterPacketCRCError: return "outer payload bit flip rejected by outer CRC32" raise AssertionError("outer CRC did not reject corruption") def zero_bad_restores_all_frames() -> str: frame_count = len(composites) for mode in FEC_MODES: schedule = schedules[mode.name] loss_flags = np.zeros( len(schedule.units), dtype=np.bool_ ) repetition = ( simulate_baseline(schedule, loss_flags, frame_count) if mode.parity_count == 0 else simulate_fec(schedule, loss_flags, frame_count) ) if ( repetition.atomic_composite_frames_completed != frame_count ): raise AssertionError( f"zero-Bad failed for {mode.name}" ) return "all modes restored all 63 frames without Bad" def fixed_seed_is_reproducible() -> str: first = simulate_condition( schedules["8+2"], len(composites), 0.05, 399_399, 3 ) second = simulate_condition( schedules["8+2"], len(composites), 0.05, 399_399, 3 ) if first != second: raise AssertionError("same seed changed aggregate results") return "identical seed produced identical result fields" def incomplete_composite_is_not_published() -> str: receiver = CompositeReassembler() first_frame = [ packet.inner_packet for packet in schedules["none"].source_packets if packet.composite_frame_id == 0 ] published = 0 for packet in first_frame[:-1]: published += int(receiver.ingest(packet) is not None) if published: raise AssertionError("incomplete composite was published") return "missing ROI fragment prevented atomic publication" def queue_includes_parity() -> str: baseline = schedules["none"] protected = schedules["8+4"] parity_units = sum( unit.unit.is_parity for unit in protected.units ) if parity_units <= 0: raise AssertionError("8+4 schedule has no parity") if len(protected.units) != ( len(baseline.source_packets) + parity_units ): raise AssertionError("parity is absent from FIFO units") if ( protected.total_transmitted_bytes <= baseline.total_transmitted_bytes ): raise AssertionError("parity did not increase wire bytes") if protected.duration_seconds < baseline.duration_seconds: raise AssertionError("parity shortened transmitter schedule") return ( f"{parity_units} parity packets included in FIFO timing " "and queue metrics" ) tests.extend( [ ("gf256_arithmetic", gf_arithmetic), ("fec_without_loss", encode_decode_without_loss), ("recover_any_r_erasures", recover_any_r_losses), ("r_plus_one_not_guaranteed", r_plus_one_is_not_guaranteed), ( "restored_inner_packet_and_crc", recovered_inner_is_exact_and_crc_valid, ), ("outer_crc_detection", outer_crc_detects_corruption), ("zero_bad_100_percent", zero_bad_restores_all_frames), ("fixed_seed_reproducibility", fixed_seed_is_reproducible), ( "atomic_incomplete_composite", incomplete_composite_is_not_published, ), ("queue_counts_parity", queue_includes_parity), ] ) test_results = [] for name, test in tests: try: detail = test() except Exception as error: test_results.append( FunctionalTestResult(name, False, str(error)) ) else: test_results.append( FunctionalTestResult(name, True, detail) ) failed = [test for test in test_results if not test.passed] if failed: raise RuntimeError( "Lab030 functional checks failed: " + "; ".join( f"{test.name}: {test.detail}" for test in failed ) ) return test_results def validate_results(results: list[SimulationResult]) -> None: if len(results) != 16: raise RuntimeError(f"expected 16 rows, got {len(results)}") keys = { (result.mode, result.mean_bad_duration_ms) for result in results } if len(keys) != 16: raise RuntimeError("Lab030 result rows are not unique") if any( not 0.0 <= result.composite_success_rate <= 1.0 for result in results ): raise RuntimeError("composite success is outside 0...1") def save_csv(results: list[SimulationResult]) -> 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 result in results: raw = asdict(result) writer.writerow( { field: ( f"{raw[field]:.12g}" if isinstance(raw[field], float) else raw[field] ) for field in CSV_FIELDS } ) def save_line_plot( results: list[SimulationResult], getter: Callable[[SimulationResult], float], ylabel: str, title: str, path: Path, ) -> None: x_values = [ duration * 1000.0 for duration in MEAN_BAD_DURATIONS_SECONDS ] figure, axis = plt.subplots(figsize=(9, 5.5)) for mode in FEC_MODES: values = [ getter(result_lookup(results, duration_ms)[mode.name]) for duration_ms in x_values ] axis.plot( x_values, values, marker="o", linewidth=2, label=mode.label, ) axis.set_xscale("log") axis.set_xticks(x_values) axis.set_xticklabels([f"{value:g}" for value in x_values]) axis.set_xlabel("Средняя длительность Bad, мс") axis.set_ylabel(ylabel) axis.set_title(title) axis.grid(True, which="both", alpha=0.3) axis.legend() figure.tight_layout() figure.savefig(path, dpi=160) plt.close(figure) def save_plots(results: list[SimulationResult]) -> None: save_line_plot( results, lambda result: result.composite_success_rate * 100.0, "Полностью восстановленные составные кадры, %", "Lab030. Эффективность пакетного FEC", COMPOSITE_SUCCESS_PLOT_PATH, ) save_line_plot( results, lambda result: result.p95_no_new_image_duration_seconds, "P95 отсутствия нового изображения, с", "Lab030. Прерывания видеопотока", NO_IMAGE_PLOT_PATH, ) representative = result_lookup(results, 200.0) x = np.arange(len(FEC_MODES)) rates = [ representative[mode.name].outer_fec_stream_bitrate_kbps for mode in FEC_MODES ] overhead = [ representative[mode.name].service_and_parity_percent for mode in FEC_MODES ] figure, rate_axis = plt.subplots(figsize=(9, 5.5)) overhead_axis = rate_axis.twinx() rate_bars = rate_axis.bar( x - 0.18, rates, 0.36, color="tab:blue", label="Wire rate", ) overhead_bars = overhead_axis.bar( x + 0.18, overhead, 0.36, color="tab:orange", label="Service + parity", ) rate_axis.axhline( CONTROL_STREAM_BITRATE_KBPS, color="tab:red", linestyle="--", label="300 kbit/s", ) rate_axis.set_xticks(x) rate_axis.set_xticklabels([mode.label for mode in FEC_MODES]) rate_axis.set_ylabel("Предлагаемая скорость, кбит/с") overhead_axis.set_ylabel("Служебные и parity-данные, %") rate_axis.set_title("Lab030. Скорость и избыточность") rate_axis.grid(True, axis="y", alpha=0.3) rate_axis.legend( [rate_bars, overhead_bars, rate_axis.lines[0]], ["Wire rate", "Service + parity", "300 kbit/s"], loc="best", ) figure.tight_layout() figure.savefig(STREAM_RATE_PLOT_PATH, dpi=160) plt.close(figure) delays = [ representative[mode.name].p95_publication_delay_seconds for mode in FEC_MODES ] queues = [ representative[mode.name].max_queue_length_packets for mode in FEC_MODES ] figure, delay_axis = plt.subplots(figsize=(9, 5.5)) queue_axis = delay_axis.twinx() delay_bars = delay_axis.bar( x - 0.18, delays, 0.36, color="tab:blue", label="P95 delay", ) queue_bars = queue_axis.bar( x + 0.18, queues, 0.36, color="tab:orange", label="Max queue", ) delay_axis.set_xticks(x) delay_axis.set_xticklabels([mode.label for mode in FEC_MODES]) delay_axis.set_ylabel("P95 задержки публикации, с") queue_axis.set_ylabel("Максимальная очередь, пакетов") delay_axis.set_title( "Lab030. Задержка и очередь при Bad 200 мс" ) delay_axis.grid(True, axis="y", alpha=0.3) delay_axis.legend( [delay_bars, queue_bars], ["P95 publication delay", "Max queue"], loc="best", ) figure.tight_layout() figure.savefig(DELAY_QUEUE_PLOT_PATH, dpi=160) plt.close(figure) success = [ representative[mode.name].composite_success_rate * 100.0 for mode in FEC_MODES ] recovery = [ representative[ mode.name ].fec_affected_block_recovery_rate * 100.0 for mode in FEC_MODES ] figure, axis = plt.subplots(figsize=(9, 5.5)) axis.bar( x - 0.18, success, 0.36, label="Composite success" ) axis.bar( x + 0.18, recovery, 0.36, label="Affected FEC blocks recovered" ) axis.set_xticks(x) axis.set_xticklabels([mode.label for mode in FEC_MODES]) axis.set_ylabel("Доля, %") axis.set_title("Lab030. Сравнение режимов при Bad 200 мс") axis.grid(True, axis="y", alpha=0.3) axis.legend() figure.tight_layout() figure.savefig(MODE_COMPARISON_PLOT_PATH, dpi=160) plt.close(figure) def mode_rate_table( schedules: dict[str, ModeSchedule], ) -> list[str]: lines = [ ( "mode | source packets | parity packets | last block k+r | " "JPEG kbit/s | inner kbit/s | outer kbit/s | overhead | " "queue mean/max | timeline s" ), ( "----:|---------------:|---------------:|---------------:|" "------------:|-------------:|-------------:|---------:|" "---------------:|----------:" ), ] for mode in FEC_MODES: schedule = schedules[mode.name] parity_packets = sum( unit.unit.is_parity for unit in schedule.units ) last_block = ( f"{schedule.blocks[-1].source_count}+" f"{schedule.blocks[-1].parity_count}" if schedule.blocks else "none" ) jpeg_rate = ( schedule.total_jpeg_bytes * 8 / schedule.source_duration_seconds / 1000 ) inner_rate = ( schedule.total_inner_bytes * 8 / schedule.source_duration_seconds / 1000 ) outer_rate = ( schedule.total_transmitted_bytes * 8 / schedule.source_duration_seconds / 1000 ) overhead = ( ( schedule.total_transmitted_bytes - schedule.total_jpeg_bytes ) / schedule.total_transmitted_bytes * 100 ) lines.append( f"{mode.label} | {len(schedule.source_packets)} | " f"{parity_packets} | {last_block} | " f"{jpeg_rate:.3f} | {inner_rate:.3f} | " f"{outer_rate:.3f} | {overhead:.3f}% | " f"{schedule.mean_queue_length_packets:.3f}/" f"{schedule.max_queue_length_packets} | " f"{schedule.duration_seconds:.6f}" ) return lines def result_table(results: list[SimulationResult]) -> list[str]: lines = [ ( "Bad ms | mode | Bad actual | lost src/parity | recovered pkt | " "blocks recovered/failed | block recovery | composite | " "BASE-only/ROI-only | no image mean/p95/max s | " "delay mean/p95/max s | video kbit/s" ), ( "------:|-----:|-----------:|----------------:|--------------:|" "------------------------:|---------------:|----------:|" "------------------:|-------------------------:|" "----------------------:|------------:" ), ] for duration in MEAN_BAD_DURATIONS_SECONDS: by_mode = result_lookup(results, duration * 1000.0) for mode in FEC_MODES: result = by_mode[mode.name] lines.append( f"{duration * 1000.0:.0f} | {mode.label} | " f"{result.actual_bad_time_fraction * 100:.3f}% | " f"{result.lost_source_packets}/" f"{result.lost_parity_packets} | " f"{result.recovered_source_packets} | " f"{result.fec_recovered_blocks}/" f"{result.fec_unrecoverable_blocks} | " f"{result.fec_affected_block_recovery_rate * 100:.3f}% | " f"{result.composite_success_rate * 100:.3f}% | " f"{result.base_only_frames}/{result.roi_only_frames} | " f"{result.mean_no_new_image_duration_seconds:.3f}/" f"{result.p95_no_new_image_duration_seconds:.3f}/" f"{result.max_no_new_image_duration_seconds:.3f} | " f"{result.mean_publication_delay_seconds:.3f}/" f"{result.p95_publication_delay_seconds:.3f}/" f"{result.max_publication_delay_seconds:.3f} | " f"{result.effective_delivered_video_bitrate_kbps:.3f}" ) return lines def write_report( metadata: VideoMetadata, composites: list[EncodedComposite], schedules: dict[str, ModeSchedule], results: list[SimulationResult], tests: list[FunctionalTestResult], ) -> None: lines = [ "Lab030. Пакетное избыточное кодирование стираний", "", "Исходный профиль и неизменный внутренний транспорт", f"- Видео: {SOURCE_VIDEO_PATH}", ( f"- {len(composites)} реальных пар BASE/ROI, 3 fps; " "BASE 240x135 grayscale JPEG Q23, " "ROI 320x180 grayscale JPEG Q33." ), ( "- Внутренний Lab028 packet не изменён: payload 512 байт, " "32-byte header, packet CRC32 и object CRC32." ), "", "GF(256) и блочный код", ( f"- Примитивный полином: 0x{GF_PRIMITIVE_POLYNOMIAL:X} " "(x^8+x^4+x^3+x^2+1)." ), ( "- Матрица Вандермонда n×k умножается на обратную верхнюю " "k×k матрицу и становится систематической." ), ( "- Любые k доступных строк систематической генераторной " "матрицы восстанавливают k исходных символов." ), ( "- Нулевое дополнение используется только в математике; " "восстановленный внутренний пакет обрезается по payload_length " "Lab028 и проходит собственный CRC." ), "", "Внешний пакет FEC", f"- struct: {OUTER_HEADER_FORMAT}", ( f"- Размер: {OUTER_HEADER_SIZE} байта, network byte order, " "padding отсутствует." ), ( "- Layout: magic[4]@0, version:u8@4, flags:u8@5, " "header_size:u16@6, block_id:u32@8, symbol_index:u16@12, " "k:u16@14, r:u16@16, symbol_size:u16@18, " "data_length:u16@20, reserved16:u16@22, " "reserved32:u32@24, outer_crc32:u32@28." ), ( "- flags bit0: parity. CRC32 вычисляется по заголовку с " "нулевым outer_crc32 и данным внешнего символа." ), "", "Формирование блоков и передача", ( "- Последовательный поток внутренних пакетов BASE→ROI делится " "на k=8; блок может пересекать границу кадра." ), ( "- Сначала передаются systematic symbols, затем parity symbols " "этого блока; перемежение отсутствует." ), ( "- Последний блок содержит 6 исходных пакетов. Для него " "r_last=max(1, ceil(k_last*r/8)): режимы 8+1, 8+2, 8+4 " "получают соответственно 1, 2 и 3 parity." ), ( "- Режим Без FEC передаёт исходные Lab028 packets без внешней " "обёртки; остальные режимы используют внешний заголовок." ), ( "- FIFO учитывает время генерации кадра, все внешние пакеты, " "фактический размер и скорость 300 кбит/с. Очередь не " "сбрасывается на границах кадров." ), "", "Скорость, избыточность и очередь", *mode_rate_table(schedules), "", "Временная модель", ( "- Экспоненциальные Good/Bad интервалы Lab029B, Bad≈2%, " "mean Bad 10/50/200/1000 мс, 200 повторов, fixed seeds " f"{SEED_BASE}...{SEED_BASE + 3}." ), ( "- Пакет теряется при любом пересечении его передачи с Bad." ), ( "- FEC исправляет стирания; CRC обнаруживает повреждения, но " "битовые ошибки отдельно не добавляются." ), "", "Функциональные проверки", ] lines.extend( f"- {'PASS' if test.passed else 'FAIL'} {test.name}: {test.detail}" for test in tests ) lines.extend( [ "", "Результаты Monte Carlo", *result_table(results), "", "Допущения и ограничения", ( "- Не реализованы перемежение, ARQ, повторные передачи, " "команды управления, телеметрия и реальный SDR." ), ( "- Любое частичное пересечение Bad уничтожает весь внешний " "или базовый внутренний пакет." ), ( "- Принятые systematic packets немедленно поступают " "reassembler; восстановленные стирания становятся доступны " "при получении k символов блока." ), ( "- Publication delay измеряется от frame_id/3 до атомарной " "выдачи BASE+ROI." ), ( "- Queue length измеряется в моменты генерации пакетов и " "включает ожидающие и обслуживаемый пакет." ), ( "- Окончательный режим FEC автоматически не выбирается." ), "", "Артефакты", f"- CSV: {CSV_PATH}", f"- Composite success: {COMPOSITE_SUCCESS_PLOT_PATH}", f"- No-image duration: {NO_IMAGE_PLOT_PATH}", f"- Stream rate/overhead: {STREAM_RATE_PLOT_PATH}", f"- Delay/queue: {DELAY_QUEUE_PLOT_PATH}", f"- Mode comparison: {MODE_COMPARISON_PLOT_PATH}", ( "- JPEG, внутренние/внешние packets и бинарные дампы " "не сохранялись." ), "", ] ) REPORT_PATH.write_text("\n".join(lines), encoding="utf-8") def validate_outputs() -> None: for path in ( CSV_PATH, REPORT_PATH, COMPOSITE_SUCCESS_PLOT_PATH, NO_IMAGE_PLOT_PATH, STREAM_RATE_PLOT_PATH, DELAY_QUEUE_PLOT_PATH, MODE_COMPARISON_PLOT_PATH, ): if not path.exists() or path.stat().st_size <= 0: raise RuntimeError(f"missing or empty output: {path}") def main() -> None: print("Lab030: loading real 512-byte-payload Lab028 stream...") metadata, composites = load_video_profile(SOURCE_VIDEO_PATH) profile = prepare_profiles(composites)[INNER_PAYLOAD_SIZE] schedules = build_mode_schedules( metadata, composites, profile ) for mode in FEC_MODES: schedule = schedules[mode.name] print( f" {mode.label}: units={len(schedule.units)}, " f"offered=" f"{schedule.total_transmitted_bytes * 8 / metadata.duration_seconds / 1000:.3f} " f"kbit/s, timeline={schedule.duration_seconds:.6f} s" ) print( f"Running 16 conditions, " f"{MONTE_CARLO_REPETITIONS} repetitions each..." ) results = run_monte_carlo(schedules, len(composites)) validate_results(results) print("Running Lab030 functional checks...") tests = run_functional_tests( composites, schedules, results ) for test in tests: print(f" PASS {test.name}: {test.detail}") OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) save_csv(results) save_plots(results) write_report( metadata, composites, schedules, results, tests ) validate_outputs() print("Representative Bad=200 ms results:") representative = result_lookup(results, 200.0) for mode in FEC_MODES: result = representative[mode.name] print( f" {mode.label}: composite=" f"{result.composite_success_rate:.6f}, " f"blocks={result.fec_recovered_blocks}/" f"{result.fec_unrecoverable_blocks}, " f"delay_p95={result.p95_publication_delay_seconds:.6f} s" ) print(f"CSV: {CSV_PATH}") print(f"Report: {REPORT_PATH}") print("Lab030 completed successfully.") if __name__ == "__main__": main()