SDR-Rover/tests/lab040_persistent_emergency.py

"""Lab040: persistent emergency intent and safe movement reauthorization."""

from __future__ import annotations

import csv
from dataclasses import asdict, dataclass
from enum import IntEnum
from pathlib import Path
from typing import Iterable

import cv2
import matplotlib
import numpy as np

matplotlib.use("Agg")
import matplotlib.pyplot as plt

from protocol.control_failsafe import ControlState, SAFE_STATE, decode_control_state, encode_control_state
from protocol.emergency_ack import EmergencyAckReceiver, EmergencyIdentity, acknowledged_identity, build_emergency_ack
from protocol.link_packet import Direction, LinkPacket, TrafficClass, decode_link_packet, encode_link_packet
from protocol.persistent_emergency import GroundState, PersistentEmergencyController
from protocol.safe_reset import (
    ResetAcknowledgement,
    ResetCommand,
    ResetDecision,
    ResetReason,
    STREAM_RESET_ACK,
    SafeResetReceiver,
    build_reset_ack,
    build_reset_packet,
    decode_reset_ack,
)
from protocol.two_stage_failsafe import (
    DecelerationProfile,
    SafetyState,
    ThresholdPolicy,
    TwoStageFailsafe,
    integrate_kinematics,
)
from tests.lab033_priority_channel_scheduler import STREAM_CONTROL, STREAM_EMERGENCY
from tests.lab037_lossy_full_link import bad_intervals
from tests.lab038_control_failsafe import DURATION_SECONDS, Frame, Unit, Workload, build_workload, packet_is_lost


OUTPUT_DIRECTORY = Path("data/processed/lab040")
SUMMARY_CSV = OUTPUT_DIRECTORY / "lab040_summary.csv"
EMERGENCY_CSV = OUTPUT_DIRECTORY / "lab040_emergency_metrics.csv"
RESET_CSV = OUTPUT_DIRECTORY / "lab040_reset_metrics.csv"
MOTION_CSV = OUTPUT_DIRECTORY / "lab040_motion_metrics.csv"
LOAD_CSV = OUTPUT_DIRECTORY / "lab040_load_metrics.csv"
REPORT_PATH = OUTPUT_DIRECTORY / "lab040_report.txt"
PLOT_PATHS = (
    OUTPUT_DIRECTORY / "lab040_emergency_delivery.png",
    OUTPUT_DIRECTORY / "lab040_unsafe_recovery.png",
    OUTPUT_DIRECTORY / "lab040_latch_time.png",
    OUTPUT_DIRECTORY / "lab040_stop_path.png",
    OUTPUT_DIRECTORY / "lab040_reset_latency.png",
    OUTPUT_DIRECTORY / "lab040_protection_load.png",
    OUTPUT_DIRECTORY / "lab040_video_telemetry.png",
    OUTPUT_DIRECTORY / "lab040_tradeoff.png",
)

REPETITIONS = 200
CONTROL_PERIOD_SECONDS = 0.050
REPEAT_PERIOD_SECONDS = 0.050
OPERATOR_EMERGENCY_SECONDS = 30.0
OPERATOR_RESET_SECONDS = 60.0
NEW_MOTION_DELAY_SECONDS = 2.0
EVENT_ID = 1
EMERGENCY_SEQUENCE = 1
RESET_REQUEST_ID = 1
RESET_SEQUENCE = 1
SEED_OFFSET = 400_000_000
EPSILON = 1e-12
POLICY = ThresholdPolicy("two_150_250", 0.150, 0.250)
PROFILE = DecelerationProfile("nominal", 1.0, 3.0)
SPEEDS_KMH = (5.0, 15.0, 25.0)


class Mode(IntEnum):
    LIMITED_LAB038 = 1
    PERSISTENT_NO_RESET = 2
    PERSISTENT_SAFE_RESET = 3


MODE_NAMES = {
    Mode.LIMITED_LAB038: "limited_lab038",
    Mode.PERSISTENT_NO_RESET: "persistent_no_reset",
    Mode.PERSISTENT_SAFE_RESET: "persistent_safe_reset",
}
MODE_LABELS = {
    "limited_lab038": "Ограниченное повторение",
    "persistent_no_reset": "Постоянное намерение",
    "persistent_safe_reset": "Постоянное намерение и сброс",
}


@dataclass(frozen=True)
class ChannelCondition:
    name: str
    label: str
    rate_kbps: float
    mean_bad_ms: float | None
    repetitions: int


CHANNELS = (
    ChannelCondition("no_loss", "Без потерь, 300 кбит/с", 300.0, None, 1),
    ChannelCondition("300_50", "300 кбит/с, помеха 50 мс", 300.0, 50.0, REPETITIONS),
    ChannelCondition("260_200", "260 кбит/с, помеха 200 мс", 260.0, 200.0, REPETITIONS),
    ChannelCondition("230_200", "230 кбит/с, помеха 200 мс", 230.0, 200.0, REPETITIONS),
    ChannelCondition("230_1000", "230 кбит/с, помеха 1000 мс", 230.0, 1000.0, REPETITIONS),
)


@dataclass(frozen=True)
class TransportTrial:
    control_receives: tuple[tuple[float, int, ControlState], ...]
    emergency_delivered: bool
    emergency_first_receive_seconds: float | None
    emergency_copies: int
    emergency_copies_lost: int
    emergency_ack_seconds: float | None
    emergency_acks_lost: int
    all_first_500ms_copies_lost: bool
    emergency_intent_fraction: float
    positive_commands_after_emergency: int
    permitted_commands_after_emergency: int
    time_to_motion_command_suppression_seconds: float
    reset_requests_created: int
    reset_requests_transmitted: int
    reset_requests_accepted: int
    reset_requests_rejected: int
    reset_rejection_reasons: tuple[tuple[str, int], ...]
    reset_first_accept_seconds: float | None
    reset_ack_seconds: float | None
    reset_acks_lost: int
    reset_duplicates: int
    reset_reacks: int
    new_motion_command_seconds: float | None
    automatic_old_command_restorations: int
    offered_bytes: tuple[tuple[str, int], ...]
    video_published: int
    telemetry_delivered: int
    max_queue_packets: int
    negative_time_order: int


@dataclass(frozen=True)
class MotionTrial:
    distance_to_braking_m: float
    distance_to_stop_m: float
    time_to_stop_seconds: float
    maximum_speed_after_emergency_mps: float
    stopped: bool
    reaccelerations_before_reset: int
    braking_releases_before_reset_ack: int
    movement_while_intent: int
    recovery_after_lost_emergency: int
    watchdog_stopped_after_lost_emergency: int
    unsafe_recovery: int
    stop_to_resume_seconds: float
    distance_after_unsafe_resume_m: float
    movement_before_reset_ack: int
    actual_resume_seconds: float
    negative_speed_cases: int
    target_speed_exceeded_cases: int


@dataclass(frozen=True)
class SummaryMetrics:
    channel_condition: str
    mode: str
    initial_speed_kmh: float
    repetitions: int
    emergency_delivery_fraction: float
    emergency_intent_time_fraction: float
    unsafe_recovery_fraction: float
    full_stop_fraction: float
    movement_while_intent_cases: int
    movement_before_reset_ack_cases: int
    positive_commands_after_emergency_mean: float
    permitted_commands_after_emergency_mean: float
    total_extra_load_kbps: float
    video_published_fraction: float
    telemetry_delivered_fraction: float
    maximum_queue_packets: int


@dataclass(frozen=True)
class EmergencyMetrics:
    channel_condition: str
    mode: str
    initial_speed_kmh: float
    repetitions: int
    delivered_fraction: float
    mean_latch_delay_ms: float
    p95_latch_delay_ms: float
    max_latch_delay_ms: float
    mean_copies: float
    mean_lost_copies: float
    mean_ack_delay_ms: float
    acknowledgements_lost_mean: float
    intent_time_fraction: float
    positive_commands_after_emergency_mean: float
    permitted_commands_after_emergency_mean: float
    time_to_command_suppression_mean_ms: float
    all_first_500ms_lost: int


@dataclass(frozen=True)
class ResetMetrics:
    channel_condition: str
    mode: str
    initial_speed_kmh: float
    repetitions: int
    requests_created_mean: float
    requests_transmitted_mean: float
    requests_accepted_mean: float
    requests_rejected_mean: float
    rejected_not_latched: int
    rejected_moving: int
    rejected_stale_link: int
    rejected_event_mismatch: int
    rejected_old_sequence: int
    rejected_nonzero_request: int
    rejected_movement_permitted: int
    mean_accept_delay_ms: float
    mean_reset_ack_delay_ms: float
    p95_reset_ack_delay_ms: float
    reset_acks_lost_mean: float
    repeated_requests_mean: float
    duplicates_suppressed_mean: float
    reset_reacks_mean: float
    mean_ack_to_new_command_seconds: float
    mean_time_to_actual_resume_seconds: float
    movement_before_reset_ack_cases: int
    automatic_old_command_restorations: int


@dataclass(frozen=True)
class MotionMetrics:
    channel_condition: str
    mode: str
    initial_speed_kmh: float
    repetitions: int
    mean_distance_to_braking_m: float
    p95_distance_to_braking_m: float
    max_distance_to_braking_m: float
    mean_distance_to_stop_m: float
    p95_distance_to_stop_m: float
    max_distance_to_stop_m: float
    mean_time_to_stop_seconds: float
    p95_time_to_stop_seconds: float
    maximum_speed_after_emergency_mps: float
    full_stop_fraction: float
    reaccelerations_before_reset: int
    braking_releases_before_reset_ack: int
    movement_while_intent: int
    recovery_after_lost_emergency: int
    watchdog_stops_after_lost_emergency: int
    unsafe_recoveries: int
    mean_stop_to_resume_seconds: float
    mean_distance_after_unsafe_resume_m: float
    negative_speed_cases: int
    target_speed_exceeded_cases: int


