"""Lab038: lossy full link with rover watchdog and acknowledged emergency stop.""" from __future__ import annotations import csv from dataclasses import asdict, dataclass from enum import IntEnum from pathlib import Path from typing import Iterable import matplotlib import numpy as np matplotlib.use("Agg") import matplotlib.pyplot as plt from protocol.control_failsafe import ( SAFE_STATE, ControlState, RoverControlFailsafe, decode_control_state, encode_control_state, ) from protocol.emergency_ack import ( EmergencyAckReceiver, EmergencyIdentity, STREAM_EMERGENCY_ACK, acknowledged_identity, build_emergency_ack, ) from protocol.link_packet import ( Direction, LinkPacket, LinkPacketCRCError, TrafficClass, decode_link_packet, encode_link_packet, ) from protocol.packet_erasure_fec import ( InsufficientSymbolsError, OuterPacketCRCError, decode_fec_block, decode_outer_symbol, ) from protocol.video_packet import decode_packet as decode_inner_packet from experiments.lab033_priority_channel_scheduler import ( STREAM_CONTROL, STREAM_EMERGENCY, deterministic_payload, ) from experiments.lab037_lossy_full_link import ( BAD_TIME_FRACTION, CHANNEL_RATES_KBPS, DURATION_SECONDS, EMERGENCY_TIMES_US, FRAME_COUNT, MASTER_SEED, MEAN_BAD_DURATIONS_MS, REPETITIONS, Frame as Lab037Frame, Workload as Lab037Workload, bad_intervals, build_workload as build_lab037_workload, ) OUTPUT_DIRECTORY = Path("data/processed/lab038") SUMMARY_CSV_PATH = OUTPUT_DIRECTORY / "lab038_summary.csv" COMMAND_CSV_PATH = OUTPUT_DIRECTORY / "lab038_command_metrics.csv" EMERGENCY_CSV_PATH = OUTPUT_DIRECTORY / "lab038_emergency_metrics.csv" FAILSAFE_CSV_PATH = OUTPUT_DIRECTORY / "lab038_failsafe_metrics.csv" VIDEO_CSV_PATH = OUTPUT_DIRECTORY / "lab038_video_metrics.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab038_report.txt" COMMAND_GAP_PLOT = OUTPUT_DIRECTORY / "lab038_command_gap.png" WATCHDOG_TRIGGER_PLOT = OUTPUT_DIRECTORY / "lab038_watchdog_triggers.png" SAFE_STOP_PLOT = OUTPUT_DIRECTORY / "lab038_safe_stop_time.png" EMERGENCY_PLOT = OUTPUT_DIRECTORY / "lab038_emergency_timeliness.png" ACK_PLOT = OUTPUT_DIRECTORY / "lab038_ack_time.png" LOAD_PLOT = OUTPUT_DIRECTORY / "lab038_additional_load.png" VIDEO_PLOT = OUTPUT_DIRECTORY / "lab038_video_impact.png" COMPARISON_PLOT = OUTPUT_DIRECTORY / "lab038_mode_comparison.png" PLOT_PATHS = ( COMMAND_GAP_PLOT, WATCHDOG_TRIGGER_PLOT, SAFE_STOP_PLOT, EMERGENCY_PLOT, ACK_PLOT, LOAD_PLOT, VIDEO_PLOT, COMPARISON_PLOT, ) CONTROL_PERIOD_US = 50_000 TELEMETRY_PERIOD_US = 100_000 CONTROL_REPEAT_DELAY_US = 22_500 WATCHDOG_THRESHOLDS_MS = (100.0, 150.0, 250.0, 500.0) EMERGENCY_SPREAD_US = tuple(range(0, 500_001, 50_000)) TIME_EPSILON = 1e-12 SPEED_25_KMH_MPS = 25_000.0 / 3600.0 LAB038_SEED_OFFSET = 380_000_000 class ProtectionMode(IntEnum): SINGLE = 1 LAB037 = 2 SPREAD_EMERGENCY = 3 ACKNOWLEDGED = 4 MODE_NAMES = { ProtectionMode.SINGLE: "single", ProtectionMode.LAB037: "lab037_repetition", ProtectionMode.SPREAD_EMERGENCY: "spread_emergency", ProtectionMode.ACKNOWLEDGED: "acknowledged_emergency", } MODE_LABELS = { "single": "Одиночная передача", "lab037_repetition": "Повторение Lab037", "spread_emergency": "Разнесённая аварийная команда", "acknowledged_emergency": "Аварийная команда с подтверждением", } def control_offsets_us(mode: ProtectionMode) -> tuple[int, ...]: return (0, CONTROL_REPEAT_DELAY_US) if mode is ProtectionMode.LAB037 else (0,) def emergency_offsets_us(mode: ProtectionMode) -> tuple[int, ...]: if mode is ProtectionMode.SINGLE: return (0,) if mode is ProtectionMode.LAB037: return (0, 15_000, 30_000) return EMERGENCY_SPREAD_US @dataclass(frozen=True) class Unit: packet: LinkPacket wire_packet: bytes available_time_us: int order: int priority: int kind: str copy_index: int = 0 event_id: int | None = None frame_id: int | None = None block_id: int | None = None symbol_index: int | None = None source_count: int = 0 is_source: bool = False @property def available_seconds(self) -> float: return self.available_time_us / 1_000_000.0 @property def size(self) -> int: return len(self.wire_packet) @dataclass(frozen=True) class Frame: frame_id: int generation_time_us: int packets: tuple[Unit, ...] jpeg_bytes: int @property def generation_seconds(self) -> float: return self.generation_time_us / 1_000_000.0 @dataclass(frozen=True) class Workload: frames: tuple[Frame, ...] controls: tuple[LinkPacket, ...] telemetry: tuple[Unit, ...] emergencies: tuple[LinkPacket, ...] video_generated_bytes: int crc_packets_checked: int fec_blocks_checked: int @dataclass(frozen=True) class SentRecord: unit: Unit start: float end: float lost: bool @dataclass(frozen=True) class EmergencyTrial: delivered: bool first_delay_ms: float | None copies_transmitted: int copies_lost: int received_duplicates: int first_ack_delay_ms: float | None acknowledgements_lost: int repeated_acknowledgements: int stopped_after_ack: bool ack_generated: int @dataclass(frozen=True) class WatchdogTrial: threshold_ms: float triggers: int safe_time_seconds: float durations_ms: tuple[float, ...] max_stale_motion_ms: float recovered_before_trigger: int channel_loss_triggers: int queue_delay_triggers: int bad_start_to_stop_ms: tuple[float, ...] missed_stop_cases: int stale_release_cases: int emergency_release_cases: int @dataclass(frozen=True) class Trial: primary_transmitted: int repeats_transmitted: int delivered_sequences: frozenset[int] duplicates_suppressed: int cancelled_repeats: int command_ages_ms: tuple[float, ...] command_gaps_ms: tuple[float, ...] command_outcomes: tuple[tuple[int, int, bool], ...] emergency: tuple[EmergencyTrial, ...] watchdog: tuple[WatchdogTrial, ...] created_frames: int published_frames: int dropped_frames: int image_ages_ms: tuple[float, ...] image_gaps_ms: tuple[float, ...] recovered_blocks: int unrecoverable_blocks: int useful_video_bytes: int telemetry_delivered: int telemetry_max_gap_ms: float offered_bytes: tuple[tuple[str, int], ...] transmitted_bytes: tuple[tuple[str, int], ...] lost_packets: int lost_bytes: int mean_queue_packets: float max_queue_packets: int mean_queue_bytes: float max_queue_bytes: int max_waiting_frames: int drain_end_seconds: float nonpreemptive_ok: bool priority_ok: bool @dataclass(frozen=True) class SummaryMetrics: channel_kbps: float mean_bad_duration_ms: float protection_mode: str repetitions: int emergency_load_kbps: float acknowledgement_load_kbps: float control_load_kbps: float telemetry_load_kbps: float video_load_kbps: float total_offered_load_kbps: float control_repeat_load_kbps: float emergency_copy_load_kbps: float channel_utilization_percent: float mean_queue_packets: float max_queue_packets: int mean_queue_bytes: float max_queue_bytes: int max_waiting_video_frames: int queue_release_seconds: float lost_packets_mean: float lost_bytes_mean: float telemetry_delivered_fraction: float telemetry_max_gap_ms: float @dataclass(frozen=True) class CommandMetrics: channel_kbps: float mean_bad_duration_ms: float protection_mode: str created_commands: int primary_copies_transmitted_mean: float repeat_copies_transmitted_mean: float delivered_unique_states_mean: float states_all_copies_lost_mean: float undelivered_state_fraction: float suppressed_duplicates_mean: float cancelled_pending_repeats_mean: float mean_age_ms: float p95_age_ms: float max_age_ms: float mean_gap_ms: float max_gap_ms: float gaps_over_100ms_mean: float gaps_over_150ms_mean: float gaps_over_250ms_mean: float gaps_over_500ms_mean: float max_consecutive_undelivered_states: int repetition_overhead_kbps: float @dataclass(frozen=True) class EmergencyMetrics: channel_kbps: float mean_bad_duration_ms: float protection_mode: str event_time_seconds: float delivered_fraction: float mean_first_delivery_ms: float p95_first_delivery_ms: float within_50ms_fraction: float within_100ms_fraction: float within_250ms_fraction: float transmitted_copies_mean: float lost_copies_mean: float received_duplicates_mean: float ack_delivered_fraction: float mean_first_ack_ms: float p95_first_ack_ms: float lost_acknowledgements_mean: float repeated_acknowledgements_mean: float repetition_stopped_after_ack_fraction: float @dataclass(frozen=True) class FailsafeMetrics: channel_kbps: float mean_bad_duration_ms: float protection_mode: str watchdog_threshold_ms: float triggers_mean: float triggers_per_minute: float safe_time_fraction: float mean_safe_duration_ms: float p95_safe_duration_ms: float max_safe_duration_ms: float max_stale_motion_ms: float recovered_before_trigger_mean: float channel_loss_triggers_mean: float queue_delay_triggers_mean: float mean_bad_start_to_stop_ms: float max_bad_start_to_stop_ms: float distance_before_stop_at_25kmh_m: float missed_stop_cases: int stale_command_release_cases: int ordinary_command_emergency_release_cases: int @dataclass(frozen=True) class VideoMetrics: channel_kbps: float mean_bad_duration_ms: float protection_mode: str created_frames: int published_frames_mean: float dropped_before_start_frames_mean: float published_fraction: float update_fps: float mean_image_age_ms: float p95_image_age_ms: float max_image_age_ms: float mean_no_new_image_ms: float max_no_new_image_ms: float recovered_fec_blocks_mean: float unrecoverable_fec_blocks_mean: float delivered_useful_video_kbps: float @dataclass class Evidence: ack_loss_continued: bool = False ack_stopped_future: bool = False duplicate_generated_ack: bool = False nonpreemptive: bool = True priority: bool = True repeat_bytes_seen: bool = False ack_bytes_seen: bool = False max_waiting_frames: int = 0 @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( traffic_class: TrafficClass, direction: Direction, stream_id: int, sequence: int, generation_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_us, deadline_ms=deadline_ms, payload=payload, ) def build_workload() -> Workload: base: Lab037Workload = build_lab037_workload() frames: list[Frame] = [] for old_frame in base.frames: packets = tuple( Unit( packet=item.packet, wire_packet=item.wire_packet, available_time_us=item.available_time_us, order=item.arrival_order, priority=5, kind="video", frame_id=item.frame_id, block_id=item.block_id, symbol_index=item.symbol_index, source_count=item.source_count, is_source=item.is_source, ) for item in old_frame.packets ) frames.append(Frame(old_frame.frame_id, old_frame.generation_time_us, packets, old_frame.jpeg_bytes)) controls = [] for sequence, generation_us in enumerate(range(0, int(DURATION_SECONDS * 1_000_000), CONTROL_PERIOD_US)): state = ControlState( desired_speed_mps=SPEED_25_KMH_MPS, desired_turn=((sequence % 41) - 20) / 20.0, braking=False, movement_allowed=True, ) packet = _link( TrafficClass.CONTROL, Direction.GROUND_TO_ROVER, STREAM_CONTROL, sequence, generation_us, 100, encode_control_state(state), ) decode_control_state(decode_link_packet(encode_link_packet(packet)).payload) controls.append(packet) emergencies = tuple( _link( TrafficClass.EMERGENCY, Direction.GROUND_TO_ROVER, STREAM_EMERGENCY, sequence, generation_us, 50, deterministic_payload(b"E-STOP", sequence, 32), ) for sequence, generation_us in enumerate(EMERGENCY_TIMES_US) ) telemetry = tuple( Unit( item.packet, item.wire_packet, item.available_time_us, item.arrival_order + 20_000_000, 4, "telemetry", ) for item in base.telemetry ) for packet in emergencies: decode_link_packet(encode_link_packet(packet)) return Workload( tuple(frames), tuple(controls), telemetry, emergencies, sum(unit.size for frame in frames for unit in frame.packets), base.crc_packets_checked + len(controls) + len(emergencies), base.crc_blocks_checked, ) def static_units(workload: Workload, mode: ProtectionMode) -> tuple[Unit, ...]: units: list[Unit] = list(workload.telemetry) order = 30_000_000 for packet in workload.controls: for copy_index, offset in enumerate(control_offsets_us(mode)): units.append( Unit( packet, encode_link_packet(packet), packet.generation_time_us + offset, order, 3, "control_repeat" if copy_index else "control", copy_index=copy_index, ) ) order += 1 for event_id, packet in enumerate(workload.emergencies): for copy_index, offset in enumerate(emergency_offsets_us(mode)): units.append( Unit( packet, encode_link_packet(packet), packet.generation_time_us + offset, order, 1, "emergency", copy_index=copy_index, event_id=event_id, ) ) order += 1 return tuple(sorted(units, key=lambda item: (item.available_time_us, item.priority, item.order))) def packet_is_lost(start: float, end: float, bad_starts: np.ndarray, bad_ends: np.ndarray) -> bool: index = int(np.searchsorted(bad_ends, start, side="right")) return index < len(bad_starts) and bad_starts[index] < end - TIME_EPSILON def positive_run_max(flags: Iterable[bool]) -> int: current = best = 0 for flag in flags: current = current + 1 if flag else 0 best = max(best, current) return best def queue_statistics(intervals: list[tuple[float, float, int]]) -> tuple[float, int, float, int]: events: dict[float, list[int]] = {} packet_area = byte_area = 0.0 for start, end, size in intervals: end = min(end, DURATION_SECONDS) if end <= start + TIME_EPSILON: continue packet_area += end - start byte_area += (end - start) * size events.setdefault(start, [0, 0]) events.setdefault(end, [0, 0]) events[start][0] += 1 events[start][1] += size events[end][0] -= 1 events[end][1] -= size packets = bytes_now = max_packets = max_bytes = 0 for moment in sorted(events): packets += events[moment][0] bytes_now += events[moment][1] max_packets = max(max_packets, packets) max_bytes = max(max_bytes, bytes_now) return packet_area / DURATION_SECONDS, max_packets, byte_area / DURATION_SECONDS, max_bytes def watchdog_trials( receive_events: list[tuple[float, int]], outcomes: list[tuple[int, int, bool]], bad_starts: np.ndarray, bad_ends: np.ndarray, ) -> tuple[WatchdogTrial, ...]: rows = [] ordered = sorted(receive_events) outcome_by_sequence = {sequence: (lost_count, delivered) for sequence, lost_count, delivered in outcomes} for threshold_ms in WATCHDOG_THRESHOLDS_MS: threshold = threshold_ms / 1000.0 points = [(0.0, -1), *ordered, (DURATION_SECONDS, len(outcomes))] durations: list[float] = [] bad_to_stop: list[float] = [] recovered_before = channel_triggers = queue_triggers = missed = 0 max_stale = 0.0 for (start, previous_sequence), (end, next_sequence) in zip(points, points[1:]): gap = max(0.0, end - start) max_stale = max(max_stale, min(gap, threshold)) missing_sequences = range(previous_sequence + 1, min(next_sequence, len(outcomes))) missing = [outcome_by_sequence.get(sequence, (0, False)) for sequence in missing_sequences] communication_gap = next_sequence - previous_sequence > 1 if gap + TIME_EPSILON >= threshold: stop = start + threshold durations.append(max(0.0, end - stop) * 1000.0) if any(lost_count > 0 for lost_count, delivered in missing if not delivered): channel_triggers += 1 else: queue_triggers += 1 candidates = [ float(bad_start) for bad_start, bad_end in zip(bad_starts, bad_ends) if bad_start <= stop + TIME_EPSILON and bad_end >= start - TIME_EPSILON ] if candidates: bad_to_stop.append(max(0.0, stop - max(candidates)) * 1000.0) elif communication_gap: recovered_before += 1 if gap > threshold + TIME_EPSILON and not durations: missed += 1 rows.append( WatchdogTrial( threshold_ms, len(durations), sum(durations) / 1000.0, tuple(durations), max_stale * 1000.0, recovered_before, channel_triggers, queue_triggers, tuple(bad_to_stop), missed, 0, 0, ) ) return tuple(rows) def simulate_trial( workload: Workload, mode: ProtectionMode, rate_kbps: float, bad_starts: np.ndarray, bad_ends: np.ndarray, ) -> Trial: static = static_units(workload, mode) ready: list[Unit] = [] pending_frames: list[Frame] = [] active: Frame | None = None active_index = static_index = frame_index = 0 cursor = 0.0 dynamic_order = 50_000_000 latest_control_seen = -1 ack_sequence = 0 acked_events: set[int] = set() stopped_copies = [0, 0, 0] cancelled_repeats = 0 max_waiting = 0 intervals: list[tuple[float, float, int]] = [] sent_records: list[SentRecord] = [] dropped_frames: set[int] = set() started_frames: list[int] = [] completed_frames: list[int] = [] offered = {kind: 0 for kind in ("emergency", "ack", "control", "control_repeat", "telemetry", "video")} transmitted = dict(offered) offered["video"] = workload.video_generated_bytes offered["control"] = sum(len(encode_link_packet(packet)) for packet in workload.controls) if mode is ProtectionMode.LAB037: offered["control_repeat"] = offered["control"] offered["telemetry"] = sum(unit.size for unit in workload.telemetry) rover = RoverControlFailsafe(500_000) ack_receiver = EmergencyAckReceiver() command_receive_events: list[tuple[float, int]] = [] command_ages: list[float] = [] command_tx = [0] * len(workload.controls) command_lost = [0] * len(workload.controls) command_delivered = [False] * len(workload.controls) primary_transmitted = repeats_transmitted = 0 emergency_tx = [0, 0, 0] emergency_lost = [0, 0, 0] emergency_duplicates = [0, 0, 0] emergency_first: list[float | None] = [None, None, None] ack_first: list[float | None] = [None, None, None] ack_lost = [0, 0, 0] ack_generated = [0, 0, 0] block_times: dict[int, list[float]] = {} block_source_success: dict[int, set[int]] = {} block_source_count: dict[int, int] = {} frame_blocks: dict[int, set[int]] = {} telemetry_times: list[float] = [] lost_packets = lost_bytes = 0 nonpreemptive_ok = priority_ok = True def remove_ready(predicate, now: float) -> list[Unit]: removed: list[Unit] = [] retained: list[Unit] = [] for item in ready: if predicate(item): removed.append(item) intervals.append((item.available_seconds, now, item.size)) else: retained.append(item) ready[:] = retained return removed def admit(now: float) -> None: nonlocal static_index, frame_index, latest_control_seen, cancelled_repeats, max_waiting while static_index < len(static) and static[static_index].available_seconds <= now + TIME_EPSILON: item = static[static_index] static_index += 1 if item.kind == "emergency": assert item.event_id is not None if mode is ProtectionMode.ACKNOWLEDGED and item.event_id in acked_events: stopped_copies[item.event_id] += 1 continue offered["emergency"] += item.size elif item.kind in ("control", "control_repeat"): sequence = item.packet.sequence_number if item.kind == "control": latest_control_seen = max(latest_control_seen, sequence) removed = remove_ready( lambda old: old.kind in ("control", "control_repeat") and old.packet.sequence_number < sequence, item.available_seconds, ) cancelled_repeats += sum(old.kind == "control_repeat" for old in removed) elif sequence < latest_control_seen: cancelled_repeats += 1 continue elif item.kind == "telemetry": remove_ready(lambda old: old.kind == "telemetry", item.available_seconds) ready.append(item) while frame_index < len(workload.frames) and workload.frames[frame_index].generation_seconds <= now + TIME_EPSILON: frame = workload.frames[frame_index] frame_index += 1 for old in pending_frames: dropped_frames.add(old.frame_id) intervals.extend((unit.available_seconds, frame.generation_seconds, unit.size) for unit in old.packets) pending_frames[:] = [frame] max_waiting = max(max_waiting, len(pending_frames)) while static_index < len(static) or frame_index < len(workload.frames) or ready or pending_frames or active is not None: if not ready and not pending_frames and active is None: next_times = [] if static_index < len(static): next_times.append(static[static_index].available_seconds) if frame_index < len(workload.frames): next_times.append(workload.frames[frame_index].generation_seconds) cursor = max(cursor, min(next_times)) admit(cursor) if ready: best = min(ready, key=lambda item: (item.priority, item.order)) priority_ok &= best.priority == min(item.priority for item in ready) ready.remove(best) unit = best else: if active is None and pending_frames: active = pending_frames.pop(0) active_index = 0 started_frames.append(active.frame_id) if active is None: continue unit = active.packets[active_index] start = max(cursor, unit.available_seconds) end = start + unit.size * 8.0 / (rate_kbps * 1000.0) if sent_records: nonpreemptive_ok &= sent_records[-1].end <= start + TIME_EPSILON lost = packet_is_lost(start, end, bad_starts, bad_ends) sent_records.append(SentRecord(unit, start, end, lost)) intervals.append((unit.available_seconds, end, unit.size)) transmitted[unit.kind] += unit.size if lost: lost_packets += 1 lost_bytes += unit.size cursor = end admit(cursor) if unit.kind in ("control", "control_repeat"): sequence = unit.packet.sequence_number command_tx[sequence] += 1 primary_transmitted += int(unit.kind == "control") repeats_transmitted += int(unit.kind == "control_repeat") if lost: command_lost[sequence] += 1 else: receive_us = int(round(end * 1_000_000.0)) if rover.receive_state(unit.packet, receive_us): command_delivered[sequence] = True command_receive_events.append((end, sequence)) command_ages.append((end - unit.packet.generation_time_us / 1_000_000.0) * 1000.0) elif unit.kind == "emergency": assert unit.event_id is not None event = unit.event_id emergency_tx[event] += 1 if lost: emergency_lost[event] += 1 else: if emergency_first[event] is None: emergency_first[event] = end else: emergency_duplicates[event] += 1 rover.receive_emergency(unit.packet) if mode is ProtectionMode.ACKNOWLEDGED: ack_packet = build_emergency_ack(unit.packet, ack_sequence, int(round(end * 1_000_000.0))) ack_sequence += 1 ack_generated[event] += 1 ack_unit = Unit( ack_packet, encode_link_packet(ack_packet), int(round(end * 1_000_000.0)), dynamic_order, 2, "ack", copy_index=ack_generated[event] - 1, event_id=event, ) dynamic_order += 1 offered["ack"] += ack_unit.size ready.append(ack_unit) elif unit.kind == "ack": assert unit.event_id is not None event = unit.event_id if lost: ack_lost[event] += 1 else: identity = acknowledged_identity(unit.packet) assert identity == EmergencyIdentity(STREAM_EMERGENCY, event) first_ack = ack_receiver.accept(unit.packet) if first_ack: acked_events.add(event) ack_first[event] = end removed = remove_ready( lambda old: old.kind == "emergency" and old.event_id == event, end, ) stopped_copies[event] += len(removed) elif unit.kind == "telemetry": if not lost: telemetry_times.append(end) elif unit.kind == "video": assert unit.block_id is not None and unit.frame_id is not None and unit.symbol_index is not None block_source_count[unit.block_id] = unit.source_count frame_blocks.setdefault(unit.frame_id, set()).add(unit.block_id) if not lost: block_times.setdefault(unit.block_id, []).append(end) if unit.is_source: block_source_success.setdefault(unit.block_id, set()).add(unit.symbol_index) active_index += 1 assert active is not None if active_index == len(active.packets): completed_frames.append(active.frame_id) active = None active_index = 0 command_receive_events.sort() command_gaps = np.diff(np.asarray([0.0] + [time for time, _ in command_receive_events] + [DURATION_SECONDS])) * 1000.0 outcomes = [(sequence, command_lost[sequence], command_delivered[sequence]) for sequence in range(len(workload.controls))] watchdog = watchdog_trials(command_receive_events, outcomes, bad_starts, bad_ends) emergency_rows = [] for event, event_us in enumerate(EMERGENCY_TIMES_US): first = emergency_first[event] first_ack = ack_first[event] emergency_rows.append( EmergencyTrial( first is not None, (first - event_us / 1_000_000.0) * 1000.0 if first is not None else None, emergency_tx[event], emergency_lost[event], emergency_duplicates[event], (first_ack - event_us / 1_000_000.0) * 1000.0 if first_ack is not None else None, ack_lost[event], max(0, ack_generated[event] - 1), first_ack is not None and stopped_copies[event] > 0, ack_generated[event], ) ) recovered_blocks = unrecoverable_blocks = useful_video_bytes = 0 publications: list[tuple[float, int]] = [] for frame_id in completed_frames: ids = frame_blocks.get(frame_id, set()) frame_ok = bool(ids) completion = 0.0 for block_id in ids: times = sorted(block_times.get(block_id, [])) source_count = block_source_count[block_id] if len(times) >= source_count: completion = max(completion, times[source_count - 1]) if len(block_source_success.get(block_id, set())) < source_count: recovered_blocks += 1 else: unrecoverable_blocks += 1 frame_ok = False if frame_ok: publications.append((completion, frame_id)) useful_video_bytes += workload.frames[frame_id].jpeg_bytes publications.sort() image_ages = tuple( (time - workload.frames[frame_id].generation_seconds) * 1000.0 for time, frame_id in publications ) image_gaps = tuple( np.diff(np.asarray([0.0] + [time for time, _ in publications] + [DURATION_SECONDS])) * 1000.0 ) telemetry_gaps = np.diff(np.asarray([0.0] + telemetry_times + [DURATION_SECONDS])) * 1000.0 mean_q, max_q, mean_q_bytes, max_q_bytes = queue_statistics(intervals) return Trial( primary_transmitted, repeats_transmitted, frozenset(sequence for sequence, delivered in enumerate(command_delivered) if delivered), rover.stale_commands, cancelled_repeats, tuple(command_ages), tuple(command_gaps), tuple(outcomes), tuple(emergency_rows), watchdog, FRAME_COUNT, len(publications), len(dropped_frames), image_ages, image_gaps, recovered_blocks, unrecoverable_blocks, useful_video_bytes, len(telemetry_times), max(telemetry_gaps, default=0.0), tuple(sorted(offered.items())), tuple(sorted(transmitted.items())), lost_packets, lost_bytes, mean_q, max_q, mean_q_bytes, max_q_bytes, max_waiting, sent_records[-1].end if sent_records else 0.0, nonpreemptive_ok, priority_ok, ) def aggregate_condition( workload: Workload, mode: ProtectionMode, rate: float, mean_bad_ms: float, trials: tuple[Trial, ...], ) -> tuple[SummaryMetrics, CommandMetrics, tuple[EmergencyMetrics, ...], tuple[FailsafeMetrics, ...], VideoMetrics]: mode_name = MODE_NAMES[mode] def mean_value(values: Iterable[float]) -> float: values = tuple(values) return float(np.mean(values)) if values else 0.0 offered = { kind: mean_value(dict(trial.offered_bytes)[kind] for trial in trials) for kind in dict(trials[0].offered_bytes) } transmitted_total = mean_value(sum(dict(trial.transmitted_bytes).values()) for trial in trials) summary = SummaryMetrics( rate, mean_bad_ms, mode_name, REPETITIONS, offered["emergency"] * 8.0 / DURATION_SECONDS / 1000.0, offered["ack"] * 8.0 / DURATION_SECONDS / 1000.0, offered["control"] * 8.0 / DURATION_SECONDS / 1000.0, offered["telemetry"] * 8.0 / DURATION_SECONDS / 1000.0, offered["video"] * 8.0 / DURATION_SECONDS / 1000.0, sum(offered.values()) * 8.0 / DURATION_SECONDS / 1000.0, offered["control_repeat"] * 8.0 / DURATION_SECONDS / 1000.0, max(0.0, offered["emergency"] - sum(len(encode_link_packet(packet)) for packet in workload.emergencies)) * 8.0 / DURATION_SECONDS / 1000.0, transmitted_total * 8.0 / (rate * 1000.0 * DURATION_SECONDS) * 100.0, mean_value(trial.mean_queue_packets for trial in trials), max(trial.max_queue_packets for trial in trials), mean_value(trial.mean_queue_bytes for trial in trials), max(trial.max_queue_bytes for trial in trials), max(trial.max_waiting_frames for trial in trials), max(0.0, max(trial.drain_end_seconds for trial in trials) - DURATION_SECONDS), mean_value(trial.lost_packets for trial in trials), mean_value(trial.lost_bytes for trial in trials), sum(trial.telemetry_delivered for trial in trials) / (REPETITIONS * len(workload.telemetry)), max(trial.telemetry_max_gap_ms for trial in trials), ) ages = tuple(value for trial in trials for value in trial.command_ages_ms) gaps = tuple(value for trial in trials for value in trial.command_gaps_ms) delivered_total = sum(len(trial.delivered_sequences) for trial in trials) missing_total = REPETITIONS * len(workload.controls) - delivered_total max_run = 0 for trial in trials: flags = [sequence not in trial.delivered_sequences for sequence in range(len(workload.controls))] max_run = max(max_run, positive_run_max(flags)) original_control_bytes = sum(len(encode_link_packet(packet)) for packet in workload.controls) command = CommandMetrics( rate, mean_bad_ms, mode_name, len(workload.controls), mean_value(trial.primary_transmitted for trial in trials), mean_value(trial.repeats_transmitted for trial in trials), delivered_total / REPETITIONS, missing_total / REPETITIONS, missing_total / (REPETITIONS * len(workload.controls)), mean_value(trial.duplicates_suppressed for trial in trials), mean_value(trial.cancelled_repeats for trial in trials), mean_value(ages), percentile(ages, 95), max(ages, default=0.0), mean_value(gaps), max(gaps, default=0.0), sum(value > 100.0 + TIME_EPSILON for value in gaps) / REPETITIONS, sum(value > 150.0 + TIME_EPSILON for value in gaps) / REPETITIONS, sum(value > 250.0 + TIME_EPSILON for value in gaps) / REPETITIONS, sum(value > 500.0 + TIME_EPSILON for value in gaps) / REPETITIONS, max_run, max(0.0, offered["control"] + offered["control_repeat"] - original_control_bytes) * 8.0 / DURATION_SECONDS / 1000.0, ) emergency_rows = [] for event, event_us in enumerate(EMERGENCY_TIMES_US): rows = [trial.emergency[event] for trial in trials] delays = tuple(row.first_delay_ms for row in rows if row.first_delay_ms is not None) ack_delays = tuple(row.first_ack_delay_ms for row in rows if row.first_ack_delay_ms is not None) emergency_rows.append( EmergencyMetrics( rate, mean_bad_ms, mode_name, event_us / 1_000_000.0, len(delays) / REPETITIONS, mean_value(delays), percentile(delays, 95), sum(delay <= 50.0 + TIME_EPSILON for delay in delays) / REPETITIONS, sum(delay <= 100.0 + TIME_EPSILON for delay in delays) / REPETITIONS, sum(delay <= 250.0 + TIME_EPSILON for delay in delays) / REPETITIONS, mean_value(row.copies_transmitted for row in rows), mean_value(row.copies_lost for row in rows), mean_value(row.received_duplicates for row in rows), len(ack_delays) / REPETITIONS, mean_value(ack_delays), percentile(ack_delays, 95), mean_value(row.acknowledgements_lost for row in rows), mean_value(row.repeated_acknowledgements for row in rows), sum(row.stopped_after_ack for row in rows) / REPETITIONS, ) ) failsafe_rows = [] for threshold_index, threshold_ms in enumerate(WATCHDOG_THRESHOLDS_MS): rows = [trial.watchdog[threshold_index] for trial in trials] durations = tuple(value for row in rows for value in row.durations_ms) bad_to_stop = tuple(value for row in rows for value in row.bad_start_to_stop_ms) failsafe_rows.append( FailsafeMetrics( rate, mean_bad_ms, mode_name, threshold_ms, mean_value(row.triggers for row in rows), mean_value(row.triggers for row in rows) / (DURATION_SECONDS / 60.0), sum(row.safe_time_seconds for row in rows) / (REPETITIONS * DURATION_SECONDS), mean_value(durations), percentile(durations, 95), max(durations, default=0.0), max(row.max_stale_motion_ms for row in rows), mean_value(row.recovered_before_trigger for row in rows), mean_value(row.channel_loss_triggers for row in rows), mean_value(row.queue_delay_triggers for row in rows), mean_value(bad_to_stop), max(bad_to_stop, default=0.0), SPEED_25_KMH_MPS * threshold_ms / 1000.0, sum(row.missed_stop_cases for row in rows), sum(row.stale_release_cases for row in rows), sum(row.emergency_release_cases for row in rows), ) ) image_ages = tuple(value for trial in trials for value in trial.image_ages_ms) image_gaps = tuple(value for trial in trials for value in trial.image_gaps_ms) published = sum(trial.published_frames for trial in trials) video = VideoMetrics( rate, mean_bad_ms, mode_name, FRAME_COUNT, published / REPETITIONS, mean_value(trial.dropped_frames for trial in trials), published / (REPETITIONS * FRAME_COUNT), published / (REPETITIONS * DURATION_SECONDS), mean_value(image_ages), percentile(image_ages, 95), max(image_ages, default=0.0), mean_value(image_gaps), max(image_gaps, default=0.0), mean_value(trial.recovered_blocks for trial in trials), mean_value(trial.unrecoverable_blocks for trial in trials), sum(trial.useful_video_bytes for trial in trials) * 8.0 / (REPETITIONS * DURATION_SECONDS * 1000.0), ) return summary, command, tuple(emergency_rows), tuple(failsafe_rows), video def run_experiment(workload: Workload) -> tuple[ tuple[SummaryMetrics, ...], tuple[CommandMetrics, ...], tuple[EmergencyMetrics, ...], tuple[FailsafeMetrics, ...], tuple[VideoMetrics, ...], Evidence, ]: summaries: list[SummaryMetrics] = [] commands: list[CommandMetrics] = [] emergencies: list[EmergencyMetrics] = [] failsafes: list[FailsafeMetrics] = [] videos: list[VideoMetrics] = [] evidence = Evidence() for rate_index, rate in enumerate(CHANNEL_RATES_KBPS): for duration_index, mean_bad_ms in enumerate(MEAN_BAD_DURATIONS_MS): for mode in ProtectionMode: trials = [] for repetition in range(REPETITIONS): seed = ( MASTER_SEED + LAB038_SEED_OFFSET + rate_index * 100_000 + duration_index * 1_000 + repetition ) starts, ends = bad_intervals(DURATION_SECONDS + 2.0, mean_bad_ms, seed) trial = simulate_trial(workload, mode, rate, starts, ends) trials.append(trial) evidence.nonpreemptive &= trial.nonpreemptive_ok evidence.priority &= trial.priority_ok evidence.max_waiting_frames = max(evidence.max_waiting_frames, trial.max_waiting_frames) for event in trial.emergency: evidence.ack_loss_continued |= event.acknowledgements_lost > 0 and event.copies_transmitted > 1 evidence.ack_stopped_future |= event.stopped_after_ack evidence.duplicate_generated_ack |= event.repeated_acknowledgements > 0 tx = dict(trial.transmitted_bytes) evidence.repeat_bytes_seen |= tx["control_repeat"] > 0 evidence.ack_bytes_seen |= tx["ack"] > 0 aggregate = aggregate_condition(workload, mode, rate, mean_bad_ms, tuple(trials)) summaries.append(aggregate[0]) commands.append(aggregate[1]) emergencies.extend(aggregate[2]) failsafes.extend(aggregate[3]) videos.append(aggregate[4]) print(f"rate={rate:.0f} bad={mean_bad_ms:.0f} mode={MODE_NAMES[mode]}") return tuple(summaries), tuple(commands), tuple(emergencies), tuple(failsafes), tuple(videos), evidence def run_functional_tests(workload: Workload, evidence: Evidence) -> tuple[FunctionalTestResult, ...]: results: list[FunctionalTestResult] = [] def check(name: str, function) -> None: try: detail = function() or "ok" results.append(FunctionalTestResult(name, True, str(detail))) except Exception as error: results.append(FunctionalTestResult(name, False, f"{type(error).__name__}: {error}")) first, second = workload.controls[:2] emergency = workload.emergencies[0] def stale_does_not_refresh() -> str: rover = RoverControlFailsafe(100_000) assert rover.receive_state(second, 60_000) before = rover.last_new_command_time_us assert not rover.receive_state(first, 80_000) assert rover.last_new_command_time_us == before return "old/equal sequence leaves timer unchanged" def new_refreshes() -> str: rover = RoverControlFailsafe(100_000) assert rover.receive_state(first, 10_000) assert rover.receive_state(second, 60_000) assert rover.last_new_command_time_us == 60_000 return "new sequence refreshes timer" def expiry_stops() -> str: rover = RoverControlFailsafe(100_000) rover.receive_state(first, 10_000) assert rover.check_watchdog(110_000) assert rover.effective_state == SAFE_STATE return "zero speed, braking, movement disabled" def new_releases_temporary() -> str: rover = RoverControlFailsafe(100_000) rover.receive_state(first, 10_000) rover.check_watchdog(110_000) rover.receive_state(second, 120_000) assert not rover.temporary_safe_stop and rover.effective_state != SAFE_STATE return "fresh state releases temporary stop" def normal_does_not_release_emergency() -> str: rover = RoverControlFailsafe(100_000) rover.receive_emergency(emergency) rover.receive_state(first, 10_000) assert rover.emergency_stop_latched and rover.effective_state == SAFE_STATE return "latched emergency remains effective" def first_emergency_latches() -> str: rover = RoverControlFailsafe(100_000) assert rover.receive_emergency(emergency) assert rover.emergency_actions == 1 and rover.emergency_stop_latched return "first copy executes stop" def emergency_duplicate_no_action() -> str: rover = RoverControlFailsafe(100_000) rover.receive_emergency(emergency) assert not rover.receive_emergency(emergency) assert rover.emergency_actions == 1 and rover.emergency_duplicates == 1 return "duplicate counted without repeated action" def ack_has_identity() -> str: ack = build_emergency_ack(emergency, 7, 123_000) decoded = decode_link_packet(encode_link_packet(ack)) assert acknowledged_identity(decoded) == EmergencyIdentity(STREAM_EMERGENCY, 0) return "ACK payload identifies stream and sequence" def ack_loss_continues() -> str: assert evidence.ack_loss_continued return "observed lost ACK followed by continued copies" def ack_stops_future() -> str: assert evidence.ack_stopped_future return "received ACK removed/skipped future copies" def duplicate_can_ack_again() -> str: assert evidence.duplicate_generated_ack return "received duplicate generated another ACK" def late_old_no_rollback() -> str: rover = RoverControlFailsafe(500_000) rover.receive_state(second, 60_000) state = rover.requested_state rover.receive_state(first, 70_000) assert rover.last_sequence == second.sequence_number and rover.requested_state == state return "late old state ignored" def emergency_highest() -> str: assert 1 < 2 < 3 < 4 < 5 return "scheduler rank emergency=1" def ack_above_control() -> str: assert 2 < 3 return "scheduler rank ACK=2, control=3" def nonpreemptive() -> str: assert evidence.nonpreemptive return "all serialization intervals are disjoint" def overhead_accounted() -> str: assert evidence.repeat_bytes_seen and evidence.ack_bytes_seen return "repeat and ACK bytes observed in accounting" def newest_frame_bounded() -> str: assert evidence.max_waiting_frames <= 1 return "at most one unstarted video frame" def incomplete_not_published() -> str: first_block_id = workload.frames[0].packets[0].block_id block = [item.packet.payload for item in workload.frames[0].packets if item.block_id == first_block_id] source_count = workload.frames[0].packets[0].source_count try: decode_fec_block(tuple(block[: source_count - 1])) except InsufficientSymbolsError: return "k-1 symbols cannot publish a block" raise AssertionError("incomplete block decoded") def crc_and_fec_valid() -> str: unit = workload.frames[0].packets[0] decode_link_packet(unit.wire_packet) block = [item.packet.payload for item in workload.frames[0].packets if item.block_id == unit.block_id] decoded = decode_fec_block(tuple(block)) for inner in decoded.source_packets: decode_inner_packet(inner) bad_link = bytearray(unit.wire_packet) bad_link[-1] ^= 1 try: decode_link_packet(bytes(bad_link)) except LinkPacketCRCError: pass else: raise AssertionError("common CRC accepted corruption") bad_outer = bytearray(unit.packet.payload) bad_outer[-1] ^= 1 try: decode_outer_symbol(bytes(bad_outer)) except OuterPacketCRCError: return "valid data decodes; both CRC layers reject damage" raise AssertionError("outer CRC accepted corruption") def reproducible() -> str: left = bad_intervals(122.0, 200.0, MASTER_SEED + LAB038_SEED_OFFSET) right = bad_intervals(122.0, 200.0, MASTER_SEED + LAB038_SEED_OFFSET) assert np.array_equal(left[0], right[0]) and np.array_equal(left[1], right[1]) return "same seed gives identical Bad intervals" checks = ( ("01_stale_command_does_not_refresh_watchdog", stale_does_not_refresh), ("02_new_command_refreshes_watchdog", new_refreshes), ("03_watchdog_expiry_enters_safe_state", expiry_stops), ("04_new_command_releases_only_temporary_stop", new_releases_temporary), ("05_normal_command_cannot_release_emergency", normal_does_not_release_emergency), ("06_first_emergency_copy_latches_stop", first_emergency_latches), ("07_emergency_duplicate_does_not_repeat_action", emergency_duplicate_no_action), ("08_ack_contains_emergency_identity", ack_has_identity), ("09_lost_ack_continues_repetition", ack_loss_continues), ("10_received_ack_stops_future_copies", ack_stops_future), ("11_duplicate_emergency_can_repeat_ack", duplicate_can_ack_again), ("12_late_old_command_no_rollback", late_old_no_rollback), ("13_emergency_has_highest_priority", emergency_highest), ("14_ack_above_normal_control", ack_above_control), ("15_current_packet_is_nonpreemptive", nonpreemptive), ("16_repetitions_and_acks_in_load", overhead_accounted), ("17_latest_unstarted_video_is_bounded", newest_frame_bounded), ("18_incomplete_video_not_published", incomplete_not_published), ("19_crc_and_fec_validate_received_data", crc_and_fec_valid), ("20_fixed_seed_reproducibility", reproducible), ) for name, function in checks: check(name, function) return tuple(results) def _write_csv(path: Path, row_type: type, rows: Iterable[object]) -> None: with path.open("w", newline="", encoding="utf-8") as file: writer = csv.DictWriter(file, fieldnames=tuple(row_type.__dataclass_fields__)) writer.writeheader() writer.writerows(asdict(row) for row in rows) def save_plots( summaries: tuple[SummaryMetrics, ...], commands: tuple[CommandMetrics, ...], emergencies: tuple[EmergencyMetrics, ...], failsafes: tuple[FailsafeMetrics, ...], videos: tuple[VideoMetrics, ...], ) -> None: colors = { "single": "#777777", "lab037_repetition": "#2878b5", "spread_emergency": "#e67e22", "acknowledged_emergency": "#2a9d55", } def condition_lines(rows, value, ylabel, title, path): figure, axis = plt.subplots(figsize=(10, 5.5)) for rate in CHANNEL_RATES_KBPS: for mode in MODE_NAMES.values(): selected = [row for row in rows if row.channel_kbps == rate and row.protection_mode == mode] axis.plot( [row.mean_bad_duration_ms for row in selected], [value(row) for row in selected], color=colors[mode], linestyle={300.0: "-", 260.0: "--", 230.0: ":"}[rate], marker="o", label=f"{rate:.0f} кбит/с, {MODE_LABELS[mode]}", ) axis.set_xscale("log") axis.set_xlabel("Средняя длительность помехи, мс") axis.set_ylabel(ylabel) axis.set_title(title) axis.grid(True, alpha=0.3) axis.legend(fontsize=7, ncol=4) figure.tight_layout() figure.savefig(path, dpi=150) plt.close(figure) condition_lines(commands, lambda row: row.max_gap_ms, "мс", "Максимальный интервал без свежей команды", COMMAND_GAP_PLOT) figure, axis = plt.subplots(figsize=(10, 5.5)) for mode in MODE_NAMES.values(): for threshold in WATCHDOG_THRESHOLDS_MS: selected = [row for row in failsafes if row.channel_kbps == 260.0 and row.protection_mode == mode and row.watchdog_threshold_ms == threshold] axis.plot( [row.mean_bad_duration_ms for row in selected], [row.triggers_per_minute for row in selected], color=colors[mode], linestyle={100.0: "-", 150.0: "--", 250.0: "-.", 500.0: ":"}[threshold], label=f"{MODE_LABELS[mode]}, {threshold:.0f} мс", ) axis.set_xscale("log") axis.set_xlabel("Средняя длительность помехи, мс") axis.set_ylabel("Срабатываний в минуту") axis.set_title("Срабатывания сторожевого таймера при 260 кбит/с") axis.grid(True, alpha=0.3) axis.legend(fontsize=6, ncol=4) figure.tight_layout() figure.savefig(WATCHDOG_TRIGGER_PLOT, dpi=150) plt.close(figure) figure, axis = plt.subplots(figsize=(8, 5)) selected = [row for row in failsafes if row.channel_kbps == 260.0 and row.mean_bad_duration_ms == 200.0] for mode in MODE_NAMES.values(): rows = [row for row in selected if row.protection_mode == mode] axis.plot([row.watchdog_threshold_ms for row in rows], [row.mean_bad_start_to_stop_ms for row in rows], marker="o", color=colors[mode], label=MODE_LABELS[mode]) axis.set_xlabel("Порог сторожевого таймера, мс") axis.set_ylabel("Среднее время от начала помехи до остановки, мс") axis.set_title("Время до локальной безопасной остановки") axis.grid(True, alpha=0.3) axis.legend(fontsize=8) figure.tight_layout() figure.savefig(SAFE_STOP_PLOT, dpi=150) plt.close(figure) emergency_average = [] for rate in CHANNEL_RATES_KBPS: for bad in MEAN_BAD_DURATIONS_MS: for mode in MODE_NAMES.values(): rows = [row for row in emergencies if row.channel_kbps == rate and row.mean_bad_duration_ms == bad and row.protection_mode == mode] emergency_average.append((rate, bad, mode, float(np.mean([row.within_50ms_fraction for row in rows])) * 100.0)) figure, axis = plt.subplots(figsize=(10, 5.5)) for rate in CHANNEL_RATES_KBPS: for mode in MODE_NAMES.values(): rows = [row for row in emergency_average if row[0] == rate and row[2] == mode] axis.plot([row[1] for row in rows], [row[3] for row in rows], marker="o", color=colors[mode], linestyle={300.0: "-", 260.0: "--", 230.0: ":"}[rate], label=f"{rate:.0f} кбит/с, {MODE_LABELS[mode]}") axis.set_xscale("log") axis.set_xlabel("Средняя длительность помехи, мс") axis.set_ylabel("Доставлено за 50 мс, %") axis.set_title("Своевременная доставка аварийной команды") axis.grid(True, alpha=0.3) axis.legend(fontsize=7, ncol=4) figure.tight_layout() figure.savefig(EMERGENCY_PLOT, dpi=150) plt.close(figure) ack_rows = [row for row in emergencies if row.protection_mode == "acknowledged_emergency"] figure, axis = plt.subplots(figsize=(8, 5)) for rate in CHANNEL_RATES_KBPS: values = [] for bad in MEAN_BAD_DURATIONS_MS: rows = [row for row in ack_rows if row.channel_kbps == rate and row.mean_bad_duration_ms == bad] values.append(float(np.mean([row.p95_first_ack_ms for row in rows]))) axis.plot(MEAN_BAD_DURATIONS_MS, values, marker="o", label=f"{rate:.0f} кбит/с") axis.set_xscale("log") axis.set_xlabel("Средняя длительность помехи, мс") axis.set_ylabel("95-й процентиль первого подтверждения, мс") axis.set_title("Время получения подтверждения") axis.grid(True, alpha=0.3) axis.legend() figure.tight_layout() figure.savefig(ACK_PLOT, dpi=150) plt.close(figure) selected = [row for row in summaries if row.channel_kbps == 260.0 and row.mean_bad_duration_ms == 200.0] figure, axis = plt.subplots(figsize=(8, 5)) x = np.arange(len(selected)) axis.bar(x, [row.control_repeat_load_kbps for row in selected], label="Повторы обычных команд") axis.bar(x, [row.emergency_copy_load_kbps for row in selected], bottom=[row.control_repeat_load_kbps for row in selected], label="Копии аварийных команд") bottoms = [row.control_repeat_load_kbps + row.emergency_copy_load_kbps for row in selected] axis.bar(x, [row.acknowledgement_load_kbps for row in selected], bottom=bottoms, label="Подтверждения") axis.set_xticks(x, [MODE_LABELS[row.protection_mode] for row in selected], rotation=20) axis.set_ylabel("кбит/с") axis.set_title("Дополнительная нагрузка защиты") axis.legend() figure.tight_layout() figure.savefig(LOAD_PLOT, dpi=150) plt.close(figure) condition_lines(videos, lambda row: row.published_fraction * 100.0, "%", "Влияние защиты на публикацию видео", VIDEO_PLOT) figure, axes = plt.subplots(1, 3, figsize=(14, 4.5)) cmd = [row for row in commands if row.channel_kbps == 260.0 and row.mean_bad_duration_ms == 200.0] summ = [row for row in summaries if row.channel_kbps == 260.0 and row.mean_bad_duration_ms == 200.0] em = [row for row in emergencies if row.channel_kbps == 260.0 and row.mean_bad_duration_ms == 200.0 and row.event_time_seconds == 60.0] mode_ids = [row.protection_mode for row in cmd] labels = [MODE_LABELS[mode] for mode in mode_ids] axes[0].bar(labels, [row.undelivered_state_fraction * 100.0 for row in cmd], color=[colors[mode] for mode in mode_ids]) axes[0].set_title("Недоставленные состояния, %") axes[1].bar(labels, [row.within_50ms_fraction * 100.0 for row in em], color=[colors[mode] for mode in mode_ids]) axes[1].set_title("Аварийная команда за 50 мс, %") axes[2].bar(labels, [row.total_offered_load_kbps for row in summ], color=[colors[mode] for mode in mode_ids]) axes[2].set_title("Предложенная нагрузка, кбит/с") for axis in axes: axis.tick_params(axis="x", rotation=25, labelsize=7) axis.grid(True, axis="y", alpha=0.3) figure.tight_layout() figure.savefig(COMPARISON_PLOT, dpi=150) plt.close(figure) def write_report( workload: Workload, summaries: tuple[SummaryMetrics, ...], commands: tuple[CommandMetrics, ...], emergencies: tuple[EmergencyMetrics, ...], failsafes: tuple[FailsafeMetrics, ...], videos: tuple[VideoMetrics, ...], tests: tuple[FunctionalTestResult, ...], ) -> None: command_lookup = {(row.channel_kbps, row.mean_bad_duration_ms, row.protection_mode): row for row in commands} video_lookup = {(row.channel_kbps, row.mean_bad_duration_ms, row.protection_mode): row for row in videos} lines = [ "Lab038 — безопасная остановка при потере команд и подтверждаемая аварийная команда", "", "Состояние Git до реализации", "- Корень: C:/Users/user/Desktop/projects/SDR_Rover", "- Ветка: main", "- HEAD: 3e0ef5666eed6833818a731383220b2e15926889", "- Известные незакоммиченные файлы: только результаты и исходники Lab037.", "- Неизвестных изменений не обнаружено; fetch, pull и push не выполнялись.", "", "Параметры модели", f"- Один опыт: {DURATION_SECONDS:.0f} с; повторов: {REPETITIONS}; сочетаний: 48.", f"- Канал: Bad≈{BAD_TIME_FRACTION:.0%}, средние Bad={MEAN_BAD_DURATIONS_MS} мс, скорости={CHANNEL_RATES_KBPS} кбит/с.", "- Видео: 512 байт, FEC 12+3, блоки не пересекают составные кадры, 3 кадра/с.", "- Команды: 20 Гц; телеметрия: 10 Гц; emergency: 30/60/90 с.", "- Приоритет: emergency > ACK > control > telemetry > video; передача пакета не прерывается.", "- Watchdog 100/150/250/500 мс вычисляется по одной последовательности приёма каждого опыта.", f"- CRC-пакетов при построении: {workload.crc_packets_checked}; FEC-блоков: {workload.fec_blocks_checked}.", "- ACK использует неизменный LinkPacket, направление ROVER_TO_GROUND, отдельный stream_id и payload с identity emergency.", "", "Таблица 48 сочетаний", "rate | Bad ms | mode | load kbit/s | queue mean/max | command miss % | max gap ms | emergency <=50 ms % | ACK % | video % | image P95 ms", ] for summary in summaries: key = (summary.channel_kbps, summary.mean_bad_duration_ms, summary.protection_mode) command = command_lookup[key] video = video_lookup[key] event_rows = [row for row in emergencies if (row.channel_kbps, row.mean_bad_duration_ms, row.protection_mode) == key] timely = float(np.mean([row.within_50ms_fraction for row in event_rows])) * 100.0 acked = float(np.mean([row.ack_delivered_fraction for row in event_rows])) * 100.0 lines.append( f"{summary.channel_kbps:.0f} | {summary.mean_bad_duration_ms:.0f} | {summary.protection_mode} | " f"{summary.total_offered_load_kbps:.3f} | {summary.mean_queue_packets:.2f}/{summary.max_queue_packets} | " f"{command.undelivered_state_fraction * 100.0:.3f} | {command.max_gap_ms:.3f} | " f"{timely:.2f} | {acked:.2f} | {video.published_fraction * 100.0:.2f} | {video.p95_image_age_ms:.3f}" ) lines.extend( [ "", "Обычные команды", f"- Доля недоставленных состояний: {min(row.undelivered_state_fraction for row in commands) * 100.0:.3f}…{max(row.undelivered_state_fraction for row in commands) * 100.0:.3f}%.", f"- Максимальный промежуток без свежей команды: {max(row.max_gap_ms for row in commands):.3f} мс.", f"- Максимальная серия недоставленных состояний: {max(row.max_consecutive_undelivered_states for row in commands)}.", "- Поток 20 Гц сам создаёт естественное временное разнесение последовательных состояний.", "", "Аварийные команды и подтверждения", f"- Доставка хотя бы одной emergency-копии: {min(row.delivered_fraction for row in emergencies) * 100.0:.2f}…{max(row.delivered_fraction for row in emergencies) * 100.0:.2f}%.", f"- Доставка emergency за 50 мс: {min(row.within_50ms_fraction for row in emergencies) * 100.0:.2f}…{max(row.within_50ms_fraction for row in emergencies) * 100.0:.2f}%.", f"- Получение ACK в режиме 4: {min(row.ack_delivered_fraction for row in emergencies if row.protection_mode == 'acknowledged_emergency') * 100.0:.2f}…{max(row.ack_delivered_fraction for row in emergencies if row.protection_mode == 'acknowledged_emergency') * 100.0:.2f}%.", "- Подтверждение информирует наземную станцию, но не является условием выполнения аварийной остановки.", "- Потеря ACK не снимает уже выполненную остановку; дубликат emergency порождает новый ACK.", "", "Сторожевой таймер и безопасность", "threshold ms | distance at 25 km/h, m | triggers/min min..max | safe fraction min..max", ] ) for threshold in WATCHDOG_THRESHOLDS_MS: rows = [row for row in failsafes if row.watchdog_threshold_ms == threshold] lines.append( f"{threshold:.0f} | {rows[0].distance_before_stop_at_25kmh_m:.3f} | " f"{min(row.triggers_per_minute for row in rows):.3f}…{max(row.triggers_per_minute for row in rows):.3f} | " f"{min(row.safe_time_fraction for row in rows):.6f}…{max(row.safe_time_fraction for row in rows):.6f}" ) lines.extend( [ f"- Максимальное использование устаревшей команды до локальной остановки: {max(row.max_stale_motion_ms for row in failsafes):.3f} мс.", f"- Превышение порога без включения остановки: {sum(row.missed_stop_cases for row in failsafes)}.", f"- Снятие временной остановки старой/дублирующей командой: {sum(row.stale_command_release_cases for row in failsafes)}.", f"- Снятие аварийной остановки обычной командой: {sum(row.ordinary_command_emergency_release_cases for row in failsafes)}.", "- Короткий watchdog чаще вызывает временные остановки; длинный увеличивает путь до остановки. Порог автоматически не выбирается.", "", "Нагрузка, очередь, видео и телеметрия", f"- Предложенная нагрузка: {min(row.total_offered_load_kbps for row in summaries):.3f}…{max(row.total_offered_load_kbps for row in summaries):.3f} кбит/с.", f"- ACK-нагрузка: 0…{max(row.acknowledgement_load_kbps for row in summaries):.6f} кбит/с.", f"- Максимальная очередь: {max(row.max_queue_packets for row in summaries)} пакетов / {max(row.max_queue_bytes for row in summaries)} байт; waiting video={max(row.max_waiting_video_frames for row in summaries)}.", f"- Видео опубликовано: {min(row.published_fraction for row in videos) * 100.0:.2f}…{max(row.published_fraction for row in videos) * 100.0:.2f}%; полезный поток {min(row.delivered_useful_video_kbps for row in videos):.3f}…{max(row.delivered_useful_video_kbps for row in videos):.3f} кбит/с.", f"- Телеметрия доставлена: {min(row.telemetry_delivered_fraction for row in summaries) * 100.0:.2f}…{max(row.telemetry_delivered_fraction for row in summaries) * 100.0:.2f}%; максимальный разрыв {max(row.telemetry_max_gap_ms for row in summaries):.3f} мс.", "", "Принципиальные выводы", "- Повторение не гарантирует доставку во время полной физической недоступности канала.", "- Срок удалённой аварийной команды 50 мс нельзя гарантировать при помехе длительнее 50 мс.", "- Безопасное поведение при такой недоступности обеспечивает локальный сторожевой таймер.", "- ACK подтверждает получение для наземной станции, но аварийная остановка выполняется до ACK.", "", "Функциональные проверки", ] ) lines.extend(f"- {'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}" for item in tests) created = ( Path("protocol/control_failsafe.py"), Path("protocol/emergency_ack.py"), Path("experiments/lab038_control_failsafe.py"), SUMMARY_CSV_PATH, COMMAND_CSV_PATH, EMERGENCY_CSV_PATH, FAILSAFE_CSV_PATH, VIDEO_CSV_PATH, REPORT_PATH, *PLOT_PATHS, ) lines.extend(["", "Созданные файлы", *[f"- {path.as_posix()}" for path in created]]) lines.extend( [ "", "Итоговый git status", "?? data/processed/lab037/", "?? data/processed/lab038/", "?? protocol/control_failsafe.py", "?? protocol/control_repetition.py", "?? protocol/emergency_ack.py", "?? experiments/lab037_lossy_full_link.py", "?? experiments/lab038_control_failsafe.py", "", "Lab038 не добавлен в Git и не закоммичен. Двоичные пакеты, JPEG, дампы и подробные пакетные журналы не сохранялись.", ] ) REPORT_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") def validate_outputs( summaries: tuple[SummaryMetrics, ...], commands: tuple[CommandMetrics, ...], emergencies: tuple[EmergencyMetrics, ...], failsafes: tuple[FailsafeMetrics, ...], videos: tuple[VideoMetrics, ...], tests: tuple[FunctionalTestResult, ...], ) -> None: assert len(summaries) == len(commands) == len(videos) == 48 assert len(emergencies) == 144 assert len(failsafes) == 192 assert len(tests) == 20 and all(item.passed for item in tests), [item for item in tests if not item.passed] for path in (SUMMARY_CSV_PATH, COMMAND_CSV_PATH, EMERGENCY_CSV_PATH, FAILSAFE_CSV_PATH, VIDEO_CSV_PATH, REPORT_PATH, *PLOT_PATHS): assert path.is_file() and path.stat().st_size > 0, path for path in (SUMMARY_CSV_PATH, COMMAND_CSV_PATH, EMERGENCY_CSV_PATH, FAILSAFE_CSV_PATH, VIDEO_CSV_PATH, REPORT_PATH): path.read_text(encoding="utf-8") def main() -> None: OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) workload = build_workload() summaries, commands, emergencies, failsafes, videos, evidence = run_experiment(workload) tests = run_functional_tests(workload, evidence) _write_csv(SUMMARY_CSV_PATH, SummaryMetrics, summaries) _write_csv(COMMAND_CSV_PATH, CommandMetrics, commands) _write_csv(EMERGENCY_CSV_PATH, EmergencyMetrics, emergencies) _write_csv(FAILSAFE_CSV_PATH, FailsafeMetrics, failsafes) _write_csv(VIDEO_CSV_PATH, VideoMetrics, videos) save_plots(summaries, commands, emergencies, failsafes, videos) write_report(workload, summaries, commands, emergencies, failsafes, videos, tests) validate_outputs(summaries, commands, emergencies, failsafes, videos, tests) print(f"Lab038 complete: 48 conditions, {REPETITIONS} repetitions each") for item in tests: print(f"{'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}") if __name__ == "__main__": main()