1646 lines
70 KiB
Python
1646 lines
70 KiB
Python
"""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 tests.lab033_priority_channel_scheduler import (
|
||
STREAM_CONTROL,
|
||
STREAM_EMERGENCY,
|
||
deterministic_payload,
|
||
)
|
||
from tests.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("tests/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",
|
||
"?? tests/lab037_lossy_full_link.py",
|
||
"?? tests/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()
|