@dataclass(frozen=True)
class LoadMetrics:
    channel_condition: str
    mode: str
    initial_speed_kmh: float
    repetitions: int
    emergency_copy_load_kbps: float
    emergency_ack_load_kbps: float
    zero_control_load_kbps: float
    reset_request_load_kbps: float
    reset_ack_load_kbps: float
    total_extra_load_kbps: float
    video_published_fraction: float
    telemetry_delivered_fraction: float
    maximum_queue_packets: int


@dataclass(frozen=True)
class FunctionalTestResult:
    name: str
    passed: bool
    detail: str


def _mean(values: Iterable[float]) -> float:
    values = tuple(values)
    return float(np.mean(values)) if values else 0.0


def _percentile(values: Iterable[float], value: float) -> float:
    values = tuple(values)
    return float(np.percentile(values, value)) if values else 0.0


def _packet(
    traffic_class: TrafficClass,
    direction: Direction,
    stream_id: int,
    sequence_number: int,
    generation_seconds: float,
    payload: bytes,
) -> LinkPacket:
    packet = LinkPacket(
        traffic_class=traffic_class,
        direction=direction,
        stream_id=stream_id,
        sequence_number=sequence_number,
        generation_time_us=int(round(generation_seconds * 1_000_000.0)),
        deadline_ms=50 if traffic_class is TrafficClass.EMERGENCY else 100,
        payload=payload,
    )
    return decode_link_packet(encode_link_packet(packet))


def _unit(packet: LinkPacket, available_seconds: float, order: int, priority: int, kind: str) -> Unit:
    return Unit(
        packet,
        encode_link_packet(packet),
        int(round(available_seconds * 1_000_000.0)),
        order,
        priority,
        kind,
    )


def simulate_transport(
    workload: Workload,
    mode: Mode,
    rate_kbps: float,
    bad_starts: np.ndarray,
    bad_ends: np.ndarray,
) -> TransportTrial:
    ready: list[Unit] = []
    pending_frames: list[Frame] = []
    active_frame: Frame | None = None
    active_index = 0
    telemetry_index = frame_index = 0
    control_sequence = 0
    control_next = 0.0
    emergency_next: float | None = None
    reset_next: float | None = None
    new_motion_due: float | None = None
    operator_emergency_done = operator_reset_done = new_motion_done = False
    order = 80_000_000
    cursor = previous_end = 0.0
    negative_time_order = 0
    maximum_queue = 0
    emergency_acknowledged = False
    emergency_first: float | None = None
    emergency_ack_first: float | None = None
    emergency_copies = emergency_lost = emergency_ack_lost = 0
    emergency_window_transmitted = emergency_window_lost = 0
    emergency_ack_sequence = 0
    reset_ack_sequence = 0
    reset_first_accept: float | None = None
    reset_ack_first: float | None = None
    reset_requests_transmitted = reset_ack_lost = 0
    new_motion_command_time: float | None = None
    positive_after = permitted_after = 0
    first_suppressed_time: float | None = None
    control_receives: list[tuple[float, int, ControlState]] = []
    last_control_receive = 0.0
    telemetry_delivered = video_published = 0
    block_successes: dict[int, int] = {}
    offered = {
        "emergency": 0,
        "emergency_ack": 0,
        "control": 0,
        "zero_control": 0,
        "reset": 0,
        "reset_ack": 0,
        "telemetry": 0,
        "video": workload.video_generated_bytes,
    }
    ground = PersistentEmergencyController(1.0) if mode is not Mode.LIMITED_LAB038 else None
    rover = TwoStageFailsafe(POLICY, PROFILE)
    reset_receiver = SafeResetReceiver()
    emergency_ack_receiver = EmergencyAckReceiver()
    rejection_counts = {reason.name.lower(): 0 for reason in ResetReason if reason is not ResetReason.ACCEPTED}
    emergency_packet = _packet(
        TrafficClass.EMERGENCY,
        Direction.GROUND_TO_ROVER,
        STREAM_EMERGENCY,
        EMERGENCY_SEQUENCE,
        OPERATOR_EMERGENCY_SECONDS,
        b"E-STOP-LAB040-PERSISTENT-INTENT",
    )

    def remove_ready(kind: str) -> None:
        ready[:] = [item for item in ready if item.kind != kind]

    def queue_size() -> int:
        active_remaining = len(active_frame.packets) - active_index if active_frame is not None else 0
        waiting = sum(len(frame.packets) for frame in pending_frames)
        return len(ready) + active_remaining + waiting

    def add_ready(item: Unit) -> None:
        nonlocal maximum_queue
        if item.kind == "control":
            remove_ready("control")
        elif item.kind == "telemetry":
            remove_ready("telemetry")
        ready.append(item)
        maximum_queue = max(maximum_queue, queue_size())

    def next_generation_time() -> float | None:
        candidates: list[float] = []
        if control_sequence < int(DURATION_SECONDS / CONTROL_PERIOD_SECONDS):
            candidates.append(control_next)
        if telemetry_index < len(workload.telemetry):
            candidates.append(workload.telemetry[telemetry_index].available_seconds)
        if frame_index < len(workload.frames):
            candidates.append(workload.frames[frame_index].generation_seconds)
        if not operator_emergency_done:
            candidates.append(OPERATOR_EMERGENCY_SECONDS)
        if mode is Mode.PERSISTENT_SAFE_RESET and not operator_reset_done:
            candidates.append(OPERATOR_RESET_SECONDS)
        if emergency_next is not None and emergency_next < DURATION_SECONDS - EPSILON:
            candidates.append(emergency_next)
        if reset_next is not None and reset_next < DURATION_SECONDS - EPSILON:
            candidates.append(reset_next)
        if new_motion_due is not None and not new_motion_done and new_motion_due < DURATION_SECONDS - EPSILON:
            candidates.append(new_motion_due)
        return min(candidates) if candidates else None

    def admit_until(limit: float, inclusive: bool) -> None:
        nonlocal control_sequence, control_next, telemetry_index, frame_index, order
        nonlocal operator_emergency_done, operator_reset_done, new_motion_done
        nonlocal emergency_next, reset_next, new_motion_command_time
        nonlocal emergency_copies, emergency_window_transmitted, reset_requests_transmitted
        nonlocal positive_after, permitted_after, first_suppressed_time, new_motion_due
        nonlocal maximum_queue
        while True:
            event_time = next_generation_time()
            if event_time is None:
                return
            due = event_time <= limit + EPSILON if inclusive else event_time < limit - EPSILON
            if not due:
                return

            if not operator_emergency_done and abs(event_time - OPERATOR_EMERGENCY_SECONDS) <= EPSILON:
                operator_emergency_done = True
                emergency_next = OPERATOR_EMERGENCY_SECONDS
                if ground is not None:
                    ground.request_emergency(EVENT_ID, STREAM_EMERGENCY, EMERGENCY_SEQUENCE)

            if (
                mode is Mode.PERSISTENT_SAFE_RESET
                and not operator_reset_done
                and abs(event_time - OPERATOR_RESET_SECONDS) <= EPSILON
            ):
                operator_reset_done = True
                assert ground is not None
                ground.request_reset(RESET_REQUEST_ID, RESET_SEQUENCE)
                reset_next = OPERATOR_RESET_SECONDS

            if new_motion_due is not None and not new_motion_done and abs(event_time - new_motion_due) <= EPSILON:
                new_motion_done = True
                assert ground is not None
                ground.new_operator_motion(1, 1.0)
                new_motion_command_time = event_time

            if (
                control_sequence < int(DURATION_SECONDS / CONTROL_PERIOD_SECONDS)
                and abs(event_time - control_next) <= EPSILON
            ):
                sequence = control_sequence
                state = ControlState(1.0, 0.0, False, True) if ground is None else ground.control_state()
                packet = _packet(
                    TrafficClass.CONTROL,
                    Direction.GROUND_TO_ROVER,
                    STREAM_CONTROL,
                    sequence,
                    control_next,
                    encode_control_state(state),
                )
                item = _unit(packet, control_next, order, 3, "control")
                order += 1
                offered["control"] += item.size
                if state == SAFE_STATE:
                    offered["zero_control"] += item.size
                    if control_next >= OPERATOR_EMERGENCY_SECONDS and first_suppressed_time is None:
                        first_suppressed_time = control_next
                intent_active = (
                    control_next >= OPERATOR_EMERGENCY_SECONDS
                    and (
                        ground is None
                        or ground.emergency_intent
                    )
                )
                if intent_active and state.desired_speed_mps > 0.0:
                    positive_after += 1
                if intent_active and state.movement_allowed:
                    permitted_after += 1
                add_ready(item)
                control_sequence += 1
                control_next = control_sequence * CONTROL_PERIOD_SECONDS

            if emergency_next is not None and abs(event_time - emergency_next) <= EPSILON:
                allowed = not emergency_acknowledged and (
                    mode is not Mode.LIMITED_LAB038
                    or emergency_next <= OPERATOR_EMERGENCY_SECONDS + 0.500 + EPSILON
                )
                current_tick = emergency_next
                emergency_next += REPEAT_PERIOD_SECONDS
                if mode is Mode.LIMITED_LAB038 and emergency_next > OPERATOR_EMERGENCY_SECONDS + 0.500 + EPSILON:
                    emergency_next = None
                if emergency_acknowledged:
                    emergency_next = None
                if allowed:
                    item = _unit(emergency_packet, current_tick, order, 1, "emergency")
                    order += 1
                    emergency_copies += 1
                    offered["emergency"] += item.size
                    if current_tick <= OPERATOR_EMERGENCY_SECONDS + 0.500 + EPSILON:
                        emergency_window_transmitted += 1
                    add_ready(item)

            if reset_next is not None and abs(event_time - reset_next) <= EPSILON:
                assert ground is not None
                current_tick = reset_next
                if ground.state is GroundState.GROUND_RESET_REQUESTED:
                    command = ResetCommand(EVENT_ID, RESET_REQUEST_ID, 0.0, False)
                    packet = build_reset_packet(command, RESET_SEQUENCE, int(round(current_tick * 1_000_000.0)))
                    item = _unit(packet, current_tick, order, 1, "reset")
                    order += 1
                    reset_requests_transmitted += 1
                    offered["reset"] += item.size
                    add_ready(item)
                    reset_next += REPEAT_PERIOD_SECONDS
                else:
                    reset_next = None

            while (
                telemetry_index < len(workload.telemetry)
                and abs(workload.telemetry[telemetry_index].available_seconds - event_time) <= EPSILON
            ):
                old = workload.telemetry[telemetry_index]
                item = Unit(old.packet, old.wire_packet, old.available_time_us, order, 4, "telemetry")
                order += 1
                telemetry_index += 1
                offered["telemetry"] += item.size
                add_ready(item)

            while (
                frame_index < len(workload.frames)
                and abs(workload.frames[frame_index].generation_seconds - event_time) <= EPSILON
            ):
                frame = workload.frames[frame_index]
                frame_index += 1
                pending_frames[:] = [frame]
                maximum_queue = max(maximum_queue, queue_size())

    while True:
        if not ready and active_frame is None and not pending_frames:
            next_time = next_generation_time()
            if next_time is None:
                break
            cursor = max(cursor, next_time)
            admit_until(cursor, True)
        if ready:
            item = min(ready, key=lambda candidate: (candidate.priority, candidate.order))
            ready.remove(item)
        else:
            if active_frame is None and pending_frames:
                active_frame = pending_frames.pop(0)
                active_index = 0
            if active_frame is None:
                continue
            item = active_frame.packets[active_index]
        start = max(cursor, item.available_seconds)
        end = start + item.size * 8.0 / (rate_kbps * 1000.0)
        negative_time_order += int(start < previous_end - EPSILON)
        previous_end = end
        lost = packet_is_lost(start, end, bad_starts, bad_ends)
        admit_until(end, False)
        cursor = end

        if item.kind == "control":
            if not lost:
                state = decode_control_state(item.packet.payload)
                if rover.receive_command(item.packet.sequence_number, end):
                    control_receives.append((end, item.packet.sequence_number, state))
                    last_control_receive = end
        elif item.kind == "emergency":
            if lost:
                emergency_lost += 1
                if item.available_seconds <= OPERATOR_EMERGENCY_SECONDS + 0.500 + EPSILON:
                    emergency_window_lost += 1
            else:
                if emergency_first is None:
                    emergency_first = end
                rover.receive_emergency(item.packet.stream_id, item.packet.sequence_number)
                ack_packet = build_emergency_ack(item.packet, emergency_ack_sequence, int(round(end * 1_000_000.0)))
                emergency_ack_sequence += 1
                ack_item = _unit(ack_packet, end, order, 2, "emergency_ack")
                order += 1
                offered["emergency_ack"] += ack_item.size
                add_ready(ack_item)
        elif item.kind == "emergency_ack":
            if lost:
                emergency_ack_lost += 1
            else:
                identity = acknowledged_identity(item.packet)
                if emergency_ack_receiver.accept(item.packet):
                    emergency_acknowledged = True
                    emergency_ack_first = end
                    remove_ready("emergency")
                    emergency_next = None
                    if ground is not None:
                        ground.receive_emergency_ack(identity)
        elif item.kind == "reset":
            if not lost:
                decision = reset_receiver.process(
                    item.packet,
                    rover=rover,
                    actual_speed_mps=0.0,
                    last_link_age_seconds=max(0.0, end - last_control_receive),
                    emergency_event_id=EVENT_ID if emergency_first is not None else None,
                )
                if decision.performed and reset_first_accept is None:
                    reset_first_accept = end
                if not decision.acknowledgement.accepted:
                    rejection_counts[decision.acknowledgement.reason.name.lower()] += 1
                ack_packet = build_reset_ack(decision, reset_ack_sequence, int(round(end * 1_000_000.0)))
                reset_ack_sequence += 1
                ack_item = _unit(ack_packet, end, order, 2, "reset_ack")
                order += 1
                offered["reset_ack"] += ack_item.size
                add_ready(ack_item)
        elif item.kind == "reset_ack":
            if lost:
                reset_ack_lost += 1
            else:
                acknowledgement = decode_reset_ack(item.packet.payload)
                assert item.packet.stream_id == STREAM_RESET_ACK
                assert ground is not None
                accepted = ground.receive_reset_ack(
                    acknowledgement.emergency_event_id,
                    acknowledgement.reset_request_id,
                    acknowledgement.accepted,
                )
                if accepted:
                    reset_ack_first = end
                    reset_next = None
                    remove_ready("reset")
                    new_motion_due = end + NEW_MOTION_DELAY_SECONDS
        elif item.kind == "telemetry":
            telemetry_delivered += int(not lost)
        elif item.kind == "video":
            if not lost and item.block_id is not None:
                block_successes[item.block_id] = block_successes.get(item.block_id, 0) + 1
            active_index += 1
            assert active_frame is not None
            if active_index == len(active_frame.packets):
                block_requirements = {
                    unit.block_id: unit.source_count
                    for unit in active_frame.packets
                    if unit.block_id is not None
                }
                if all(block_successes.get(block_id, 0) >= source_count for block_id, source_count in block_requirements.items()):
                    video_published += 1
                active_frame = None
                active_index = 0
        admit_until(end, True)
        maximum_queue = max(maximum_queue, queue_size())

    if ground is None:
        intent_end = min(DURATION_SECONDS, OPERATOR_EMERGENCY_SECONDS + 0.500)
    elif ground.emergency_intent:
        intent_end = DURATION_SECONDS
    else:
        intent_end = reset_ack_first if reset_ack_first is not None else DURATION_SECONDS
    intent_fraction = max(0.0, intent_end - OPERATOR_EMERGENCY_SECONDS) / DURATION_SECONDS
    suppression = (
        max(0.0, first_suppressed_time - OPERATOR_EMERGENCY_SECONDS)
        if first_suppressed_time is not None
        else DURATION_SECONDS - OPERATOR_EMERGENCY_SECONDS
    )
    all_window_lost = (
        emergency_window_transmitted > 0
        and emergency_window_lost == emergency_window_transmitted
        and (emergency_first is None or emergency_first > OPERATOR_EMERGENCY_SECONDS + 0.500)
    )
    return TransportTrial(
        tuple(control_receives),
        emergency_first is not None,
        emergency_first,
        emergency_copies,
        emergency_lost,
        emergency_ack_first,
        emergency_ack_lost,
        all_window_lost,
        intent_fraction,
        positive_after,
        permitted_after,
        suppression,
        int(mode is Mode.PERSISTENT_SAFE_RESET),
        reset_requests_transmitted,
        reset_receiver.accepted,
        reset_receiver.rejected,
        tuple(sorted(rejection_counts.items())),
        reset_first_accept,
        reset_ack_first,
        reset_ack_lost,
        reset_receiver.duplicates,
        reset_receiver.reacknowledgements,
        new_motion_command_time,
        0,
        tuple(sorted(offered.items())),
        video_published,
        telemetry_delivered,
        maximum_queue,
        negative_time_order,
    )


