train_sim.py

Go to the documentation of this file.

#!/usr/bin/env python3
"""Train sim (ADR-029) - a real, dual-homed UDP client. Connects to BOTH
sites' C (c-west and c-east) at once and keeps both sockets open for the
whole run, so a site failover is invisible from this process's own point
of view (see simcore.dual_link's own header) - it sends its own
P0/136/146 to both sites unconditionally and simply acts on whichever
site actually answers (the STANDBY site's own C has nothing to forward
back, since its A/B pair isn't computing decisions - see monitor_c.c).
 
Message names/keys are this project's own (message_catalog.json, right
next to this script) - real Subset-026 message numbers where a real
counterpart exists ("3" = real message 3 Movement Authority, "146" =
real message 146 Acknowledgement - see that file's own
"_subset026Reference" for the real spec text), and clearly-labeled
project-specific names where none does ("P0" handshake -
neither corresponds to any real Subset-026 message). These are still
encoded with today's simplified flat fields, not the real nested
Subset-026 packet structure - see message_catalog.json's own per-message
comments and root CLAUDE.md's Subset-026 note for what that gap actually
is. Msg 136 (Train Position Report) is the one exception, switched this
session on direct request: this sim's autonomous cyclic reporting AND
the REPORT/sendTrainMessage commands all send the REAL Subset-026 Msg
136/Packet 0 now (rbc_ertms_wire.py's own "M136" encoder, "ertms"
wireFormat via message_catalog.json's implicit "M"-prefixed-name
default), not a flat kind - see send_position_report()'s own doc.
 
Configuration: one JSON file (common/sim_config.py, SIM_CONFIG_PATH env
var, default /app/config.json) supplies everything instance-specific -
trainIndex (0 or 1, selects which of this project's
RTE_EXAMPLE_MAX_TRAINS train slots this instance is - train_id is
trainIndex + 1), siteTag (display-only, kept for log-prefix consistency
with every other role - not a site selector, this sim talks to both
sites regardless), rbcWest/rbcEast (host+port to reach each site's C),
controlServer (bindHost+port for the command channel below), and
simPeriodSeconds. See sims/train/config/*.json for the actual files and
docker-compose.yml's own `volumes:` entries for how each container gets
its own. Message vocabulary (which messages/fields this sim may send or
receive) comes from message_catalog.json, right next to this script -
safeAPITestEnv/doc/design/SIM_INTERFACES.md documents the same table in
prose.
 
Train lifecycle is test-managed, not process-managed: the TCP transport
itself connects (and reconnects-forever, dual_link.py's own convention)
the moment this process starts, same as always, but NEITHER the P0
handshake NOR the autonomous cyclic "136" position reporting fires until
a test explicitly connects this train (CONNECT / the connectTrain JSON
command below) - a test that never connects a given train leaves it
genuinely absent from the RBC (ab_gp_train.c's own RBC_MSG_P0/
RBC_MSG_M136 handling never creates that train's session at all), not
just quiet on this sim's own side. Once connected, position reporting
keeps running autonomously (a real train reports position continuously,
not on command) until disconnectTrain (below) - a test harness
(safeAPITestEnv/robot/) can also drive specific position reports on
demand via the command channel (common/control_server.py) regardless of
connection state, e.g. to deterministically hit the D_LRBG single-
channel-fault value without waiting on the random walk.
 
Commands (controlServer.port, config file - see common/control_server.py):
  REPORT <d_lrbg>   Sends one message "136" (position report) immediately
                    with the given position (does not wait for the next
                    autonomous tick; also resets the autonomous random
                    walk to continue from this value). Works regardless
                    of connection state (an explicit test-driven send,
                    not the autonomous loop) - a test using this without
                    ever connecting the train is exercising the RBC's own
                    documented lenient "M136 arriving before/without a
                    P0" tolerance. Replies "OK\\n".
  PING              Replies "PONG\\n".
  CONNECT           Explicitly connects this train to the RBC: sends P0
                    now (not waiting for the next reconnect edge) and
                    starts autonomous "136" reporting. Replies "OK\\n".
                    See the structured connectTrain/disconnectTrain
                    commands below for the nidEngine-explicit,
                    multi-train-aware equivalents this file's own test
                    suite actually uses.
  DISCONNECT        Closes BOTH site connections right now (simulating
                    the train losing radio contact with the RBC
                    entirely) - reconnects automatically afterward, same
                    as any other transport-level disconnect. Deliberately
                    does NOT touch connected_to_rbc or send
                    RBC_MSG_TRAIN_DISCONNECT - this simulates losing
                    radio contact, not a formal disconnect; the RBC-side
                    session is untouched and autonomous reporting resumes
                    the moment the socket reconnects. Replies "OK\\n".
  DISCONNECTWITHONLINE
                    Closes the connection to whichever site this sim
                    currently believes is ONLINE (see "online-site
                    tracking" below) - simulating losing contact with
                    the active RBC specifically. Replies "OK\\n", or
                    "ERR <reason>\\n" if no site has been heard from yet
                    (nothing to call "online").
  DISCONNECTWITHSTANDBY
                    Same as DISCONNECTWITHONLINE but for the OTHER site
                    (whichever this sim does NOT currently believe is
                    ONLINE) - simulating losing contact with the standby
                    RBC specifically, while the active one stays
                    reachable. Same ERR case as DISCONNECTWITHONLINE.
 
Online-site tracking: this sim has no explicit ONLINE/STANDBY signal
from the wire (the RBC side doesn't tell a train which of A/B/site is
currently active) - "online" here is inferred as whichever site most
recently sent this sim ANY decodable message (updated in the main
receive loop, below), the same de-facto signal the sim already uses
operationally (only the ONLINE site's own C ever answers - this file's
own header). Purely a heuristic for the two DISCONNECTWITH* commands
above; not itself sent anywhere or asserted on by anything else.
 
Structured JSON commands (safeAPITestEnv/doc/design/DESIGN.md,
SIM_INTERFACES.md - same port, see common/control_server.py's own doc
for how the two command styles share one line-based channel):
  sendMessage   {"cmd":"sendMessage","nidEngine":N,"message":"136","fields":{...}}
                Sends one message NOW (any message this train can
                originate - P0/136/146, message_catalog.json) with the
                given fields; unset fields default to 0. Does NOT touch
                the autonomous cycle counter/random-walk state the way
                REPORT does - a purely explicit, one-shot send.
  getMessage    {"cmd":"getMessage","nidEngine":N,"message":"3"}
                Returns the most recently received message of that kind
                for that nidEngine ("PositionReportAck" or "3" - the
                only kinds this sim ever receives), or an ERR if none
                has arrived yet.
  connectTrain  {"cmd":"connectTrain","nidEngine":N,"nidLrbg":M,"dLrbg":D}
                Explicitly connects train N to the RBC: sends P0 now and
                starts this train's autonomous "136" reporting (see this
                file's own "Train lifecycle" paragraph above) - the
                nidEngine-explicit structured equivalent of the plain
                CONNECT command, and what safeAPITestEnv/robot/'s own
                "Connect Train" keyword actually calls. "nidLrbg"/"dLrbg"
                are both optional (either, neither, or both) - override
                this train's starting balise/position instead of
                _default_lrbg()'s own west/east heuristic, applied
                before the first real "136" - see handle_connect_train()'s
                own doc for why both state["position_m"] and
                train_vars["d_lrbg"] get set. Replies {"status":"OK"}.
  disconnectTrain
                {"cmd":"disconnectTrain","nidEngine":N}
                Sends RBC_MSG_TRAIN_DISCONNECT (rbc_wire_types.h) for
                train N and stops its autonomous reporting - the RBC-side
                session is actually removed (ab_gp_train.c releases every
                route the train held, then removes it), not just starved
                of traffic. Does NOT close the TCP transport (see
                DISCONNECT above for that, a different, radio-silence
                concept) - "Disconnect Train" in the Robot library.
                Replies {"status":"OK"}.
  setTrainVariable
                {"cmd":"setTrainVariable","nidEngine":N,"variable":"d_lrbg","value":123}
                Stores one SS026-style field for this train (nid_lrbg/
                d_lrbg/v_train/m_mode/m_level), persisting until
                overwritten - sendTrainMessage (below) pulls whatever a
                given message's own catalog entry declares out of this
                store at send time. Replies {"status":"OK"}.
  sendTrainMessage
                {"cmd":"sendTrainMessage","nidEngine":N,"message":"136"}
                Sends message N NOW, built from whatever setTrainVariable
                has stored for this train (message_catalog.json's own
                field list for that message) - an explicit, one-shot send
                like sendMessage above, but from the persistent variable
                store rather than an inline fields dict. Works regardless
                of connection state, same as REPORT/sendMessage. Replies
                {"status":"OK"} or {"status":"ERR","reason":"..."}.
"""
 
