status_report.py

Go to the documentation of this file.

"""ADR-038 own-perspective status heartbeat for a sim.
 
Each sim instance reports its OWN view of the world - which RBC sites it
is currently connected to, plus a few caller-supplied extras - once at
startup and then every PERIOD_S seconds. Each report is BOTH logged
locally (so `each sim shows connection statuses from own perspective`,
ADR-038) AND pushed to both sites over the sim's existing dual-homed
link as an opaque RBC_MSG_SIM_STATUS envelope frame; the owning Role C
gateway (train/il/ctc handler) recognises that kind, does NOT forward it
to A/B, and republishes the ASCII payload to the C-side status broker.
 
Kind-blind by construction: the caller owns what the payload keys mean;
this module only handles the cadence, the log line and the send. Mirrors
rbc_wire.encode_opaque()'s own transport-vs-application split.
"""
 
import time
 
from simcore import rbc_wire
 
#: rbc_wire_types.h RBC_MSG_SIM_STATUS - kept in sync with BOTH C copies.
SIM_STATUS_KIND = 19
 
#: Startup + every PERIOD_S (ADR-038: "Startup + every 5s").
PERIOD_S = 5.0
 
 
class StatusReporter:
    """Cadence + encode + log + send for one sim's status heartbeat."""
 
    def __init__(self, log, link, unit, log_source, kind=SIM_STATUS_KIND,
                 period_s=PERIOD_S, clock=time.monotonic, frame_size=rbc_wire.ENVELOPE_SIZE):
        """@param frame_size: passed straight through to rbc_wire.encode_opaque() -
        override if @link's own channel uses a wider fixed slot than the
        default ENVELOPE_SIZE (see that function's own doc)."""
        self._log = log
        self._link = link
        self._unit = unit
        self._source = log_source
        self._kind = kind
        self._period = period_s
        self._clock = clock
        self._frame_size = frame_size
        self._next = 0.0
 
    def _payload(self, extra):
        sites = set(self._link.connected_sites())
        west_up = "WEST" in sites
        east_up = "EAST" in sites
        if west_up and east_up:
            conn_desc = "Both_Up"
        elif west_up:
            conn_desc = "West_Up"
        elif east_up:
            conn_desc = "East_Up"
        else:
            conn_desc = "Both_Down"
 
        parts = [
            f"unit={self._unit}",
            "up=1",
            f"west={1 if west_up else 0}",
            f"east={1 if east_up else 0}",
            f"conn_status={conn_desc}",
        ]
        for key, value in (extra or {}).items():
            parts.append(f"{key}={value}")
        return " ".join(parts)
 
    def emit(self, extra=None):
        """Report now, unconditionally - use once at startup."""
        payload = self._payload(extra)
        self._log.info("[GENERAL] [%s] [internal] [STATUS] [self] [%s]", self._source, payload)
        self._link.send_all(rbc_wire.encode_opaque(self._kind, payload, frame_size=self._frame_size))
        self._next = self._clock() + self._period
 
    def tick(self, extra=None):
        """Report iff PERIOD_S has elapsed since the last report."""
        if self._clock() >= self._next:
            self.emit(extra)