def simulate_motion(trial: TransportTrial, mode: Mode, initial_speed_kmh: float) -> MotionTrial:
    initial_speed = initial_speed_kmh / 3.6
    controls = tuple(item for item in trial.control_receives if item[0] <= DURATION_SECONDS + EPSILON)
    control_index = 0
    current = 0.0
    speed = target_speed = initial_speed
    distance = 0.0
    operator_distance = 0.0
    fsm = TwoStageFailsafe(POLICY, PROFILE)
    emergency_done = reset_done = False
    emergency_time = trial.emergency_first_receive_seconds
    reset_time = trial.reset_first_accept_seconds
    braking_start_time: float | None = None
    braking_start_distance: float | None = None
    stop_time: float | None = None
    stop_distance: float | None = None
    resume_time: float | None = None
    resume_distance: float | None = None
    max_speed_after = initial_speed
    reaccelerations = braking_releases = movement_intent = movement_before_ack = 0
    recovery_after_lost = unsafe_recovery = 0
    negative_speed_cases = target_exceeded_cases = 0
    positive_segment_active = False
    intent_end = (
        OPERATOR_EMERGENCY_SECONDS + 0.500
        if mode is Mode.LIMITED_LAB038
        else trial.reset_ack_seconds if mode is Mode.PERSISTENT_SAFE_RESET and trial.reset_ack_seconds is not None
        else DURATION_SECONDS
    )

    while current < DURATION_SECONDS - EPSILON:
        candidates: list[tuple[float, int, str]] = [(DURATION_SECONDS, 9, "end")]
        if control_index < len(controls):
            candidates.append((controls[control_index][0], 1, "control"))
        if emergency_time is not None and not emergency_done and emergency_time <= DURATION_SECONDS + EPSILON:
            candidates.append((emergency_time, 0, "emergency"))
        if reset_time is not None and not reset_done and reset_time <= DURATION_SECONDS + EPSILON:
            candidates.append((reset_time, 0, "reset"))
        if current < OPERATOR_EMERGENCY_SECONDS - EPSILON:
            candidates.append((OPERATOR_EMERGENCY_SECONDS, 0, "operator"))
        if fsm.state is SafetyState.NORMAL:
            assert POLICY.stage1_seconds is not None
            first = fsm.last_fresh_time_seconds + POLICY.stage1_seconds
            second = fsm.last_fresh_time_seconds + POLICY.stage2_seconds
            if first > current + EPSILON:
                candidates.append((first, 3, "stage1"))
            if second > current + EPSILON:
                candidates.append((second, 2, "stage2"))
        elif fsm.state is SafetyState.STAGE1_DECELERATION:
            second = fsm.last_fresh_time_seconds + POLICY.stage2_seconds
            if second > current + EPSILON:
                candidates.append((second, 2, "stage2"))
        next_time, _, event_type = min(candidates, key=lambda item: (item[0], item[1]))
        duration = max(0.0, next_time - current)
        acceleration = fsm.acceleration_mps2(speed, target_speed)
        if acceleration > 0.0 and current >= OPERATOR_EMERGENCY_SECONDS - EPSILON and braking_start_time is not None:
            if not positive_segment_active:
                reaccelerations += 1
                if resume_time is None:
                    resume_time = current
                    resume_distance = distance
                if current < intent_end - EPSILON:
                    movement_intent += 1
                if mode is Mode.PERSISTENT_SAFE_RESET and (
                    trial.reset_ack_seconds is None or current < trial.reset_ack_seconds - EPSILON
                ):
                    movement_before_ack += 1
                    braking_releases += 1
                if trial.all_first_500ms_copies_lost:
                    recovery_after_lost = 1
                    unsafe_recovery = int(mode is Mode.LIMITED_LAB038)
            positive_segment_active = True
        else:
            positive_segment_active = False
        step = integrate_kinematics(
            speed,
            acceleration,
            duration,
            target_speed_mps=target_speed if acceleration > 0.0 else None,
        )
        if current >= OPERATOR_EMERGENCY_SECONDS - EPSILON:
            max_speed_after = max(max_speed_after, speed, step.final_speed_mps)
        if step.zero_after_seconds is not None and stop_time is None and current >= OPERATOR_EMERGENCY_SECONDS - EPSILON:
            stop_time = current + step.zero_after_seconds
            stop_distance = distance + step.distance_m
        distance += step.distance_m
        speed = step.final_speed_mps
        current = next_time
        negative_speed_cases += int(speed < -EPSILON)
        target_exceeded_cases += int(speed > initial_speed + 1e-9)

        if event_type == "end":
            break
        if event_type == "operator":
            operator_distance = distance
            if fsm.state in (
                SafetyState.STAGE1_DECELERATION,
                SafetyState.STAGE2_BRAKING,
                SafetyState.EMERGENCY_LATCHED,
            ):
                braking_start_time = current
                braking_start_distance = distance
            continue
        if event_type == "emergency":
            emergency_done = True
            target_speed = 0.0
            fsm.receive_emergency(STREAM_EMERGENCY, EMERGENCY_SEQUENCE)
            if braking_start_time is None:
                braking_start_time = current
                braking_start_distance = distance
            continue
        if event_type == "reset":
            reset_done = True
            fsm.state = SafetyState.NORMAL
            target_speed = 0.0
            continue
        if event_type == "control":
            receive_time, sequence, state = controls[control_index]
            control_index += 1
            accepted = fsm.receive_command(sequence, receive_time)
            if not accepted:
                continue
            if fsm.state is SafetyState.EMERGENCY_LATCHED:
                target_speed = 0.0
            elif state.braking or not state.movement_allowed or state.desired_speed_mps <= 0.0:
                target_speed = 0.0
                fsm.state = SafetyState.STAGE2_BRAKING
                if current >= OPERATOR_EMERGENCY_SECONDS - EPSILON and braking_start_time is None:
                    braking_start_time = current
                    braking_start_distance = distance
            else:
                target_speed = initial_speed
                fsm.state = SafetyState.NORMAL
            continue
        if event_type in ("stage1", "stage2"):
            fsm.apply_watchdog(current)
            if current >= OPERATOR_EMERGENCY_SECONDS - EPSILON and braking_start_time is None:
                braking_start_time = current
                braking_start_distance = distance

    stopped = stop_time is not None and stop_distance is not None
    distance_to_braking = (
        max(0.0, braking_start_distance - operator_distance)
        if braking_start_distance is not None
        else 0.0
    )
    distance_to_stop = max(0.0, stop_distance - operator_distance) if stop_distance is not None else 0.0
    time_to_stop = max(0.0, stop_time - OPERATOR_EMERGENCY_SECONDS) if stop_time is not None else 0.0
    distance_after_resume = max(0.0, distance - resume_distance) if unsafe_recovery and resume_distance is not None else 0.0
    stop_to_resume = (
        max(0.0, resume_time - stop_time)
        if unsafe_recovery and resume_time is not None and stop_time is not None
        else 0.0
    )
    actual_resume = (
        max(0.0, resume_time - OPERATOR_RESET_SECONDS)
        if mode is Mode.PERSISTENT_SAFE_RESET and resume_time is not None
        else 0.0
    )
    return MotionTrial(
        distance_to_braking,
        distance_to_stop,
        time_to_stop,
        max_speed_after,
        stopped,
        reaccelerations if mode is not Mode.PERSISTENT_SAFE_RESET else 0,
        braking_releases,
        movement_intent,
        recovery_after_lost,
        int(trial.all_first_500ms_copies_lost and stopped and emergency_time is None),
        unsafe_recovery,
        stop_to_resume,
        distance_after_resume,
        movement_before_ack,
        actual_resume,
        negative_speed_cases,
        target_exceeded_cases,
    )


