rbc_ertms_wire.py

Go to the documentation of this file.

"""Real Subset-026 ERTMS message codec - Python mirror of
RBC_GP/src/application/AB/GP/ertms/train/ab_gp_ertms_train_codec.c
(byte shape and order only). Keep this file in exact sync with that C
file by hand; there is no shared schema between the two languages, same
"by-hand port, not a generated binding" precedent rbc_wire.py's own
module doc already states for the flat ADR-029 scheme.
 
**Deliberately a DIFFERENT wire shape from rbc_wire.py, not a variant of
it** - this is not an oversight, don't "fix" the two into consistency:
  - rbc_wire.py: one fixed 96-byte struct, little-endian, every field at
    a fixed offset regardless of kind - mirrors rbc_wire_types.h/
    rbc_wire.c's own flat ADR-029 envelope.
  - This module: real Subset-026 message-specific shapes, BIG-endian,
    genuinely variable length (optional packets each self-identify with
    their own leading NID_PACKET byte) - mirrors ab_gp_ertms_train_codec.c,
    a byte-aligned TLV codec (not literally the real spec's own sub-byte
    bit-packing - see that C file's own header doc for why: a deliberate,
    already-made MISRA-friendly simplification this module also follows).
 
Both scheme's bytes are wrapped identically at the TRANSPORT level by one
new envelope kind, RBC_MSG_ERTMS_ENVELOPE (rbc_wire_types.h) - see that
kind's own doc: byte 0 = the kind (22 - moved from 21 after a real
collision with this project's own flat TRAIN_CONNECT=21, see
rbc_wire_types.h's/rbc_messages.py's own comments), byte 1 = train_id, bytes 2..
= this module's own encode()/decode() output verbatim. rbc_messages.py's
MessageCatalog is what actually builds/parses that 2-byte prefix (its own
"ertms" wireFormat branch) - this module only ever sees/produces the
bytes AFTER that prefix, same "RBC_Test_Sim_Core doesn't know kind numbers"
boundary rbc_wire.py's own module doc draws for the flat scheme.
 
Real NID_MESSAGE/NID_PACKET numbers only, never this project's own flat-
scheme project-specific numbering (rbc_wire.py's `kind` byte) - see
ab_gp_ertms_train_interface.h/ab_gp_ertms_train_packets.h for the C-side
`#define`s these mirror.
"""
 
import struct
 
# ---- Real NID_MESSAGE numbers (Subset-026 Ch. 8) --------------------------
# Train -> RBC (this module's own ENCODE side - what RBC_Test_Sim_Train originates)
MSG_VALIDATED_TRAIN_DATA = 129
MSG_REQUEST_FOR_SHUNTING = 130
MSG_MA_REQUEST = 132
MSG_POSITION_REPORT = 136
MSG_ACKNOWLEDGEMENT = 146
MSG_ACK_EMERGENCY_STOP = 147
MSG_INIT_SESSION = 155
MSG_TERMINATE_SESSION = 156
MSG_SESSION_ESTABLISHED = 159
# RBC -> Train (this module's own DECODE side - what RBC_Test_Sim_Train receives)
MSG_MOVEMENT_AUTHORITY = 3
MSG_UNCONDITIONAL_EMERGENCY_STOP = 16
MSG_REVOCATION_EMERGENCY_STOP = 18
MSG_GENERAL_MESSAGE = 24
MSG_SH_REFUSED = 27
MSG_SH_AUTHORISED = 28
 
# ---- Real NID_PACKET numbers (Subset-026 Ch. 7) ----------------------------
PACKET_NATIONAL_VALUES = 3
PACKET_LINKING_INFO = 5
PACKET_LEVEL_2_3_MA = 15
PACKET_GRADIENT_PROFILE = 21
PACKET_STATIC_SPEED_PROFILE = 27
PACKET_MA_REQUEST_PARAMS = 57
PACKET_POSITION_REPORT_PARAMS = 58
PACKET_TEMPORARY_SPEED_RESTRICT = 65
PACKET_TSR_REVOCATION = 66
PACKET_TEXT_MESSAGE = 72
 
