Report runtime status so Cockpit can show what ACARSCOT is hearing

src/acarscot/classes.py

@@ -11,6 +11,45 @@
 import acarscot
 
 
+class _NoStatus:
+    """Stand-in for pytak.StatusWriter on a pytak too old to have one.
+
+    AryaOS boxes are updated as packages, so a gateway can land on a host whose
+    pytak predates StatusWriter (added in 7.4.0). Failing to import would take
+    the gateway down over its telemetry helper, which is exactly backwards:
+    moving CoT is the job, reporting on it is not.
+
+    Degrading here is safe because it is VISIBLE. With nothing writing
+    /run/acarscot/status.json, the Cockpit plugin reports "No status from this
+    gateway ... may be running a pytak too old to report status" rather than
+    rendering an empty feed as though the band were quiet.
+    """
+
+    def count(self, *args, **kwargs) -> None:
+        return None
+
+    def record(self, *args, **kwargs) -> None:
+        return None
+
+    def set(self, *args, **kwargs) -> None:
+        return None
+
+    def write(self, *args, **kwargs) -> bool:
+        return False
+
+
+# Resolved at import so a missing StatusWriter is a startup-time decision
+# rather than an AttributeError on the first message.
+_StatusWriter = getattr(pytak, "StatusWriter", None)
+
+
+def make_status(app_name: str, version: str):
+    """Return a status writer, or a no-op if this pytak has none."""
+    if _StatusWriter is None:
+        return _NoStatus()
+    return _StatusWriter(app_name, version=version)
+
+
 def listen_host_port(listen_url: str):
     """Split a listen URL into (host, port).
 
@@ -46,6 +85,10 @@ def __init__(self, queue, config):
         self._no_position = 0
         self._emitted = 0
 
+        # Runtime status for Cockpit. Systemd gives us /run/acarscot via
+        # RuntimeDirectory=, so this lands where the plugin looks for it.
+        self.status = make_status("acarscot", acarscot.__version__)
+
     async def handle_data(self, data: bytes) -> None:
         """One acarsdec JSON message."""
         try:
@@ -57,17 +100,39 @@ async def handle_data(self, data: bytes) -> None:
         if not isinstance(msg, dict):
             return
 
+        self.status.count("rx")
+
         state = self.cache.update(msg)
         if state is None:
             # No usable tail number. Counted rather than logged per-message:
             # unknown-tail traffic is routine and would drown the journal.
+            self.status.count("no_tail")
+            self.status.write()
             return
 
         event = acarscot.aircraft_to_cot(state, self.config, self.cache)
+
+        # Record EVERY message with a tail, not just the ones that produce a
+        # marker. Most ACARS carries no position -- 1829 of 1841 in a measured
+        # capture -- so a feed showing only plotted aircraft would sit empty on
+        # a receiver that is working perfectly, which reads as a fault. The
+        # `placed` flag distinguishes the two without hiding either.
+        self.status.record(
+            tail=state.get("tail"),
+            flight=state.get("flight"),
+            label=msg.get("label"),
+            freq=msg.get("freq"),
+            level=msg.get("level"),
+            placed=event is not None,
+        )
+        self.status.set(tracked=len(self.cache))
+
         if event is None:
             # The COMMON case, not an error. Most ACARS messages carry no
             # position, so this is what the majority of traffic does.
             self._no_position += 1
+            self.status.count("no_position")
+            self.status.write()
             if self._no_position % 100 == 0:
                 self._logger.info(
                     "ACARS: %s messages with no plottable position, %s events emitted, "
@@ -79,6 +144,8 @@ async def handle_data(self, data: bytes) -> None:
             return
 
         self._emitted += 1
+        self.status.count("emitted")
+        self.status.write()
         await self.put_queue(event)
 
     async def run(self, number_of_iterations=-1):
@@ -92,9 +159,22 @@ async def run(self, number_of_iterations=-1):
         transport, _ = await loop.create_datagram_endpoint(
             lambda: _JSONDatagramProtocol(self), local_addr=(host, port)
         )
+
+        # Write immediately, before any traffic arrives. ACARS can be silent
+        # for minutes on a quiet band, and without this the management UI would
+        # report "no status from this gateway" -- indistinguishable from a
+        # gateway that failed to start -- for that whole time.
+        self.status.set(listen=f"udp://{host}:{port}")
+        self.status.write(force=True)
+
         try:
             while True:
-                await asyncio.sleep(3600)
+                # Heartbeat. The UI decides freshness from whether this file
+                # keeps changing, so an idle-but-healthy gateway MUST keep
+                # writing; otherwise silence on the band would be reported as
+                # a wedged service.
+                await asyncio.sleep(5)
+                self.status.write(force=True)
         finally:
             transport.close()
 

tests/test_acarscot.py

@@ -331,3 +331,146 @@ def test_wildcard_bind(self):
         from acarscot.classes import listen_host_port
 
         assert listen_host_port(acarscot.DEFAULT_LISTEN_URL) == ("0.0.0.0", 5555)
+
+
+@pytest.mark.skipif(
+    not hasattr(__import__("pytak"), "StatusWriter"),
+    reason="installed pytak predates StatusWriter (added 7.4.0)",
+)
+class TestStatusSurface:
+    """The data the Cockpit plugin reads.
+
+    The gateway already counted these things; they just went nowhere. The risk
+    with a status surface is that it reports the shape of the code rather than
+    what actually happened, so these assert against real messages.
+
+    These drive the coroutine with asyncio.run() rather than pytest-asyncio.
+    That is not a style choice: pytest-asyncio is not installed here, and when
+    I first wrote these as bare `async def` tests pytest SKIPPED all four while
+    still reporting them in the pass count. Four tests that could not fail.
+    """
+
+    def _worker(self, tmp_path):
+        import logging
+
+        import pytak
+
+        from acarscot.classes import ACARSWorker
+
+        worker = ACARSWorker.__new__(ACARSWorker)
+        worker.config = {}
+        worker.cache = AircraftCache()
+        worker._no_position = 0
+        worker._emitted = 0
+        worker._logger = logging.getLogger("test")
+        worker.status = pytak.StatusWriter(
+            "acarscot-test", path=str(tmp_path / "status.json")
+        )
+        worker.put_queue = _noop_put
+        return worker
+
+    def _handle(self, worker, payload):
+        import asyncio
+        import json
+
+        raw = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
+        asyncio.run(worker.handle_data(raw))
+
+    def _doc(self, worker):
+        import json
+
+        with open(worker.status.path) as handle:
+            return json.load(handle)
+
+    def test_positionless_message_still_appears_in_the_feed(self, tmp_path):
+        """The whole point of showing decodes rather than only plotted tracks.
+
+        1829 of 1841 real messages carried no position. A feed showing only
+        plotted aircraft would sit empty on a perfectly working receiver, which
+        an operator reads as a fault.
+        """
+        worker = self._worker(tmp_path)
+        self._handle(worker, MSG_ENGINE)
+
+        doc = self._doc(worker)
+        assert doc["counters"]["rx"] == 1
+        assert doc["counters"]["no_position"] == 1
+        entry = doc["recent"][0]
+        assert entry["tail"] == "N890UA"
+        assert entry["label"] == "H1"
+        assert entry["placed"] is False
+
+    def test_plotted_message_is_marked_placed(self, tmp_path):
+        worker = self._worker(tmp_path)
+        self._handle(worker, MSG_POSITION)
+
+        doc = self._doc(worker)
+        assert doc["counters"]["emitted"] == 1
+        assert doc["recent"][0]["placed"] is True
+        assert doc["recent"][0]["flight"] == "OO5738"
+
+    def test_unknown_tail_counted_but_not_shown_as_a_contact(self, tmp_path):
+        """It was received, but it is not an aircraft we can name."""
+        worker = self._worker(tmp_path)
+        self._handle(worker, MSG_NO_TAIL)
+
+        doc = self._doc(worker)
+        assert doc["counters"]["rx"] == 1
+        assert doc["counters"]["no_tail"] == 1
+        assert doc["recent"] == []
+
+    def test_undecodable_input_is_not_counted_as_received(self, tmp_path):
+        """Garbage on the socket is not an ACARS message."""
+        import os
+
+        worker = self._worker(tmp_path)
+        self._handle(worker, b"{not json")
+        assert not os.path.exists(worker.status.path)
+
+    def test_tracked_count_is_reported(self, tmp_path):
+        worker = self._worker(tmp_path)
+        self._handle(worker, MSG_ENGINE)
+        self._handle(worker, MSG_POSITION)
+
+        # Writes are rate-limited to once a second, so two messages in the same
+        # second leave the file holding the first one's figures. That is by
+        # design -- a gateway at hundreds of messages a second must not spend
+        # its time serialising JSON -- and the run loop's 5s heartbeat is what
+        # reconciles it. Forcing the write here is standing in for that
+        # heartbeat, not working around a bug.
+        worker.status.write(force=True)
+        assert self._doc(worker)["tracked"] == 2
+
+    def test_rate_limiting_means_the_file_can_lag_briefly(self, tmp_path):
+        """Documents the above, so nobody 'fixes' it into a per-message write."""
+        worker = self._worker(tmp_path)
+        self._handle(worker, MSG_ENGINE)
+        self._handle(worker, MSG_POSITION)
+        assert self._doc(worker)["tracked"] == 1  # lags until the heartbeat
+
+
+class TestStatusDegradesVisibly:
+    """A pytak without StatusWriter must not take the gateway down."""
+
+    def test_no_op_status_when_pytak_is_too_old(self, monkeypatch):
+        from acarscot import classes
+
+        monkeypatch.setattr(classes, "_StatusWriter", None)
+        status = classes.make_status("acarscot", "0.1.0")
+
+        # Every call the worker makes must be safe on the stand-in.
+        status.count("rx")
+        status.record(tail="N1")
+        status.set(tracked=1)
+        assert status.write() is False
+
+    def test_real_writer_used_when_available(self):
+        from acarscot import classes
+
+        if classes._StatusWriter is None:
+            pytest.skip("installed pytak has no StatusWriter")
+        assert not isinstance(classes.make_status("x", "0"), classes._NoStatus)
+
+
+async def _noop_put(event):
+    return None