def aggregate_combination(
    workload: Workload,
    condition: ChannelCondition,
    mode: Mode,
    speed_kmh: float,
    transports: tuple[TransportTrial, ...],
) -> tuple[SummaryMetrics, EmergencyMetrics, ResetMetrics, MotionMetrics, LoadMetrics]:
    motions = tuple(simulate_motion(trial, mode, speed_kmh) for trial in transports)
    mode_name = MODE_NAMES[mode]
    latch_delays = tuple(
        (trial.emergency_first_receive_seconds - OPERATOR_EMERGENCY_SECONDS) * 1000.0
        for trial in transports
        if trial.emergency_first_receive_seconds is not None
    )
    emergency_ack_delays = tuple(
        (trial.emergency_ack_seconds - OPERATOR_EMERGENCY_SECONDS) * 1000.0
        for trial in transports
        if trial.emergency_ack_seconds is not None
    )
    reset_accept_delays = tuple(
        (trial.reset_first_accept_seconds - OPERATOR_RESET_SECONDS) * 1000.0
        for trial in transports
        if trial.reset_first_accept_seconds is not None
    )
    reset_ack_delays = tuple(
        (trial.reset_ack_seconds - OPERATOR_RESET_SECONDS) * 1000.0
        for trial in transports
        if trial.reset_ack_seconds is not None
    )
    ack_to_motion = tuple(
        trial.new_motion_command_seconds - trial.reset_ack_seconds
        for trial in transports
        if trial.new_motion_command_seconds is not None and trial.reset_ack_seconds is not None
    )
    reason_totals = {reason.name.lower(): 0 for reason in ResetReason if reason is not ResetReason.ACCEPTED}
    for trial in transports:
        for reason, count in trial.reset_rejection_reasons:
            reason_totals[reason] += count
    mean_bytes = {
        kind: _mean(dict(trial.offered_bytes)[kind] for trial in transports)
        for kind in dict(transports[0].offered_bytes)
    }
    loads = {kind: value * 8.0 / DURATION_SECONDS / 1000.0 for kind, value in mean_bytes.items()}
    extra_load = loads["emergency"] + loads["emergency_ack"] + loads["reset"] + loads["reset_ack"]
    video_fraction = _mean(trial.video_published for trial in transports) / len(workload.frames)
    telemetry_fraction = _mean(trial.telemetry_delivered for trial in transports) / len(workload.telemetry)
    summary = SummaryMetrics(
        condition.name,
        mode_name,
        speed_kmh,
        len(transports),
        _mean(float(trial.emergency_delivered) for trial in transports),
        _mean(trial.emergency_intent_fraction for trial in transports),
        _mean(motion.unsafe_recovery for motion in motions),
        _mean(float(motion.stopped) for motion in motions),
        sum(motion.movement_while_intent for motion in motions),
        sum(motion.movement_before_reset_ack for motion in motions),
        _mean(trial.positive_commands_after_emergency for trial in transports),
        _mean(trial.permitted_commands_after_emergency for trial in transports),
        extra_load,
        video_fraction,
        telemetry_fraction,
        max(trial.max_queue_packets for trial in transports),
    )
    emergency = EmergencyMetrics(
        condition.name,
        mode_name,
        speed_kmh,
        len(transports),
        summary.emergency_delivery_fraction,
        _mean(latch_delays),
        _percentile(latch_delays, 95),
        max(latch_delays, default=0.0),
        _mean(trial.emergency_copies for trial in transports),
        _mean(trial.emergency_copies_lost for trial in transports),
        _mean(emergency_ack_delays),
        _mean(trial.emergency_acks_lost for trial in transports),
        summary.emergency_intent_time_fraction,
        summary.positive_commands_after_emergency_mean,
        summary.permitted_commands_after_emergency_mean,
        _mean(trial.time_to_motion_command_suppression_seconds for trial in transports) * 1000.0,
        sum(trial.all_first_500ms_copies_lost for trial in transports),
    )
    reset = ResetMetrics(
        condition.name,
        mode_name,
        speed_kmh,
        len(transports),
        _mean(trial.reset_requests_created for trial in transports),
        _mean(trial.reset_requests_transmitted for trial in transports),
        _mean(trial.reset_requests_accepted for trial in transports),
        _mean(trial.reset_requests_rejected for trial in transports),
        reason_totals["not_latched"],
        reason_totals["moving"],
        reason_totals["stale_link"],
        reason_totals["event_mismatch"],
        reason_totals["old_sequence"],
        reason_totals["nonzero_request"],
        reason_totals["movement_permitted"],
        _mean(reset_accept_delays),
        _mean(reset_ack_delays),
        _percentile(reset_ack_delays, 95),
        _mean(trial.reset_acks_lost for trial in transports),
        _mean(max(0, trial.reset_requests_transmitted - 1) for trial in transports),
        _mean(trial.reset_duplicates for trial in transports),
        _mean(trial.reset_reacks for trial in transports),
        _mean(ack_to_motion),
        _mean(motion.actual_resume_seconds for motion in motions),
        sum(motion.movement_before_reset_ack for motion in motions),
        sum(trial.automatic_old_command_restorations for trial in transports),
    )
    distance_braking = tuple(motion.distance_to_braking_m for motion in motions)
    distance_stop = tuple(motion.distance_to_stop_m for motion in motions if motion.stopped)
    time_stop = tuple(motion.time_to_stop_seconds for motion in motions if motion.stopped)
    motion = MotionMetrics(
        condition.name,
        mode_name,
        speed_kmh,
        len(transports),
        _mean(distance_braking),
        _percentile(distance_braking, 95),
        max(distance_braking, default=0.0),
        _mean(distance_stop),
        _percentile(distance_stop, 95),
        max(distance_stop, default=0.0),
        _mean(time_stop),
        _percentile(time_stop, 95),
        max(item.maximum_speed_after_emergency_mps for item in motions),
        summary.full_stop_fraction,
        sum(item.reaccelerations_before_reset for item in motions),
        sum(item.braking_releases_before_reset_ack for item in motions),
        sum(item.movement_while_intent for item in motions),
        sum(item.recovery_after_lost_emergency for item in motions),
        sum(item.watchdog_stopped_after_lost_emergency for item in motions),
        sum(item.unsafe_recovery for item in motions),
        _mean(item.stop_to_resume_seconds for item in motions if item.unsafe_recovery),
        _mean(item.distance_after_unsafe_resume_m for item in motions if item.unsafe_recovery),
        sum(item.negative_speed_cases for item in motions),
        sum(item.target_speed_exceeded_cases for item in motions),
    )
    load = LoadMetrics(
        condition.name,
        mode_name,
        speed_kmh,
        len(transports),
        loads["emergency"],
        loads["emergency_ack"],
        loads["zero_control"],
        loads["reset"],
        loads["reset_ack"],
        extra_load,
        video_fraction,
        telemetry_fraction,
        summary.maximum_queue_packets,
    )
    return summary, emergency, reset, motion, load