_HEADER_T2T = struct.Struct(">BHII")  # nid_message, l_message, t_train, nid_engine (11 bytes)
_HEADER_C2T = struct.Struct(">BHIBHH")  # nid_message, l_message, t_train, m_ack, nid_c, nid_bg (12 bytes)
# Full Packet 0 field set, in rbc_packet_0_position_report_t's own declared
# order (ab_gp_ertms_train_packets.h) - q_scale, nid_c, nid_bg, d_lrbg,
# q_dirlrbg, q_dlrbg, l_doubtover, l_doubtunder, v_train, m_mode, m_level
# (23 bytes). Extended this session, on direct request, from a reduced
# subset (nid_c/nid_bg/d_lrbg/v_train/m_mode/m_level only, 12 bytes) that
# never modeled q_scale/q_dirlrbg/q_dlrbg/l_doubtover/l_doubtunder at all -
# see ab_gp_ertms_train_packet_codec.c's own rbc_packet_read_0_position_
# report() doc, which this MUST stay byte-identical to.
_POSITION_REPORT = struct.Struct(">BHHIBBIIHBB")
_POSITION_REPORT_PACKET1 = struct.Struct(">HHHHIHBB")  # nid_c, nid_bg, prv_nid_c, prv_nid_bg,
                                                        # d_lrbg, v_train, m_mode, m_level (16 bytes) -
                                                        # Packet 1 does NOT carry q_scale/q_dirlrbg/
                                                        # q_dlrbg/l_doubtover/l_doubtunder (no such
                                                        # fields in rbc_packet_1_position_report_2bg_t).
 
# Real Subset-026 Q_SCALE/Q_DIRLRBG/Q_DLRBG enum values (ab_gp_ertms_types.h) -
# mirrored here as plain ints (no shared C/Python enum binding, same
# "by-hand port" precedent this whole module already follows).
Q_SCALE_10CM = 0
Q_SCALE_1M = 1
Q_SCALE_10M = 2
Q_DIRLRBG_REVERSE = 0
Q_DIRLRBG_NOMINAL = 1
Q_DIRLRBG_UNKNOWN = 2
Q_DLRBG_REVERSE = 0
Q_DLRBG_NOMINAL = 1
Q_DLRBG_UNKNOWN = 2
 
# Real Subset-026 NID_PACKET values for Msg 136 specifically (Packet 0 vs
# Packet 1 - see ab_gp_ertms_train_packets.h's own RBC_PACKET_0_POSITION_
# REPORT/RBC_PACKET_1_POSITION_REPORT_2BG). Msg 129/132 never carry this
# leading byte at all - only Msg 136's own decode case reads one, mirror
# that exactly here (see rbc_train_codec_decode_train_to_track()'s own
# RBC_MSG_TRAIN_TO_TRACK_POSITION_REPORT case).
PACKET_0_POSITION_REPORT = 0
PACKET_1_POSITION_REPORT_2BG = 1
 
 
def _pack_position_report(pos):
    """@param pos: dict with the full Packet 0 field set - q_scale, nid_c,
    nid_bg, d_lrbg, q_dirlrbg, q_dlrbg, l_doubtover, l_doubtunder, v_train,
    m_mode, m_level - matching rbc_packet_read_0_position_report()'s own
    field order exactly (ab_gp_ertms_train_packet_codec.c)."""
    return _POSITION_REPORT.pack(pos["q_scale"], pos["nid_c"], pos["nid_bg"], pos["d_lrbg"],
                                 pos["q_dirlrbg"], pos["q_dlrbg"], pos["l_doubtover"], pos["l_doubtunder"],
                                 pos["v_train"], pos["m_mode"], pos["m_level"])
 
 
def _unpack_position_report(buf, offset):
    """Unused today (this sim only ENCODES Packet 0, as the message
    originator - see _pack_position_report()'s own doc) - kept for
    encode/decode symmetry with the rest of this module."""
    (q_scale, nid_c, nid_bg, d_lrbg, q_dirlrbg, q_dlrbg, l_doubtover, l_doubtunder,
     v_train, m_mode, m_level) = _POSITION_REPORT.unpack_from(buf, offset)
    return {"q_scale": q_scale, "nid_c": nid_c, "nid_bg": nid_bg, "d_lrbg": d_lrbg,
            "q_dirlrbg": q_dirlrbg, "q_dlrbg": q_dlrbg, "l_doubtover": l_doubtover,
            "l_doubtunder": l_doubtunder, "v_train": v_train, "m_mode": m_mode,
            "m_level": m_level}, offset + _POSITION_REPORT.size
 
 