import logging
import os
import random
import signal
import sys
import time
 
from simcore import sim_config
from simcore.control_server import ControlServer
from simcore.dual_link import SITES, DualHomedLink
from simcore.safecomm import link_from_config as _safecomm_link_from_config
from simcore import saf_train_link
from simcore.rbc_messages import MessageCatalog
from simcore.status_report import StatusReporter
 
from . import rbc_ertms_wire as ertms_codec
from . import ertms_adapters
from . import site_data
 
CATALOG = MessageCatalog(
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "message_catalog.json"),
    ertms_codec=ertms_codec,
    ertms_encoders=ertms_adapters.ENCODERS,
    ertms_flatteners=ertms_adapters.FLATTENERS,
)
 
# RBC_TRAIN_ENVELOPE_WIRE_SIZE (common_config.h) - the Train relay
# channel's own physical slot size, wider than the flat scheme's own
# ENVELOPE_SIZE (96) to also fit a real Subset-026 ertms message
# (RBC_MSG_ERTMS_ENVELOPE, rbc_wire_types.h). GP's own receive side
# requires an EXACT match to whatever it requested to read (a genuine
# wire-protocol-violation check, not framing trivia - see
# rte_posix_backend_netlink.c's own backend_read()), so EVERY send on
# this channel - flat scheme included, not just the new ertms one - must
# be padded to this exact size, not just the "real" content length.
TRAIN_WIRE_SIZE = 512
 
 
def _pad(frame):
    """Zero-pads @frame up to TRAIN_WIRE_SIZE - both wire families
    already put their own real length in their own header (l_message),
    so the receiving side reads exactly the meaningful part regardless
    of this trailing padding; this padding exists purely to satisfy the
    transport's own exact-datagram-size expectation on this channel."""
    if len(frame) > TRAIN_WIRE_SIZE:
        raise ValueError(f"encoded frame is {len(frame)} bytes, exceeds TRAIN_WIRE_SIZE={TRAIN_WIRE_SIZE}")
    return frame + bytes(TRAIN_WIRE_SIZE - len(frame))
 
 
# This one sim process can host up to this many DISTINCT nid_engine
# values over its lifetime (a soft capacity guard, not a pre-sized
# roster - see main()'s own doc on why there is no roster at all now).
# Purely a sanity bound on handle_connect_train() - nowhere near GP's
# own current RTE_EXAMPLE_MAX_TRAINS=2U (common_config.h); this sim
# is deliberately more permissive than today's RBC-side cap, so it's
# never itself the limiting factor.
MAX_TRAINS = 100
 
 
def _default_lrbg(tid):
    """A train's own starting LRBG before any real position report has
    set one - purely a plausible default so an early sendTrainMessage/136
    isn't sending nid_lrbg=0. Same "train 1 near the west end, everything
    else near the east end" heuristic this file always used, just now a
    per-train function instead of a single value precomputed from a
    static roster's first entry (there is no roster to read "first
    entry" from any more)."""
    return 1001 if tid == 1 else 1009
 
 
