ctc_sim.py

Go to the documentation of this file.

#!/usr/bin/env python3
"""CTC (Centralized Traffic Control) sim (ADR-029) - a real, dual-homed
TCP client. A single instance oversees both sites (unlike train/il, which
run one pair per train slot) - dual-homed to both sites' CTC port the
same way train_sim.py/il_sim.py are (see dual_link.py's own header),
receive-only: this scenario never has CTC send anything back to the RBC,
it only observes the one-way status pushes (train connected/MA granted/
MA extended) and logs them, tagged by train_id.
 
Configuration: one JSON file (common/sim_config.py, SIM_CONFIG_PATH env
var, default /app/config.json) - rbcWest/rbcEast and controlServer only
(no trainIndex - CTC is not per-train). See config/ctc.json, right next
to this script. Message vocabulary comes from message_catalog.json, also
right next to this script.
 
Commands (controlServer.port, config file - see common/control_server.py):
  PING   Replies "PONG\\n" - liveness check.
 
Structured JSON commands (tests/robot/design/DESIGN.md,
SIM_INTERFACES.md - same port, see common/control_server.py's own doc).
No sendMessage - CTC never transmits in this scenario:
  getMessage   {"cmd":"getMessage","nidEngine":N,"message":"CTC_MA_GRANTED"}
               Returns the most recently received indication of that
               kind concerning train nidEngine, or an ERR if none has
               arrived yet.
"""
 
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
 
CATALOG = MessageCatalog(os.path.join(os.path.dirname(os.path.abspath(__file__)), "message_catalog.json"))
 
 
_RELAY_KEEPALIVE_KIND = 9
 
# CTC_ALARM's own msg_type field (message_catalog.json) carries
# rbc_alarm_severity_t (rbc_wire_types.h) as a raw uint32 - this is the
# Python-side mirror of that same small, closed vocabulary (0=INFO/
# 1=WARNING/2=ERROR), used only to pick this sim's own log level below.
_ALARM_SEVERITY_LOG = {0: logging.INFO, 1: logging.WARNING, 2: logging.ERROR}
_ALARM_SEVERITY_NAME = {0: "INFO", 1: "WARNING", 2: "ERROR"}
 
 
def _is_keepalive(data):
    return len(data) >= 1 and data[0] == _RELAY_KEEPALIVE_KIND
 
 
def _decode_ma_offset(raw_offset):
    """ma_length rides the wire as a SIGNED value's raw two's-complement uint32 bit pattern
    (CTC_TRAIN_INFO's own message_catalog.json comment) - converts it back to a real signed
    Python int. Pure extraction (2026-09-24, for unit-testability - this used to be inline in
    main()'s own receive loop, matching the "pure extraction, no logic changed" precedent
    RBC_GP's own C code already uses for the same reason, e.g. rte_voter.c's
    voter_group_matching_responses())."""
    return raw_offset - 0x100000000 if raw_offset >= 0x80000000 else raw_offset
 
 
