rbc_messages.py

Go to the documentation of this file.

"""Structured message layer on top of rbc_wire.py's raw envelope codec
(safeAPITestEnv/doc/design/DESIGN.md) - translates between the JSON
sendMessage/getMessage control-port commands (control_server.py) and
real rbc_envelope_t bytes (rbc_wire.py). rbc_wire.py itself is
deliberately untouched by this module (DESIGN.md section 6 - "what does
NOT change") - this is a new layer above it, not a replacement: every
sim still sends/receives the exact same wire bytes it always has, this
module only adds a named, field-validated way to build/decode them.
 
The message/packet/field vocabulary itself is DATA, not code - loaded
from each sim's own message_catalog.json (sims/train/, sims/il/,
sims/ctc/ - NOT one shared catalog, deliberately: a sim's own catalog
only lists the messages THAT sim actually sends/receives, e.g. il's
never mentions M3 at all since IL neither sends nor receives it - see
each catalog file's own header). Adding/adjusting a message's fields or
allowed values is a JSON edit, not a Python change.
 
MessageCatalog is a class, not a module-level singleton, specifically
so each sim can load its OWN catalog file independently - see this
module's previous, single-shared-catalog version's own history if
comparing; that shape did not allow one sim's catalog to differ from
another's, which is exactly what per-sim catalogs need.
 
**Two coexisting wire families, per-message "wireFormat"** (added
alongside the original flat scheme, not replacing it - the existing
08-18 Robot suite still depends on the flat family working exactly as
before): a message with no "wireFormat" key (every pre-existing catalog
entry) or `"wireFormat": "flat"` uses the original path (rbc_wire.py,
kind-based, always ENVELOPE_SIZE bytes, generic and shared - SimCore
owns the shape, no sim-specific knowledge needed). `"wireFormat": "ertms"`
uses a real Subset-026 codec instead - genuinely variable length,
message-specific shapes (Packet 15/21/27 for an MA, Packet 3/57/58/65/66
for a General Message, etc.), which is Subset-026-the-Train-RBC-air-gap-
protocol specifically - IL and CTC never speak it in this codebase (same
"a sim's own catalog only lists the messages THAT sim uses" principle
above), so unlike rbc_wire.py this is NOT something SimCore can own
generically. **This module stays sim-agnostic**: it does not import any
ertms codec itself - a caller whose OWN catalog declares "ertms"
messages must inject two things at construction time:
  - `ertms_codec`: the codec module (e.g. TrainRBCSim's own
    rbc_ertms_wire.py) - exposes decode(buf) -> (nid_message, decoded_dict).
  - `ertms_encoders`: {catalog_message_name: (nid_engine, fields_dict,
    t_train) -> bytes} - one small adapter per message this sim sends,
    mapping ITS OWN flat {field: int} fields dict (the same shape every
    JSON sendMessage command already uses for the flat scheme) onto the
    codec's own per-message-shaped encode_msg_*() signature.
  - `ertms_flatteners`: {nid_message: decoded_dict -> flat_fields_dict} -
    one small adapter per message this sim receives, flattening the
    codec's own rich nested decode() output into the same flat
    {field: value} shape decode_to_message() already returns for the
    flat scheme.
Both default to empty/None - a catalog with no "ertms" messages (IL's,
CTC's) never needs any of this. Wire bytes for both families are wrapped
identically by one shared outer kind, declared once per catalog file as
a top-level `"_ertmsWrapperKind"` key (RBC_MSG_ERTMS_ENVELOPE=22 today -
moved from 21 after a real collision with a catalog's own flat
TRAIN_CONNECT=21, see __init__()'s own comment on the default below) -
kept catalog-data-driven like every other kind number this module
touches, never hardcoded here (see this module's own precedent for
`kind`/`nidMessage` above), read once at load time, absent/None if a
catalog defines no "ertms" messages at all.
"""
 
import json
import os
 
from . import rbc_wire as _wire
 
 
class UnknownMessageError(ValueError):
    """@message (sendMessage/getMessage) is not in this catalog."""
 
 
class UnknownFieldError(ValueError):
    """@fields (sendMessage) names a field that message's own packet
    does not define in this catalog."""
 
 
class InvalidFieldValueError(ValueError):
    """@fields (sendMessage) gives a field a value outside its catalog-
    defined allowed range/allowedValues."""
 
 
