il_wire.py

Go to the documentation of this file.

"""Byte-identical Python mirror of safeAPIRBC2oo2GA/src/AB/il/ab_ga_il_wire.h -
the dedicated IL <-> RBC messages (wire kinds 25/26/27).
 
These do NOT use SimCore's generic flat 96-byte envelope codec
(rbc_wire.py / MessageCatalog); each has its own small packed layout,
still shipped inside the fixed 96-byte relay slot (rest zero). Keep this
in exact lockstep with ab_ga_il_wire.h - there is no shared schema, same
"by-hand port" precedent as SimCore/rbc_wire.py for the flat scheme and
TrainRBCSim/rbc_ertms_wire.py for the ERTMS scheme.
 
All multi-byte fields little-endian. See ab_ga_il_wire.h for the
authoritative byte-offset table.
"""
 
import struct
 
from simcore.rbc_wire import ENVELOPE_SIZE
 
# ---- wire kinds (rbc_wire_types.h) ---------------------------------------
KIND_IL_ROUTE_CMD = 25
KIND_IL_STATUS = 26
KIND_IL_ROUTE_INDICATION = 27
 
# ---- enum values (rbc_wire_types.h, rbc_il_* enums) --------------------
ROUTE_TYPE_NO_ROUTE = 0
ROUTE_TYPE_LOCKED = 1
 
ROUTE_STATUS_NO_STATUS = 0
ROUTE_STATUS_FS = 1
ROUTE_STATUS_OS = 2
ROUTE_STATUS_SH = 3
ROUTE_STATUS_USED = 4
ROUTE_STATUS_DEGRADED = 5
 
DEGRADED_STATUS_NO_STATUS = 0
DEGRADED_STATUS_LOW = 1
DEGRADED_STATUS_MEDIUM = 2
DEGRADED_STATUS_HIGH = 3
 
IL_STATUS_DOWN = 0
IL_STATUS_RESTARTING = 1
IL_STATUS_UP = 2
 
EMERGENCY_TYPE_NONE = -1
EMERGENCY_TYPE_SAM = 0
EMERGENCY_TYPE_CEM = 1
EMERGENCY_TYPE_UEM = 2
 
_NAME_BY_KIND = {
    KIND_IL_ROUTE_CMD: "IL_ROUTE_CMD",
    KIND_IL_STATUS: "IL_STATUS",
    KIND_IL_ROUTE_INDICATION: "IL_ROUTE_INDICATION",
}
 
 
def _slot(payload):
    """Zero-pads @payload out to the full 96-byte relay slot."""
    if len(payload) > ENVELOPE_SIZE:
        raise ValueError("IL message payload exceeds the relay slot")
    return payload + b"\x00" * (ENVELOPE_SIZE - len(payload))
 
 
def encode_route_cmd(route_type=ROUTE_TYPE_NO_ROUTE, route_status=ROUTE_STATUS_NO_STATUS,
                     degraded_status=DEGRADED_STATUS_NO_STATUS, release_route_request=0,
                     first_route_in_path=0, route_id=-1, start_signal=-1, end_signal=-1):
    """RBC_MSG_IL_ROUTE_CMD (IL -> RBC). Fields default to the SPEC
    defaults ("first value is default value"). route_id/start_signal/
    end_signal are int32, -1 = unset."""
    body = struct.pack(
        "<BBBBBBBB iii",
        KIND_IL_ROUTE_CMD,
        0,  # reserved
        route_type & 0xFF,
        route_status & 0xFF,
        degraded_status & 0xFF,
        1 if release_route_request else 0,
        1 if first_route_in_path else 0,
        0,  # reserved
        int(route_id), int(start_signal), int(end_signal),
    )
    return _slot(body)
 
 
def encode_il_status(il_status=IL_STATUS_DOWN):
    """RBC_MSG_IL_STATUS (IL -> RBC)."""
    body = struct.pack("<BBH", KIND_IL_STATUS, il_status & 0xFF, 0)
    return _slot(body)
 
 
def decode_route_indication(frame):
    """RBC_MSG_IL_ROUTE_INDICATION (RBC -> IL) -> flat dict, or None if
    @frame is not this kind / too short. emergency_type is a SIGNED byte
    (-1 = none). route_fsm_state is ab_ga_route_state_t (0..6:
    NO_ROUTE/LOCKED/FS/OS/SH/USED/DEGRADED)."""
    if len(frame) < 24 or frame[0] != KIND_IL_ROUTE_INDICATION:
        return None
    (_kind, _rsv1, train_in_route, train_has_ma_in_route, emergency_in_route,
     emergency_type_i8, route_fsm_state, route_fsm_is_stub) = struct.unpack_from("<BBBBBbBB", frame, 0)
    route_id, start_signal, end_signal, allocated_train_nid_engine = struct.unpack_from("<iiii", frame, 8)
    return {
        "route_id": route_id,
        "start_signal": start_signal,
        "end_signal": end_signal,
        "allocated_train_nid_engine": allocated_train_nid_engine,
        "train_in_route": int(bool(train_in_route)),
        "train_has_ma_in_route": int(bool(train_has_ma_in_route)),
        "emergency_in_route": int(bool(emergency_in_route)),
        "emergency_type": emergency_type_i8,
        "route_fsm_state": route_fsm_state,
        "route_fsm_is_stub": int(bool(route_fsm_is_stub)),
    }
 
 
ROUTE_FSM_STATE_NAMES = ["NO_ROUTE", "LOCKED", "FS", "OS", "SH", "USED", "DEGRADED"]
 
 
def route_fsm_state_name(v):
    return ROUTE_FSM_STATE_NAMES[v] if 0 <= v < len(ROUTE_FSM_STATE_NAMES) else "#%d" % v
 
 
def kind_name(kind_byte):
    return _NAME_BY_KIND.get(kind_byte)