def main():
    config = sim_config.load()
    control_host = sim_config.require(config, "controlServer", "bindHost")
    control_port = sim_config.require(config, "controlServer", "port")
 
    # "[%(levelname)s] " prefix - see RBC_Test_Sim_Train/src/train/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("ctc")
 
    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)
 
    # Most-recently-received indication per (nid_engine, message name) -
    # tests/robot/design/DESIGN.md section 4.2. Indexed by nid_engine
    # (the TRAIN the indication is about, not a CTC identity - there is
    # only one CTC instance - SIM_INTERFACES.md's own note).
    last_received = {}
    site_health = {"WEST": {"last_seen": None, "msg_count": 0}, "EAST": {"last_seen": None, "msg_count": 0}}
    train_status = {}
    alarms = []
    start_time = time.monotonic()
 
    def handle_ping(_args):
        return "PONG\n"
 
    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_get_status(_request):
        connected_sites = link.connected_sites()
        sites_res = {}
        for s in ("WEST", "EAST"):
            tgt = targets.get(s, ("127.0.0.1", 0))
            sites_res[s] = {
                "connected": s in connected_sites,
                "host": tgt[0],
                "port": tgt[1],
                "last_seen": site_health.get(s, {}).get("last_seen"),
                "msg_count": site_health.get(s, {}).get("msg_count", 0),
            }
        return {
            "status": "OK",
            "sites": sites_res,
            "trains": train_status,
            "alarms": alarms[-20:],
            "stats": {
                "indications_seen": seen,
                "connected_sites": connected_sites,
                "uptime_s": round(time.monotonic() - start_time, 1),
            },
        }
 
    def handle_clear_messages(_request):
        last_received.clear()
        train_status.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). Same as clearMessages plus alarms -
        clearMessages deliberately left alarms alone (a running audit
        trail useful mid-test), but a between-tests Cleanup should not
        leave a prior test's alarms visible to the next one."""
        last_received.clear()
        train_status.clear()
        alarms.clear()
        return {"status": "OK"}
 
    control = ControlServer(
        log,
        control_port,
        {"PING": handle_ping},
        json_handlers={
            "getMessage": handle_get_message,
            "getStatus": handle_get_status,
            "getRbcStatus": handle_get_status,
            "clearMessages": handle_clear_messages,
            "cleanup": handle_cleanup,
        },
        bind_host=control_host,
    )
 
    running = True
 
    def _stop(signum, _frame):
        nonlocal running
        log.info("[GENERAL] [CTC] [internal] [SIGNAL] [CTC sim shutting down] [signal=%d]", signum)
        running = False
 
    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)
 
    log.info("[GENERAL] [CTC] [internal] [INIT] [CTC sim starting - overseeing WEST, EAST] []")
    seen = 0
 
    # ADR-038: own-perspective status - startup + every 5s, to both sites.
    status = StatusReporter(log, link, "SIM-CTC", "ctc")
    status.emit()
 
    while running:
        link.ensure_connected()
        control.poll(timeout=0.0)
        status.tick()
        for site, data in link.poll_recv(timeout=0.2):
            if _is_keepalive(data):
                continue
            decoded = CATALOG.decode_to_message(data)
            if decoded is None:
                log.warning("[CTC] [internal] [MALFORMED] [Malformed indication dropped] [site=%s]", site)
                continue
            nid_engine, message, fields = decoded
            last_received.setdefault(nid_engine, {})[message] = fields
            seen += 1
            now_ts = time.time()
            site_health.setdefault(site, {})["last_seen"] = now_ts
            site_health[site]["msg_count"] = site_health[site].get("msg_count", 0) + 1
 
            fields_str = " ".join(f"{k}={v}" for k, v in sorted(fields.items()))
            rbc_label = f"RBC {site.title()}"
            if message == "CTC_CONNECTED":
                train_status[nid_engine] = {
                    "connected": True,
                    "d_lrbg": fields.get("d_lrbg", 0),
                    "site": site,
                    "last_seen": now_ts,
                }
                log.info("[IO] [%s] [CTC] [CTC_CONNECTED] [CTC Connected indication received] [nid_engine=%d %s]",
                         rbc_label, nid_engine, fields_str)
            elif message == "CTC_MA_GRANTED":
                t_entry = train_status.setdefault(nid_engine, {"connected": True})
                t_entry["ma_length"] = fields.get("ma_length")
                t_entry["ma_seq"] = fields.get("ma_seq")
                t_entry["site"] = site
                t_entry["last_seen"] = now_ts
                log.info("[IO] [%s] [CTC] [CTC_MA_GRANTED] [CTC Movement Authority Granted received] [nid_engine=%d %s]",
                         rbc_label, nid_engine, fields_str)
            elif message == "CTC_MA_EXTENDED":
                t_entry = train_status.setdefault(nid_engine, {"connected": True})
                t_entry["ma_length"] = fields.get("ma_length")
                t_entry["ma_seq"] = fields.get("ma_seq")
                t_entry["site"] = site
                t_entry["last_seen"] = now_ts
                log.info("[IO] [%s] [CTC] [CTC_MA_EXTENDED] [CTC Movement Authority Extended received] [nid_engine=%d %s]",
                         rbc_label, nid_engine, fields_str)
            elif message == "CTC_TRAIN_INFO":
                t_entry = train_status.setdefault(nid_engine, {"connected": True})
                t_entry["d_lrbg"] = fields.get("d_lrbg")
                t_entry["ma_status"] = "GRANTED" if fields.get("msg_type") == 1 else "NONE"
                t_entry["ma_target_signal"] = fields.get("route_len")
                t_entry["front_signal"] = fields.get("t_train")
                t_entry["dist_to_front_signal"] = fields.get("cycle")
                t_entry["ma_seq"] = fields.get("ma_seq")
                ma_offset = _decode_ma_offset(fields.get("ma_length", 0))
                t_entry["ma_offset"] = ma_offset
                t_entry["ma_length_from_balise"] = -ma_offset
                t_entry["nid_lrbg"] = fields.get("nid_lrbg")
                t_entry["v_train"] = fields.get("v_train")
                t_entry["m_mode"] = fields.get("m_mode")
                t_entry["site"] = site
                t_entry["last_seen"] = now_ts
                log.info("[IO] [%s] [CTC] [CTC_TRAIN_INFO] [CTC Train Info received] [nid_engine=%d %s]",
                         rbc_label, nid_engine, fields_str)
            elif message == "CTC_ALARM":
                severity = fields.get("msg_type")
                level = _ALARM_SEVERITY_LOG.get(severity, logging.WARNING)
                severity_name = _ALARM_SEVERITY_NAME.get(severity, f"UNKNOWN({severity})")
                alarms.append({
                    "site": site,
                    "severity": severity_name,
                    "severity_code": severity,
                    "ma_seq": fields.get("ma_seq"),
                    "cycle": fields.get("cycle"),
                    "t_train": fields.get("t_train"),
                    "timestamp": now_ts,
                })
                # The severity-differentiated log level here (not always
                # log.info like every other kind above) is the actual
                # point of this message kind - see ctc_sim.py's own
                # module doc: a dispatcher watching this sim's own log
                # output needs to see an ERROR-severity alarm stand out
                # from routine indications, not just another INFO line.
                log.log(level, "[IO] [%s] [CTC] [CTC_ALARM] [%s alarm from RBC] [%s]",
                         rbc_label, severity_name, fields_str)
            else:
                log.info("[IO] [%s] [CTC] [%s] [CTC indication received] [nid_engine=%d %s]",
                         rbc_label, message, nid_engine, fields_str)
 
    log.info("[GENERAL] [CTC] [internal] [INIT] [CTC sim stopped] [indications_seen=%d]", seen)
 
 
if __name__ == "__main__":
    main()