class WrongDirectionError(ValueError):
    """build_envelope() was asked to build a message this catalog marks
    "direction": "receive" - this sim is not the one that originates
    it (e.g. asking the Train sim to build an M3, which only the RBC
    itself ever sends)."""
 
 
class MessageCatalog:
    def __init__(self, catalog_path, ertms_codec=None, ertms_encoders=None, ertms_flatteners=None):
        """@param catalog_path: path to this sim's own message_catalog.json.
        Callers should pass an absolute path built from their own
        __file__ (e.g. os.path.join(os.path.dirname(os.path.abspath(
        __file__)), "message_catalog.json")) rather than a bare relative
        name, since a relative path here would otherwise depend on the
        process's own current working directory at startup, not on
        where the catalog file actually lives on disk.
        @param ertms_codec, ertms_encoders, ertms_flatteners: only needed
        if this catalog declares any "wireFormat": "ertms" message - see
        this module's own header doc for what each one is. Omitted
        (None/{}) for a catalog with no ertms messages (raises at load
        time below if that combination is wrong either way)."""
        path = catalog_path
        ertms_encoders = ertms_encoders or {}
        ertms_flatteners = ertms_flatteners or {}
        with open(path, "r", encoding="utf-8") as f:
            raw = json.load(f)
        messages = {}
        packets_catalog = {}
        has_ertms_messages = False
        import re
        for name, spec in raw.get("messages", {}).items():
            if name.startswith("P") and (name[1:].isdigit() or name in (
                "P0", "P1", "P2", "P11", "P12", "P15", "P65", "P66", "P68", "P72", "P131", "P136"
            )):
                packets_catalog[name] = spec
                continue
 
            wire_format = spec.get("wireFormat")
            if wire_format is None:
                if "kind" in spec and not name.startswith("M"):
                    wire_format = "flat"
                else:
                    wire_format = "ertms"
 
            if wire_format not in ("flat", "ertms"):
                raise ValueError(f"{path}: {name!r} has wireFormat {wire_format!r} - must be \"flat\" or \"ertms\"")
 
            direction = spec.get("direction")
            if not direction:
                digits = re.findall(r'\d+', name)
                if digits:
                    num = int(digits[0])
                    direction = "send" if num >= 120 else "receive"
                elif name in ("TRAIN_CONNECT", "TRAIN_DISCONNECT"):
                    direction = "send"
                else:
                    direction = "send"
 
            if direction not in ("send", "receive"):
                raise ValueError(f"{path}: {name!r} has direction {direction!r} - must be \"send\" or \"receive\"")
 
            if wire_format == "ertms":
                nid_message = spec.get("nidMessage")
                if nid_message is None:
                    digits = re.findall(r'\d+', name)
                    if digits:
                        nid_message = int(digits[0])
                    else:
                        nid_message = 0
                if not isinstance(nid_message, int) or isinstance(nid_message, bool) or not (0 <= nid_message <= 255):
                    raise ValueError(f"{path}: {name!r} (wireFormat=ertms) nidMessage={nid_message!r} "
                                     "is not a plausible wire byte (0-255) - required, not optional, for ertms messages")
 
                raw_fields = spec.get("fields", {})
                if isinstance(raw_fields, list):
                    all_fields = {f: {} for f in raw_fields}
                elif isinstance(raw_fields, dict):
                    all_fields = raw_fields
                else:
                    all_fields = {}
 
                has_ertms_messages = True
                msg_entry = {
                    "kind": None,
                    "nid_message": nid_message,
                    "direction": direction,
                    "packets": {},
                    "fields": all_fields,
                    "wire_format": "ertms",
                }
                messages[name] = msg_entry
                if name.startswith("M") and name[1:].isdigit():
                    messages[name[1:]] = msg_entry
                elif name.isdigit():
                    messages[f"M{name}"] = msg_entry
                continue
 
            packets = spec.get("packets", {})
            if not packets:
                raise ValueError(
                    f"{path}: {name!r} must define at least one packet in 'packets'"
                )
            kind = spec["kind"]
            if not isinstance(kind, int) or isinstance(kind, bool) or not (0 <= kind <= 255):
                raise ValueError(f"{path}: {name!r} kind={kind!r} is not a plausible wire kind byte (0-255)")
            nid_message = spec.get("nidMessage", 0)
            if not isinstance(nid_message, int) or isinstance(nid_message, bool) or not (0 <= nid_message <= 255):
                raise ValueError(f"{path}: {name!r} nidMessage={nid_message!r} is not a plausible wire byte (0-255)")
 
            all_fields = {}
            for pkt_name, pkt_spec in packets.items():
                if isinstance(pkt_spec, dict) and "fields" in pkt_spec:
                    all_fields.update(pkt_spec["fields"])
 
            msg_entry = {
                "kind": kind,
                "nid_message": nid_message,
                "direction": direction,
                "packets": packets,
                "fields": all_fields,
                "wire_format": "flat",
            }
            messages[name] = msg_entry
            if name == "TRAIN_CONNECT":
                messages["P0"] = msg_entry
            elif name == "P0":
                messages["TRAIN_CONNECT"] = msg_entry
 
        ertms_wrapper_kind = raw.get("_ertmsWrapperKind")
        if has_ertms_messages and ertms_wrapper_kind is None:
            # RBC_MSG_ERTMS_ENVELOPE (rbc_wire_types.h, both C-side copies)
            # - deliberately 22, NOT 21: TrainRBCSim's own
            # message_catalog.json defines a flat "TRAIN_CONNECT"/"P0"
            # message with kind=21 (added independently, after this value
            # already claimed 21) - a real, found-live collision (a plain
            # connect handshake was being routed into the ERTMS decoder
            # and rejected as malformed). Keep this default numerically
            # synced with the C-side enum if it ever moves again.
            ertms_wrapper_kind = 22
        if ertms_wrapper_kind is not None:
            if not isinstance(ertms_wrapper_kind, int) or isinstance(ertms_wrapper_kind, bool) \
                    or not (0 <= ertms_wrapper_kind <= 255):
                raise ValueError(f"{path}: _ertmsWrapperKind={ertms_wrapper_kind!r} is not a plausible wire byte (0-255)")
        if has_ertms_messages and ertms_codec is None:
            raise ValueError(f"{path}: has wireFormat=ertms message(s) but no ertms_codec was injected "
                             "(MessageCatalog(catalog_path, ertms_codec=..., ...) - see this module's own doc)")
 
        self._path = path
        self._messages = messages
        self._packets = packets_catalog
        self._ertms_codec = ertms_codec
        self._ertms_encoders = ertms_encoders
        self._ertms_flatteners = ertms_flatteners
        self._ertms_wrapper_kind = ertms_wrapper_kind
        self._name_by_kind = {spec["kind"]: name for name, spec in messages.items() if spec["wire_format"] == "flat"}
        # Deliberately excludes the bare-digit alias keys (e.g. "24") this
        # loader also registers for every "M<nid>"-named ertms message
        # (see the aliasing just above) - a real, found-live bug: without
        # this filter, a dict comprehension over ALL of messages.items()
        # picks whichever of the two keys for the same nid_message was
        # inserted LAST (Python dict insertion order), which is always
        # the bare-digit alias (registered immediately after its own
        # canonical name in the loop above) - so decode_to_message() was
        # silently returning "24" instead of "M24" for every incoming
        # real Subset-026 Msg 24, even though callers only ever look it
        # up by the canonical "M24" name (message_catalog.json's own
        # documented convention, and the only form a Robot test would
        # sensibly request). Restricting this reverse map to "M"-prefixed
        # keys makes the canonical name win unconditionally, regardless
        # of dict insertion order.
        self._name_by_nid_message_ertms = {spec["nid_message"]: name for name, spec in messages.items()
                                           if (spec["wire_format"] == "ertms") and name.startswith("M")}
 
    def message_names(self):
        """@return every message name in this catalog, sorted."""
        return sorted(self._messages)
 
    def is_ertms(self, message):
        """@return True if @message is a "wireFormat": "ertms" catalog
        entry, False if "flat". Lets a caller decide whether
        fields_for(message)'s own field-NAME list is safe to use as a
        real, populated-variable-store filter (true for flat messages,
        whose "fields" are this project's own snake_case variable names)
        or is purely Subset-026 documentation text unrelated to any
        variable a caller would actually have stored (true for ertms
        messages - see message_catalog.json's own field lists for e.g.
        "M129"/"M132": ["NID_MESSAGE", "L_MESSAGE", ..., "Train data
        (Packet type 11)"], never matching real stored field names like
        "l_train"/"v_maxtrain"). Raises UnknownMessageError if @message
        is not in this catalog at all."""
        if message not in self._messages:
            raise UnknownMessageError(f"unknown message {message!r} - known: {self.message_names()}")
        return self._messages[message]["wire_format"] == "ertms"
 
    def build_ertms_envelope_raw(self, ertms_codec_bytes, train_id):
        """@return the wire datagram wrapping already-encoded ertms_codec
        bytes with this catalog's own _ertmsWrapperKind+train_id prefix -
        the same 2-byte prefix build_envelope() applies for a catalog-
        registered "wireFormat": "ertms" message, exposed for a sim's own
        internally-driven sends (e.g. a periodic heartbeat) that have no
        catalog entry of their own name to look up an encoder by. Raises
        if this catalog has no ertms wrapper kind configured at all."""
        if self._ertms_wrapper_kind is None:
            raise ValueError(f"{self._path} has no _ertmsWrapperKind configured - "
                             "cannot build a raw ertms envelope")
        return bytes([self._ertms_wrapper_kind, train_id & 0xFF]) + ertms_codec_bytes
 
    def packet_names(self):
        """@return every packet name in this catalog, sorted."""
        return sorted(self._packets)
 
    def packets_for(self, message):
        """@return dict of packet_name -> packet_spec for @message."""
        if message not in self._messages:
            raise UnknownMessageError(f"unknown message {message!r} - known: {self.message_names()}")
        return self._messages[message]["packets"]
 
    def fields_for(self, message, packet=None):
        """@return {field_name: spec_dict} for @message. If packet is specified,
        returns fields for that specific packet; otherwise returns all fields for the message."""
        if message not in self._messages:
            raise UnknownMessageError(f"unknown message {message!r} - known: {self.message_names()}")
        if packet is not None:
            packets = self._messages[message]["packets"]
            if packet in packets:
                return packets[packet].get("fields", {})
            if packet == message:
                return self._messages[message]["fields"]
            raise UnknownFieldError(f"message {message!r} has no packet {packet!r} - known: {list(packets)}")
        return self._messages[message]["fields"]
 
    def build_envelope(self, message, train_id, fields):
        """@return the raw wire datagram for @message/@train_id/@fields (a
        dict of field name -> int, may omit fields to leave them 0).
        Raises UnknownMessageError/WrongDirectionError/UnknownFieldError/
        InvalidFieldValueError on a bad request - a typo'd name, a
        message this sim does not originate, or an out-of-catalog value
        (e.g. msg_type=99, not 24 or 15) must fail loudly here, not be
        silently sent onto the wire as garbage.
 
        For a "wireFormat": "ertms" message, delegates to this catalog's
        own injected ertms_encoders[message] adapter (byte length is
        genuinely message-specific, not this method's own concern the
        way the flat scheme's l_message computation below is) and wraps
        the result with the shared 2-byte _ertmsWrapperKind+train_id
        prefix - see this module's own header doc."""
        if message not in self._messages:
            raise UnknownMessageError(f"unknown message {message!r} - known: {self.message_names()}")
        if self._messages[message]["direction"] != "send":
            raise WrongDirectionError(f"{message} is direction=receive in {self._path} - this sim does not send it")
        if self._messages[message]["wire_format"] == "ertms":
            encoder = self._ertms_encoders.get(message) or self._ertms_encoders.get(message.lstrip("M")) or self._ertms_encoders.get(f"M{message}")
            if encoder is None:
                raise ValueError(f"no ertms_encoders[{message!r}] was injected for ERTMS send message")
            codec_bytes = encoder(train_id, fields, fields.get("t_train", 0))
            return bytes([self._ertms_wrapper_kind, train_id & 0xFF]) + codec_bytes
 
        field_specs = self.fields_for(message)
        unknown = set(fields) - set(field_specs)
        if unknown:
            raise UnknownFieldError(
                f"{message} does not have field(s) {sorted(unknown)} - allowed: {sorted(field_specs)}"
            )
        for field, value in fields.items():
            _validate_value(message, field, field_specs[field], value)
        # REQ-RBC-001 (revised): l_message is genuinely calculated here,
        # from this message's OWN catalog field list, not hand-set by a
        # caller - HEADER(12) + BASE_CONTENT(24) is every existing
        # message's own shape (unchanged), plus whichever kind-specific
        # extra content this message's catalog fields actually declare -
        # mirrors rbc_wire.c's own content_size_for_kind(), the C side's
        # equally-hardcoded (not catalog-driven there) equivalent. At
        # most one of these applies per message today (each catalog
        # message maps to exactly one kind, which has exactly one extra-
        # content shape - see rbc_wire_types.h's own doc).
        extra_content = 0
        if "t_train_ack" in field_specs:
            extra_content = 4  # RBC_MSG_M146's own extra content (rbc_wire.c's RBC_WIRE_CONTENT_SIZE_M146)
        elif "nid_lrbg" in field_specs:
            # RBC_MSG_M136's own extra content (rbc_wire.c's RBC_WIRE_CONTENT_SIZE_M136) - the
            # (unused, always-zero for this kind) t_train_ack slot(4) it sits after on the wire,
            # PLUS the 10 real Packet 0 fields(40) - see rbc_wire.c's own comment: fields are
            # always written at the same fixed offsets for every kind, so M136's own fields start
            # right AFTER t_train_ack's slot, not instead of it.
            extra_content = 4 + 40
        elif ("start_signal" in field_specs) or ("end_signal" in field_specs):
            # RBC_MSG_ROUTE_ADD/_ROUTE_RELEASE/_TRAIN_POSITION_IN_ROUTE's own extra content
            # (IL<->RBC route-identity pass) - these fields sit at offsets 80-95, past the
            # (unused-for-these-kinds) t_train_ack(4)+M136 block(40) - so the whole gap up to
            # them (44) plus their own 16 bytes is "extra content" the same way M136's own
            # block is, even though these kinds never touch t_train_ack/M136's own fields.
            extra_content = 4 + 40 + 16
        else:
            extra_content = 0
        l_message = 12 + 24 + extra_content
        return _wire.encode(
            self._messages[message]["kind"],
            train_id,
            nid_message=self._messages[message]["nid_message"],
            l_message=l_message,
            **fields,
        )
 
    def decode_to_message(self, data):
        """@return (train_id, message_name, fields_dict) for a raw
        received datagram, or None if malformed/unrecognized by THIS
        catalog (mirrors rbc_wire.decode()'s own contract -
        REQ-RBC-002's Python side) - including a structurally valid
        envelope of a kind this sim's own catalog simply does not list
        (e.g. a ROUTE_ADD arriving at the Train sim, which should never
        happen but is treated the same as any other unrecognized kind,
        not a crash). fields_dict holds only the fields THIS message's
        own packet defines, not the whole flat envelope.
 
        First checks data[0] against this catalog's own _ertmsWrapperKind
        (if any "ertms" messages are declared) and routes to the injected
        ertms_codec + ertms_flatteners for that family; falls through to
        the original flat-scheme path (_wire.decode()) otherwise - a
        single call site works for either family, same as build_envelope()."""
        if (self._ertms_wrapper_kind is not None) and (len(data) >= 2) and (data[0] == self._ertms_wrapper_kind):
            train_id = data[1]
            try:
                nid_message, decoded = self._ertms_codec.decode(data[2:])
            except Exception:
                # ErtmsDecodeError (or any other decode failure) - same
                # "malformed/unrecognized is None, not a crash" contract
                # the flat branch below already has for _wire.decode().
                return None
            name = self._name_by_nid_message_ertms.get(nid_message)
            if name is None:
                return None
            flattener = self._ertms_flatteners.get(nid_message)
            fields = flattener(decoded) if flattener is not None else decoded
            return train_id, name, fields
 
        env = _wire.decode(data)
        if env is None:
            return None
        name = self._name_by_kind.get(env["kind"])
        if name is None:
            return None
        fields = {field: env[field] for field in self._messages[name]["fields"]}
        return env["train_id"], name, fields
 
 
def _validate_value(message, field, spec, value):
    if not isinstance(value, int) or isinstance(value, bool):
        raise InvalidFieldValueError(f"{message}.{field}: value {value!r} is not an integer")
    allowed_values = spec.get("allowedValues")
    if allowed_values is not None:
        if value not in allowed_values:
            raise InvalidFieldValueError(f"{message}.{field}: {value} not in allowed values {allowed_values}")
        return
    lo, hi = spec.get("min"), spec.get("max")
    if lo is not None and value < lo:
        raise InvalidFieldValueError(f"{message}.{field}: {value} < minimum {lo}")
    if hi is not None and value > hi:
        raise InvalidFieldValueError(f"{message}.{field}: {value} > maximum {hi}")