il_sim.py

Go to the documentation of this file.

#!/usr/bin/env python3
"""Interlocking (IL) sim (ADR-029) - a real, dual-homed TCP client that
sets/releases routes for one train's session, ON COMMAND from a test
harness (tests/robot/) rather than a fixed wall-clock schedule - see
control_server.py's own header for the command channel design. Dual-homed
the same way train_sim.py is (see dual_link.py's own header) - sends
ROUTE_ADD/ROUTE_RELEASE to both sites unconditionally, since only
whichever site is currently ONLINE will actually act on it.
 
IL<->RBC route-identity pass: routes are now identified on the wire by
their own (start_signal, end_signal) pair (the real signal pair a route
runs between - see rbc_wire_types.h's own doc), not a synthetic id, and
IL now genuinely RECEIVES something meaningful too -
TrainPositionInRoute, RBC's own report of whether a route currently has
a train authorized through it. Before this pass IL was send-only in
practice (see this file's own git history) - the main loop's own
poll_recv() drained and ignored everything; it now decodes real inbound
messages the same way train_sim.py/ctc_sim.py already do.
 
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 IL instance sets routes for -
train_id is trainIndex + 1; il-west sets routes for train-west's session,
il-east for train-east's, by this shared convention - ADR-029),
siteTag, rbcWest/rbcEast, controlServer, and routeLengthM (the default
ADD_ROUTE grant, replaces the old RBC_ROUTE_LENGTH_M constant). See
sims/il/config/*.json for the actual files. Message vocabulary comes
from message_catalog.json, right next to this script.
 
Commands (controlServer.port, config file - see common/control_server.py):
  ADD_ROUTE [length_m]   Sends one ROUTE_ADD (default length
                         config's own routeLengthM if omitted) - no route
                         identity, see this handler's own doc for why
                         that's a real limitation now. Replies "OK\\n".
  PING                   Replies "PONG\\n" - liveness check for a test
                         harness before it starts asserting anything.
  SET_IL_STATUS <s>      Sets + sends this IL's own RBC_MSG_IL_STATUS
                         (s = down|restarting|up). IL announces "up" at
                         startup and re-announces every
                         ROUTE_STATUS_PERIOD_S; this overrides it for
                         tests of the RBC's DOWN/RESTARTING handling.
                         Replies "OK\\n".
 
Structured JSON commands (tests/robot/design/DESIGN.md,
SIM_INTERFACES.md - same port, see common/control_server.py's own doc):
  sendMessage   {"cmd":"sendMessage","nidEngine":N,"message":"ROUTE_ADD",
                 "fields":{"route_len":500,"start_signal":100,
                           "end_signal":101,"route_type":0}}
                Sends one ROUTE_ADD/ROUTE_RELEASE/etc NOW with the given
                fields - the structured equivalent of ADD_ROUTE above
                (route_len defaults to config's own routeLengthM if the
                field is left unset for ROUTE_ADD, same as ADD_ROUTE's
                own default behavior; route identity fields have no
                default - the caller must supply them for a real grant).
                message "IL_ROUTE_CMD" / "IL_STATUS" use the dedicated
                packed IL protocol (il_wire.py, mirrors ab_ga_il_wire.h)
                instead of the flat codec: IL_ROUTE_CMD fields
                route_type/route_status/degraded_status/
                release_route_request/first_route_in_path (enum ints,
                first value = default) + route_id/start_signal/end_signal
                (-1 = unset); IL_STATUS field il_status (0/1/2).
  getMessage    {"cmd":"getMessage","nidEngine":N,"message":"TrainPositionInRoute"}
                Returns the most recently received packet of that kind
                for that nidEngine, or an ERR if none has arrived yet -
                same "most recent value, not a queue" contract
                train_sim.py's own getMessage already has (DESIGN.md
                section 4.2).
"""
 
import logging
import os
import signal
import sys
import time
 
from simcore import sim_config
from simcore.control_server import ControlServer
from simcore.dual_link import DualHomedLink
from simcore.rbc_messages import MessageCatalog
from simcore.rbc_wire import ENVELOPE_SIZE
from simcore.status_report import StatusReporter
 
from . import site_data
from . import il_wire
 
# Phase D: periodic all-route-status broadcast cadence (2s, per direct
# instruction - "IL sim shall send all route status each 2s").
ROUTE_STATUS_PERIOD_S = 2.0
 
