ertms_adapters.py

Go to the documentation of this file.

"""Train sim's own adapters between message_catalog.json's flat
{field: int} fields dict (the same shape every JSON sendMessage command
already uses for the flat scheme) and rbc_ertms_wire.py's real, per-
message-shaped encode_msg_*()/decode() functions - injected into
SimCore's MessageCatalog at construction time (see that module's own
header doc for why this lives here, not there: Subset-026 is the
Train<->RBC air-gap protocol specifically, not something IL/CTC ever
speak in this codebase, so it's this sim's own application-layer
concern, same as message_catalog.json itself already is per-sim).
"""
 
from . import rbc_ertms_wire as ertms
 
 
def _position_report_from_fields(fields):
    # q_scale/q_dirlrbg/q_dlrbg/l_doubtover/l_doubtunder added this session
    # (ab_gp_ertms_train_packet_codec.c's own Packet 0 wire extension) -
    # default to "1m scale, nominal direction/orientation, zero doubt
    # radius" (the least eventful reading a real EVC would report absent
    # any actual doubt-window/orientation scenario this sim models) rather
    # than 0/0/0, which would read as Q_SCALE_10CM/Q_DIRLRBG_REVERSE/
    # Q_DLRBG_REVERSE - real enum values with a real (wrong) meaning, not
    # an obviously-unset placeholder.
    return {
        "q_scale": fields.get("q_scale", ertms.Q_SCALE_1M),
        "nid_c": fields.get("nid_c", 0),
        "nid_bg": fields.get("nid_bg", 0),
        "d_lrbg": fields.get("d_lrbg", 0),
        "q_dirlrbg": fields.get("q_dirlrbg", ertms.Q_DIRLRBG_NOMINAL),
        "q_dlrbg": fields.get("q_dlrbg", ertms.Q_DLRBG_NOMINAL),
        "l_doubtover": fields.get("l_doubtover", 0),
        "l_doubtunder": fields.get("l_doubtunder", 0),
        "v_train": fields.get("v_train", 0),
        "m_mode": fields.get("m_mode", 0),
        "m_level": fields.get("m_level", 0),
    }
 
 
def _position_report_packet1_from_fields(fields):
    """Same shape as _position_report_from_fields() plus Packet 1's own
    extra NID_PRVLRBG pair (prv_nid_c/prv_nid_bg) - defaults to mirroring
    nid_c/nid_bg when not explicitly given, since a stationary heartbeat's
    "previous" LRBG is typically its own last-reported one."""
    base = _position_report_from_fields(fields)
    base["prv_nid_c"] = fields.get("prv_nid_c", base["nid_c"])
    base["prv_nid_bg"] = fields.get("prv_nid_bg", base["nid_bg"])
    return base
 
 