def main():
    config = sim_config.load()
    site_tag = sim_config.require(config, "siteTag")
    period = float(config.get("simPeriodSeconds", 2.0))
    control_host = sim_config.require(config, "controlServer", "bindHost")
    control_port = sim_config.require(config, "controlServer", "port")
 
    # "[%(levelname)s] " prefix on every line - on direct request: the
    # test env's own merged-log/live-stream listener
    # (ContainerLogAttachListener.py) used to have to GUESS a level for
    # every line (defaulting to INFO when none was detectable), since
    # this format string used to strip Python's own real level
    # information entirely before it ever reached stdout/the log file -
    # log.warning()/log.error() calls were textually indistinguishable
    # from log.info() ones. This emits the sim's own REAL level as data,
    # collected (not reconstructed) by the test env from here on.
    logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s", stream=sys.stdout)
    log = logging.getLogger("train")
 
    # A "commServer" config block routes the RBC link through the
    # RBC_Test_CommServer access-point (Platform_Protocol_SafeComm path): one TCP link, with
    # RBC_Test_CommServer owning the WEST/EAST dual-homing. Absent -> the direct
    # dual-homed UDP link to each site's role-C gateway, as before.
    link = _safecomm_link_from_config(log, config, TRAIN_WIRE_SIZE, sim_config)
    if link is not None:
        log.info("train link: via RBC_Test_CommServer (Platform_Protocol_SafeComm path)")
    else:
        targets = {
            "WEST": (sim_config.require(config, "rbcWest", "host"), sim_config.require(config, "rbcWest", "port")),
            "EAST": (sim_config.require(config, "rbcEast", "host"), sim_config.require(config, "rbcEast", "port")),
        }
        link = DualHomedLink(log, targets, TRAIN_WIRE_SIZE)
    # Subset-037 SaF on the ERTMS traffic (SAFE_COMM_INTEGRATION.dox plan steps 2/3): the same two variables that turn
    # it on in RBC_GP's A/B processes. Unset (the default) leaves the link exactly as before.
    saf_rbc_id = os.environ.get("RTE_SAFECOMM_TRAIN_RBC_ID", "").strip()
    if saf_rbc_id:
        keys_path = os.environ.get("RTE_SAFECOMM_TRAIN_KEYS_FILE", "").strip()
        if not keys_path:
            log.error("[SAF] RTE_SAFECOMM_TRAIN_RBC_ID set without RTE_SAFECOMM_TRAIN_KEYS_FILE - refusing to start")
            sys.exit(2)
        link = saf_train_link.SafTrainLink(link, log, int(saf_rbc_id, 0), saf_train_link.load_keys_file(keys_path),
                                           TRAIN_WIRE_SIZE)
        log.info("train link: ERTMS traffic protected by Subset-037 SaF (rbc_etcs_id=%s)", saf_rbc_id)
    was_connected = {site: False for site in SITES}
    # Per-train autonomous-walk state (ADR-036: one process, many trains -
    # up to MAX_TRAINS). Deliberately starts EMPTY, not pre-sized from any
    # config roster: this process is connected to the RBC (transport-wise)
    # from the moment it starts, but hosts NO train until a test explicitly
    # names one via connectTrain - "no default trains" (per direct
    # instruction) - a train's own state is created lazily, the first time
    # it's actually named, by _ensure_train() below.
    state = {}
    online_site = {"value": None}  # heuristic - see this file's own "Online-site tracking" doc above
 
    # Train lifecycle is test-managed, not process-managed: this TCP
    # transport connects (and reconnects-forever) on its own the moment
    # the process starts, same as always, but NEITHER the P0 handshake
    # NOR the autonomous cyclic M136 reporting fires until an explicit
    # connect (CONNECT / connectTrain, below) sets this true - a test
    # that never connects this train leaves it genuinely absent from the
    # RBC (no session ever created there - see ab_gp_train.c's own
    # RBC_MSG_P0/RBC_MSG_M136 handling), not just quiet on this sim's
    # own side. disconnectTrain (below) sets it back to false AND sends
    # RBC_MSG_TRAIN_DISCONNECT so the RBC-side session is actually
    # removed, not just starved of further traffic.
    # Per-train connect state (nid_engine -> bool). Was a single shared
    # {"value": bool} until this fix - found live via a real symptom: a
    # test that only ever calls `Connect Train ${TRAIN_1_NID_ENGINE}`
    # (e.g. 12_ertms_shunting_mode_procedure.robot) was ALSO producing
    # autonomous M136 traffic for train 2, because activeTrains=2 means
    # this one sim process manages BOTH train 1 and train 2 (ADR-036
    # multiplexing), and the single shared flag being set true by
    # connecting train 1 made the autonomous-report loop below (which
    # iterates every train in train_ids) start reporting for train 2 as
    # well - a train that specific test never connected at all. Symmetric
    # bug on the other end too: disconnecting train 1 was silently
    # stopping train 2's reporting as well. This file's own header doc
    # already documented per-train semantics ("connectTrain: ... starts
    # THIS train's autonomous reporting") - the implementation just never
    # actually matched that doc until now.
    connected_to_rbc = {}
 
    # Real Subset-026 mission-phase tracking (new, per the "critical
    # architecture issue" request): "IDLE" (default / after a fresh
    # connect or M155) means active-mission - the existing 2s Packet-0
    # cadence below applies. "POST_EOM" (set when this sim itself sends
    # M150/M156 via handle_send_message) means the train has finished its
    # mission but stayed transport-connected - a SEPARATE 6s Packet-1
    # heartbeat cadence applies instead (real Subset-026 Packet 1 is the
    # degraded/no-longer-actively-tracked LRBG-orientation variant, a
    # good fit for a stationary train no longer under active MA). Purely
    # local bookkeeping for logging/cadence-selection - the RBC's own
    # ab_gp_ertms_session_state_t (ab_gp_proc_dispatch.c's handle_msg_136)
    # stays the sole authority on what a heartbeat is actually allowed to
    # do session-state-wise.
    # Also starts empty (see `state`'s own doc just above) - populated the
    # same lazy way, and read everywhere via .get(tid, ...) already, so an
    # unconnected/never-seen train simply behaves as "IDLE" with no
    # explicit entry needed.
    ertms_post_eom = {}
    HEARTBEAT_PERIOD_S = 6.0
    next_heartbeat = {}
 
    # Resolved once, not per (train, site) iteration below - and NEVER
    # falls back to a real ERTMS message name (e.g. "M155"): TRAIN_CONNECT/
    # P0 are a low-level transport-connect signal, not a session-init
    # request - those are different messages with different meaning, not
    # interchangeable names for the same thing. If neither exists in a
    # catalog, that is a real configuration problem worth failing loudly
    # on, not silently sending the wrong message instead (found live:
    # the previous fallback chain would have done exactly that).
    _connect_msg_name = "TRAIN_CONNECT" if "TRAIN_CONNECT" in CATALOG.message_names() else (
        "P0" if "P0" in CATALOG.message_names() else None)
    if _connect_msg_name is None:
        log.error("[IO] [%s] [internal] [CONFIG] [catalog has neither TRAIN_CONNECT nor P0 - "
                  "train connect handshake cannot be sent]", f"train-{site_tag.lower()}")
 
    def send_p0_now(target_nid):
        """Sends the connect handshake for exactly ONE train, @target_nid -
        no implicit "every train" fan-out any more (there is no roster to
        fan out over - see main()'s own "no default trains" doc). The
        legacy plain CONNECT command (handle_connect below) now passes a
        single hardcoded nid_engine explicitly, same as connectTrain
        already did."""
        if _connect_msg_name is None:
            return
        for site in SITES:
            s = link.sockets[site]
            if s is not None:
                try:
                    s.sendall(_pad(CATALOG.build_envelope(_connect_msg_name, target_nid, {})))
                    log.info("[IO] [Train %d] [RBC %s] [%s] [ERTMS Train connection handshake sent] [train_id=%d]",
                             target_nid, site.title(), _connect_msg_name, target_nid)
                except OSError:
                    pass
 
    # Most-recently-received message per (nid_engine, message name) -
    # safeAPITestEnv/doc/design/DESIGN.md section 4.2. This sim only ever
    # sees its own train_id on the wire, so nid_engine here is always
    # train_id, but keyed by it anyway (not a bare dict-by-message) for
    # the same forward-looking reason SIM_INTERFACES.md gives for
    # getMessage always taking nidEngine explicitly.
    last_received = {}
 
    # Per-train SS026-variable store, {nid_engine: {field_name: value}} -
    # persists across calls until a test overwrites it (setTrainVariable),
    # NOT cleared after a send the way the old Set Message/Set Packet
    # builder pattern was. sendTrainMessage compiles a message by pulling
    # whatever fields THAT message's own catalog entry declares
    # (CATALOG.fields_for()) out of this store at send time - a variable
    # nobody has set yet is simply absent (build_envelope() defaults an
    # unset field to 0, same as the plain sendMessage command already
    # does). Field names here are this catalog's own (message_catalog.json
    # "fields" keys, e.g. "d_lrbg"/"cycle") - the same simplified,
    # flat-encodable vocabulary "messages" uses, not yet the real SS026
    # UPPER_CASE variable names in "_subset026Reference" (those aren't
    # encodable on today's wire protocol at all - see that key's own doc).
    train_vars = {}
 
    def _reset_train(tid):
        state[tid] = {"position_m": 0, "cycle": 0}
        train_vars[tid] = {
            "nid_lrbg": _default_lrbg(tid),
            "d_lrbg": 0,
            "v_train": 0,
            "m_mode": 0,
            "m_level": 2,
        }
        last_received.pop(tid, None)
 
    def _ensure_train(tid):
        """Lazily creates @tid's own state/train_vars entries the first
        time it's actually named (connectTrain, or the legacy REPORT/
        CONNECT commands) - see main()'s own "no default trains" doc.
        Idempotent - safe to call on every touch, not just the first."""
        state.setdefault(tid, {"position_m": 0, "cycle": 0})
        train_vars.setdefault(tid, {
            "nid_lrbg": _default_lrbg(tid),
            "d_lrbg": 0,
            "v_train": 0,
            "m_mode": 0,
            "m_level": 2,
        })
 
    # Read fresh from the real RailML file (site_data.py) rather than
    # hardcoded here - was, by direct comparison, a byte-for-byte
    # duplicate of that file's own <baliseGroup> data. See site_data.py's
    # own doc for the fallback/logging posture if the file can't be read
    # (e.g. a Docker image without the sibling RBC_SA checkout).
    BALISE_NAMES = site_data.balise_names()
 
    def get_balise_name(nid):
        if not nid:
            return ""
        return BALISE_NAMES.get(int(nid), f"BG_{nid}")
 
    def send_position_report(tid, d_lrbg, nid_lrbg=None):
        """Sends this train's cyclic position report as a REAL Subset-026
        Msg 136 / Packet 0 (CATALOG's own "M136" ertms entry -
        ab_gp_ertms_train_message_codec.c on the RBC side), not the old
        flat scheme's own "136" kind - switched this session, on direct
        request, for every caller of this shared function (the autonomous
        cyclic loop below AND the REPORT/handle_report command), so there
        is exactly one M136-sending code path, not two. GP's own real
        Msg 136 handler (handle_msg_136(), ab_gp_proc_dispatch.c) already
        works standalone (no prior Msg 155 session handshake required) and
        already replies with a real Msg 24 - see that function's own doc.
        "cycle" stays purely local bookkeeping (this sim's own REPORT-
        command/logging counter) - real Msg 136 has no counter field of
        its own to carry it on the wire (message_catalog.json's own "M136"
        reference entry doc: "cycle... is a project-invented counter with
        no real-message-136 equivalent")."""
        _ensure_train(tid)
        st = state[tid]
        st["cycle"] += 1
        st["position_m"] = d_lrbg
        vars_for_train = train_vars.setdefault(tid, {})
        vars_for_train["d_lrbg"] = d_lrbg
        if nid_lrbg is not None:
            vars_for_train["nid_lrbg"] = nid_lrbg
 
        cur_lrbg = vars_for_train.get("nid_lrbg", _default_lrbg(tid))
        cur_v = vars_for_train.get("v_train", 0)
        cur_mode = vars_for_train.get("m_mode", 0)
        cur_level = vars_for_train.get("m_level", 2)
 
        # Full real Packet 0 field set (ertms_adapters.py's own
        # _position_report_from_fields() supplies the same defaults for
        # q_scale/q_dirlrbg/q_dlrbg/l_doubtover/l_doubtunder when a
        # sendMessage/sendTrainMessage command builds M136 instead of this
        # autonomous path) - nid_c defaults to 0 (this sim never models a
        # real multi-country NID_C), nid_bg is this sim's own flat
        # "nid_lrbg" balise-ID convention.
        wire_fields = {
            "q_scale": vars_for_train.get("q_scale", ertms_codec.Q_SCALE_1M),
            "nid_c": vars_for_train.get("nid_c", 0),
            "nid_bg": cur_lrbg,
            "d_lrbg": d_lrbg,
            "q_dirlrbg": vars_for_train.get("q_dirlrbg", ertms_codec.Q_DIRLRBG_NOMINAL),
            "q_dlrbg": vars_for_train.get("q_dlrbg", ertms_codec.Q_DLRBG_NOMINAL),
            "l_doubtover": vars_for_train.get("l_doubtover", 0),
            "l_doubtunder": vars_for_train.get("l_doubtunder", 0),
            "v_train": cur_v,
            "m_mode": cur_mode,
            "m_level": cur_level,
        }
        link.send_all(_pad(CATALOG.build_envelope("M136", tid, wire_fields)))
 
        # Returned for this function's own two callers (REPORT/handle_report
        # and the cyclic loop) to log - "cycle"/"nid_lrbg" kept as the
        # names those callers/get_balise_name() already expect, alongside
        # every real wire field actually sent.
        payload = dict(wire_fields)
        payload["cycle"] = st["cycle"]
        payload["nid_lrbg"] = cur_lrbg
        return payload
 
    def handle_report(args):
        # Legacy, non-nidEngine-aware command - always targets train 1
        # (there is no roster/"first train" any more to derive this from;
        # see main()'s own "no default trains" doc). Use REPORT's own
        # structured equivalent, sendTrainMessage with an explicit
        # nidEngine, for anything beyond train 1.
        tid = 1
        d_lrbg = int(args[0])
        nid_lrbg = int(args[1]) if len(args) > 1 else None
        payload = send_position_report(tid, d_lrbg, nid_lrbg)
        fields_str = " ".join(f"{k}={v}" for k, v in sorted(payload.items()))
        bname = get_balise_name(payload.get("nid_lrbg"))
        balise_tag = f" balise={bname}" if bname else ""
        log.info("[IO] [Train %d] [RBC West, RBC East] [M136] [ERTMS Train position report commanded] "
                 "[train_id=%d] [P0] [%s%s]",
                 tid, tid, fields_str, balise_tag)
        return "OK\n"
 
    def handle_ping(_args):
        return "PONG\n"
 
    def handle_connect(_args):
        # Legacy, non-nidEngine-aware command - connects train 1 only
        # (there is no roster to fan out over any more; see main()'s own
        # "no default trains" doc). Use connectTrain for any other train.
        tid = 1
        _ensure_train(tid)
        connected_to_rbc[tid] = True
        link.ensure_connected()
        send_p0_now(tid)
        return "OK\n"
 
    def handle_disconnect(_args):
        for site in SITES:
            link.disconnect(site)
        return "OK\n"
 
    def handle_disconnect_with_online(_args):
        site = online_site["value"]
        if site is None:
            return "ERR no site currently believed online\n"
        link.disconnect(site)
        return "OK\n"
 
    def handle_disconnect_with_standby(_args):
        site = online_site["value"]
        if site is None:
            return "ERR no site currently believed online\n"
        standby = "EAST" if site == "WEST" else "WEST"
        link.disconnect(standby)
        return "OK\n"
 
    def handle_send_message(request):
        nid_engine = request["nidEngine"]
        message = request["message"]
        fields = request.get("fields", {})
        link.send_all(_pad(CATALOG.build_envelope(message, nid_engine, fields)))
        fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
        log.info("[IO] [Train %d] [RBC West, RBC East] [%s] [ERTMS message sent] [train_id=%d %s]",
                 nid_engine, message, nid_engine, fields_str)
        # Real mission-phase tracking (see this file's own ertms_post_eom
        # doc): M150 (End of Mission) or M156 (Terminate Session) flip this
        # train to the post-EoM 6s Packet-1 heartbeat cadence; M155
        # (Initiate Session) - a fresh mission starting - flips it back.
        if message in ("M150", "M156"):
            ertms_post_eom[nid_engine] = True
            next_heartbeat[nid_engine] = time.monotonic() + HEARTBEAT_PERIOD_S
        elif message == "M155":
            ertms_post_eom[nid_engine] = False
        return {"status": "OK"}
 
    def handle_get_message(request):
        nid_engine = request["nidEngine"]
        message = str(request["message"])
        packets = last_received.get(nid_engine, {}).get(message)
        # "24"/"15"/"M15" are legacy bridges to the flat scheme's own
        # combined M24_M15 answer (kind=RBC_MSG_M24_M15) - none of them
        # is a real message_catalog.json name. "M24" deliberately does
        # NOT get this fallback any more: it is now a real, distinct
        # ertms wireFormat message name (real Subset-026 Msg 24) that can
        # genuinely arrive later than this call - falling back to the
        # flat scheme's own unrelated answer here would make a caller
        # waiting for the REAL Msg 24 (Wait For Train Message's own
        # retry loop) see an immediate, wrong "success" instead of
        # actually waiting for it.
        if packets is None and message in ("24", "15", "M15"):
            packets = last_received.get(nid_engine, {}).get("PositionReportAck")
        if packets is None:
            return {"status": "ERR", "reason": f"no {message} received yet for nidEngine {nid_engine}"}
        result_packets = {message: packets}
        try:
            for pkt_name in CATALOG.packets_for(message):
                result_packets[pkt_name] = packets
        except Exception:
            pass
        if "msg_type" in packets:
            result_packets["PositionReportAck"] = packets
            result_packets["M24_M15"] = packets
            result_packets["24"] = packets
            result_packets[str(packets["msg_type"])] = packets
            result_packets[f"M{packets['msg_type']}"] = packets
        return {"status": "OK", "message": message, "packets": result_packets}
 
    def handle_connect_train(request):
        nid_engine = request["nidEngine"]
        # MAX_TRAINS is a capacity guard on DISTINCT trains ever named,
        # not on currently-connected count - re-connecting an
        # already-known nid_engine never counts against it.
        if nid_engine not in state and len(state) >= MAX_TRAINS:
            return {"status": "ERR",
                    "reason": f"this sim already hosts {MAX_TRAINS} distinct trains (MAX_TRAINS)"}
        _reset_train(nid_engine)
        # Optional starting position override, on direct request - a test
        # that needs a train to start somewhere other than
        # _default_lrbg()'s own west/east heuristic (e.g. right at a
        # specific route's own start balise) can now say so explicitly,
        # instead of connecting then separately calling Set Train
        # Variable/sendTrainMessage before the first real M136. Both
        # optional and independent - either, neither, or both. Sets
        # state[nid_engine]["position_m"] too (not just train_vars),
        # since that field - not train_vars["d_lrbg"] - is what the
        # cyclic position-report loop below actually advances from; see
        # send_position_report()'s own doc for why both need to agree.
        if "nidLrbg" in request and request["nidLrbg"] is not None:
            train_vars[nid_engine]["nid_lrbg"] = int(request["nidLrbg"])
        if "dLrbg" in request and request["dLrbg"] is not None:
            d_lrbg = int(request["dLrbg"])
            train_vars[nid_engine]["d_lrbg"] = d_lrbg
            state[nid_engine]["position_m"] = d_lrbg
        connected_to_rbc[nid_engine] = True
        link.ensure_connected()
        send_p0_now(nid_engine)
        # A fresh connect always starts a train back in active-mission
        # phase (not post-EoM) - matches session->in_use being reset fresh
        # RBC-side on a genuine (re)connect.
        ertms_post_eom[nid_engine] = False
        log.info("[GENERAL] [Train %d] [internal] [CONNECT] [Train connected to RBC] [train_id=%d]",
                 nid_engine, nid_engine)
        return {"status": "OK"}
 
    def handle_disconnect_train(request):
        nid_engine = request["nidEngine"]
        for site in SITES:
            s = link.sockets[site]
            if s is not None:
                try:
                    s.sendall(_pad(CATALOG.build_envelope("TRAIN_DISCONNECT", nid_engine, {})))
                except OSError:
                    pass
        connected_to_rbc[nid_engine] = False
        ertms_post_eom[nid_engine] = False
        _reset_train(nid_engine)
        log.info("[IO] [Train %d] [RBC West, RBC East] [TRAIN_DISCONNECT] [ERTMS Train disconnected from RBC] [train_id=%d]",
                 nid_engine, nid_engine)
        return {"status": "OK"}
 
    def handle_set_train_variable(request):
        nid_engine = request["nidEngine"]
        variable = request["variable"]
        value = request["value"]
        train_vars.setdefault(nid_engine, {})[variable] = value
        # d_lrbg is genuinely two places at once: the variable store here
        # (what an explicit Send Train Message pulls fields from) AND
        # state[nid_engine]["position_m"] (what the autonomous cyclic M136
        # reporter below self-increments from every `period` seconds,
        # entirely independently of this store). A real bug, found live:
        # without this, Set Train Variable ... d_lrbg <N> had no visible
        # effect on the next periodic report at all - the cyclic reporter
        # just kept incrementing from its own prior position_m, and only
        # an immediately-following explicit Send Train Message call
        # (handle_send_train_message()'s own identical position_m write)
        # ever bridged the two, racing against the next autonomous tick.
        # Keeping both in sync here means Set Train Variable alone is
        # enough, same as every other variable.
        if variable == "d_lrbg":
            state.setdefault(nid_engine, {"position_m": 0, "cycle": 0})["position_m"] = value
        log.info("[GENERAL] [Train %d] [internal] [VAR_STORE] [Train variable set] [train_id=%d %s=%s]",
                 nid_engine, nid_engine, variable, value)
        return {"status": "OK"}
 
    def handle_send_train_message(request):
        nonlocal next_report
        nid_engine = request["nidEngine"]
        message = request["message"]
        packet = request.get("packet", message)
        stored = train_vars.get(nid_engine, {})
        try:
            # ertms-format messages (M129/M132/...) get the WHOLE stored
            # var set, unfiltered - a real, found-live bug otherwise:
            # their own catalog "fields" list is Subset-026 documentation
            # text ("Train data (Packet type 11)"), not real snake_case
            # variable names, so filtering against it (the flat-message
            # path below) always produced an EMPTY fields dict - e.g.
            # Send Train Message M129 silently sent l_train=0/v_maxtrain=0
            # regardless of what Set Train Variable had actually stored,
            # which then failed GP's own train-data validation
            # (ab_gp_proc_train_data_validate()) silently, leaving
            # ctx->train_data.valid permanently false and the config-
            # carrying Msg 24 (Packet 3/57/58) never sent. ertms encoders
            # already do their own fields.get(key, default) extraction
            # (ertms_adapters.py), so passing the raw store is safe - it's
            # exactly what those lambdas expect.
            if CATALOG.is_ertms(message):
                fields = dict(stored)
            else:
                field_specs = CATALOG.fields_for(message, packet if packet != message else None)
                fields = {name: stored[name] for name in field_specs if name in stored}
        except Exception as e:
            return {"status": "ERR", "reason": str(e)}
        if message == "136" or message == "PositionReport" or message == "M136":
            train_st = state.setdefault(nid_engine, {"position_m": 0, "cycle": 0})
            if "cycle" not in fields:
                train_st["cycle"] += 1
                fields["cycle"] = train_st["cycle"]
            if "nid_lrbg" not in fields:
                fields["nid_lrbg"] = _default_lrbg(nid_engine)
            if "d_lrbg" in fields:
                train_st["position_m"] = fields["d_lrbg"]
            # Per direct request: the autonomous cyclic report still fires
            # every `period` (simPeriodSeconds) seconds, but a Robot-
            # triggered explicit send (this handler) sends immediately AND
            # restarts that timer from now - without this, an explicit
            # send could be followed by a redundant autonomous report just
            # a fraction of a period later, instead of a clean full period
            # after the explicit one.
            next_report = time.monotonic() + period
        if message in ("132", "M132", "146", "M146"):
            last_received.get(nid_engine, {}).pop("M24_CONFIG", None)
        if message in ("M150", "M156", "150", "156"):
            ertms_post_eom[nid_engine] = True
            next_heartbeat[nid_engine] = time.monotonic() + HEARTBEAT_PERIOD_S
        elif message in ("M155", "155"):
            ertms_post_eom[nid_engine] = False
        link.send_all(_pad(CATALOG.build_envelope(message, nid_engine, fields)))
        fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
        msg_tag = f"M{message}" if message.isdigit() else message
        log.info("[IO] [%s] [c-west,c-east] [%s] [ERTMS message sent from stored variables] [train_id=%d %s]",
                 f"train-{site_tag.lower()}", msg_tag, nid_engine, fields_str)
        return {"status": "OK"}
 
    def handle_clear_messages(_request):
        last_received.clear()
        return {"status": "OK"}
 
    def handle_cleanup(_request):
        """Comprehensive one-call reset of every train this sim has ever
        named, for every EVER-connected train (RBC_MSG_TRAIN_DISCONNECT,
        the same wire action disconnectTrain sends, so the RBC side's own
        per-train session/db state resets too - the same signal
        ab_ga_ctc.c's on-change cache now relies on, see RBC_Template/ISSUES.md)
        followed by a full local reset (_reset_train() - train_vars,
        position/cycle, message history) plus ertms_post_eom/
        next_heartbeat, which disconnectTrain's own per-call version
        already covers per train but this does in one round trip for all
        of them. Deliberately does NOT touch the underlying TCP link
        (link.sockets) - only application-level state, so a test suite's
        own [Teardown] can call this without paying a reconnect cost.
        Per direct request: sims shall forget historical data between
        tests; the connection itself can stay up."""
        for tid in list(state.keys()):
            if connected_to_rbc.get(tid):
                for site in SITES:
                    s = link.sockets[site]
                    if s is not None:
                        try:
                            s.sendall(_pad(CATALOG.build_envelope("TRAIN_DISCONNECT", tid, {})))
                        except OSError:
                            pass
            connected_to_rbc[tid] = False
            ertms_post_eom[tid] = False
            next_heartbeat.pop(tid, None)
            _reset_train(tid)
        last_received.clear()
        log.info("[GENERAL] [%s] [internal] [CLEANUP] [All train state reset, link kept up] []",
                 f"train-{site_tag.lower()}")
        return {"status": "OK"}
 
    control = ControlServer(
        log,
        control_port,
        {
            "REPORT": handle_report,
            "PING": handle_ping,
            "CONNECT": handle_connect,
            "DISCONNECT": handle_disconnect,
            "DISCONNECTWITHONLINE": handle_disconnect_with_online,
            "DISCONNECTWITHSTANDBY": handle_disconnect_with_standby,
        },
        json_handlers={
            "sendMessage": handle_send_message,
            "getMessage": handle_get_message,
            "connectTrain": handle_connect_train,
            "disconnectTrain": handle_disconnect_train,
            "setTrainVariable": handle_set_train_variable,
            "sendTrainMessage": handle_send_train_message,
            "clearMessages": handle_clear_messages,
            "cleanup": handle_cleanup,
        },
        bind_host=control_host,
    )
 
    running = True
 
    def _stop(signum, _frame):
        nonlocal running
        log.info("[GENERAL] [%s] [internal] [SIGNAL] [Train sim shutting down] [signal=%d]",
                 f"train-{site_tag.lower()}", signum)
        running = False
 
    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)
 
    log.info("[GENERAL] [%s] [internal] [INIT] [Train sim starting] "
             "[no trains connected yet - up to %d supported, see connectTrain period=%.1fs]",
             f"train-{site_tag.lower()}", MAX_TRAINS, period)
    next_report = 0.0
 
    # ADR-038: own-perspective status - startup + every 5s, to both sites.
    status = StatusReporter(log, link, f"SIM-TRAIN/{site_tag}", f"train-{site_tag.lower()}",
                            frame_size=TRAIN_WIRE_SIZE)
    status.emit({"trains": sum(1 for v in connected_to_rbc.values() if v)})
 
    while running:
        link.ensure_connected()
        for site in SITES:
            now_connected = link.sockets[site] is not None
            # Re-sends P0 on a RECONNECT after a transient socket drop
            # (was_connected[site] false -> true) once this train has
            # already been explicitly connected - the initial connect
            # itself is handled synchronously by handle_connect/
            # handle_connect_train's own send_p0_now() call, not here.
            if now_connected and not was_connected[site]:
                s = link.sockets[site]
                for tid in list(connected_to_rbc.keys()):
                    if not connected_to_rbc.get(tid):
                        continue
                    try:
                        s.sendall(_pad(CATALOG.build_envelope("P0", tid, {})))
                        log.info("[IO] [Train %d] [RBC %s] [P0] [ERTMS Train connection handshake sent] [train_id=%d]",
                                 tid, site.title(), tid)
                    except OSError:
                        pass
            was_connected[site] = now_connected
 
        control.poll(timeout=0.0)
        status.tick({"trains": sum(1 for v in connected_to_rbc.values() if v)})
 
        now = time.monotonic()
        if now >= next_report:
            for tid in list(connected_to_rbc.keys()):
                if not connected_to_rbc.get(tid):
                    continue
                # Post-EoM trains stop the flat/legacy simPeriodSeconds
                # cadence entirely - they get their own HEARTBEAT_PERIOD_S
                # real-Subset-026 Packet-1 heartbeat below instead, not
                # both at once.
                if ertms_post_eom.get(tid):
                    continue
                cur_vars = train_vars.get(tid, {})
                cur_speed = cur_vars.get("v_train", 0)
                if cur_speed > 0:
                    delta_m = int(round((cur_speed * 1000.0 / 3600.0) * period))
                    new_position = state[tid]["position_m"] + max(1, delta_m)
                else:
                    new_position = state[tid]["position_m"]
                payload = send_position_report(tid, new_position)
                fields_str = " ".join(f"{k}={v}" for k, v in sorted(payload.items()))
                bname = get_balise_name(payload.get("nid_lrbg"))
                balise_tag = f" balise={bname}" if bname else ""
                # Msg 136 carries Packet 0 OR Packet 1, never both (Subset-026
                # 8.6.4 - see ab_gp_ertms_train_messages.h's own rbc_msg_136_
                # position_report_s doc) - this cyclic report is the Packet 0
                # (plain position report) variant, tagged [P0] so the message
                # tag [M136] stays one consistent name across both variants
                # instead of a second ad-hoc tag (was [M136_P1] below for the
                # other variant, now [M136] ... [P1] to match).
                log.info("[IO] [Train %d] [RBC West, RBC East] [M136] [ERTMS Train periodic position report] "
                         "[train_id=%d] [P0] [%s%s]",
                         tid, tid, fields_str, balise_tag)
            next_report = now + period
 
        # Post-EoM idle heartbeat (real Subset-026 Msg 136 / Packet 1,
        # 6s cadence) - see this file's own ertms_post_eom doc. Sent
        # directly through rbc_ertms_wire.py + MessageCatalog's new
        # build_ertms_envelope_raw(), NOT via CATALOG.build_envelope(),
        # since "M136_P1" is deliberately not a message_catalog.json
        # entry (that file must stay unmodified) - this is this sim's own
        # internally-driven cadence, not something a Robot test commands
        # by name.
        for tid in list(connected_to_rbc.keys()):
            if connected_to_rbc.get(tid) and ertms_post_eom.get(tid) and now >= next_heartbeat.get(tid, 0.0):
                cur_vars = train_vars.get(tid, {})
                cur_lrbg_c = cur_vars.get("nid_c", 0)
                cur_lrbg_bg = cur_vars.get("nid_lrbg", _default_lrbg(tid))
                packet1_report = {
                    "nid_c": cur_lrbg_c, "nid_bg": cur_lrbg_bg,
                    "prv_nid_c": cur_lrbg_c, "prv_nid_bg": cur_lrbg_bg,
                    "d_lrbg": state[tid]["position_m"],
                    "v_train": cur_vars.get("v_train", 0),
                    "m_mode": cur_vars.get("m_mode", 0),
                    "m_level": cur_vars.get("m_level", 2),
                }
                codec_bytes = ertms_codec.encode_msg_136_packet1(tid, packet1_report)
                link.send_all(_pad(CATALOG.build_ertms_envelope_raw(codec_bytes, tid)))
                bname = get_balise_name(cur_lrbg_bg)
                balise_tag = f" balise={bname}" if bname else ""
                # Same [M136] message tag as the Packet 0 cyclic report
                # above, now tagged [P1] instead of a separate [M136_P1]
                # message name - Msg 136 is one message with two mutually
                # exclusive packet variants, not two different messages.
                log.info("[IO] [Train %d] [RBC West, RBC East] [M136] [ERTMS post-EoM idle heartbeat sent] "
                         "[train_id=%d] [P1] [d_lrbg=%d nid_lrbg=%d%s]",
                         tid, tid, state[tid]["position_m"], cur_lrbg_bg, balise_tag)
                next_heartbeat[tid] = now + HEARTBEAT_PERIOD_S
 
        for site, data in link.poll_recv(timeout=0.2):
            decoded = CATALOG.decode_to_message(data)
            if decoded is None:
                log.warning("[Train 1] [internal] [MALFORMED] [Malformed envelope dropped] [site=%s]", site)
                continue
            nid_engine, message, fields = decoded
            if message == "M24":
                if fields.get("packetsPresent"):
                    last_received.setdefault(nid_engine, {})["M24_CONFIG"] = fields
                    last_received.setdefault(nid_engine, {})["M24"] = fields
                elif "M24_CONFIG" not in last_received.get(nid_engine, {}):
                    last_received.setdefault(nid_engine, {})["M24"] = fields
            else:
                last_received.setdefault(nid_engine, {})[message] = fields
            online_site["value"] = site
            fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
            target_train_label = f"Train {nid_engine}"
            rbc_label = f"RBC {site.title()}"
            if message == "PositionReportAck":
                msg_type_name = f"M{fields.get('msg_type', '24')}"
                log.info("[IO] [%s] [%s] [%s] [ERTMS Position report ack received] [nid_engine=%d %s]",
                         rbc_label, target_train_label, msg_type_name, nid_engine, fields_str)
            elif message in ("3", "M3"):
                # Per direct request: log each Subset-026 packet on its own
                # line with its own real fields/values, not one flattened
                # summary line - rbc_ertms_wire.py's own decode_msg_3()
                # doc: Packet 15 (Movement Authority) is always present;
                # Packets 21 (Gradient Profile), 27 (Static Speed Profile),
                # 5 (Linking Information) are optional, only logged when
                # this particular grant actually carried them.
                packets_present = fields.get("packetsPresent") or ["P15"]
                log.info("[IO] [%s] [%s] [M3] [ERTMS Movement Authority received] [nid_engine=%d packets=%s]",
                         rbc_label, target_train_label, nid_engine, ",".join(packets_present))
                log.info("[IO] [%s] [%s] [P15] [Movement Authority packet] "
                         "[nid_engine=%d ma_seq=%s ma_length=%s v_loa=%s t_loa_s=%s]",
                         rbc_label, target_train_label, nid_engine,
                         fields.get("ma_seq"), fields.get("ma_length"), fields.get("v_loa"), fields.get("t_loa_s"))
                if "P21" in packets_present:
                    log.info("[IO] [%s] [%s] [P21] [Gradient Profile packet] "
                             "[nid_engine=%d segment_count=%s segments=%s]",
                             rbc_label, target_train_label, nid_engine,
                             fields.get("gradient_segment_count"), fields.get("gradient_segments"))
                if "P27" in packets_present:
                    log.info("[IO] [%s] [%s] [P27] [Static Speed Profile packet] "
                             "[nid_engine=%d segment_count=%s segments=%s]",
                             rbc_label, target_train_label, nid_engine,
                             fields.get("ssp_segment_count"), fields.get("ssp_segments"))
                if "P5" in packets_present:
                    log.info("[IO] [%s] [%s] [P5] [Linking Information packet] "
                             "[nid_engine=%d link_count=%s links=%s]",
                             rbc_label, target_train_label, nid_engine,
                             fields.get("linking_link_count"), fields.get("linking_links"))
                # Ack as the train the MA was actually addressed to.
                link.send_all(_pad(CATALOG.build_envelope("146", nid_engine, {"ma_seq": fields["ma_seq"]})))
                log.info("[IO] [%s] [%s] [M146] [ERTMS Movement Authority Ack sent] [train_id=%d ma_seq=%d]",
                         target_train_label, rbc_label, nid_engine, fields["ma_seq"])
            else:
                msg_tag = message if message.startswith("M") else (f"M{message}" if message.isdigit() else message)
                log.info("[IO] [%s] [%s] [%s] [ERTMS message received from RBC] [nid_engine=%d %s]",
                         rbc_label, target_train_label, msg_tag, nid_engine, fields_str)
 
    log.info("stopped after %d cycle(s)", sum(st["cycle"] for st in state.values()))
 
 
if __name__ == "__main__":
    main()