def run_experiment(
    workload: Workload,
) -> tuple[
    tuple[SummaryMetrics, ...],
    tuple[EmergencyMetrics, ...],
    tuple[ResetMetrics, ...],
    tuple[MotionMetrics, ...],
    tuple[LoadMetrics, ...],
    dict[tuple[str, str], tuple[TransportTrial, ...]],
]:
    transport_sets: dict[tuple[str, str], tuple[TransportTrial, ...]] = {}
    for condition_index, condition in enumerate(CHANNELS):
        for mode in Mode:
            rows = []
            for repetition in range(condition.repetitions):
                if condition.mean_bad_ms is None:
                    starts = ends = np.asarray([], dtype=float)
                else:
                    seed = SEED_OFFSET + condition_index * 100_000 + repetition
                    starts, ends = bad_intervals(DURATION_SECONDS + 2.0, condition.mean_bad_ms, seed)
                rows.append(simulate_transport(workload, mode, condition.rate_kbps, starts, ends))
            transport_sets[(condition.name, MODE_NAMES[mode])] = tuple(rows)
            print(f"transport={condition.name}/{MODE_NAMES[mode]} repetitions={condition.repetitions}")
    summaries: list[SummaryMetrics] = []
    emergency_rows: list[EmergencyMetrics] = []
    reset_rows: list[ResetMetrics] = []
    motion_rows: list[MotionMetrics] = []
    load_rows: list[LoadMetrics] = []
    for condition in CHANNELS:
        for mode in Mode:
            transports = transport_sets[(condition.name, MODE_NAMES[mode])]
            for speed in SPEEDS_KMH:
                aggregate = aggregate_combination(workload, condition, mode, speed, transports)
                summaries.append(aggregate[0])
                emergency_rows.append(aggregate[1])
                reset_rows.append(aggregate[2])
                motion_rows.append(aggregate[3])
                load_rows.append(aggregate[4])
        print(f"aggregate={condition.name} combinations=9")
    return (
        tuple(summaries),
        tuple(emergency_rows),
        tuple(reset_rows),
        tuple(motion_rows),
        tuple(load_rows),
        transport_sets,
    )