def _finalize(nid_message, body_after_header, nid_engine, t_train):
    """Builds the full Train-to-Track frame and patches l_message with the
    real total length, exactly matching
    rbc_train_codec_encode_track_to_train()'s own "l_message computed
    here, not caller-supplied" fix on the C side - never trust a
    placeholder length, always the real encoded size."""
    header = _HEADER_T2T.pack(nid_message, 0, t_train, nid_engine)
    frame = bytearray(header + body_after_header)
    l_message = len(frame)
    frame[1:3] = struct.pack(">H", l_message)
    return bytes(frame)
 
 
# ---- Train -> RBC encoders (what RBC_Test_Sim_Train sends) ------------------------
 
def encode_msg_155(nid_engine, t_train=0):
    """Msg 155: Initiation of a Communication Session - header only."""
    return _finalize(MSG_INIT_SESSION, b"", nid_engine, t_train)
 
 
def encode_msg_156(nid_engine, t_train=0):
    """Msg 156: Termination of a Communication Session - header only."""
    return _finalize(MSG_TERMINATE_SESSION, b"", nid_engine, t_train)
 
 
def encode_msg_159(nid_engine, m_version, t_train=0):
    """Msg 159: Session Established."""
    return _finalize(MSG_SESSION_ESTABLISHED, struct.pack(">B", m_version), nid_engine, t_train)
 
 
def encode_msg_146(nid_engine, t_train_ack, t_train=0):
    """Msg 146: Acknowledgement - real Subset-026 146 acks by TIMESTAMP of
    the acked message, not a sequence number (see message_catalog.json's
    own _subset026Reference comment on this)."""
    return _finalize(MSG_ACKNOWLEDGEMENT, struct.pack(">I", t_train_ack), nid_engine, t_train)
 
 
def encode_msg_136(nid_engine, position_report, t_train=0):
    """Msg 136: Train Position Report, Packet 0 - the normal active-mission
    cadence. Real Subset-026 Msg 136 self-identifies which packet variant
    it carries with a leading NID_PACKET byte (see
    rbc_train_codec_decode_train_to_track()'s own Msg 136 case) - this
    encoder writes PACKET_0_POSITION_REPORT then the Packet 0 body."""
    body = struct.pack(">B", PACKET_0_POSITION_REPORT) + _pack_position_report(position_report)
    return _finalize(MSG_POSITION_REPORT, body, nid_engine, t_train)
 
 
def encode_msg_136_packet1(nid_engine, packet1_report, t_train=0):
    """Msg 136: Train Position Report, Packet 1 (2-balise-group variant) -
    the post-EoM idle heartbeat cadence (see train_sim.py's own post-EoM
    phase tracking). @param packet1_report: dict with nid_c, nid_bg,
    prv_nid_c, prv_nid_bg, d_lrbg, v_train, m_mode, m_level - same field
    set read_position_report_packet1() reads on the C side."""
    body = struct.pack(">B", PACKET_1_POSITION_REPORT_2BG) + _POSITION_REPORT_PACKET1.pack(
        packet1_report["nid_c"], packet1_report["nid_bg"],
        packet1_report["prv_nid_c"], packet1_report["prv_nid_bg"],
        packet1_report["d_lrbg"], packet1_report["v_train"],
        packet1_report["m_mode"], packet1_report["m_level"])
    return _finalize(MSG_POSITION_REPORT, body, nid_engine, t_train)
 
 
def encode_msg_132(nid_engine, q_marqstreason, position_report, t_train=0):
    """Msg 132: MA Request - Q_MARQSTREASON then Packet 0."""
    body = struct.pack(">B", q_marqstreason) + _pack_position_report(position_report)
    return _finalize(MSG_MA_REQUEST, body, nid_engine, t_train)
 
 