# Catalog message name -> (nid_engine, flat_fields_dict, t_train) -> bytes.
ENCODERS = {
    "M155": lambda nid_engine, fields, t_train: ertms.encode_msg_155(nid_engine, t_train),
    "M156": lambda nid_engine, fields, t_train: ertms.encode_msg_156(nid_engine, t_train),
    "M159": lambda nid_engine, fields, t_train: ertms.encode_msg_159(
        nid_engine, fields.get("m_version", 0), t_train),
    "M146": lambda nid_engine, fields, t_train: ertms.encode_msg_146(
        nid_engine, fields.get("t_train_ack", 0), t_train),
    "M136": lambda nid_engine, fields, t_train: ertms.encode_msg_136(
        nid_engine, _position_report_from_fields(fields), t_train),
    "M136_P1": lambda nid_engine, fields, t_train: ertms.encode_msg_136_packet1(
        nid_engine, _position_report_packet1_from_fields(fields), t_train),
    "M132": lambda nid_engine, fields, t_train: ertms.encode_msg_132(
        nid_engine, fields.get("q_marqstreason", 0), _position_report_from_fields(fields), t_train),
    "M129": lambda nid_engine, fields, t_train: ertms.encode_msg_129(
        nid_engine, _position_report_from_fields(fields), fields.get("l_train", 0),
        fields.get("v_maxtrain", 0), fields.get("m_loadinggauge", 0), fields.get("m_axleload", 0),
        bool(fields.get("m_airtight", 0)), t_train),
    "M130": lambda nid_engine, fields, t_train: ertms.encode_msg_130(
        nid_engine, _position_report_from_fields(fields), t_train),
    "M147": lambda nid_engine, fields, t_train: ertms.encode_msg_147(
        nid_engine, fields.get("nid_em", 0), fields.get("q_emergency_stop", 0),
        _position_report_from_fields(fields), t_train),
    "M150": lambda nid_engine, fields, t_train: ertms.encode_msg_150(
        nid_engine, _position_report_from_fields(fields), t_train),
    "M154": lambda nid_engine, fields, t_train: ertms.encode_msg_154(nid_engine, t_train),
    "M157": lambda nid_engine, fields, t_train: ertms.encode_msg_157(
        nid_engine, fields.get("q_status", 0), _position_report_from_fields(fields), t_train),
    "M158": lambda nid_engine, fields, t_train: ertms.encode_msg_158(
        nid_engine, fields.get("nid_textmessage", 0), _position_report_from_fields(fields), t_train),
}
 
 
def _flatten_msg3(decoded):
    """Msg 3 (Movement Authority) -> one flat {field: value} dict, same
    shape decode_to_message() already returns for the flat scheme, plus
    "packetsPresent" (this message's own optional-packet list) so a
    Robot test can assert presence, not just field values. Packet-
    specific fields are prefixed by packet (gradient_*/ssp_*/linking_*)
    since e.g. gradient and ssp both have their own "segment_count" -
    Msg 3 in this flow ALWAYS carries both at once (Packet 15+21+27),
    so this collision is real, not theoretical."""
    flat = dict(decoded["ma"])
    if decoded["has_gradient"]:
        flat["gradient_segment_count"] = decoded["gradient"]["segment_count"]
        flat["gradient_segments"] = decoded["gradient"]["segments"]
    if decoded["has_ssp"]:
        flat["ssp_segment_count"] = decoded["ssp"]["segment_count"]
        flat["ssp_segments"] = decoded["ssp"]["segments"]
    if decoded["has_linking"]:
        flat["linking_link_count"] = decoded["linking"]["link_count"]
        flat["linking_links"] = decoded["linking"]["links"]
    flat["packetsPresent"] = decoded["packets_present"]
    return flat
 
 
def _flatten_msg24(decoded):
    """Msg 24 (General Message) -> one flat dict, same shape/rationale as
    _flatten_msg3() above. NOTE a real, currently-latent collision: P65
    (tsr) and P66 (tsr_revoke) both have their own "nid_tsr" field - not
    prefixed below since this project's own SoM/Running/EoM flow never
    carries both in the same Msg 24, but a future scenario that does
    (dynamic TSR work) would need to prefix these the same way
    gradient/ssp are above."""
    flat = {}
    if decoded["has_national_values"]:
        flat.update(decoded["national_values"])
    if decoded["has_ma_request_params"]:
        flat.update(decoded["ma_request_params"])
    if decoded["has_pr_params"]:
        flat.update(decoded["pr_params"])
    if decoded["has_tsr"]:
        flat.update(decoded["tsr"])
    if decoded["has_tsr_revoke"]:
        flat.update(decoded["tsr_revoke"])
    if decoded["has_text_message"]:
        flat.update(decoded["text_message"])
    flat["packetsPresent"] = decoded["packets_present"]
    return flat
 
 
# nid_message (int) -> decoded_dict -> flat_fields_dict.
FLATTENERS = {
    ertms.MSG_MOVEMENT_AUTHORITY: _flatten_msg3,
    ertms.MSG_GENERAL_MESSAGE: _flatten_msg24,
}