def run_functional_tests() -> tuple[FunctionalTestResult, ...]:
    results: list[FunctionalTestResult] = []

    def check(name: str, function) -> None:
        try:
            detail = function() or "проверка выполнена"
            results.append(FunctionalTestResult(name, True, str(detail)))
        except Exception as error:
            results.append(FunctionalTestResult(name, False, f"{type(error).__name__}: {error}"))

    def ground_controller() -> PersistentEmergencyController:
        controller = PersistentEmergencyController(2.0)
        controller.request_emergency(EVENT_ID, STREAM_EMERGENCY, EMERGENCY_SEQUENCE)
        return controller

    def latched_rover() -> TwoStageFailsafe:
        rover = TwoStageFailsafe(POLICY, PROFILE)
        rover.receive_emergency(STREAM_EMERGENCY, EMERGENCY_SEQUENCE)
        return rover

    def reset_packet(
        *,
        event_id: int = EVENT_ID,
        request_id: int = RESET_REQUEST_ID,
        sequence: int = RESET_SEQUENCE,
        speed: float = 0.0,
        permitted: bool = False,
    ) -> LinkPacket:
        return build_reset_packet(
            ResetCommand(event_id, request_id, speed, permitted),
            sequence,
            int(OPERATOR_RESET_SECONDS * 1_000_000),
        )

    def emergency_intent_latches() -> str:
        controller = ground_controller()
        assert controller.emergency_intent and controller.state is GroundState.GROUND_EMERGENCY_REQUESTED
        return "действие оператора фиксирует аварийное намерение наземной станции"

    def no_positive_after_intent() -> str:
        state = ground_controller().control_state()
        assert state.desired_speed_mps == 0.0 and state.braking
        return "аварийное намерение заменяет движение безопасным нулевым состоянием с торможением"

    def movement_false_until_reset() -> str:
        controller = ground_controller()
        assert not controller.control_state().movement_allowed
        controller.receive_emergency_ack(EmergencyIdentity(STREAM_EMERGENCY, EMERGENCY_SEQUENCE))
        assert not controller.control_state().movement_allowed
        return "после подтверждения аварийной команды движение остаётся запрещённым"

    def emergency_loss_keeps_intent() -> str:
        controller = ground_controller()
        starts = np.asarray([OPERATOR_EMERGENCY_SECONDS - 0.001])
        ends = np.asarray([OPERATOR_EMERGENCY_SECONDS + 0.510])
        size = len(encode_link_packet(_packet(TrafficClass.EMERGENCY, Direction.GROUND_TO_ROVER, STREAM_EMERGENCY, 1, 30.0, b"x")))
        duration = size * 8.0 / 300_000.0
        assert all(
            packet_is_lost(30.0 + offset, 30.0 + offset + duration, starts, ends)
            for offset in np.arange(0.0, 0.501, 0.050)
        )
        assert controller.emergency_intent
        return "управляемая помеха теряет все копии за первые 500 мс, не снимая аварийное намерение"

    def emergency_ack_keeps_intent() -> str:
        controller = ground_controller()
        rover = latched_rover()
        command = _packet(TrafficClass.EMERGENCY, Direction.GROUND_TO_ROVER, STREAM_EMERGENCY, 1, 30.0, b"x")
        ack = build_emergency_ack(command, 0, 30_001_000)
        starts = np.asarray([30.001])
        ends = np.asarray([30.010])
        assert packet_is_lost(30.001, 30.003, starts, ends)
        assert controller.emergency_intent and rover.state is SafetyState.EMERGENCY_LATCHED
        controller.receive_emergency_ack(acknowledged_identity(ack))
        assert controller.emergency_intent and controller.state is GroundState.GROUND_EMERGENCY_CONFIRMED
        return "потеря первого подтверждения и доставка повтора сохраняют аварийное намерение"

    def first_emergency_latches_rover() -> str:
        rover = latched_rover()
        assert rover.state is SafetyState.EMERGENCY_LATCHED and rover.emergency_actions == 1
        return "первая аварийная команда фиксирует аварийное состояние ровера"

    def ordinary_does_not_unlatch() -> str:
        rover = latched_rover()
        rover.receive_command(1, 0.1)
        assert rover.state is SafetyState.EMERGENCY_LATCHED
        return "обычная команда не снимает аварийное состояние ровера"

    def late_old_command_rejected() -> str:
        rover = latched_rover()
        rover.receive_command(10, 0.1)
        assert not rover.receive_command(9, 0.2)
        assert rover.state is SafetyState.EMERGENCY_LATCHED
        return "запоздалая команда движения со старым номером последовательности отклоняется"

    def reject_moving() -> str:
        receiver = SafeResetReceiver()
        decision = receiver.process(reset_packet(), rover=latched_rover(), actual_speed_mps=0.1, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert decision.acknowledgement.reason is ResetReason.MOVING
        return "движущийся ровер отклоняет запрос сброса"

    def reject_stale_link() -> str:
        receiver = SafeResetReceiver()
        decision = receiver.process(reset_packet(), rover=latched_rover(), actual_speed_mps=0.0, last_link_age_seconds=0.251, emergency_event_id=EVENT_ID)
        assert decision.acknowledgement.reason is ResetReason.STALE_LINK
        return "при устаревшем состоянии связи запрос сброса отклоняется"

    def reject_wrong_event() -> str:
        receiver = SafeResetReceiver()
        decision = receiver.process(reset_packet(event_id=2), rover=latched_rover(), actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert decision.acknowledgement.reason is ResetReason.EVENT_MISMATCH
        return "запрос сброса с неверным идентификатором аварии отклоняется"

    def reject_old_sequence() -> str:
        receiver = SafeResetReceiver()
        receiver.process(reset_packet(sequence=2), rover=latched_rover(), actual_speed_mps=1.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        decision = receiver.process(reset_packet(sequence=1), rover=latched_rover(), actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert decision.acknowledgement.reason is ResetReason.OLD_SEQUENCE
        return "запрос сброса со старым номером последовательности отклоняется"

    def reject_nonzero_speed_request() -> str:
        receiver = SafeResetReceiver()
        decision = receiver.process(reset_packet(speed=0.1), rover=latched_rover(), actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert decision.acknowledgement.reason is ResetReason.NONZERO_REQUEST
        permitted = SafeResetReceiver().process(
            reset_packet(permitted=True),
            rover=latched_rover(),
            actual_speed_mps=0.0,
            last_link_age_seconds=0.0,
            emergency_event_id=EVENT_ID,
        )
        assert permitted.acknowledgement.reason is ResetReason.MOVEMENT_PERMITTED
        return "ненулевая скорость или разрешение движения приводят к отклонению сброса"

    def successful_reset_is_zero() -> str:
        rover = latched_rover()
        receiver = SafeResetReceiver()
        decision = receiver.process(reset_packet(), rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert decision.performed and rover.state is SafetyState.NORMAL
        assert decode_reset_ack(build_reset_ack(decision, 0, 60_001_000).payload).accepted
        return "принятый сброс выводит из аварийного состояния только в состояние нулевой скорости"

    def no_movement_before_reset_ack() -> str:
        controller = ground_controller()
        controller.request_reset(RESET_REQUEST_ID, RESET_SEQUENCE)
        assert controller.state is GroundState.GROUND_RESET_REQUESTED
        assert controller.control_state() == SAFE_STATE
        return "до подтверждения сброса наземная станция сохраняет безопасное нулевое задание"

    def lost_reset_ack_repeats() -> str:
        rover = latched_rover()
        receiver = SafeResetReceiver()
        packet = reset_packet()
        first = receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        starts = np.asarray([60.001])
        ends = np.asarray([60.010])
        assert packet_is_lost(60.001, 60.003, starts, ends)
        repeated = receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert first.performed and repeated.duplicate and repeated.acknowledgement.accepted
        return "управляемая потеря первого подтверждения сброса вызывает повтор запроса"

    def duplicate_does_not_reset_twice() -> str:
        rover = latched_rover()
        receiver = SafeResetReceiver()
        packet = reset_packet()
        receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        duplicate = receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert not duplicate.performed and receiver.accepted == 1
        return "принятый дубликат не выполняет сброс повторно"

    def duplicate_reacks() -> str:
        rover = latched_rover()
        receiver = SafeResetReceiver()
        packet = reset_packet()
        receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        duplicate = receiver.process(packet, rover=rover, actual_speed_mps=0.0, last_link_age_seconds=0.0, emergency_event_id=EVENT_ID)
        assert duplicate.acknowledgement.accepted and receiver.reacknowledgements == 1
        return "принятый дубликат вызывает повторное подтверждение сброса"

    def no_automatic_old_motion() -> str:
        controller = ground_controller()
        controller.request_reset(RESET_REQUEST_ID, RESET_SEQUENCE)
        controller.receive_reset_ack(EVENT_ID, RESET_REQUEST_ID, True)
        assert controller.control_state() == SAFE_STATE
        assert controller.reject_old_operator_motion(0)
        return "подтверждение сброса не восстанавливает сохранённую команду движения"

    def new_operator_command_required() -> str:
        controller = ground_controller()
        controller.request_reset(RESET_REQUEST_ID, RESET_SEQUENCE)
        controller.receive_reset_ack(EVENT_ID, RESET_REQUEST_ID, True)
        try:
            controller.new_operator_motion(0, 1.0)
            raise AssertionError("принят старый номер последовательности команды оператора")
        except ValueError:
            pass
        state = controller.new_operator_motion(1, 1.0)
        assert state.movement_allowed and state.desired_speed_mps > 0.0
        return "для движения требуется новый номер последовательности команды оператора"

    def emergency_latch_against_ordinary() -> str:
        rover = latched_rover()
        rover.receive_command(100, 1.0)
        assert rover.state is SafetyState.EMERGENCY_LATCHED
        return "состояние EMERGENCY_LATCHED не снимается обычной командой управления"

    def watchdog_without_emergency() -> str:
        rover = TwoStageFailsafe(POLICY, PROFILE)
        assert rover.apply_watchdog(0.150) is SafetyState.STAGE1_DECELERATION
        assert rover.apply_watchdog(0.250) is SafetyState.STAGE2_BRAKING
        return "локальный двухступенчатый сторожевой таймер работает без доставки аварийной команды"

    def reproducible() -> str:
        left = bad_intervals(122.0, 1000.0, SEED_OFFSET + 123)
        right = bad_intervals(122.0, 1000.0, SEED_OFFSET + 123)
        assert np.array_equal(left[0], right[0]) and np.array_equal(left[1], right[1])
        return "фиксированное начальное значение генератора воспроизводит интервалы помех"

    def operator_during_braking_has_zero_reaction_distance() -> str:
        trial = TransportTrial(
            control_receives=((29.800, 1, ControlState(1.0, 0.0, False, True)),),
            emergency_delivered=False,
            emergency_first_receive_seconds=None,
            emergency_copies=0,
            emergency_copies_lost=0,
            emergency_ack_seconds=None,
            emergency_acks_lost=0,
            all_first_500ms_copies_lost=False,
            emergency_intent_fraction=0.0,
            positive_commands_after_emergency=0,
            permitted_commands_after_emergency=0,
            time_to_motion_command_suppression_seconds=0.0,
            reset_requests_created=0,
            reset_requests_transmitted=0,
            reset_requests_accepted=0,
            reset_requests_rejected=0,
            reset_rejection_reasons=(),
            reset_first_accept_seconds=None,
            reset_ack_seconds=None,
            reset_acks_lost=0,
            reset_duplicates=0,
            reset_reacks=0,
            new_motion_command_seconds=None,
            automatic_old_command_restorations=0,
            offered_bytes=(),
            video_published=0,
            telemetry_delivered=0,
            max_queue_packets=0,
            negative_time_order=0,
        )
        motion = simulate_motion(trial, Mode.LIMITED_LAB038, 25.0)
        assert motion.distance_to_braking_m == 0.0
        return "действие оператора при уже активном торможении даёт нулевой путь до начала реакции"

    checks = (
        ("01. Оператор фиксирует аварийное намерение", emergency_intent_latches),
        ("02. После аварии нет положительной заданной скорости", no_positive_after_intent),
        ("03. Движение запрещено до сброса", movement_false_until_reset),
        ("04. Потеря аварийной команды сохраняет намерение", emergency_loss_keeps_intent),
        ("05. Подтверждение аварии сохраняет намерение", emergency_ack_keeps_intent),
        ("06. Первая аварийная команда фиксирует состояние ровера", first_emergency_latches_rover),
        ("07. Обычная команда не снимает аварийное состояние", ordinary_does_not_unlatch),
        ("08. Старая команда движения отклоняется", late_old_command_rejected),
        ("09. Сброс при движении отклоняется", reject_moving),
        ("10. Сброс при устаревшей связи отклоняется", reject_stale_link),
        ("11. Сброс для неверного события отклоняется", reject_wrong_event),
        ("12. Сброс со старым номером последовательности отклоняется", reject_old_sequence),
        ("13. Сброс с ненулевой скоростью отклоняется", reject_nonzero_speed_request),
        ("14. После успешного сброса скорость остаётся нулевой", successful_reset_is_zero),
        ("15. До подтверждения сброса движение отсутствует", no_movement_before_reset_ack),
        ("16. Потеря подтверждения сброса вызывает повтор", lost_reset_ack_repeats),
        ("17. Дубликат не выполняет сброс дважды", duplicate_does_not_reset_twice),
        ("18. На дубликат повторно отправляется подтверждение", duplicate_reacks),
        ("19. Старая команда движения не восстанавливается", no_automatic_old_motion),
        ("20. Для движения требуется новая команда оператора", new_operator_command_required),
        ("21. Аварийное состояние устойчиво к обычным командам", emergency_latch_against_ordinary),
        ("22. Сторожевой таймер работает при потере аварийной команды", watchdog_without_emergency),
        ("23. Интервалы помех воспроизводимы", reproducible),
        ("24. Действие оператора во время торможения", operator_during_braking_has_zero_reaction_distance),
    )
    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", encoding="utf-8", newline="") 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, ...],
    emergencies: tuple[EmergencyMetrics, ...],
    resets: tuple[ResetMetrics, ...],
    motions: tuple[MotionMetrics, ...],
    loads: tuple[LoadMetrics, ...],
) -> None:
    colors = {
        "limited_lab038": "#d95f02",
        "persistent_no_reset": "#2878b5",
        "persistent_safe_reset": "#2a9d55",
    }
    channel_names = [condition.name for condition in CHANNELS]
    channel_labels = [condition.label for condition in CHANNELS]
    x = np.arange(len(CHANNELS))

    figure, axis = plt.subplots(figsize=(10, 5))
    for index, mode in enumerate(MODE_NAMES.values()):
        rows = [row for row in emergencies if row.mode == mode and row.initial_speed_kmh == 5.0]
        axis.bar(x + (index - 1) * 0.24, [row.delivered_fraction * 100.0 for row in rows], 0.24, label=MODE_LABELS[mode], color=colors[mode])
    axis.set_xticks(x, channel_labels, rotation=15)
    axis.set_ylabel("Доставка аварийной команды, %")
    axis.set_title("Итоговая доставка аварийного намерения")
    axis.set_ylim(90, 101)
    axis.grid(True, axis="y", alpha=0.3)
    axis.legend()
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[0], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(9, 5))
    for speed in SPEEDS_KMH:
        rows = [row for row in motions if row.mode == "limited_lab038" and row.initial_speed_kmh == speed]
        axis.plot(channel_labels, [row.unsafe_recoveries / row.repetitions * 100.0 for row in rows], marker="o", label=f"{speed:.0f} км/ч")
    axis.set_ylabel("Небезопасные восстановления, %")
    axis.set_title("Риск ограниченного представления аварийного намерения")
    axis.tick_params(axis="x", rotation=15)
    axis.grid(True, alpha=0.3)
    axis.legend(title="Начальная скорость")
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[1], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(9, 5))
    for mode in MODE_NAMES.values():
        rows = [row for row in emergencies if row.mode == mode and row.initial_speed_kmh == 25.0]
        axis.plot(channel_labels, [row.p95_latch_delay_ms for row in rows], marker="o", label=MODE_LABELS[mode], color=colors[mode])
    axis.set_ylabel("95-й процентиль времени фиксации, мс")
    axis.set_title("Время до состояния аварийной остановки")
    axis.tick_params(axis="x", rotation=15)
    axis.grid(True, alpha=0.3)
    axis.legend()
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[2], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(9, 5))
    for mode in MODE_NAMES.values():
        rows = [row for row in motions if row.mode == mode and row.initial_speed_kmh == 25.0]
        axis.plot(channel_labels, [row.max_distance_to_stop_m for row in rows], marker="o", label=MODE_LABELS[mode], color=colors[mode])
    axis.set_ylabel("Максимальный путь до остановки, м")
    axis.set_title("Движение после действия оператора при 25 км/ч")
    axis.tick_params(axis="x", rotation=15)
    axis.grid(True, alpha=0.3)
    axis.legend()
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[3], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(9, 5))
    rows = [row for row in resets if row.mode == "persistent_safe_reset" and row.initial_speed_kmh == 25.0]
    axis.bar(channel_labels, [row.p95_reset_ack_delay_ms for row in rows], color="#2a9d55")
    axis.set_ylabel("95-й процентиль времени подтверждения, мс")
    axis.set_title("Безопасное подтверждение сброса")
    axis.tick_params(axis="x", rotation=15)
    axis.grid(True, axis="y", alpha=0.3)
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[4], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(10, 5))
    for index, mode in enumerate(MODE_NAMES.values()):
        rows = [row for row in loads if row.mode == mode and row.initial_speed_kmh == 5.0]
        axis.bar(x + (index - 1) * 0.24, [row.total_extra_load_kbps for row in rows], 0.24, label=MODE_LABELS[mode], color=colors[mode])
    axis.set_xticks(x, channel_labels, rotation=15)
    axis.set_ylabel("Дополнительная нагрузка, кбит/с")
    axis.set_title("Нагрузка постоянного намерения и безопасного сброса")
    axis.grid(True, axis="y", alpha=0.3)
    axis.legend()
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[5], dpi=150)
    plt.close(figure)

    figure, axes = plt.subplots(1, 2, figsize=(12, 5))
    for mode in MODE_NAMES.values():
        rows = [row for row in loads if row.mode == mode and row.initial_speed_kmh == 5.0]
        axes[0].plot(channel_labels, [row.video_published_fraction * 100.0 for row in rows], marker="o", label=MODE_LABELS[mode], color=colors[mode])
        axes[1].plot(channel_labels, [row.telemetry_delivered_fraction * 100.0 for row in rows], marker="o", label=MODE_LABELS[mode], color=colors[mode])
    axes[0].set_title("Опубликованные видеокадры")
    axes[0].set_ylabel("Доля, %")
    axes[1].set_title("Доставленная телеметрия")
    axes[1].set_ylabel("Доля, %")
    for axis in axes:
        axis.tick_params(axis="x", rotation=20)
        axis.grid(True, alpha=0.3)
    axes[1].legend(fontsize=8)
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[6], dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(8, 5))
    for mode in MODE_NAMES.values():
        rows = [row for row in summaries if row.mode == mode]
        axis.scatter(
            _mean(row.total_extra_load_kbps for row in rows),
            _mean(row.unsafe_recovery_fraction for row in rows) * 100.0,
            s=100,
            color=colors[mode],
            label=MODE_LABELS[mode],
        )
    axis.set_xlabel("Средняя дополнительная нагрузка, кбит/с")
    axis.set_ylabel("Небезопасные восстановления, %")
    axis.set_title("Компромисс нагрузки и сохранения аварийного намерения")
    axis.grid(True, alpha=0.3)
    axis.legend()
    figure.tight_layout()
    figure.savefig(PLOT_PATHS[7], dpi=150)
    plt.close(figure)


