sai.py

Go to the documentation of this file.

"""SAI - Python mirror of the Safe Application Intermediate sub-layer (Subset-098 v4.0.0 SS5.4),
EC-counter variant. Byte-identical to ``Platform_Protocol_SafeComm`` (``src/sai/sai.c``), the same way
``simcore.saf`` mirrors ``src/saf/saf.c``.
 
Sits functionally *above* a connected :class:`simcore.saf.SafSession`: it wraps every user PDU in a SAI
header (message type, 16-bit sequence number, EC counter; the Triple-Time-Stamp fields are zeroed - only
EC is implemented, same scope limit as the C module) before handing it to ``SafSession.send_pdu``, and
unwraps/checks it on ``recv_pdu``. This supplies the delay/re-ordering/repetition defence the Safety
Layer's own DT PDU doesn't carry (Subset-037-2 Table 12 has no sequence field at all).
 
Implements SS5.4.7 (sequence window) and SS5.4.9 (EC counter freshness: the EC-start exchange, the
ratio/Ex bookkeeping, and the Delta-based 0..3 state check). Does NOT implement Triple Time Stamping
(SS5.4.8), the long-term drift correction of SS5.4.9.5, or the separate delay-detection heartbeat of
SS5.4.9.6 - same scope as ``sai.c``; see Platform_Protocol_SafeComm/docs/safety/SUBSET_MAPPING.md.
"""
from simcore import saf as _saf
 
MSG_EC_START = 0x81
MSG_EC_DATA = 0x86
 
HEADER_LEN = 19  # msgtype(1) + seq(2) + TTS(12, zeroed) + EC(4)
START_EXTRA_LEN = 6  # Version(4) + EC period(2)
 
 
class SaiError(Exception):
    """A SAI failure; ``status`` names the matching C ``SAFECOMM_STATUS_*`` value."""
 
    def __init__(self, status, detail=""):
        super().__init__(f"{status}{': ' + detail if detail else ''}")
        self.status = status
 
 
def _ec_state_from_delta(delta):
    """SS5.4.9.4.7: the current EC state depends only on the latest Delta."""
    if delta <= 0:
        return 0
    if delta == 1:
        return 1
    if delta == 2:
        return 2
    return 3  # alarm
 
 
class SaiSession:
    """One SAI connection, layered over an already-connected :class:`simcore.saf.SafSession`.
 
    ``ec_period_ms`` is this side's own EC (execution cycle) period. ``seq_window`` is N of SS5.4.7 (up
    to ``seq_window - 1`` missing messages tolerated; default 1, no gaps allowed). ``ec_alarm_state`` is
    the Delta at/above which freshness fails (default 3, matching the spec's own worked example).
    """
 
    def __init__(self, saf_session, ec_period_ms, *, seq_window=1, ec_alarm_state=3):
        if ec_period_ms <= 0:
            raise ValueError("ec_period_ms must be positive")
        self.saf = saf_session
        self.ec_period_ms = ec_period_ms
        self.seq_window = seq_window or 1
        self.ec_alarm_state = ec_alarm_state or 3
 
        self.seq_tx = 0
        self.seq_rx = 0
        self.seq_rx_valid = False
        self.ec_tx = 0
        self.peer_ec_period_ms = None
        self.ratio = None
        self.expected_ec = None
        self.last_rx_ec = 0
        self.ec_initialized = False
        self.ec_state = 0
 
    def _header(self, msg_type, seq, ec):
        return bytes([msg_type]) + seq.to_bytes(2, "big") + bytes(12) + ec.to_bytes(4, "big")
 
    def start(self):
        """Run the EC initialisation exchange (SS5.4.9.3): send our ExecutionCycleStart, wait for the
        peer's, and compute the EC ratio. ``self.saf`` must already be in DATA (its own handshake done)."""
        out = (self._header(MSG_EC_START, self.seq_tx, self.ec_tx)
               + (1).to_bytes(4, "big") + min(self.ec_period_ms, 0xFFFF).to_bytes(2, "big"))
        self.saf.send_pdu(out)
        self.seq_tx = (self.seq_tx + 1) & 0xFFFF
 
        try:
            frame = self.saf.recv_pdu()
        except _saf.SafError as exc:
            raise SaiError(exc.status, str(exc)) from exc
        if len(frame) != HEADER_LEN + START_EXTRA_LEN or frame[0] != MSG_EC_START:
            raise SaiError("AUTH_FAILED", "EC start shape")
 
        self.last_rx_ec = int.from_bytes(frame[15:19], "big")
        self.peer_ec_period_ms = int.from_bytes(frame[23:25], "big")
        if self.peer_ec_period_ms == 0:
            raise SaiError("INVALID_PARAM", "peer EC period")
        self.seq_rx = int.from_bytes(frame[1:3], "big")
        self.seq_rx_valid = True
 
        # SS5.4.9.4.3: R = receiver (our) EC period / sender (peer) EC period.
        # SS5.4.9.4.4: update Ex once immediately, using the peer's initial EC counter.
        self.ratio = self.ec_period_ms / self.peer_ec_period_ms
        self.expected_ec = self.last_rx_ec + self.ratio
        self.ec_initialized = True
        self.ec_state = 0
 
    def send(self, pdu):
        frame = self._header(MSG_EC_DATA, self.seq_tx, self.ec_tx) + bytes(pdu)
        self.saf.send_pdu(frame)
        self.seq_tx = (self.seq_tx + 1) & 0xFFFF
 
    def recv(self):
        """Receive, unwrap and check one SAI-protected PDU. Raises :class:`SaiError`."""
        try:
            frame = self.saf.recv_pdu()
        except _saf.SafError as exc:
            raise SaiError(exc.status, str(exc)) from exc
        if len(frame) < HEADER_LEN or frame[0] != MSG_EC_DATA:
            raise SaiError("STATE", "unexpected SAI message type")
 
        # SS5.4.7.2: sequence number check with up to (seq_window - 1) tolerated gaps.
        rx_seq = int.from_bytes(frame[1:3], "big")
        if self.seq_rx_valid:
            diff = (rx_seq - self.seq_rx) & 0xFFFF
            if diff >= self.seq_window:
                raise SaiError("SEQ_ERROR", f"got {rx_seq}, expected {self.seq_rx}..+{self.seq_window - 1}")
        self.seq_rx = (rx_seq + 1) & 0xFFFF
        self.seq_rx_valid = True
 
        # SS5.4.9.4: freshness.
        self.last_rx_ec = int.from_bytes(frame[15:19], "big")
        if self.ec_initialized:
            delta = int(self.expected_ec - self.last_rx_ec)
            self.ec_state = _ec_state_from_delta(delta)
            if self.ec_state >= self.ec_alarm_state:
                raise SaiError("TIMEOUT", "EC freshness alarm")
 
        return frame[HEADER_LEN:]
 
    def tick(self):
        """Advance our own EC by one: call this once per ``ec_period_ms`` elapsed on the caller's clock.
        Updates the expected-EC estimate and the freshness state (SS5.4.9.4.4-7), reporting the state
        computed BEFORE advancing Ex for the next cycle (so the first tick after a fresh receive is 0)."""
        if not self.ec_initialized:
            return
        delta = int(self.expected_ec - self.last_rx_ec)
        self.ec_state = _ec_state_from_delta(delta)
        self.expected_ec += self.ratio
        self.ec_tx += 1