"""Lab036: long-duration whole-frame scheduling with changing capacity.""" from __future__ import annotations import csv from dataclasses import asdict, dataclass, replace import hashlib from pathlib import Path import subprocess from typing import Iterable import cv2 import matplotlib import numpy as np matplotlib.use("Agg") import matplotlib.pyplot as plt from protocol.link_packet import ( Direction, LinkPacket, TrafficClass, decode_link_packet, encode_link_packet, ) from protocol.packet_erasure_fec import ( decode_fec_block, decode_outer_symbol, encode_fec_block, ) from protocol.video_age_policy import AgePolicyPacket from protocol.video_frame_scheduler import FramePolicy, VideoFrameGroup from protocol.video_packet import decode_packet as decode_inner_packet from tests.lab028_video_packetization import ( COMPOSITE_FPS, EncodedComposite, packets_for_composite, ) from tests.lab033_priority_channel_scheduler import ( STREAM_CONTROL, STREAM_EMERGENCY, STREAM_TELEMETRY, STREAM_VIDEO, build_workload as build_lab033_workload, deterministic_payload, ) from tests.lab034_stale_video_drop import parity_for_partial OUTPUT_DIRECTORY = Path("data/processed/lab036") SUMMARY_CSV_PATH = OUTPUT_DIRECTORY / "lab036_summary.csv" VIDEO_CSV_PATH = OUTPUT_DIRECTORY / "lab036_video_metrics.csv" CONTROL_CSV_PATH = OUTPUT_DIRECTORY / "lab036_control_metrics.csv" RECOVERY_CSV_PATH = OUTPUT_DIRECTORY / "lab036_recovery_metrics.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab036_report.txt" AGE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_image_age_timeline.png" QUEUE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_queue_timeline.png" MINUTE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_minute_update_rate.png" RECOVERY_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_recovery.png" CONTROL_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_control_delay.png" OUTCOME_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_frame_outcomes.png" GROWTH_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_queue_growth_theory.png" COMPARISON_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_policy_comparison.png" PLOT_PATHS = ( AGE_PLOT_PATH, QUEUE_PLOT_PATH, MINUTE_PLOT_PATH, RECOVERY_PLOT_PATH, CONTROL_PLOT_PATH, OUTCOME_PLOT_PATH, GROWTH_PLOT_PATH, COMPARISON_PLOT_PATH, ) LAB035_COMMIT = "22c2eeab8cc794e67378485f49c236a622f814a7" DURATION_SECONDS = 600.0 FRAME_COUNT = int(DURATION_SECONDS * COMPOSITE_FPS) VIDEO_PAYLOAD_BYTES = 512 SOURCE_BLOCK_SIZE = 12 NOMINAL_PARITY_COUNT = 3 CONTROL_PERIOD_US = 50_000 TELEMETRY_PERIOD_US = 100_000 EMERGENCY_TIMES_US = (60_000_000, 180_000_000, 310_000_000, 480_000_000) TIME_EPSILON_SECONDS = 1e-9 @dataclass(frozen=True) class SpeedProfile: name: str label: str kind: str constant_kbps: float | None = None def rate_kbps(self, time_seconds: float) -> float: if self.kind == "constant": assert self.constant_kbps is not None return self.constant_kbps if self.kind == "step": if time_seconds < 120.0: return 300.0 if time_seconds < 300.0: return 230.0 return 300.0 if self.kind == "variable": if time_seconds >= DURATION_SECONDS: return 230.0 phase = time_seconds % 120.0 return 300.0 if phase < 40.0 else 260.0 if phase < 80.0 else 230.0 raise ValueError(f"unknown speed profile: {self.kind}") @property def mean_rate_kbps(self) -> float: if self.kind == "constant": assert self.constant_kbps is not None return self.constant_kbps if self.kind == "step": return (120.0 * 300.0 + 180.0 * 230.0 + 300.0 * 300.0) / DURATION_SECONDS return (300.0 + 260.0 + 230.0) / 3.0 @dataclass(frozen=True) class PolicyDefinition: name: str label: str policy: FramePolicy SPEED_PROFILES = ( SpeedProfile("constant_300", "Постоянно 300", "constant", 300.0), SpeedProfile("constant_260", "Постоянно 260", "constant", 260.0), SpeedProfile("constant_230", "Постоянно 230", "constant", 230.0), SpeedProfile("step_300_230_300", "300→230→300", "step"), SpeedProfile("variable_300_260_230", "Цикл 300/260/230", "variable"), ) POLICIES = ( PolicyDefinition("no_drop", "Без удаления", FramePolicy.NO_DROP), PolicyDefinition("latest_only", "Самый свежий", FramePolicy.LATEST_ONLY), PolicyDefinition("two_waiting", "Два ожидающих", FramePolicy.TWO_WAITING), ) PROFILE_BY_NAME = {item.name: item for item in SPEED_PROFILES} POLICY_BY_NAME = {item.name: item for item in POLICIES} @dataclass(frozen=True) class LongWorkload: frames: tuple[VideoFrameGroup, ...] high_priority: tuple[AgePolicyPacket, ...] jpeg_bytes_by_frame: tuple[int, ...] original_frame_count: int video_wire_bytes: int nonvideo_wire_bytes: int crc_packets_checked: int crc_blocks_checked: int repeated_headers_are_fresh: bool @property def offered_wire_bytes(self) -> int: return self.video_wire_bytes + self.nonvideo_wire_bytes @dataclass(frozen=True) class DroppedFrame: frame: VideoFrameGroup time_seconds: float reason: str @dataclass(frozen=True) class Replacement: removed: AgePolicyPacket replacement: AgePolicyPacket time_seconds: float @dataclass(frozen=True) class ScheduledPacket: item: AgePolicyPacket start_seconds: float end_seconds: float rate_kbps: float blocked_by: AgePolicyPacket | None blocking_delay_seconds: float @dataclass(frozen=True) class LongSchedule: profile: SpeedProfile policy: PolicyDefinition transmitted: tuple[ScheduledPacket, ...] dropped: tuple[DroppedFrame, ...] replacements: tuple[Replacement, ...] started_frame_ids: tuple[int, ...] completed_frame_ids: tuple[int, ...] publication_times: dict[int, float] digest: str bit_accounting_error: float rate_rule_ok: bool frames_do_not_interleave: bool high_priority_between_video_packets: bool @dataclass(frozen=True) class QueueStatistics: mean_packets: float max_packets: int mean_bytes: float max_bytes: int mean_waiting_frames: float max_waiting_frames: int minute_packets: tuple[int, ...] minute_bytes: tuple[int, ...] timeline_seconds: tuple[float, ...] timeline_packets: tuple[int, ...] timeline_bytes: tuple[int, ...] remaining_packets: int remaining_bytes: int remaining_frames: int drain_seconds: float growth_bytes_per_second: float growth_bytes_per_minute: float excess_queue_clear_time_seconds: float @dataclass(frozen=True) class SummaryMetrics: scenario: str policy: str duration_seconds: float mean_channel_kbps: float offered_load_kbps: float speed_reserve_kbps: float theoretical_queue_growth_bytes_per_second: float theoretical_minimum_skip_fraction: float theoretical_sustainable_fps: float actual_queue_growth_bytes_per_second: float actual_queue_growth_bytes_per_minute: float unbounded_queue_growth: bool mean_queue_packets: float max_queue_packets: int mean_queue_bytes: float max_queue_bytes: int mean_waiting_video_frames: float max_waiting_video_frames: int queue_packets_each_minute: str queue_bytes_each_minute: str queue_at_600s_packets: int queue_at_600s_bytes: int frames_remaining_at_600s: int additional_drain_seconds: float transmitted_packets: int transmitted_bytes: int replaced_state_packets: int @dataclass(frozen=True) class VideoMetrics: scenario: str policy: str created_frames: int started_frames: int published_frames: int published_by_600s_frames: int dropped_before_start_frames: int published_fraction: float actual_update_fps: float update_fps_each_minute: str mean_publication_delay_ms: float p95_publication_delay_ms: float max_publication_delay_ms: float mean_display_age_ms: float p95_display_age_ms: float max_display_age_ms: float display_age_over_500ms_fraction: float display_age_over_1000ms_fraction: float mean_no_update_duration_ms: float p95_no_update_duration_ms: float max_no_update_duration_ms: float max_consecutive_missing_frames: int mean_publication_gap_ms: float max_publication_gap_ms: float transmitted_video_bytes: int wasted_transmitted_video_bytes: int delivered_useful_video_kbps: float @dataclass(frozen=True) class ControlMetrics: scenario: str policy: str control_mean_delay_ms: float control_p95_delay_ms: float control_max_delay_ms: float control_over_100ms: int control_max_receive_gap_ms: float emergency_delays_ms: str emergency_max_delay_ms: float all_emergency_under_50ms: bool telemetry_mean_delay_ms: float telemetry_p95_delay_ms: float telemetry_max_delay_ms: float telemetry_over_500ms: int @dataclass(frozen=True) class RecoveryMetrics: scenario: str policy: str applicable: bool image_age_at_300s_ms: float time_to_age_below_1000ms_seconds: float time_to_age_below_500ms_seconds: float excess_queue_clear_seconds: float frames_dropped_after_recovery: int old_frames_published_after_recovery: int @dataclass(frozen=True) class ScenarioResult: summary: SummaryMetrics video: VideoMetrics control: ControlMetrics recovery: RecoveryMetrics minute_update_fps: tuple[float, ...] age_timeline_seconds: tuple[float, ...] age_timeline_ms: tuple[float, ...] queue_timeline_seconds: tuple[float, ...] queue_timeline_bytes: tuple[int, ...] digest: str accounting_ok: bool started_frames_complete: bool no_interleave: bool priority_between_video: bool rate_rule_ok: bool bit_accounting_error: float @dataclass(frozen=True) class FunctionalTestResult: name: str passed: bool detail: str def percentile(values: Iterable[float], q: float) -> float: values = tuple(values) return float(np.percentile(values, q)) if values else 0.0 def _link_packet( traffic_class: TrafficClass, direction: Direction, stream_id: int, sequence: int, generation_time_us: int, deadline_ms: int, payload: bytes, ) -> LinkPacket: return LinkPacket( traffic_class=traffic_class, direction=direction, stream_id=stream_id, sequence_number=sequence, generation_time_us=generation_time_us, deadline_ms=deadline_ms, payload=payload, ) def _validated_link(packet: LinkPacket) -> None: decoded = decode_link_packet(encode_link_packet(packet)) assert decoded.traffic_class is packet.traffic_class assert decoded.sequence_number == packet.sequence_number assert decoded.generation_time_us == packet.generation_time_us assert decoded.payload == packet.payload def build_long_workload() -> LongWorkload: """Repeat the 63 real JPEG-size pairs while rebuilding every header/CRC.""" lab033 = build_lab033_workload() originals = tuple(lab033.composites) if len(originals) != 63: raise AssertionError("Lab036 requires the 63-frame source profile") frames: list[VideoFrameGroup] = [] jpeg_sizes: list[int] = [] video_sequence = 0 block_id = 0 crc_packets = 0 crc_blocks = 0 repeated_inner_headers: dict[int, bytes] = {} for frame_id in range(FRAME_COUNT): original = originals[frame_id % len(originals)] composite = EncodedComposite( composite_frame_id=frame_id, source_frame_index=original.source_frame_index, base_jpeg=original.base_jpeg, roi_jpeg=original.roi_jpeg, ) base_packets, roi_packets = packets_for_composite(composite, VIDEO_PAYLOAD_BYTES) inner_packets = tuple(base_packets + roi_packets) if frame_id in (0, len(originals)): repeated_inner_headers[frame_id] = inner_packets[0] generation_us = int(round(frame_id / COMPOSITE_FPS * 1_000_000.0)) frame_packets: list[AgePolicyPacket] = [] for start in range(0, len(inner_packets), SOURCE_BLOCK_SIZE): chunk = inner_packets[start:start + SOURCE_BLOCK_SIZE] parity_count = ( NOMINAL_PARITY_COUNT if len(chunk) == SOURCE_BLOCK_SIZE else parity_for_partial(len(chunk)) ) outer_packets = encode_fec_block(chunk, block_id, parity_count) decoded_block = decode_fec_block(tuple(outer_packets)) assert decoded_block.source_packets == chunk for inner in decoded_block.source_packets: assert decode_inner_packet(inner).composite_frame_id == frame_id crc_blocks += 1 for outer_wire in outer_packets: decode_outer_symbol(outer_wire) packet = _link_packet( TrafficClass.VIDEO, Direction.ROVER_TO_GROUND, STREAM_VIDEO, video_sequence, generation_us, 0, outer_wire, ) _validated_link(packet) frame_packets.append( AgePolicyPacket(packet, video_sequence, generation_us, frame_id) ) video_sequence += 1 crc_packets += 1 block_id += 1 frames.append(VideoFrameGroup(frame_id, generation_us, tuple(frame_packets))) jpeg_sizes.append(len(original.base_jpeg) + len(original.roi_jpeg)) high_packets: list[AgePolicyPacket] = [] for sequence, generation_us in enumerate( range(0, int(DURATION_SECONDS * 1_000_000), CONTROL_PERIOD_US) ): packet = _link_packet( TrafficClass.CONTROL, Direction.GROUND_TO_ROVER, STREAM_CONTROL, sequence, generation_us, 100, deterministic_payload(b"CONTROL", sequence, 32), ) high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) for sequence, generation_us in enumerate( range(0, int(DURATION_SECONDS * 1_000_000), TELEMETRY_PERIOD_US) ): packet = _link_packet( TrafficClass.TELEMETRY, Direction.ROVER_TO_GROUND, STREAM_TELEMETRY, sequence, generation_us, 500, deterministic_payload(b"TELEM", sequence, 64), ) high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) for sequence, generation_us in enumerate(EMERGENCY_TIMES_US): packet = _link_packet( TrafficClass.EMERGENCY, Direction.GROUND_TO_ROVER, STREAM_EMERGENCY, sequence, generation_us, 50, deterministic_payload(b"E-STOP", sequence, 32), ) high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) high_packets.sort( key=lambda item: ( item.available_time_us, int(item.packet.traffic_class), item.packet.stream_id, item.packet.sequence_number, ) ) high_packets = [ replace(item, arrival_order=video_sequence + index) for index, item in enumerate(high_packets) ] for item in high_packets: _validated_link(item.packet) crc_packets += 1 return LongWorkload( frames=tuple(frames), high_priority=tuple(high_packets), jpeg_bytes_by_frame=tuple(jpeg_sizes), original_frame_count=len(originals), video_wire_bytes=sum(frame.wire_size_bytes for frame in frames), nonvideo_wire_bytes=sum(item.wire_size_bytes for item in high_packets), crc_packets_checked=crc_packets, crc_blocks_checked=crc_blocks, repeated_headers_are_fresh=( repeated_inner_headers[0] != repeated_inner_headers[len(originals)] ), ) def schedule_long( workload: LongWorkload, profile: SpeedProfile, policy: PolicyDefinition, ) -> LongSchedule: frame_arrivals = workload.frames high_arrivals = workload.high_priority ready_high: list[AgePolicyPacket] = [] pending_frames: list[VideoFrameGroup] = [] transmitted: list[ScheduledPacket] = [] dropped: list[DroppedFrame] = [] replacements: list[Replacement] = [] started: list[int] = [] completed: list[int] = [] publications: dict[int, float] = {} blocker_by_order: dict[int, tuple[AgePolicyPacket, float]] = {} active: VideoFrameGroup | None = None active_index = 0 cursor = 0.0 frame_index = 0 high_index = 0 def enqueue_high(item: AgePolicyPacket) -> None: if item.packet.traffic_class in (TrafficClass.CONTROL, TrafficClass.TELEMETRY): retained = [] for old in ready_high: if ( old.packet.traffic_class is item.packet.traffic_class and old.packet.stream_id == item.packet.stream_id ): replacements.append(Replacement(old, item, item.available_time_seconds)) else: retained.append(old) ready_high[:] = retained ready_high.append(item) def admit(now: float, current: AgePolicyPacket | None = None, start: float = 0.0) -> None: nonlocal frame_index, high_index while ( high_index < len(high_arrivals) and high_arrivals[high_index].available_time_seconds <= now + TIME_EPSILON_SECONDS ): item = high_arrivals[high_index] high_index += 1 if current is not None and item.available_time_seconds > start + TIME_EPSILON_SECONDS: blocker_by_order[item.arrival_order] = ( current, max(0.0, now - item.available_time_seconds), ) enqueue_high(item) while ( frame_index < len(frame_arrivals) and frame_arrivals[frame_index].generation_time_seconds <= now + TIME_EPSILON_SECONDS ): frame = frame_arrivals[frame_index] frame_index += 1 arrival = frame.generation_time_seconds if policy.policy is FramePolicy.LATEST_ONLY: dropped.extend( DroppedFrame(old, arrival, "replaced_by_newest") for old in pending_frames ) pending_frames[:] = [frame] elif policy.policy is FramePolicy.TWO_WAITING: pending_frames.append(frame) while len(pending_frames) > 2: dropped.append(DroppedFrame(pending_frames.pop(0), arrival, "waiting_limit")) else: pending_frames.append(frame) while ( frame_index < len(frame_arrivals) or high_index < len(high_arrivals) or ready_high or pending_frames or active is not None ): if not ready_high and not pending_frames and active is None: next_times = [] if frame_index < len(frame_arrivals): next_times.append(frame_arrivals[frame_index].generation_time_seconds) if high_index < len(high_arrivals): next_times.append(high_arrivals[high_index].available_time_seconds) cursor = max(cursor, min(next_times)) admit(cursor) selected: AgePolicyPacket | None = None if ready_high: selected = min( ready_high, key=lambda item: (int(item.packet.traffic_class), item.arrival_order), ) ready_high.remove(selected) else: if active is None and pending_frames: active = pending_frames.pop(0) active_index = 0 started.append(active.composite_frame_id) if active is not None: selected = active.packets[active_index] if selected is None: continue start = max(cursor, selected.available_time_seconds) rate_kbps = profile.rate_kbps(start) end = start + selected.wire_size_bytes * 8.0 / (rate_kbps * 1000.0) admit(end, selected, start) blocker, blocking_delay = blocker_by_order.get(selected.arrival_order, (None, 0.0)) transmitted.append( ScheduledPacket(selected, start, end, rate_kbps, blocker, blocking_delay) ) cursor = end if selected.packet.traffic_class is TrafficClass.VIDEO: assert active is not None active_index += 1 if active_index == len(active.packets): frame_id = active.composite_frame_id completed.append(frame_id) publications[frame_id] = end active = None active_index = 0 digest = hashlib.sha256() bit_error = 0.0 rate_rule_ok = True compressed_video: list[int] = [] high_since_video = False priority_between = False last_video: int | None = None for sent in transmitted: digest.update( f"{sent.item.arrival_order}:{sent.start_seconds:.12f}:{sent.end_seconds:.12f}:{sent.rate_kbps:.3f};".encode() ) bit_error += abs( (sent.end_seconds - sent.start_seconds) * sent.rate_kbps * 1000.0 - sent.item.wire_size_bytes * 8.0 ) rate_rule_ok &= sent.rate_kbps == profile.rate_kbps(sent.start_seconds) frame_id = sent.item.composite_frame_id if frame_id is None: if last_video is not None: high_since_video = True else: if frame_id == last_video and high_since_video: priority_between = True if not compressed_video or compressed_video[-1] != frame_id: compressed_video.append(frame_id) last_video = frame_id high_since_video = False for item in dropped: digest.update(f"D{item.frame.composite_frame_id}:{item.time_seconds:.12f};".encode()) return LongSchedule( profile, policy, tuple(transmitted), tuple(dropped), tuple(replacements), tuple(started), tuple(completed), publications, digest.hexdigest(), bit_error, rate_rule_ok, len(compressed_video) == len(set(compressed_video)), priority_between, ) def _sample_events( events: dict[float, tuple[int, int]], sample_times: tuple[float, ...], ) -> tuple[tuple[int, ...], tuple[int, ...]]: ordered = sorted(events.items()) index = 0 packets = 0 byte_count = 0 packet_values = [] byte_values = [] for sample in sample_times: while index < len(ordered) and ordered[index][0] <= sample + TIME_EPSILON_SECONDS: packets += ordered[index][1][0] byte_count += ordered[index][1][1] index += 1 packet_values.append(packets) byte_values.append(byte_count) return tuple(packet_values), tuple(byte_values) def queue_statistics( workload: LongWorkload, schedule: LongSchedule, ) -> QueueStatistics: intervals: list[tuple[float, float, int]] = [] first_start: dict[int, float] = {} drop_time = {item.frame.composite_frame_id: item.time_seconds for item in schedule.dropped} for sent in schedule.transmitted: intervals.append((sent.item.available_time_seconds, sent.end_seconds, sent.item.wire_size_bytes)) if sent.item.composite_frame_id is not None: first_start.setdefault(sent.item.composite_frame_id, sent.start_seconds) for item in schedule.dropped: intervals.extend( (packet.available_time_seconds, item.time_seconds, packet.wire_size_bytes) for packet in item.frame.packets ) intervals.extend( (item.removed.available_time_seconds, item.time_seconds, item.removed.wire_size_bytes) for item in schedule.replacements ) events: dict[float, tuple[int, int]] = {} packet_area = 0.0 byte_area = 0.0 for start, end, size in intervals: left = max(0.0, start) right = min(DURATION_SECONDS, end) if right <= left + TIME_EPSILON_SECONDS: continue packet_area += right - left byte_area += (right - left) * size old = events.get(start, (0, 0)) events[start] = (old[0] + 1, old[1] + size) old = events.get(end, (0, 0)) events[end] = (old[0] - 1, old[1] - size) count = 0 byte_count = 0 max_count = 0 max_bytes = 0 for time, delta in sorted(events.items()): if time > DURATION_SECONDS + TIME_EPSILON_SECONDS: break count += delta[0] byte_count += delta[1] max_count = max(max_count, count) max_bytes = max(max_bytes, byte_count) # One largest frame plus coincident control, telemetry, and emergency is # the normal 300-kbit/s envelope, not residual video backlog. baseline_limit = max(len(frame.packets) for frame in workload.frames) + 3 count = 0 excess_active = False excess_clear = 300.0 for time, delta in sorted(events.items()): count += delta[0] if time < 300.0 - TIME_EPSILON_SECONDS: continue if count > baseline_limit: excess_active = True elif excess_active: excess_clear = time excess_active = False minute_times = tuple(float(value) for value in range(60, 601, 60)) minute_packets, minute_bytes = _sample_events(events, minute_times) timeline_times = tuple(float(value) for value in range(0, 601, 5)) timeline_packets, timeline_bytes = _sample_events(events, timeline_times) fit_times = np.asarray(minute_times[1:], dtype=float) fit_bytes = np.asarray(minute_bytes[1:], dtype=float) growth = float(np.polyfit(fit_times, fit_bytes, 1)[0]) if len(fit_times) > 1 else 0.0 frame_events: dict[float, int] = {} frame_area = 0.0 for frame in workload.frames: start = frame.generation_time_seconds end = first_start.get(frame.composite_frame_id, drop_time.get(frame.composite_frame_id, start)) left = max(0.0, start) right = min(DURATION_SECONDS, end) if right <= left + TIME_EPSILON_SECONDS: continue frame_area += right - left frame_events[left] = frame_events.get(left, 0) + 1 frame_events[right] = frame_events.get(right, 0) - 1 waiting = 0 max_waiting = 0 for _, delta in sorted(frame_events.items()): waiting += delta max_waiting = max(max_waiting, waiting) published_by_end = sum(time <= DURATION_SECONDS + TIME_EPSILON_SECONDS for time in schedule.publication_times.values()) dropped_by_end = sum(item.time_seconds <= DURATION_SECONDS + TIME_EPSILON_SECONDS for item in schedule.dropped) remaining_frames = FRAME_COUNT - published_by_end - dropped_by_end finish = max( [DURATION_SECONDS] + [item.end_seconds for item in schedule.transmitted] + [item.time_seconds for item in schedule.dropped] + [item.time_seconds for item in schedule.replacements] ) return QueueStatistics( packet_area / DURATION_SECONDS, max_count, byte_area / DURATION_SECONDS, max_bytes, frame_area / DURATION_SECONDS, max_waiting, minute_packets, minute_bytes, timeline_times, timeline_packets, timeline_bytes, minute_packets[-1], minute_bytes[-1], remaining_frames, max(0.0, finish - DURATION_SECONDS), growth, growth * 60.0, max(0.0, excess_clear - 300.0), ) def display_ages( publications: dict[int, float], sample_times: tuple[float, ...], ) -> tuple[float, ...]: events = sorted((time, frame_id) for frame_id, time in publications.items()) ages = [] index = 0 last_frame: int | None = None for sample in sample_times: while index < len(events) and events[index][0] <= sample + TIME_EPSILON_SECONDS: last_frame = events[index][1] index += 1 generation = 0.0 if last_frame is None else last_frame / COMPOSITE_FPS ages.append(max(0.0, sample - generation)) return tuple(ages) def _missing_run(published: set[int]) -> int: current = 0 maximum = 0 for frame_id in range(FRAME_COUNT): if frame_id in published: current = 0 else: current += 1 maximum = max(maximum, current) return maximum def video_metrics( workload: LongWorkload, schedule: LongSchedule, ) -> tuple[VideoMetrics, tuple[float, ...], tuple[float, ...], tuple[float, ...]]: publications = schedule.publication_times published = set(publications) within = sorted((time, frame_id) for frame_id, time in publications.items() if time <= DURATION_SECONDS + TIME_EPSILON_SECONDS) minute_fps = tuple( sum(left <= time < right for time, _ in within) / 60.0 for left, right in zip(range(0, 600, 60), range(60, 601, 60)) ) sample_times = tuple(float(value) / 10.0 for value in range(0, 6001)) ages = display_ages(publications, sample_times) timeline_times = tuple(float(value) for value in range(0, 601)) timeline_ages = display_ages(publications, timeline_times) publication_delays = [ time - frame_id / COMPOSITE_FPS for frame_id, time in publications.items() ] update_times = [0.0] + [time for time, _ in within] + [DURATION_SECONDS] no_update = [max(0.0, right - left) for left, right in zip(update_times, update_times[1:])] gaps = [right[0] - left[0] for left, right in zip(within, within[1:])] transmitted_video_bytes = sum( item.item.wire_size_bytes for item in schedule.transmitted if item.item.packet.traffic_class is TrafficClass.VIDEO ) useful_bytes = sum(workload.jpeg_bytes_by_frame[frame_id] for _, frame_id in within) metrics = VideoMetrics( schedule.profile.name, schedule.policy.name, FRAME_COUNT, len(schedule.started_frame_ids), len(published), len(within), len(schedule.dropped), len(published) / FRAME_COUNT, len(within) / DURATION_SECONDS, ";".join(f"{value:.6f}" for value in minute_fps), float(np.mean(publication_delays)) * 1000.0, percentile(publication_delays, 95) * 1000.0, max(publication_delays, default=0.0) * 1000.0, float(np.mean(ages)) * 1000.0, percentile(ages, 95) * 1000.0, max(ages, default=0.0) * 1000.0, sum(value > 0.5 for value in ages) / len(ages), sum(value > 1.0 for value in ages) / len(ages), float(np.mean(no_update)) * 1000.0, percentile(no_update, 95) * 1000.0, max(no_update, default=0.0) * 1000.0, _missing_run(published), float(np.mean(gaps)) * 1000.0 if gaps else 0.0, max(gaps, default=0.0) * 1000.0, transmitted_video_bytes, 0, useful_bytes * 8.0 / DURATION_SECONDS / 1000.0, ) return metrics, minute_fps, timeline_times, tuple(value * 1000.0 for value in timeline_ages) def control_metrics(schedule: LongSchedule) -> ControlMetrics: by_class = { traffic: [item for item in schedule.transmitted if item.item.packet.traffic_class is traffic] for traffic in (TrafficClass.CONTROL, TrafficClass.EMERGENCY, TrafficClass.TELEMETRY) } control = by_class[TrafficClass.CONTROL] control_delays = [item.end_seconds - item.item.frame_generation_seconds for item in control] control_gaps = [right.end_seconds - left.end_seconds for left, right in zip(control, control[1:])] telemetry_delays = [ item.end_seconds - item.item.frame_generation_seconds for item in by_class[TrafficClass.TELEMETRY] ] emergency = by_class[TrafficClass.EMERGENCY] if len(emergency) != len(EMERGENCY_TIMES_US): raise AssertionError("all four emergency commands must be transmitted") emergency_delays = [item.end_seconds - item.item.frame_generation_seconds for item in emergency] return ControlMetrics( schedule.profile.name, schedule.policy.name, float(np.mean(control_delays)) * 1000.0, percentile(control_delays, 95) * 1000.0, max(control_delays) * 1000.0, sum(value > 0.1 + TIME_EPSILON_SECONDS for value in control_delays), max(control_gaps, default=0.0) * 1000.0, ";".join(f"{value * 1000.0:.6f}" for value in emergency_delays), max(emergency_delays) * 1000.0, all(value <= 0.05 + TIME_EPSILON_SECONDS for value in emergency_delays), float(np.mean(telemetry_delays)) * 1000.0, percentile(telemetry_delays, 95) * 1000.0, max(telemetry_delays) * 1000.0, sum(value > 0.5 + TIME_EPSILON_SECONDS for value in telemetry_delays), ) def _age_at(publications: dict[int, float], time_seconds: float) -> float: available = [(time, frame_id) for frame_id, time in publications.items() if time <= time_seconds + TIME_EPSILON_SECONDS] if not available: return time_seconds _, frame_id = max(available) return max(0.0, time_seconds - frame_id / COMPOSITE_FPS) def _time_to_age( publications: dict[int, float], start_seconds: float, threshold_seconds: float, ) -> float: if _age_at(publications, start_seconds) < threshold_seconds: return 0.0 for frame_id, time in sorted(publications.items(), key=lambda item: item[1]): if time < start_seconds - TIME_EPSILON_SECONDS or time > DURATION_SECONDS + TIME_EPSILON_SECONDS: continue if time - frame_id / COMPOSITE_FPS < threshold_seconds: return time - start_seconds return -1.0 def recovery_metrics(schedule: LongSchedule, queue: QueueStatistics) -> RecoveryMetrics: applicable = schedule.profile.kind == "step" if not applicable: return RecoveryMetrics(schedule.profile.name, schedule.policy.name, False, 0.0, 0.0, 0.0, 0.0, 0, 0) dropped_after = sum( item.frame.generation_time_seconds < 300.0 and item.time_seconds >= 300.0 - TIME_EPSILON_SECONDS for item in schedule.dropped ) old_published = sum( frame_id / COMPOSITE_FPS < 300.0 and time >= 300.0 - TIME_EPSILON_SECONDS for frame_id, time in schedule.publication_times.items() ) return RecoveryMetrics( schedule.profile.name, schedule.policy.name, True, _age_at(schedule.publication_times, 300.0) * 1000.0, _time_to_age(schedule.publication_times, 300.0, 1.0), _time_to_age(schedule.publication_times, 300.0, 0.5), queue.excess_queue_clear_time_seconds, dropped_after, old_published, ) def analyze_schedule(workload: LongWorkload, schedule: LongSchedule) -> ScenarioResult: queue = queue_statistics(workload, schedule) video, minute_fps, age_times, age_values = video_metrics(workload, schedule) control = control_metrics(schedule) recovery = recovery_metrics(schedule, queue) offered_kbps = workload.offered_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 video_kbps = workload.video_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 nonvideo_kbps = workload.nonvideo_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 mean_rate = schedule.profile.mean_rate_kbps reserve = mean_rate - offered_kbps remaining_video = max(0.0, mean_rate - nonvideo_kbps) skip = max(0.0, 1.0 - remaining_video / video_kbps) unbounded = ( schedule.policy.policy is FramePolicy.NO_DROP and reserve < 0.0 and queue.growth_bytes_per_second > 0.0 ) transmitted_bytes = sum(item.item.wire_size_bytes for item in schedule.transmitted) summary = SummaryMetrics( schedule.profile.name, schedule.policy.name, DURATION_SECONDS, mean_rate, offered_kbps, reserve, max(0.0, -reserve) * 1000.0 / 8.0, skip, COMPOSITE_FPS * min(1.0, remaining_video / video_kbps), queue.growth_bytes_per_second, queue.growth_bytes_per_minute, unbounded, queue.mean_packets, queue.max_packets, queue.mean_bytes, queue.max_bytes, queue.mean_waiting_frames, queue.max_waiting_frames, ";".join(str(value) for value in queue.minute_packets), ";".join(str(value) for value in queue.minute_bytes), queue.remaining_packets, queue.remaining_bytes, queue.remaining_frames, queue.drain_seconds, len(schedule.transmitted), transmitted_bytes, len(schedule.replacements), ) input_packets = sum(len(frame.packets) for frame in workload.frames) + len(workload.high_priority) input_bytes = workload.offered_wire_bytes dropped_packets = sum(len(item.frame.packets) for item in schedule.dropped) dropped_bytes = sum(item.frame.wire_size_bytes for item in schedule.dropped) replaced_packets = len(schedule.replacements) replaced_bytes = sum(item.removed.wire_size_bytes for item in schedule.replacements) accounting_ok = ( input_packets == len(schedule.transmitted) + dropped_packets + replaced_packets and input_bytes == transmitted_bytes + dropped_bytes + replaced_bytes and FRAME_COUNT == len(schedule.completed_frame_ids) + len(schedule.dropped) ) started_complete = ( set(schedule.started_frame_ids) == set(schedule.completed_frame_ids) and not set(schedule.started_frame_ids).intersection( item.frame.composite_frame_id for item in schedule.dropped ) ) return ScenarioResult( summary, video, control, recovery, minute_fps, age_times, age_values, queue.timeline_seconds, queue.timeline_bytes, schedule.digest, accounting_ok, started_complete, schedule.frames_do_not_interleave, schedule.high_priority_between_video_packets, schedule.rate_rule_ok, schedule.bit_accounting_error, ) def run_experiment(workload: LongWorkload) -> tuple[ScenarioResult, ...]: results = [] for profile in SPEED_PROFILES: for policy in POLICIES: results.append(analyze_schedule(workload, schedule_long(workload, profile, policy))) print(f" {profile.name} / {policy.name}") return tuple(results) def run_functional_tests( workload: LongWorkload, results: tuple[ScenarioResult, ...], ) -> tuple[FunctionalTestResult, ...]: lookup = {(item.summary.scenario, item.summary.policy): item for item in results} checks: list[tuple[str, callable]] = [] def check(name): def register(function): checks.append((name, function)) return function return register @check("01_unique_repeated_frames_and_crc") def _(): assert len({frame.composite_frame_id for frame in workload.frames}) == FRAME_COUNT assert workload.repeated_headers_are_fresh assert workload.crc_packets_checked > 0 and workload.crc_blocks_checked > 0 @check("02_300kbps_latest_has_no_unnecessary_drop") def _(): assert lookup[("constant_300", "latest_only")].video.dropped_before_start_frames == 0 @check("03_260kbps_no_drop_matches_small_deficit") def _(): row = lookup[("constant_260", "no_drop")].summary assert row.theoretical_queue_growth_bytes_per_second > 0.0 assert abs(row.actual_queue_growth_bytes_per_second - row.theoretical_queue_growth_bytes_per_second) / row.theoretical_queue_growth_bytes_per_second < 0.45 @check("04_230kbps_no_drop_grows_faster") def _(): slow = lookup[("constant_230", "no_drop")].summary.actual_queue_growth_bytes_per_second marginal = lookup[("constant_260", "no_drop")].summary.actual_queue_growth_bytes_per_second assert slow > marginal * 4.0 @check("05_latest_waiting_limit") def _(): assert all(item.summary.max_waiting_video_frames <= 1 for item in results if item.summary.policy == "latest_only") @check("06_two_waiting_limit") def _(): assert all(item.summary.max_waiting_video_frames <= 2 for item in results if item.summary.policy == "two_waiting") @check("07_started_frame_never_dropped") def _(): assert all(item.started_frames_complete for item in results) @check("08_active_video_frames_do_not_interleave") def _(): assert all(item.no_interleave for item in results) @check("09_priority_between_video_packets") def _(): assert all(item.priority_between_video for item in results) @check("10_emergency_never_deleted") def _(): assert all(len(item.control.emergency_delays_ms.split(";")) == 4 for item in results) @check("11_speed_change_bit_accounting") def _(): assert all(item.rate_rule_ok and item.bit_accounting_error < 1e-3 for item in results) @check("12_latest_does_not_replay_full_history") def _(): assert lookup[("step_300_230_300", "latest_only")].recovery.old_frames_published_after_recovery <= 2 @check("13_proactive_policies_have_no_waste") def _(): assert all(item.video.wasted_transmitted_video_bytes == 0 for item in results if item.summary.policy != "no_drop") @check("14_incomplete_frame_not_published") def _(): assert all(item.video.published_frames == item.video.started_frames for item in results) @check("15_all_three_crc_layers_pass") def _(): assert workload.crc_packets_checked == sum(len(frame.packets) for frame in workload.frames) + len(workload.high_priority) @check("16_frame_accounting") def _(): assert all(item.video.created_frames == item.video.published_frames + item.video.dropped_before_start_frames for item in results) @check("17_packet_and_byte_accounting") def _(): assert all(item.accounting_ok for item in results) @check("18_reproducible") def _(): repeated = schedule_long( workload, PROFILE_BY_NAME["constant_230"], POLICY_BY_NAME["latest_only"], ) assert repeated.digest == lookup[("constant_230", "latest_only")].digest output = [] for name, function in checks: try: function() output.append(FunctionalTestResult(name, True, "PASS")) except Exception as error: output.append(FunctionalTestResult(name, False, f"{type(error).__name__}: {error}")) failed = [item.name for item in output if not item.passed] if failed: raise AssertionError("functional checks failed: " + ", ".join(failed)) return tuple(output) def save_csv(results: tuple[ScenarioResult, ...]) -> None: OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) for path, cls, rows in ( (SUMMARY_CSV_PATH, SummaryMetrics, (item.summary for item in results)), (VIDEO_CSV_PATH, VideoMetrics, (item.video for item in results)), (CONTROL_CSV_PATH, ControlMetrics, (item.control for item in results)), (RECOVERY_CSV_PATH, RecoveryMetrics, (item.recovery for item in results)), ): with path.open("w", encoding="utf-8", newline="") as file: writer = csv.DictWriter(file, fieldnames=list(cls.__dataclass_fields__)) writer.writeheader() writer.writerows(asdict(row) for row in rows) def _step_results(results: tuple[ScenarioResult, ...]) -> list[ScenarioResult]: return [item for item in results if item.summary.scenario == "step_300_230_300"] def save_plots(results: tuple[ScenarioResult, ...]) -> None: step = _step_results(results) fig, axis = plt.subplots(figsize=(11, 5.5)) for item in step: axis.plot(item.age_timeline_seconds, item.age_timeline_ms, label=POLICY_BY_NAME[item.summary.policy].label) axis.axvline(120, color="black", ls="--", alpha=.4); axis.axvline(300, color="black", ls="--", alpha=.4) axis.set(xlabel="Время, с", ylabel="Возраст, мс", title="Возраст изображения: ступенчатый сценарий") axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(AGE_PLOT_PATH, dpi=150); plt.close(fig) fig, axis = plt.subplots(figsize=(11, 5.5)) for item in step: axis.plot(item.queue_timeline_seconds, np.asarray(item.queue_timeline_bytes) / 1_000_000.0, label=POLICY_BY_NAME[item.summary.policy].label) axis.axvline(120, color="black", ls="--", alpha=.4); axis.axvline(300, color="black", ls="--", alpha=.4) axis.set(xlabel="Время, с", ylabel="Очередь, Мбайт", title="Очередь во времени") axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(QUEUE_PLOT_PATH, dpi=150); plt.close(fig) variable = [item for item in results if item.summary.scenario == "variable_300_260_230"] fig, axis = plt.subplots(figsize=(10, 5.2)) minutes = np.arange(1, 11) for item in variable: axis.plot(minutes, item.minute_update_fps, marker="o", label=POLICY_BY_NAME[item.summary.policy].label) axis.set(xlabel="Минута", ylabel="Обновлений/с", title="Частота обновления по минутам") axis.set_xticks(minutes); axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(MINUTE_PLOT_PATH, dpi=150); plt.close(fig) fig, axis = plt.subplots(figsize=(10, 5.2)) for item in step: mask = np.asarray(item.age_timeline_seconds) >= 280.0 axis.plot(np.asarray(item.age_timeline_seconds)[mask], np.asarray(item.age_timeline_ms)[mask], label=POLICY_BY_NAME[item.summary.policy].label) axis.axhline(1000, color="orange", ls="--"); axis.axhline(500, color="green", ls=":"); axis.axvline(300, color="black", ls="--") axis.set(xlabel="Время, с", ylabel="Возраст, мс", title="Восстановление после возврата к 300 кбит/с") axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(RECOVERY_PLOT_PATH, dpi=150); plt.close(fig) labels = [f"{PROFILE_BY_NAME[item.summary.scenario].label}\n{POLICY_BY_NAME[item.summary.policy].label}" for item in results] x = np.arange(len(results)) fig, axis = plt.subplots(figsize=(16, 6)) axis.bar(x, [item.control.control_p95_delay_ms for item in results]) axis.set_xticks(x, labels, rotation=55, ha="right", fontsize=8) axis.set(ylabel="P95, мс", title="Задержка обычных команд"); axis.grid(axis="y", alpha=.3) fig.tight_layout(); fig.savefig(CONTROL_PLOT_PATH, dpi=150); plt.close(fig) fig, axis = plt.subplots(figsize=(16, 6)) width = .38 axis.bar(x - width / 2, [item.video.published_frames for item in results], width, label="Опубликовано") axis.bar(x + width / 2, [item.video.dropped_before_start_frames for item in results], width, label="Удалено") axis.set_xticks(x, labels, rotation=55, ha="right", fontsize=8) axis.set(ylabel="Кадров", title="Опубликованные и удалённые кадры"); axis.legend(); axis.grid(axis="y", alpha=.3) fig.tight_layout(); fig.savefig(OUTCOME_PLOT_PATH, dpi=150); plt.close(fig) constants = [item for item in results if item.summary.scenario.startswith("constant_") and item.summary.policy == "no_drop"] rates = [item.summary.mean_channel_kbps for item in constants] fig, axis = plt.subplots(figsize=(8.5, 5.2)) axis.plot(rates, [item.summary.theoretical_queue_growth_bytes_per_second for item in constants], marker="o", label="Теория") axis.plot(rates, [max(0.0, item.summary.actual_queue_growth_bytes_per_second) for item in constants], marker="s", label="Модель") axis.set(xlabel="Скорость, кбит/с", ylabel="Рост, байт/с", title="Теоретический и фактический рост очереди") axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(GROWTH_PLOT_PATH, dpi=150); plt.close(fig) fig, axis = plt.subplots(figsize=(9, 6)) for policy in POLICIES: rows = [item for item in results if item.summary.policy == policy.name] axis.scatter([item.video.actual_update_fps for item in rows], [item.video.p95_display_age_ms for item in rows], s=70, label=policy.label) axis.set(xlabel="Обновлений/с", ylabel="P95 возраста, мс", title="Сравнение политик") axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(COMPARISON_PLOT_PATH, dpi=150); plt.close(fig) def write_report( workload: LongWorkload, results: tuple[ScenarioResult, ...], tests: tuple[FunctionalTestResult, ...], ) -> None: git_status = subprocess.run( ("git", "status", "--short", "--branch"), check=True, capture_output=True, text=True, encoding="utf-8", ).stdout.rstrip() lines = [ "Lab036. Долговременная устойчивость видеопланировщика при изменении пропускной способности", "", "1. Исходное состояние и конфигурация", f"- Commit Lab035: {LAB035_COMMIT}.", "- Рабочее дерево перед Lab036 было чистым; ветка main.", "- Модель использует 600 с, 1800 кадров, повтор реальной последовательности размеров 63 кадров, BASE 240×135 Q23, ROI 320×180 Q33 и 3 составных кадра/с.", "- Lab036 использует ровно 3 кадра/с в течение 600 с; короткий ролик Lab034 и предшествующих лабораторных давал немного отличающуюся фактическую частоту формирования кадров.", "- Длинный опыт уменьшает влияние начального и конечного участков, поэтому значения общей предложенной нагрузки Lab034 и Lab036 не обязаны побайтно совпадать.", "- Каждый повтор получает новые composite_frame_id, времена, заголовки и CRC; JPEG и пакеты на диск не записываются.", "- Используются внутренний пакет Lab028 с payload 512 байт, FEC 12+3 с выравниванием по кадрам и общий пакет Lab033.", "- Команды 20 Гц, телеметрия 10 Гц; аварийные команды создаются в 60, 180, 310 и 480 с.", "- Ошибки радиоканала отсутствуют.", "- Смена скорости не прерывает начатый пакет: новая скорость применяется только к следующему пакету после завершения текущего.", "", "2. Пятнадцать сочетаний", "scenario | policy | published/drop | fps | age P95 ms | queue max packets | queue at 600 packets | drain s | control P95 ms | emergency max ms", ] for item in results: s, v, c = item.summary, item.video, item.control lines.append( f"{s.scenario} | {s.policy} | {v.published_frames}/{v.dropped_before_start_frames} | {v.actual_update_fps:.3f} | {v.p95_display_age_ms:.3f} | {s.max_queue_packets} | {s.queue_at_600s_packets} | {s.additional_drain_seconds:.3f} | {c.control_p95_delay_ms:.3f} | {c.emergency_max_delay_ms:.3f}" ) lines.extend(["", "3. Постоянные скорости: теория и модель", "rate | policy | reserve kbps | theory growth B/s | model growth B/s | minimum skip | stable fps | minute queue packets"]) for item in results: if not item.summary.scenario.startswith("constant_"): continue s = item.summary lines.append( f"{s.mean_channel_kbps:.0f} | {s.policy} | {s.speed_reserve_kbps:.3f} | {s.theoretical_queue_growth_bytes_per_second:.3f} | {s.actual_queue_growth_bytes_per_second:.3f} | {s.theoretical_minimum_skip_fraction:.6f} | {s.theoretical_sustainable_fps:.3f} | {s.queue_packets_each_minute}" ) lines.extend(["", "4. Восстановление после ухудшения", "policy | age at 300 ms | below 1000 s | below 500 s | excess queue clear s | dropped old | published old"]) for item in _step_results(results): r = item.recovery lines.append( f"{r.policy} | {r.image_age_at_300s_ms:.3f} | {r.time_to_age_below_1000ms_seconds:.3f} | {r.time_to_age_below_500ms_seconds:.3f} | {r.excess_queue_clear_seconds:.3f} | {r.frames_dropped_after_recovery} | {r.old_frames_published_after_recovery}" ) lines.extend(["", "5. Видео, очередь, команды и телеметрия"]) latest_rows = [item for item in results if item.summary.policy == "latest_only"] lines.extend( f"- {item.summary.scenario}: fps={item.video.actual_update_fps:.3f}, age P95={item.video.p95_display_age_ms:.3f} мс, queue max={item.summary.max_queue_packets}, waiting max={item.summary.max_waiting_video_frames}, control P95/max={item.control.control_p95_delay_ms:.3f}/{item.control.control_max_delay_ms:.3f} мс, emergency={item.control.emergency_delays_ms} мс, telemetry P95={item.control.telemetry_p95_delay_ms:.3f} мс." for item in latest_rows ) lines.extend([ "- Политика самого свежего кадра ограничивает ожидающую видеоочередь одним кадром и не передаёт накопленную историю после восстановления.", "- Политика двух ожидающих кадров ограничивает очередь двумя кадрами, но увеличивает возраст изображения относительно основной политики.", "- Без удаления при дефиците скорости очередь растёт; при 300 кбит/с запас достаточен.", "- У упреждающих политик нет бесполезно переданных видеобайтов: начатый кадр всегда завершается.", "", "6. Функциональные проверки", ]) lines.extend(f"- {'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}" for item in tests) lines.extend([ f"- CRC-проверено общих пакетов: {workload.crc_packets_checked}; FEC-блоков: {workload.crc_blocks_checked}.", "", "7. Созданные файлы", "- tests/lab036_long_duration_scheduler.py", "- data/processed/lab036/lab036_summary.csv", "- data/processed/lab036/lab036_video_metrics.csv", "- data/processed/lab036/lab036_control_metrics.csv", "- data/processed/lab036/lab036_recovery_metrics.csv", "- data/processed/lab036/lab036_report.txt", ]) lines.extend(f"- {path.as_posix()}" for path in PLOT_PATHS) lines.extend(["", "8. Итоговый Git status", "- Lab036 не добавлена в индекс и не закоммичена.", "", git_status]) REPORT_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") def validate_outputs() -> None: for path in (SUMMARY_CSV_PATH, VIDEO_CSV_PATH, CONTROL_CSV_PATH, RECOVERY_CSV_PATH): with path.open(encoding="utf-8", newline="") as file: rows = list(csv.DictReader(file)) if len(rows) != 15: raise AssertionError(f"{path} must contain 15 result rows") if "Lab036" not in REPORT_PATH.read_text(encoding="utf-8"): raise AssertionError("invalid Lab036 report") for path in PLOT_PATHS: image = cv2.imread(str(path), cv2.IMREAD_UNCHANGED) if image is None or image.size == 0: raise AssertionError(f"OpenCV could not read {path}") def main() -> None: print("Lab036: building 600-second workload") workload = build_long_workload() print("Lab036: running 15 combinations") results = run_experiment(workload) tests = run_functional_tests(workload, results) save_csv(results) save_plots(results) write_report(workload, results, tests) validate_outputs() print(f"Lab036 complete: {len(results)} combinations, {len(tests)} checks") if __name__ == "__main__": main()