def write_report(
    workload: Workload,
    summaries: tuple[SummaryMetrics, ...],
    emergencies: tuple[EmergencyMetrics, ...],
    resets: tuple[ResetMetrics, ...],
    motions: tuple[MotionMetrics, ...],
    loads: tuple[LoadMetrics, ...],
    tests: tuple[FunctionalTestResult, ...],
) -> None:
    channel_labels = {condition.name: condition.label for condition in CHANNELS}
    lines = [
        "Lab040 — постоянное аварийное намерение и безопасное повторное разрешение движения",
        "",
        "Состояние Git до начала",
        "- Корень: C:/Users/user/Desktop/projects/SDR_Rover",
        "- Ветка: main",
        "- HEAD: b3ebe1637fbb401198cdbb43a862e94614be9cb0",
        "- Рабочее дерево и индекс были чистыми; относительно локально известной origin/main расхождение составляло 0/0.",
        "- Коммиты Lab038 и Lab039 присутствовали; сетевые команды Git не выполнялись.",
        "",
        "Автоматы наземной станции и ровера",
        "- Наземная станция: GROUND_NORMAL → GROUND_EMERGENCY_REQUESTED → GROUND_EMERGENCY_CONFIRMED → GROUND_RESET_REQUESTED → GROUND_MOVEMENT_REAUTHORIZED.",
        "- Подтверждение аварийной команды прекращает её копии, но не снимает аварийное намерение и не разрешает положительные команды движения.",
        "- Ровер сохраняет состояния NORMAL, STAGE1_DECELERATION, STAGE2_BRAKING и EMERGENCY_LATCHED из Lab039.",
        "- Обычная или поздняя старая команда не снимает EMERGENCY_LATCHED.",
        "",
        "Безопасный сброс",
        "- Сброс использует отдельные идентификаторы потоков и прикладные данные внутри неизменного пакета LinkPacket из Lab033.",
        "- Проверяются зафиксированное аварийное состояние, нулевая фактическая скорость, свежесть связи 250 мс, идентификатор события, номер последовательности, нулевая требуемая скорость и запрет движения.",
        "- Потерянное подтверждение сброса вызывает повтор запроса; принятый дубликат не повторяет действие, но вызывает новое подтверждение.",
        "- После подтверждения сброса сохраняется нулевая скорость; новая команда оператора формируется только через 2 секунды и имеет новый номер последовательности.",
        "",
        "Параметры",
        f"- Опыт {DURATION_SECONDS:.0f} с; аварийная команда на 30-й секунде; сброс на 60-й секунде; команды 20 Гц; повторы каждые 50 мс.",
        f"- Матрица: 5 каналов × 3 режима × 3 скорости = {len(summaries)} сочетаний; условия с потерями по {REPETITIONS} повторов.",
        "- Сторожевой таймер 150/250 мс; замедление первой ступени 1,0 м/с², второй ступени 3,0 м/с², ускорение восстановления 1,0 м/с².",
        f"- Исходная нагрузка проверена контрольной суммой и помехоустойчивым кодированием: {workload.crc_packets_checked}/{workload.fec_blocks_checked}.",
        "",
        "Компактная таблица 45 сочетаний",
        "канал | режим | скорость | доставка аварийной команды, % | время аварийного намерения, % | небезопасное восстановление, % | остановка, % | нагрузка, кбит/с | видео, % | телеметрия, % | очередь",
    ]
    for row in summaries:
        lines.append(
            f"{channel_labels[row.channel_condition]} | {MODE_LABELS[row.mode]} | {row.initial_speed_kmh:.0f} | "
            f"{row.emergency_delivery_fraction * 100.0:.3f} | {row.emergency_intent_time_fraction * 100.0:.3f} | "
            f"{row.unsafe_recovery_fraction * 100.0:.3f} | {row.full_stop_fraction * 100.0:.3f} | "
            f"{row.total_extra_load_kbps:.5f} | {row.video_published_fraction * 100.0:.3f} | "
            f"{row.telemetry_delivered_fraction * 100.0:.3f} | {row.maximum_queue_packets}"
        )
    baseline = [row for row in motions if row.mode == "limited_lab038"]
    persistent = [row for row in summaries if row.mode in ("persistent_no_reset", "persistent_safe_reset")]
    reset_only = [row for row in resets if row.mode == "persistent_safe_reset"]
    corrected_motion = [row for row in motions if row.channel_condition == "230_1000"]
    lines.extend(
        [
            "",
            "Отрицательный исходный режим",
            f"- Полностью потерянные первые 500 мс аварийной передачи: {max(row.all_first_500ms_lost for row in emergencies if row.mode == 'limited_lab038')} на одно сочетание.",
            f"- Остановки по сторожевому таймеру после такой потери: {sum(row.watchdog_stops_after_lost_emergency for row in baseline)}.",
            f"- Небезопасные восстановления движения: {sum(row.unsafe_recoveries for row in baseline)}; максимальная доля {max(row.unsafe_recoveries / row.repetitions for row in baseline) * 100.0:.3f}%.",
            f"- Среднее время от остановки до повторного движения: {max(row.mean_stop_to_resume_seconds for row in baseline):.4f} с; путь после восстановления до {max(row.mean_distance_after_unsafe_resume_m for row in baseline):.4f} м.",
            "- Ограниченный режим Lab038 служит отрицательным исходным уровнем и не принимается для рабочей архитектуры.",
            "",
            "Постоянное аварийное намерение",
            "- Итоговая доставка 100% в режимах постоянного намерения получена в конечной выборке и не является математической гарантией.",
            "- Безопасность не зависит от гарантированной доставки удалённой команды: локальный сторожевой таймер продолжает действовать.",
            "- Подтверждение аварийной команды (Emergency ACK) прекращает копии, но не снимает аварийное намерение emergency_intent.",
            f"- Положительные задания при активном аварийном намерении: {sum(row.positive_commands_after_emergency_mean for row in persistent):.0f}.",
            f"- Команды с разрешением движения при активном аварийном намерении: {sum(row.permitted_commands_after_emergency_mean for row in persistent):.0f}.",
            f"- Небезопасные восстановления в режимах 2 и 3: {sum(row.unsafe_recovery_fraction * row.repetitions for row in persistent):.0f}.",
            "",
            "Подтверждение сброса",
            f"- Принятые запросы сброса: {_mean(row.requests_accepted_mean for row in reset_only):.4f} на опыт; отклонённые: {_mean(row.requests_rejected_mean for row in reset_only):.4f}.",
            f"- Причины отказов: аварийное состояние не зафиксировано={sum(row.rejected_not_latched for row in reset_only)}, движение={sum(row.rejected_moving for row in reset_only)}, устаревшая связь={sum(row.rejected_stale_link for row in reset_only)}, неверное событие={sum(row.rejected_event_mismatch for row in reset_only)}, старый номер последовательности={sum(row.rejected_old_sequence for row in reset_only)}, ненулевая скорость={sum(row.rejected_nonzero_request for row in reset_only)}, разрешение движения={sum(row.rejected_movement_permitted for row in reset_only)}.",
            f"- Среднее число повторов сброса: {_mean(row.repeated_requests_mean for row in reset_only):.4f}; потерянных подтверждений сброса: {_mean(row.reset_acks_lost_mean for row in reset_only):.4f}.",
            f"- 95-й процентиль времени от запроса оператора до подтверждения сброса: до {max(row.p95_reset_ack_delay_ms for row in reset_only):.3f} мс.",
            f"- Движение до подтверждения сброса: {sum(row.movement_before_reset_ack_cases for row in reset_only)}; автоматическое восстановление старой команды: {sum(row.automatic_old_command_restorations for row in reset_only)}.",
            "- Подтверждение сброса (RESET_ACK) разрешает перейти только в безопасное состояние с нулевой скоростью.",
            "- Для движения после сброса требуется новая команда оператора.",
            "",
            "Точечная проверка пути до начала безопасной реакции",
            "- Канал: 230 кбит/с, средняя помеха 1000 мс; транспортный эксперимент повторно не запускался.",
            "- Если в момент действия оператора уже активно состояние STAGE1_DECELERATION, STAGE2_BRAKING или EMERGENCY_LATCHED, путь до начала реакции равен нулю и не измеряется до будущего перехода в торможение.",
            "- режим | скорость, км/ч | среднее, м | 95-й процентиль, м | максимум, м | полные остановки, % | небезопасные возобновления",
        ]
    )
    for row in corrected_motion:
        summary_row = next(
            item
            for item in summaries
            if item.channel_condition == row.channel_condition
            and item.mode == row.mode
            and item.initial_speed_kmh == row.initial_speed_kmh
        )
        lines.append(
            f"- {MODE_LABELS[row.mode]} | {row.initial_speed_kmh:.0f} | "
            f"{row.mean_distance_to_braking_m:.6f} | {row.p95_distance_to_braking_m:.6f} | "
            f"{row.max_distance_to_braking_m:.6f} | {summary_row.full_stop_fraction * 100.0:.3f} | "
            f"{row.unsafe_recoveries}"
        )
    lines.extend(
        [
            "- Сводная и двигательная таблицы согласованы по числу повторов, доле полных остановок и небезопасным возобновлениям.",
            "- Ни один график Lab040 напрямую не использует путь до начала реакции; график пути до остановки использует отдельный показатель полного пути и не требует изменения.",
            "",
            "Движение и нагрузка",
            f"- Максимальный путь до полной остановки: {max(row.max_distance_to_stop_m for row in motions):.4f} м; максимальное время: {max(row.p95_time_to_stop_seconds for row in motions):.4f} с.",
            f"- Дополнительная нагрузка защиты: {_mean(row.total_extra_load_kbps for row in loads):.5f}…{max(row.total_extra_load_kbps for row in loads):.5f} кбит/с.",
            f"- Нагрузка нулевых состояний: до {max(row.zero_control_load_kbps for row in loads):.5f} кбит/с; максимальная очередь {max(row.maximum_queue_packets for row in loads)} пакетов.",
            f"- Видео опубликовано: {min(row.video_published_fraction for row in loads) * 100.0:.3f}…{max(row.video_published_fraction for row in loads) * 100.0:.3f}%; телеметрия: {min(row.telemetry_delivered_fraction for row in loads) * 100.0:.3f}…{max(row.telemetry_delivered_fraction for row in loads) * 100.0:.3f}%.",
            "",
            "Функциональные проверки",
        ]
    )
    lines.extend(f"- {'ПРОЙДЕНО' if item.passed else 'ОШИБКА'} — {item.name}: {item.detail}" for item in tests)
    created = (
        Path("protocol/persistent_emergency.py"),
        Path("protocol/safe_reset.py"),
        Path("tests/lab040_persistent_emergency.py"),
        SUMMARY_CSV,
        EMERGENCY_CSV,
        RESET_CSV,
        MOTION_CSV,
        LOAD_CSV,
        REPORT_PATH,
        *PLOT_PATHS,
    )
    lines.extend(["", "Созданные файлы", *[f"- {path.as_posix()}" for path in created]])
    lines.extend(
        [
            "",
            "Ограничения",
            "- Параметры замедления предварительные и требуют измерения на реальном ровере.",
            "- Модель одномерная и не заменяет испытания тормозов, привода и процедуры операторского сброса.",
            "- Lab040 не добавлена в Git и не закоммичена; Lab028–Lab039 не изменены.",
        ]
    )
    REPORT_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")