def encode_msg_129(nid_engine, position_report, l_train, v_maxtrain, m_loadinggauge,
                   m_axleload, m_airtight, t_train=0):
    """Msg 129: Validated Train Data - Packet 0 then the flat train-data
    fields (same field subset ab_gp_ertms_train_codec.c's own decode
    reads)."""
    body = (_pack_position_report(position_report) +
            struct.pack(">IHBBB", l_train, v_maxtrain, m_loadinggauge, m_axleload,
                       1 if m_airtight else 0))
    return _finalize(MSG_VALIDATED_TRAIN_DATA, body, nid_engine, t_train)
 
 
def encode_msg_130(nid_engine, position_report, t_train=0):
    """Msg 130: Request for Shunting - Packet 0."""
    return _finalize(MSG_REQUEST_FOR_SHUNTING, _pack_position_report(position_report), nid_engine, t_train)
 
 
def encode_msg_147(nid_engine, nid_em, q_emergency_stop, position_report, t_train=0):
    """Msg 147: Ack Emergency Stop - NID_EM, Q_EMERGENCY_STOP then Packet 0."""
    body = struct.pack(">BB", nid_em, q_emergency_stop) + _pack_position_report(position_report)
    return _finalize(MSG_ACK_EMERGENCY_STOP, body, nid_engine, t_train)
 
 
def encode_msg_150(nid_engine, position_report, t_train=0):
    """Msg 150: End of Mission - Packet 0."""
    return _finalize(150, _pack_position_report(position_report), nid_engine, t_train)
 
 
def encode_msg_154(nid_engine, t_train=0):
    """Msg 154: No compatible version supported - header only."""
    return _finalize(154, b"", nid_engine, t_train)
 
 
def encode_msg_157(nid_engine, q_status, position_report, t_train=0):
    """Msg 157: SoM Position Report - Q_STATUS then Packet 0."""
    body = struct.pack(">B", q_status) + _pack_position_report(position_report)
    return _finalize(157, body, nid_engine, t_train)
 
 
def encode_msg_158(nid_engine, nid_textmessage, position_report, t_train=0):
    """Msg 158: Text Message Ack - NID_TEXTMESSAGE then Packet 0."""
    body = struct.pack(">B", nid_textmessage) + _pack_position_report(position_report)
    return _finalize(158, body, nid_engine, t_train)
 
 
# ---- RBC -> Train decoders (what RBC_Test_Sim_Train receives) ---------------------
 
class ErtmsDecodeError(ValueError):
    """@payload could not be decoded as a real Subset-026 track-to-train
    message - malformed length, unknown nid_message, or an unrecognized
    NID_PACKET inside a variable-length message (Msg 3/24)."""
 
 
def _decode_header_c2t(buf):
    if len(buf) < _HEADER_C2T.size:
        raise ErtmsDecodeError(f"buffer too short for track-to-train header: {len(buf)} bytes")
    nid_message, l_message, t_train, m_ack, nid_c, nid_bg = _HEADER_C2T.unpack_from(buf, 0)
    if l_message > len(buf):
        raise ErtmsDecodeError(f"l_message={l_message} exceeds buffer length {len(buf)}")
    header = {"nid_message": nid_message, "l_message": l_message, "t_train": t_train,
              "m_ack": m_ack, "nid_lrbg": {"nid_c": nid_c, "nid_bg": nid_bg}}
    return header, _HEADER_C2T.size
 
 
