control_server.py

Go to the documentation of this file.

"""Non-blocking line-based TCP command server (ADR-029) - gives any sim
built on SimCore a real TCP/IP control port, with its own dedicated
command table: `ControlServer` takes @handlers/@json_handlers as
constructor arguments rather than owning any commands itself, so
Train/IL/CTC each wire up only the commands that make sense for that
sim (e.g. Train: REPORT/PING + sendMessage/getMessage; IL: ADD_ROUTE/
PING + sendMessage; CTC: PING + getMessage only - CTC is receive-only in
this scenario) - no sim sees another sim's commands, and SimCore itself
defines none. Lets Robot Framework (safeAPITestEnv/robot/) drive a
Train/IL sim's real, test-relevant actions on demand (e.g. "add a route
now", "report this position now") instead of only ever firing on a
fixed wall-clock schedule. Robot talks to this with its own built-in
Telnet library (no extra Python dependency needed) - see
safeAPITestEnv/robot/ for the keywords that use it.
 
Owns its own small selectors.DefaultSelector, polled non-blockingly
(timeout=0) once per iteration of the sim's own main loop, right
alongside that loop's own dual_link.py poll - a second cheap select()
syscall per iteration, not a separate thread. One accepted control
connection is served at a time per sim, which is all a test harness
needs.
 
Two command styles share this one port/line-framing, distinguished by
the line's own first character:
 
- **Plain-text commands** (original ADR-029 protocol): one command per
  line, ASCII, `NAME arg1 arg2\\n`. Unrecognized commands get
  "ERR unknown command\\n"; a command's own handler decides its own
  success reply (e.g. "OK\\n"). Still used by REPORT/ADD_ROUTE/PING -
  unchanged.
- **JSON commands** (safeAPITestEnv/doc/design/DESIGN.md - structured
  sendMessage/getMessage): a line starting with `{` is parsed as one
  JSON object, `{"cmd": "<name>", ...}`; the handler receives the whole
  parsed dict and returns a dict, which is sent back JSON-encoded plus
  a trailing newline. Kept as a genuinely separate dispatch path (not
  merged into the plain-text one by, say, treating `{"cmd":...}` as a
  single whitespace-free token) specifically because a JSON payload
  routinely contains internal whitespace (`{"cmd": "sendMessage", ...}`)
  that plain-text's own `line.split()` would incorrectly tokenize.
 
Neither style is part of the RBC's own rbc_envelope_t wire protocol -
this whole channel is kept deliberately separate (different port,
different framing) so a malformed/foreign line here can never be
confused for real RBC traffic.
"""
 
import json
import selectors
import socket
 
 
class _ControlClient:
    def __init__(self, conn, addr):
        self.conn = conn
        self.addr = addr
        self.buf = b""
 
 
class ControlServer:
    def __init__(self, log, port, handlers, json_handlers=None, bind_host="0.0.0.0"):
        """@param bind_host: interface to listen on - "0.0.0.0" (every
               interface, the previous hardcoded behavior) unless the
               sim's own config file (sim_config.py) says otherwise.
           @param handlers: {command_name: fn(args: list[str]) -> str}
               - fn's return value is written back verbatim (the handler
               is responsible for its own trailing newline). Command
               names are matched case-insensitively (upper-cased before
               lookup) - @handlers' own keys should be upper-case.
           @param json_handlers: {cmd_name: fn(request: dict) -> dict} -
               fn's return value is JSON-encoded and a trailing newline
               appended automatically; fn should raise (any Exception)
               rather than return an error dict itself - the caller
               turns that into {"status": "ERR", "reason": str(exc)}
               uniformly, same "a bad command must not crash the sim"
               contract as the plain-text handlers below. Command names
               (the JSON request's own "cmd" field) are matched
               case-sensitively, unlike plain-text commands - a JSON
               API's own field values are not conventionally
               case-folded the way a human-typed line's leading word is.
        """
        self.log = log
        self.sel = selectors.DefaultSelector()
        self.handlers = handlers
        self.json_handlers = json_handlers or {}
        self._listen = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._listen.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._listen.bind((bind_host, port))
        self._listen.listen(16)
        self._listen.setblocking(False)
        self.sel.register(self._listen, selectors.EVENT_READ, data=None)
        all_commands = sorted(handlers) + sorted(self.json_handlers)
        self.log.info("control server listening on :%d (commands: %s)", port, ", ".join(all_commands))
 
    def poll(self, timeout=0.0):
        """Call once per iteration of the sim's own main loop - dispatches
        any ready accept/read event (non-blocking with @timeout=0, or a
        small blocking budget if the caller wants this to double as its
        own pacing tick)."""
        for key, mask in self.sel.select(timeout=timeout):
            if key.data is None:
                self._on_accept(mask)
            else:
                self._on_data(key.data, mask)
 
    def _on_accept(self, _mask):
        try:
            conn, addr = self._listen.accept()
        except OSError:
            return
        conn.setblocking(False)
        client = _ControlClient(conn, addr)
        self.sel.register(conn, selectors.EVENT_READ, data=client)
        self.log.info("control connection from %s", addr)
 
    def _on_data(self, client, _mask):
        try:
            chunk = client.conn.recv(4096)
        except OSError:
            self._close_client(client)
            return
        if not chunk:
            self._close_client(client)
            return
        client.buf += chunk
        while b"\n" in client.buf:
            line, client.buf = client.buf.split(b"\n", 1)
            self._handle_line(client, line.decode("ascii", errors="replace").strip())
 
    def _handle_line(self, client, line):
        if not line:
            return
        if line.startswith("{"):
            reply = self._handle_json_line(line)
            encoding = "utf-8"
        else:
            parts = line.split()
            name = parts[0].upper()
            args = parts[1:]
            handler = self.handlers.get(name)
            if handler is None:
                reply = "ERR unknown command\n"
            else:
                try:
                    reply = handler(args)
                except Exception as exc:  # noqa: BLE001 - a bad command must not crash the sim
                    self.log.warning("control command %s failed: %s", name, exc)
                    reply = f"ERR {exc}\n"
            encoding = "ascii"
        if client.conn is not None:
            try:
                client.conn.sendall(reply.encode(encoding, errors="replace"))
            except OSError:
                self._close_client(client)
 
    def _handle_json_line(self, line):
        """@return the JSON-encoded (plus trailing newline) reply string
        for one JSON-command line - see this class's own doc for the
        {"cmd": ...} request/{"status": "OK"|"ERR", ...} reply shape."""
        try:
            request = json.loads(line)
        except json.JSONDecodeError as exc:
            return json.dumps({"status": "ERR", "reason": f"malformed JSON: {exc}"}) + "\n"
        if not isinstance(request, dict) or "cmd" not in request:
            return json.dumps({"status": "ERR", "reason": "request must be a JSON object with a \"cmd\" field"}) + "\n"
        name = request["cmd"]
        handler = self.json_handlers.get(name)
        if handler is None:
            return json.dumps({"status": "ERR", "reason": f"unknown cmd {name!r}"}) + "\n"
        try:
            response = handler(request)
        except Exception as exc:  # noqa: BLE001 - a bad command must not crash the sim
            self.log.warning("control command %s failed: %s", name, exc)
            return json.dumps({"status": "ERR", "reason": str(exc)}) + "\n"
        return json.dumps(response) + "\n"
 
    def _close_client(self, client):
        if client.conn is not None:
            try:
                self.sel.unregister(client.conn)
            except (KeyError, ValueError):
                pass
            try:
                client.conn.close()
            except OSError:
                pass
            client.conn = None