def validate_outputs(
    summaries: tuple[SummaryMetrics, ...],
    emergencies: tuple[EmergencyMetrics, ...],
    resets: tuple[ResetMetrics, ...],
    motions: tuple[MotionMetrics, ...],
    loads: tuple[LoadMetrics, ...],
    tests: tuple[FunctionalTestResult, ...],
    transport_sets: dict[tuple[str, str], tuple[TransportTrial, ...]],
) -> None:
    assert len(summaries) == len(emergencies) == len(resets) == len(motions) == len(loads) == 45
    assert len(tests) == 24 and all(item.passed for item in tests), [item for item in tests if not item.passed]
    assert all(
        len(transport_sets[(condition.name, MODE_NAMES[mode])]) == condition.repetitions
        for condition in CHANNELS
        for mode in Mode
    )
    assert all(row.repetitions == (1 if row.channel_condition == "no_loss" else REPETITIONS) for row in summaries)
    assert all(trial.negative_time_order == 0 for trials in transport_sets.values() for trial in trials)
    persistent_summaries = [
        row for row in summaries if row.mode in ("persistent_no_reset", "persistent_safe_reset")
    ]
    assert sum(row.positive_commands_after_emergency_mean for row in persistent_summaries) == 0.0
    assert sum(row.permitted_commands_after_emergency_mean for row in persistent_summaries) == 0.0
    persistent_motions = [
        row for row in motions if row.mode in ("persistent_no_reset", "persistent_safe_reset")
    ]
    assert sum(row.reaccelerations_before_reset for row in persistent_motions) == 0
    assert sum(row.braking_releases_before_reset_ack for row in persistent_motions) == 0
    assert sum(row.movement_while_intent for row in persistent_motions) == 0
    assert sum(row.unsafe_recoveries for row in persistent_motions) == 0
    safe_resets = [row for row in resets if row.mode == "persistent_safe_reset"]
    assert sum(row.movement_before_reset_ack_cases for row in safe_resets) == 0
    assert sum(row.automatic_old_command_restorations for row in safe_resets) == 0
    assert sum(row.negative_speed_cases for row in motions) == 0
    assert sum(row.target_speed_exceeded_cases for row in motions) == 0

    csv_paths = (SUMMARY_CSV, EMERGENCY_CSV, RESET_CSV, MOTION_CSV, LOAD_CSV)
    for path in csv_paths:
        assert path.is_file() and path.stat().st_size > 0
        path.read_text(encoding="utf-8")
        with path.open(encoding="utf-8", newline="") as file:
            assert sum(1 for _ in csv.DictReader(file)) == 45
    assert REPORT_PATH.is_file() and REPORT_PATH.stat().st_size > 0
    REPORT_PATH.read_text(encoding="utf-8")
    for path in PLOT_PATHS:
        image = cv2.imread(str(path), cv2.IMREAD_UNCHANGED)
        assert image is not None and image.size > 0 and image.shape[0] > 100 and image.shape[1] > 100
    expected = {path.resolve() for path in (*csv_paths, REPORT_PATH, *PLOT_PATHS)}
    actual = {path.resolve() for path in OUTPUT_DIRECTORY.rglob("*") if path.is_file()}
    assert actual == expected, sorted(str(path) for path in actual ^ expected)
    forbidden = {".jpg", ".jpeg", ".pyc", ".pickle", ".pkl", ".bin", ".zip", ".tar", ".gz", ".7z"}
    assert not any(path.suffix.lower() in forbidden for path in actual)
    assert not any(path.name == "__pycache__" for path in OUTPUT_DIRECTORY.rglob("*"))


def main() -> None:
    OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True)
    workload = build_workload()
    tests = run_functional_tests()
    summaries, emergencies, resets, motions, loads, transport_sets = run_experiment(workload)
    _write_csv(SUMMARY_CSV, SummaryMetrics, summaries)
    _write_csv(EMERGENCY_CSV, EmergencyMetrics, emergencies)
    _write_csv(RESET_CSV, ResetMetrics, resets)
    _write_csv(MOTION_CSV, MotionMetrics, motions)
    _write_csv(LOAD_CSV, LoadMetrics, loads)
    save_plots(summaries, emergencies, resets, motions, loads)
    write_report(workload, summaries, emergencies, resets, motions, loads, tests)
    validate_outputs(summaries, emergencies, resets, motions, loads, tests, transport_sets)
    print(f"Lab040 complete: combinations={len(summaries)}, lossy repetitions={REPETITIONS}")
    for item in tests:
        print(f"{'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}")


if __name__ == "__main__":
    main()