def decode_msg_3(buf):
    """Msg 3: Movement Authority. @return dict with header, ma (Packet 15,
    always present), and optional gradient/ssp/linking (Packets 21/27/5) -
    same has_X/X pairing ab_gp_ertms_train_interface.h's own
    rbc_msg_3_movement_authority_t struct uses."""
    header, offset = _decode_header_c2t(buf)
    if header["nid_message"] != MSG_MOVEMENT_AUTHORITY:
        raise ErtmsDecodeError(f"decode_msg_3 called on nid_message={header['nid_message']}")
    l_message = header["l_message"]
 
    packet_id, = struct.unpack_from(">B", buf, offset)
    offset += 1
    if packet_id != PACKET_LEVEL_2_3_MA:
        raise ErtmsDecodeError(f"Msg 3 expected Packet 15 first, got {packet_id}")
    ma_seq, ma_length, v_loa, t_loa_s = struct.unpack_from(">IIHI", buf, offset)
    offset += 14
    result = {"header": header,
              "ma": {"ma_seq": ma_seq, "ma_length": ma_length, "v_loa": v_loa, "t_loa_s": t_loa_s},
              "has_gradient": False, "gradient": None,
              "has_ssp": False, "ssp": None,
              "has_linking": False, "linking": None,
              "packets_present": ["P15"]}
 
    while offset < l_message:
        packet_id, = struct.unpack_from(">B", buf, offset)
        offset += 1
        if packet_id == PACKET_GRADIENT_PROFILE:
            count, = struct.unpack_from(">B", buf, offset)
            offset += 1
            segments = []
            for _ in range(count):
                d_gradient, g_a = struct.unpack_from(">IB", buf, offset)
                offset += 5
                segments.append({"d_gradient": d_gradient, "g_a": g_a})
            result["has_gradient"] = True
            result["gradient"] = {"segment_count": count, "segments": segments}
            result["packets_present"].append("P21")
        elif packet_id == PACKET_STATIC_SPEED_PROFILE:
            count, = struct.unpack_from(">B", buf, offset)
            offset += 1
            segments = []
            for _ in range(count):
                d_static, v_static = struct.unpack_from(">IB", buf, offset)
                offset += 5
                segments.append({"d_static": d_static, "v_static": v_static})
            result["has_ssp"] = True
            result["ssp"] = {"segment_count": count, "segments": segments}
            result["packets_present"].append("P27")
        elif packet_id == PACKET_LINKING_INFO:
            count, = struct.unpack_from(">B", buf, offset)
            offset += 1
            links = []
            for _ in range(count):
                nid_bg, d_link, q_linkdir = struct.unpack_from(">HIB", buf, offset)
                offset += 7
                links.append({"nid_bg": nid_bg, "d_link": d_link, "q_linkdir": q_linkdir})
            result["has_linking"] = True
            result["linking"] = {"link_count": count, "links": links}
            result["packets_present"].append("P5")
        else:
            raise ErtmsDecodeError(f"Msg 3: unrecognized NID_PACKET {packet_id} at offset {offset - 1}")
    return result
 
 
def decode_msg_24(buf):
    """Msg 24: General Message. @return dict with header and whichever of
    Packet 3/57/58/65/66/72 were present - same has_X/X pairing
    ab_gp_ertms_train_interface.h's own rbc_msg_24_general_message_t
    struct uses."""
    header, offset = _decode_header_c2t(buf)
    if header["nid_message"] != MSG_GENERAL_MESSAGE:
        raise ErtmsDecodeError(f"decode_msg_24 called on nid_message={header['nid_message']}")
    l_message = header["l_message"]
 
    result = {"header": header,
              "has_national_values": False, "national_values": None,
              "has_ma_request_params": False, "ma_request_params": None,
              "has_pr_params": False, "pr_params": None,
              "has_tsr": False, "tsr": None,
              "has_tsr_revoke": False, "tsr_revoke": None,
              "has_text_message": False, "text_message": None,
              "packets_present": []}
 
    while offset < l_message:
        packet_id, = struct.unpack_from(">B", buf, offset)
        offset += 1
        if packet_id == PACKET_NATIONAL_VALUES:
            v_nvshunt, v_nvstff, v_nvunfit, t_nvcontact, d_nvroll = struct.unpack_from(">HHHHH", buf, offset)
            offset += 10
            result["has_national_values"] = True
            result["national_values"] = {"v_nvshunt": v_nvshunt, "v_nvstff": v_nvstff,
                                         "v_nvunfit": v_nvunfit, "t_nvcontact": t_nvcontact,
                                         "d_nvroll": d_nvroll}
            result["packets_present"].append("P3")
        elif packet_id == PACKET_MA_REQUEST_PARAMS:
            t_mar_s, d_mar_m = struct.unpack_from(">II", buf, offset)
            offset += 8
            result["has_ma_request_params"] = True
            result["ma_request_params"] = {"t_mar_s": t_mar_s, "d_mar_m": d_mar_m}
            result["packets_present"].append("P57")
        elif packet_id == PACKET_POSITION_REPORT_PARAMS:
            t_cycloc_s, d_cycloc_m = struct.unpack_from(">II", buf, offset)
            offset += 8
            result["has_pr_params"] = True
            result["pr_params"] = {"t_cycloc_s": t_cycloc_s, "d_cycloc_m": d_cycloc_m}
            result["packets_present"].append("P58")
        elif packet_id == PACKET_TEMPORARY_SPEED_RESTRICT:
            nid_tsr, d_tsr, l_tsr, v_tsr, q_front = struct.unpack_from(">HIIHB", buf, offset)
            offset += 13
            result["has_tsr"] = True
            result["tsr"] = {"nid_tsr": nid_tsr, "d_tsr": d_tsr, "l_tsr": l_tsr,
                             "v_tsr": v_tsr, "q_front": q_front}
            result["packets_present"].append("P65")
        elif packet_id == PACKET_TSR_REVOCATION:
            nid_tsr, = struct.unpack_from(">H", buf, offset)
            offset += 2
            result["has_tsr_revoke"] = True
            result["tsr_revoke"] = {"nid_tsr": nid_tsr}
            result["packets_present"].append("P66")
        elif packet_id == PACKET_TEXT_MESSAGE:
            ack_required, duration_s, text_len = struct.unpack_from(">BIB", buf, offset)
            offset += 6
            text = buf[offset:offset + text_len].decode("ascii", errors="replace")
            offset += text_len
            result["has_text_message"] = True
            result["text_message"] = {"ack_required": bool(ack_required), "duration_s": duration_s,
                                      "text": text}
            result["packets_present"].append("P72")
        else:
            raise ErtmsDecodeError(f"Msg 24: unrecognized NID_PACKET {packet_id} at offset {offset - 1}")
    return result
 
 
