Lab040: add persistent emergency intent and safe reset

2026-08-05 13:12:49 +03:00
parent b3ebe1637f
commit 88addff5b6
18 changed files with 2513 additions and 0 deletions
--- a/protocol/persistent_emergency.py
+++ b/protocol/persistent_emergency.py
@@ -0,0 +1,147 @@
+"""Постоянное аварийное намерение наземной станции для Lab040.
+
+Автомат состояний намеренно отделяет подтверждение аварийного пакета от
+снятия намерения оператора. Только принятое подтверждение сброса разрешает
+сформировать новую команду движения.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+from protocol.control_failsafe import ControlState, SAFE_STATE
+from protocol.emergency_ack import EmergencyIdentity
+
+
+class GroundState(str, Enum):
+    GROUND_NORMAL = "ground_normal"
+    GROUND_EMERGENCY_REQUESTED = "ground_emergency_requested"
+    GROUND_EMERGENCY_CONFIRMED = "ground_emergency_confirmed"
+    GROUND_RESET_REQUESTED = "ground_reset_requested"
+    GROUND_MOVEMENT_REAUTHORIZED = "ground_movement_reauthorized"
+
+
+@dataclass(frozen=True)
+class GroundSnapshot:
+    state: GroundState
+    emergency_intent: bool
+    movement_permitted: bool
+    emergency_acknowledged: bool
+    reset_acknowledged: bool
+
+
+class PersistentEmergencyController:
+    """Автомат наземной станции, не снимающий намерение по подтверждению аварии."""
+
+    def __init__(self, initial_speed_mps: float) -> None:
+        if initial_speed_mps < 0.0:
+            raise ValueError("начальная скорость не может быть отрицательной")
+        self.state = GroundState.GROUND_NORMAL
+        self.emergency_intent = False
+        self.emergency_event_id: int | None = None
+        self.emergency_stream_id: int | None = None
+        self.emergency_sequence: int | None = None
+        self.emergency_acknowledged = False
+        self.reset_request_id: int | None = None
+        self.reset_sequence: int | None = None
+        self.reset_acknowledged = False
+        self.current_motion_speed_mps = initial_speed_mps
+        self.last_operator_sequence = 0
+        self._sequence_at_emergency = -1
+
+    @property
+    def snapshot(self) -> GroundSnapshot:
+        control = self.control_state()
+        return GroundSnapshot(
+            self.state,
+            self.emergency_intent,
+            control.movement_allowed,
+            self.emergency_acknowledged,
+            self.reset_acknowledged,
+        )
+
+    def request_emergency(
+        self,
+        event_id: int,
+        stream_id: int,
+        sequence_number: int,
+    ) -> None:
+        if self.state is not GroundState.GROUND_NORMAL:
+            raise RuntimeError("аварийную команду можно запросить только из нормального состояния")
+        if event_id < 0:
+            raise ValueError("идентификатор события не может быть отрицательным")
+        self.emergency_event_id = event_id
+        self.emergency_stream_id = stream_id
+        self.emergency_sequence = sequence_number
+        self.emergency_intent = True
+        self.emergency_acknowledged = False
+        self.reset_acknowledged = False
+        self._sequence_at_emergency = self.last_operator_sequence
+        self.state = GroundState.GROUND_EMERGENCY_REQUESTED
+
+    def receive_emergency_ack(self, identity: EmergencyIdentity) -> bool:
+        expected = EmergencyIdentity(
+            self.emergency_stream_id if self.emergency_stream_id is not None else -1,
+            self.emergency_sequence if self.emergency_sequence is not None else -1,
+        )
+        if identity != expected:
+            return False
+        first = not self.emergency_acknowledged
+        self.emergency_acknowledged = True
+        if self.state is GroundState.GROUND_EMERGENCY_REQUESTED:
+            self.state = GroundState.GROUND_EMERGENCY_CONFIRMED
+        return first
+
+    def request_reset(self, reset_request_id: int, sequence_number: int) -> None:
+        if not self.emergency_intent:
+            raise RuntimeError("для сброса требуется постоянное аварийное намерение")
+        if self.state not in (
+            GroundState.GROUND_EMERGENCY_REQUESTED,
+            GroundState.GROUND_EMERGENCY_CONFIRMED,
+        ):
+            raise RuntimeError("сброс уже ожидается или завершён")
+        if reset_request_id < 0 or sequence_number < 0:
+            raise ValueError("идентификаторы сброса не могут быть отрицательными")
+        self.reset_request_id = reset_request_id
+        self.reset_sequence = sequence_number
+        self.reset_acknowledged = False
+        self.state = GroundState.GROUND_RESET_REQUESTED
+
+    def receive_reset_ack(
+        self,
+        event_id: int,
+        reset_request_id: int,
+        accepted: bool,
+    ) -> bool:
+        if self.state is not GroundState.GROUND_RESET_REQUESTED:
+            return False
+        if event_id != self.emergency_event_id or reset_request_id != self.reset_request_id:
+            return False
+        if not accepted:
+            return False
+        self.reset_acknowledged = True
+        self.emergency_intent = False
+        self.current_motion_speed_mps = 0.0
+        self.state = GroundState.GROUND_MOVEMENT_REAUTHORIZED
+        return True
+
+    def control_state(self) -> ControlState:
+        if self.state is GroundState.GROUND_NORMAL:
+            return ControlState(self.current_motion_speed_mps, 0.0, False, True)
+        return SAFE_STATE
+
+    def new_operator_motion(self, sequence_number: int, speed_mps: float) -> ControlState:
+        if self.state is not GroundState.GROUND_MOVEMENT_REAUTHORIZED:
+            raise RuntimeError("повторное движение не разрешено")
+        if sequence_number <= self._sequence_at_emergency or sequence_number <= self.last_operator_sequence:
+            raise ValueError("требуется новый номер последовательности команды оператора")
+        if speed_mps <= 0.0:
+            raise ValueError("новая скорость движения должна быть положительной")
+        self.last_operator_sequence = sequence_number
+        self.current_motion_speed_mps = speed_mps
+        self.state = GroundState.GROUND_NORMAL
+        return self.control_state()
+
+    def reject_old_operator_motion(self, sequence_number: int) -> bool:
+        return sequence_number <= self._sequence_at_emergency