# Enum-int -> name, for readable IL_ROUTE_CMD / IL_ROUTE_INDICATION log
# lines (rbc_wire_types.h's rbc_il_* enums, mirrored in il_wire.py).
_CMD_ROUTE_TYPE_NAME = {il_wire.ROUTE_TYPE_NO_ROUTE: "NoRoute", il_wire.ROUTE_TYPE_LOCKED: "Locked"}
_CMD_ROUTE_STATUS_NAME = {
    il_wire.ROUTE_STATUS_NO_STATUS: "NoStatus", il_wire.ROUTE_STATUS_FS: "FS",
    il_wire.ROUTE_STATUS_OS: "OS", il_wire.ROUTE_STATUS_SH: "SH",
    il_wire.ROUTE_STATUS_USED: "Used", il_wire.ROUTE_STATUS_DEGRADED: "Degraded",
}
 
CATALOG = MessageCatalog(os.path.join(os.path.dirname(os.path.abspath(__file__)), "message_catalog.json"))
 
# RBC_MSG_RELAY_KEEPALIVE (rbc_wire_types.h) - a pure wire-level liveness
# frame, not a real domain message, and not listed in this sim's own
# catalog (RBC_Test_Sim_Core stays kind-blind by design - this filtering is this
# sim's own application-layer job, same layer as message_catalog.json).
# Unlike train_sim.py, which rarely if ever receives one, IL's own relay
# channel gets ONE OF THESE EVERY SINGLE CYCLE regardless of whether
# real data is also sent that cycle (ab_gp_channel_send_il_keepalive_if_online(),
# ab_gp_channel.c) - without filtering it out before decode_to_message(),
# every ordinary idle cycle would log a spurious "malformed envelope"
# warning, drowning out genuinely rare real malformed frames.
_RELAY_KEEPALIVE_KIND = 9
 
 
def _is_keepalive(data):
    return len(data) >= 1 and data[0] == _RELAY_KEEPALIVE_KIND
 
 