def decode_msg_27(buf):
    """Msg 27 SH Refused (Subset-026 5.6): header + the T_TRAIN of the refused Msg 130
    (32-bit, same shape as Msg 8)."""
    header, offset = _decode_header_c2t(buf)
    if header["nid_message"] != MSG_SH_REFUSED:
        raise ErtmsDecodeError(f"decode_msg_27 called on nid_message={header['nid_message']}")
    if len(buf) < offset + 4:
        raise ErtmsDecodeError(f"Msg 27 too short: {len(buf)} bytes")
    (t_train_ref,) = struct.unpack_from(">I", buf, offset)
    return {"header": header, "t_train_ref": t_train_ref}
 
 
def decode_msg_28(buf):
    """Msg 28 SH Authorised (Subset-026 5.6): header + the T_TRAIN of the authorised Msg 130
    (RBC_GP sends no optional packets)."""
    header, offset = _decode_header_c2t(buf)
    if header["nid_message"] != MSG_SH_AUTHORISED:
        raise ErtmsDecodeError(f"decode_msg_28 called on nid_message={header['nid_message']}")
    if len(buf) < offset + 4:
        raise ErtmsDecodeError(f"Msg 28 too short: {len(buf)} bytes")
    (t_train_ref,) = struct.unpack_from(">I", buf, offset)
    return {"header": header, "t_train_ref": t_train_ref}
 
 
_DECODERS = {
    MSG_MOVEMENT_AUTHORITY: decode_msg_3,
    MSG_GENERAL_MESSAGE: decode_msg_24,
    MSG_SH_REFUSED: decode_msg_27,
    MSG_SH_AUTHORISED: decode_msg_28,
}
 
 
def decode(buf):
    """@return (nid_message, decoded_dict) by peeking buf[0], dispatching
    to the matching decode_msg_*() above. Raises ErtmsDecodeError for an
    empty buffer or an nid_message this module doesn't yet decode
    (Msg 16/18 - Emergency Stop - have real C-side encoders already but
    no Python decoder yet, out of scope for the SoM/Running/EoM flow)."""
    if len(buf) < 1:
        raise ErtmsDecodeError("empty buffer")
    nid_message = buf[0]
    decoder = _DECODERS.get(nid_message)
    if decoder is None:
        raise ErtmsDecodeError(f"no Python decoder registered for nid_message={nid_message}")
    return nid_message, decoder(buf)