def main():
    config = sim_config.load()
    site_tag = sim_config.require(config, "siteTag")
    # IL itself tracks no per-train state at all (unlike train_sim.py -
    # every real command, structured sendMessage/getMessage, already
    # takes an explicit nidEngine with no "is this train known" gate).
    # This is only for the legacy, non-structured ADD_ROUTE/RELEASE_ROUTE
    # commands, which predate nidEngine-explicit fields entirely and have
    # nowhere to take one from - always targets train 1 (no roster to
    # pick a "first configured train" from any more; see
    # RBC_Test_Sim_Train/src/train/train_sim.py's own "no default trains" doc
    # for the identical principle on the Train side). Use sendMessage
    # with an explicit nidEngine for any other train.
    train_id = 1
    route_length_m = int(config.get("routeLengthM", 500))
    control_host = sim_config.require(config, "controlServer", "bindHost")
    control_port = sim_config.require(config, "controlServer", "port")
 
    # "[%(levelname)s] " prefix - see train_sim.py's own identical change
    # for why (the test env's own listener should collect a real level,
    # not guess/default one).
    logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s", stream=sys.stdout)
    log = logging.getLogger("il")
 
    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, ENVELOPE_SIZE)
    routes_sent = 0
    # {nid_engine: {message_name: fields_dict}} - most-recently-received
    # value per (nid_engine, message), same "not a queue" convention
    # train_sim.py's own last_received already uses (DESIGN.md 4.2).
    last_received = {}
 
    # Phase D: real route table from the RailML file (site_data.py, built
    # in an earlier phase but unwired until now) + this IL instance's OWN
    # locally-tracked lock state per route - "free" until THIS IL issues
    # a ROUTE_ADD for it, "locked" until THIS IL releases it. Genuinely
    # local bookkeeping, not fed back from the RBC's own TrainPositionInRoute
    # replies (a different, RBC-side-authoritative signal already handled
    # separately above) - matches the plan's own "each defaulting to
    # free, updated locally whenever this same IL instance itself issues
    # a ROUTE_ADD/ROUTE_RELEASE" design.
    try:
        route_table = site_data.routes()
    except Exception as exc:
        log.warning("[GENERAL] [%s] [internal] [CONFIG] [could not load route table from RailML - "
                    "Phase D route-status broadcast disabled] [%s]", f"il-{site_tag.lower()}", exc)
        route_table = []
    route_lock_state = {(r["start_signal"], r["end_signal"]): False for r in route_table}
 
    def _mark_route(start_signal, end_signal, locked):
        key = (int(start_signal), int(end_signal))
        if key in route_lock_state:
            route_lock_state[key] = locked
 
    # RBC_MSG_IL_STATUS: this IL's own reported health. Announced UP on
    # startup and re-announced every ROUTE_STATUS_PERIOD_S (the RBC starts
    # assuming DOWN). Overridable via the Set IL Status keyword / control
    # command below, for tests that exercise the RBC's DOWN/RESTARTING
    # handling.
    il_status = {"value": il_wire.IL_STATUS_UP}
    _IL_STATUS_BY_NAME = {
        "down": il_wire.IL_STATUS_DOWN,
        "restarting": il_wire.IL_STATUS_RESTARTING,
        "up": il_wire.IL_STATUS_UP,
    }
 
    def _send_il_status():
        link.send_all(il_wire.encode_il_status(il_status["value"]))
 
    def handle_set_il_status(args):
        name = (args[0] if args else "up").strip().lower()
        if name not in _IL_STATUS_BY_NAME:
            return "ERR unknown status (down|restarting|up)\n"
        il_status["value"] = _IL_STATUS_BY_NAME[name]
        _send_il_status()
        log.info("[IO] [IL] [RBC West, RBC East] [IL_STATUS] [IL status set] [il_status=%d]",
                 il_status["value"])
        return "OK\n"
 
    def handle_add_route(args):
        nonlocal routes_sent
        length_m = int(args[0]) if args else route_length_m
        routes_sent += 1
        log.info(
            "route #%d added (+%dm, no route identity) for train_id=%d [commanded] - RBC will reject this: "
            "the plaintext ADD_ROUTE command has no route identity fields; use sendMessage/the Robot "
            "'Set IL Route' keyword for a real grant",
            routes_sent, length_m, train_id,
        )
        link.send_all(CATALOG.build_envelope("ROUTE_ADD", train_id, {"route_len": length_m}))
        return "OK\n"
 
    def handle_release_route(args):
        start_signal, end_signal = int(args[0]), int(args[1])
        log.info(
            "route release requested (start_signal=%d end_signal=%d) for train_id=%d [commanded]",
            start_signal, end_signal, train_id,
        )
        link.send_all(CATALOG.build_envelope(
            "ROUTE_RELEASE", train_id, {"start_signal": start_signal, "end_signal": end_signal}
        ))
        _mark_route(start_signal, end_signal, False)
        return "OK\n"
 
    def handle_ping(_args):
        return "PONG\n"
 
    def handle_send_message(request):
        nonlocal routes_sent
        nid_engine = request["nidEngine"]
        message = request["message"]
        fields = dict(request.get("fields", {}))
 
        # Dedicated IL messages (25/26) - their own packed layout, not
        # RBC_Test_Sim_Core's flat 96-byte codec. il_wire.py mirrors ab_ga_il_wire.h.
        if message == "IL_ROUTE_CMD":
            # Fully-resolved field set (every field, defaults applied) - so
            # the log shows exactly what went on the wire, not just what
            # the caller happened to pass.
            sent = {
                "route_type": int(fields.get("route_type", il_wire.ROUTE_TYPE_NO_ROUTE)),
                "route_status": int(fields.get("route_status", il_wire.ROUTE_STATUS_NO_STATUS)),
                "degraded_status": int(fields.get("degraded_status", il_wire.DEGRADED_STATUS_NO_STATUS)),
                "release_route_request": int(fields.get("release_route_request", 0)),
                "first_route_in_path": int(fields.get("first_route_in_path", 0)),
                "route_id": int(fields.get("route_id", -1)),
                "start_signal": int(fields.get("start_signal", -1)),
                "end_signal": int(fields.get("end_signal", -1)),
            }
            frame = il_wire.encode_route_cmd(**sent)
            link.send_all(frame)
            if sent["start_signal"] != -1 and sent["end_signal"] != -1:
                locked = (sent["route_type"] == il_wire.ROUTE_TYPE_LOCKED
                          and not sent["release_route_request"])
                _mark_route(sent["start_signal"], sent["end_signal"], locked)
            fields_str = " ".join(f"{k}={v}" for k, v in sorted(sent.items()))
            log.info("[IO] [IL] [RBC West, RBC East] [IL_ROUTE_CMD] [IL route command sent] "
                     "[%s type=%s status=%s]",
                     fields_str,
                     _CMD_ROUTE_TYPE_NAME.get(sent["route_type"], sent["route_type"]),
                     _CMD_ROUTE_STATUS_NAME.get(sent["route_status"], sent["route_status"]))
            return {"status": "OK"}
        if message == "IL_STATUS":
            il_status["value"] = int(fields.get("il_status", il_wire.IL_STATUS_UP))
            _send_il_status()
            log.info("[IO] [IL] [RBC West, RBC East] [IL_STATUS] [IL status sent] [il_status=%d]",
                     il_status["value"])
            return {"status": "OK"}
 
        if message == "ROUTE_ADD" and "route_len" not in fields:
            fields["route_len"] = route_length_m
        link.send_all(CATALOG.build_envelope(message, nid_engine, fields))
        if message == "ROUTE_ADD":
            routes_sent += 1
            if "start_signal" in fields and "end_signal" in fields:
                _mark_route(fields["start_signal"], fields["end_signal"], True)
        elif message == "ROUTE_RELEASE" and "start_signal" in fields and "end_signal" in fields:
            _mark_route(fields["start_signal"], fields["end_signal"], False)
        fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
        log.info("[IO] [IL] [RBC West, RBC East] [%s] [IL message sent] [nid_engine=%d %s]",
                 message, nid_engine, fields_str)
        return {"status": "OK"}
 
    def handle_get_message(request):
        nid_engine = request["nidEngine"]
        message = request["message"]
        packets = last_received.get(nid_engine, {}).get(message)
        if packets is None:
            return {"status": "ERR", "reason": f"no {message} received yet for nidEngine {nid_engine}"}
        return {"status": "OK", "message": message, "packets": {message: packets}}
 
    def handle_clear_messages(_request):
        last_received.clear()
        return {"status": "OK"}
 
    def handle_cleanup(_request):
        """Comprehensive reset of this sim's own local state (per direct
        request: sims shall forget historical data between tests, the
        connection can stay up). IL owns no per-route memory of its own
        beyond the received-message log below - which routes currently
        exist is GP's db, not this sim's - so this is currently
        equivalent to clearMessages; kept as its own named command for
        API symmetry with train_sim.py's/ctc_sim.py's own cleanup (which
        do have more to reset) and so a Robot-side "Cleanup All Sims"
        keyword can call the same verb name on every sim uniformly."""
        last_received.clear()
        return {"status": "OK"}
 
    control = ControlServer(
        log,
        control_port,
        {"ADD_ROUTE": handle_add_route, "RELEASE_ROUTE": handle_release_route, "PING": handle_ping,
         "SET_IL_STATUS": handle_set_il_status},
        json_handlers={
            "sendMessage": handle_send_message,
            "getMessage": handle_get_message,
            "cleanup": handle_cleanup,
            "clearMessages": handle_clear_messages,
        },
        bind_host=control_host,
    )
 
    running = True
 
    def _stop(signum, _frame):
        nonlocal running
        log.info("[GENERAL] [IL] [internal] [SIGNAL] [IL sim shutting down] [signal=%d]",
                 signum)
        running = False
 
    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)
 
    log.info("[GENERAL] [IL] [internal] [INIT] [IL sim starting] [port=%d]",
             control_port)
 
    # ADR-038: own-perspective status - startup + every 5s, to both sites.
    # IL tracks no per-train roster (see main()'s own doc) - this "trains"
    # field is a fixed 1 (legacy compatibility with the status frame
    # format).
    status = StatusReporter(log, link, "SIM-IL", "il")
    status.emit({"trains": 1})
    _send_il_status()  # announce health once at startup (RBC assumes DOWN)
 
    next_route_status_report = time.monotonic()
    next_il_status_report = time.monotonic()
 
    while running:
        link.ensure_connected()  # dial (or re-dial) each RBC site; bounded per call, retried forever
        control.poll(timeout=0.2)
        status.tick({"trains": 1})
 
        # Phase D: broadcast this IL's own locked/free state for EVERY
        # route it knows about, every 2s, unconditionally - not gated on
        # any train being connected (this is IL's own liveness/state
        # feed, independent of Train's cadence). One RBC_MSG_ROUTE_STATUS_REPORT
        # frame per route (same "one small frame per item" convention
        # ROUTE_ADD/ROUTE_RELEASE already use, not a batched shape).
        now = time.monotonic()
        if now >= next_il_status_report:
            _send_il_status()
            next_il_status_report = now + ROUTE_STATUS_PERIOD_S
        if route_table and now >= next_route_status_report:
            for route in route_table:
                key = (route["start_signal"], route["end_signal"])
                locked = route_lock_state.get(key, False)
                link.send_all(CATALOG.build_envelope("ROUTE_STATUS_REPORT", train_id, {
                    "start_signal": route["start_signal"],
                    "end_signal": route["end_signal"],
                    "route_status": 1 if locked else 0,
                }))
            log.info("[IO] [IL] [RBC West, RBC East] [ROUTE_STATUS_REPORT] [IL periodic route-status broadcast] "
                     "[route_count=%d locked_count=%d]",
                     len(route_table), sum(1 for v in route_lock_state.values() if v))
            next_route_status_report = now + ROUTE_STATUS_PERIOD_S
 
        for site, data in link.poll_recv(timeout=0.0):
            if _is_keepalive(data):
                continue
 
            # Dedicated RBC -> IL indication (kind 27) - own packed layout,
            # decoded before RBC_Test_Sim_Core's flat codec (il_wire.py mirrors
            # ab_ga_il_wire.h). Stored under "IL_ROUTE_INDICATION" keyed by
            # allocated_train_nid_engine (-1 when no train) so getMessage /
            # the Robot verify keyword can read it back.
            if data and data[0] == il_wire.KIND_IL_ROUTE_INDICATION:
                ind = il_wire.decode_route_indication(data)
                if ind is not None:
                    # Two lookup keys, both live: by allocated train
                    # (allocated_train_nid_engine, -1 = no train) AND by
                    # route (-(1000 + route_id), well out of any real
                    # nid_engine range) so a test can query ONE specific
                    # route's latest indication without the "-1 shared
                    # key, most-recent-wins" ambiguity when several
                    # trainless routes are toggled.
                    last_received.setdefault(ind["allocated_train_nid_engine"], {})["IL_ROUTE_INDICATION"] = ind
                    if ind["route_id"] is not None and ind["route_id"] >= 0:
                        last_received.setdefault(-(1000 + ind["route_id"]), {})["IL_ROUTE_INDICATION"] = ind
                    # Every field, plus a readable name for the FSM state.
                    fields_str = " ".join(f"{k}={v}" for k, v in sorted(ind.items()))
                    log.info("[IO] [RBC %s] [IL] [IL_ROUTE_INDICATION] [RBC route indication received] "
                             "[%s fsm_state_name=%s]",
                             site.title(), fields_str,
                             il_wire.route_fsm_state_name(ind["route_fsm_state"]))
                continue
 
            decoded = CATALOG.decode_to_message(data)
            if decoded is None:
                log.warning("[IL] [internal] [MALFORMED] [Malformed envelope dropped] [site=%s]", site)
                continue
            nid_engine, message, fields = decoded
            last_received.setdefault(nid_engine, {})[message] = fields
            fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
            if message == "TrainPositionInRoute":
                log.info("[IO] [RBC %s] [IL] [TrainPositionInRoute] [Train position in route received] [nid_engine=%d %s]",
                         site.title(), nid_engine, fields_str)
                # Phase D's own route_lock_state sync point: the RBC is the
                # real authority on whether a route is occupied, and this
                # is the ONE feedback channel IL already has for it -
                # syncing here keeps route_lock_state correct for ANY
                # reason a route freed up (an explicit ROUTE_RELEASE this
                # IL sent itself, GP's own auto-release when a train
                # disconnects, EoM session termination, etc.), not just
                # the release path _mark_route() already covers. Found
                # live: IL sim has NO disconnectTrain handler at all (only
                # train_sim.py does), so a test's own "Disconnect Train"
                # teardown was never seen by THIS process at all - without
                # this sync, route_lock_state would keep reporting a route
                # as locked after the train that held it was long gone.
                if "start_signal" in fields and "end_signal" in fields and "route_status" in fields:
                    _mark_route(fields["start_signal"], fields["end_signal"], bool(fields["route_status"]))
 
    log.info("stopped after %d route(s) added", routes_sent)
 
 
if __name__ == "__main__":
    main()