Levandor

WiFi usage strip — implementation plan

19 min read

WiFi Device Usage Strip Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Show every WiFi device on telep1 with its live down/up speed, plus total 5G WAN throughput, on the cctv pane of the camwall.

Architecture: A python3 daemon on telep-mainframe polls the OpenWrt router every 5s over one persistent SSH connection, diffs iw station dump byte counters into Mbps, resolves MACs to friendly names, renders an ASS overlay, and pushes it to the running mpv via /run/camwall.sock. The router gets no new packages and no config changes; mpv’s filter chain and video stream are never touched.

Tech Stack: Python 3.13 stdlib only, iw (already on router), mpv osd-overlay IPC, systemd.

Design: 2026-07-22-wifi-usage-strip-design

Global Constraints

  • Python 3 stdlib only. No pip installs. PIL is NOT available and must not be used.
  • No comments, docstrings, or inline annotations in any code (user’s global rule). Code must be self-explanatory through naming.
  • Do not run git commit. Commit steps are deliberately omitted from this plan per the user’s global rule. Each task ends in a verification step instead.
  • Service runs as root — /run/camwall.sock is srw------- root root. Running as levander fails with EACCES.
  • All code lives in /home/levander/wifi-usage/ on telep-mainframe.
  • Only APs phy0-ap0 and phy1-ap0 are ever queried. phy1-ap1 (telep-cc, cameras) must never be queried or referenced.
  • MACs are normalised to lowercase everywhere.
  • Station tx bytes = device download. Station rx bytes = device upload. WAN rx_bytes = download, tx_bytes = upload (no inversion for WAN).
  • Never restart, reconfigure, or reload camwall. The collector only writes to a socket.
  • Run all commands on the box: /usr/bin/ssh levander@telep-mainframe. Use sudo for root actions.

Test command (used in every task): cd /home/levander/wifi-usage && python3 -m unittest -v <module>

Task 1: Router output parsers

Files:

  • Create: /home/levander/wifi-usage/iwparse.py
  • Test: /home/levander/wifi-usage/test_iwparse.py

Interfaces:

  • Consumes: nothing

  • Produces:

    • parse_station_dump(text) -> dict[str, dict] — keys are lowercase MACs, values have rx_bytes:int, tx_bytes:int, signal:int|None, inactive_ms:int
    • parse_dhcp_leases(text) -> dict[str, str] — lowercase MAC → hostname, omitting *
    • parse_wan_counters(text) -> tuple[int, int] | None — (rx_bytes, tx_bytes)

  • Step 1: Create the directory

/usr/bin/ssh levander@telep-mainframe 'mkdir -p /home/levander/wifi-usage'
  • Step 2: Write the failing test

Create /home/levander/wifi-usage/test_iwparse.py:

import unittest
 
from iwparse import parse_dhcp_leases, parse_station_dump, parse_wan_counters
 
STATION_FIXTURE = """Station 0a:ad:c4:80:BE:18 (on phy0-ap0)
	inactive time:	50 ms
	rx bytes:	488675230
	rx packets:	583572
	tx bytes:	613423947
	tx packets:	754883
	tx retries:	71
	signal:  	-41 dBm
	signal avg:	-99 dBm
	tx bitrate:	1200.9 MBit/s 80MHz HE-MCS 11
	last ack signal:-42 dBm
	authorized:	yes
Station fe:8e:63:c3:98:8c (on phy0-ap0)
	inactive time:	440 ms
	rx bytes:	100
	rx packets:	2
	tx bytes:	200
	tx packets:	3
	signal:  	-40 dBm
"""
 
 
class TestStationDump(unittest.TestCase):
    def test_parses_both_stations(self):
        result = parse_station_dump(STATION_FIXTURE)
        self.assertEqual(len(result), 2)
 
    def test_macs_are_lowercased(self):
        result = parse_station_dump(STATION_FIXTURE)
        self.assertIn("0a:ad:c4:80:be:18", result)
 
    def test_byte_counters(self):
        result = parse_station_dump(STATION_FIXTURE)["0a:ad:c4:80:be:18"]
        self.assertEqual(result["rx_bytes"], 488675230)
        self.assertEqual(result["tx_bytes"], 613423947)
 
    def test_signal_avg_does_not_overwrite_signal(self):
        result = parse_station_dump(STATION_FIXTURE)["0a:ad:c4:80:be:18"]
        self.assertEqual(result["signal"], -41)
 
    def test_bitrate_does_not_pollute_byte_counters(self):
        result = parse_station_dump(STATION_FIXTURE)["fe:8e:63:c3:98:8c"]
        self.assertEqual(result["rx_bytes"], 100)
        self.assertEqual(result["tx_bytes"], 200)
 
    def test_inactive_ms(self):
        result = parse_station_dump(STATION_FIXTURE)["0a:ad:c4:80:be:18"]
        self.assertEqual(result["inactive_ms"], 50)
 
    def test_empty_input(self):
        self.assertEqual(parse_station_dump(""), {})
 
 
class TestLeases(unittest.TestCase):
    def test_parses_hostnames(self):
        text = "1784300000 AA:BB:CC:DD:EE:FF 192.168.1.50 marks-laptop *\n"
        self.assertEqual(parse_dhcp_leases(text), {"aa:bb:cc:dd:ee:ff": "marks-laptop"})
 
    def test_skips_star_hostname(self):
        text = "1784300000 aa:bb:cc:dd:ee:ff 192.168.1.50 * *\n"
        self.assertEqual(parse_dhcp_leases(text), {})
 
    def test_skips_short_lines(self):
        self.assertEqual(parse_dhcp_leases("garbage\n"), {})
 
 
class TestWanCounters(unittest.TestCase):
    def test_parses_two_numbers(self):
        self.assertEqual(parse_wan_counters("16440043695\n7676036048\n"), (16440043695, 7676036048))
 
    def test_returns_none_when_incomplete(self):
        self.assertIsNone(parse_wan_counters("123\n"))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 3: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_iwparse -v'

Expected: FAIL with ModuleNotFoundError: No module named 'iwparse'

  • Step 4: Write the implementation

Create /home/levander/wifi-usage/iwparse.py:

def parse_station_dump(text):
    stations = {}
    mac = None
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if line.startswith("Station "):
            mac = line.split()[1].lower()
            stations[mac] = {"rx_bytes": 0, "tx_bytes": 0, "signal": None, "inactive_ms": 0}
            continue
        if mac is None or ":" not in line:
            continue
        key, _, value = line.partition(":")
        key = key.strip()
        value = value.strip()
        if key == "rx bytes":
            stations[mac]["rx_bytes"] = int(value)
        elif key == "tx bytes":
            stations[mac]["tx_bytes"] = int(value)
        elif key == "signal":
            stations[mac]["signal"] = int(value.split()[0])
        elif key == "inactive time":
            stations[mac]["inactive_ms"] = int(value.split()[0])
    return stations
 
 
def parse_dhcp_leases(text):
    leases = {}
    for line in text.splitlines():
        parts = line.split()
        if len(parts) < 4:
            continue
        hostname = parts[3]
        if hostname and hostname != "*":
            leases[parts[1].lower()] = hostname
    return leases
 
 
def parse_wan_counters(text):
    numbers = [int(token) for token in text.split() if token.isdigit()]
    if len(numbers) < 2:
        return None
    return numbers[0], numbers[1]
  • Step 5: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_iwparse -v'

Expected: OK — 12 tests pass.

  • Step 6: Verify against real router output
/usr/bin/ssh levander@telep-mainframe '/usr/bin/ssh -i ~/.ssh/router_alarm -o StrictHostKeyChecking=no root@192.168.1.1 "iw dev phy0-ap0 station dump" > /tmp/real_dump.txt; cd /home/levander/wifi-usage && python3 -c "
import iwparse
print(iwparse.parse_station_dump(open(\"/tmp/real_dump.txt\").read()))
"'

Expected: a dict of real lowercase MACs with non-zero rx_bytes/tx_bytes.

Task 2: Rate tracker

Files:

  • Create: /home/levander/wifi-usage/rates.py
  • Test: /home/levander/wifi-usage/test_rates.py

Interfaces:

  • Consumes: nothing

  • Produces:

    • compute_mbps(prev_bytes, cur_bytes, elapsed) -> float — clamps negative deltas to 0.0
    • RateTracker with update(key, rx_bytes, tx_bytes, now) -> tuple[float, float] | None returning (rx_mbps, tx_mbps) in raw counter terms, and forget(live_keys)

  • Step 1: Write the failing test

Create /home/levander/wifi-usage/test_rates.py:

import unittest
 
from rates import MAX_GAP_SECONDS, RateTracker, compute_mbps
 
 
class TestComputeMbps(unittest.TestCase):
    def test_known_rate(self):
        self.assertAlmostEqual(compute_mbps(0, 1_250_000, 1.0), 10.0)
 
    def test_uses_elapsed_time(self):
        self.assertAlmostEqual(compute_mbps(0, 1_250_000, 5.0), 2.0)
 
    def test_counter_reset_clamps_to_zero(self):
        self.assertEqual(compute_mbps(1000, 5, 5.0), 0.0)
 
    def test_zero_elapsed_is_safe(self):
        self.assertEqual(compute_mbps(0, 100, 0.0), 0.0)
 
 
class TestRateTracker(unittest.TestCase):
    def test_first_sample_returns_none(self):
        tracker = RateTracker()
        self.assertIsNone(tracker.update("a", 0, 0, 100.0))
 
    def test_second_sample_returns_rates(self):
        tracker = RateTracker()
        tracker.update("a", 0, 0, 100.0)
        rx, tx = tracker.update("a", 1_250_000, 2_500_000, 105.0)
        self.assertAlmostEqual(rx, 2.0)
        self.assertAlmostEqual(tx, 4.0)
 
    def test_reassociation_reset_yields_zero_not_negative(self):
        tracker = RateTracker()
        tracker.update("a", 5_000_000, 5_000_000, 100.0)
        rx, tx = tracker.update("a", 10, 10, 105.0)
        self.assertEqual(rx, 0.0)
        self.assertEqual(tx, 0.0)
 
    def test_long_gap_discards_sample(self):
        tracker = RateTracker()
        tracker.update("a", 0, 0, 100.0)
        self.assertIsNone(tracker.update("a", 999_999_999, 0, 100.0 + MAX_GAP_SECONDS + 1))
 
    def test_long_gap_rebaselines_for_next_cycle(self):
        tracker = RateTracker()
        tracker.update("a", 0, 0, 100.0)
        tracker.update("a", 1_000_000, 0, 100.0 + MAX_GAP_SECONDS + 1)
        rx, _ = tracker.update("a", 1_000_000 + 1_250_000, 0, 100.0 + MAX_GAP_SECONDS + 6)
        self.assertAlmostEqual(rx, 2.0)
 
    def test_forget_drops_absent_keys(self):
        tracker = RateTracker()
        tracker.update("a", 0, 0, 100.0)
        tracker.update("b", 0, 0, 100.0)
        tracker.forget({"a"})
        self.assertIsNone(tracker.update("b", 1_250_000, 0, 105.0))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_rates -v'

Expected: FAIL with ModuleNotFoundError: No module named 'rates'

  • Step 3: Write the implementation

Create /home/levander/wifi-usage/rates.py:

MAX_GAP_SECONDS = 30.0
 
 
def compute_mbps(prev_bytes, cur_bytes, elapsed):
    if elapsed <= 0:
        return 0.0
    delta = cur_bytes - prev_bytes
    if delta < 0:
        return 0.0
    return delta * 8 / elapsed / 1_000_000
 
 
class RateTracker:
    def __init__(self):
        self.previous = {}
 
    def update(self, key, rx_bytes, tx_bytes, now):
        previous = self.previous.get(key)
        self.previous[key] = (rx_bytes, tx_bytes, now)
        if previous is None:
            return None
        prev_rx, prev_tx, prev_now = previous
        elapsed = now - prev_now
        if elapsed <= 0 or elapsed > MAX_GAP_SECONDS:
            return None
        return compute_mbps(prev_rx, rx_bytes, elapsed), compute_mbps(prev_tx, tx_bytes, elapsed)
 
    def forget(self, live_keys):
        for key in list(self.previous):
            if key not in live_keys:
                del self.previous[key]
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_rates -v'

Expected: OK — 10 tests pass.

Task 3: Name resolution

Files:

  • Create: /home/levander/wifi-usage/names.py
  • Test: /home/levander/wifi-usage/test_names.py

Interfaces:

  • Consumes: nothing

  • Produces:

    • load_trusted(path) -> dict[str, dict] — lowercased MAC keys, {} on missing/invalid file
    • resolve_name(mac, trusted, leases) -> str

  • Step 1: Write the failing test

Create /home/levander/wifi-usage/test_names.py:

import json
import os
import tempfile
import unittest
 
from names import load_trusted, resolve_name
 
TRUSTED = {"fe:8e:63:c3:98:8c": {"name": "Mark-iPhone-ja"}}
LEASES = {"aa:bb:cc:dd:ee:ff": "marks-laptop"}
 
 
class TestResolveName(unittest.TestCase):
    def test_trusted_name_wins(self):
        self.assertEqual(resolve_name("FE:8E:63:C3:98:8C", TRUSTED, LEASES), "Mark-iPhone-ja")
 
    def test_falls_back_to_dhcp_hostname(self):
        self.assertEqual(resolve_name("aa:bb:cc:dd:ee:ff", TRUSTED, LEASES), "marks-laptop")
 
    def test_falls_back_to_last4(self):
        self.assertEqual(resolve_name("0a:ad:c4:80:be:18", TRUSTED, LEASES), "eszköz-be18")
 
    def test_trusted_entry_without_name_falls_through(self):
        trusted = {"11:22:33:44:55:66": {}}
        self.assertEqual(resolve_name("11:22:33:44:55:66", trusted, {}), "eszköz-5566")
 
 
class TestLoadTrusted(unittest.TestCase):
    def test_lowercases_keys(self):
        handle = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False)
        json.dump({"FE:8E:63:C3:98:8C": {"name": "X"}}, handle)
        handle.close()
        try:
            self.assertIn("fe:8e:63:c3:98:8c", load_trusted(handle.name))
        finally:
            os.unlink(handle.name)
 
    def test_missing_file_returns_empty(self):
        self.assertEqual(load_trusted("/nonexistent/trusted.json"), {})
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_names -v'

Expected: FAIL with ModuleNotFoundError: No module named 'names'

  • Step 3: Write the implementation

Create /home/levander/wifi-usage/names.py:

import json
 
 
def load_trusted(path):
    try:
        with open(path, encoding="utf-8") as handle:
            raw = json.load(handle)
    except (OSError, ValueError):
        return {}
    if not isinstance(raw, dict):
        return {}
    return {key.lower(): value for key, value in raw.items()}
 
 
def resolve_name(mac, trusted, leases):
    key = mac.lower()
    entry = trusted.get(key)
    if isinstance(entry, dict) and entry.get("name"):
        return entry["name"]
    hostname = leases.get(key)
    if hostname:
        return hostname
    return "eszköz-" + key.replace(":", "")[-4:]
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_names -v'

Expected: OK — 6 tests pass.

  • Step 5: Verify against the real trusted.json
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -c "
from names import load_trusted, resolve_name
trusted = load_trusted(\"/home/levander/alarm/trusted.json\")
print(len(trusted), \"trusted entries\")
print(resolve_name(\"fe:8e:63:c3:98:8c\", trusted, {}))
"'

Expected: 3 trusted entries and Mark-iPhone-ja.

Task 4: ASS renderer

Files:

  • Create: /home/levander/wifi-usage/render.py
  • Test: /home/levander/wifi-usage/test_render.py

Interfaces:

  • Consumes: nothing

  • Produces:

    • render_ass(devices, wan_down, wan_up, stale=0.0) -> str where devices is a list of {"name": str, "down": float|None, "up": float|None}
    • select_tiles(devices) -> tuple[list, int]
    • format_rate(value) -> str
    • Constants SCREEN_W, SCREEN_H, MAX_TILES, STALE_WARN_SECONDS, STALE_LIMIT_SECONDS

  • Step 1: Write the failing test

Create /home/levander/wifi-usage/test_render.py:

import unittest
 
from render import (
    MAX_TILES,
    STALE_LIMIT_SECONDS,
    format_rate,
    render_ass,
    select_tiles,
)
 
 
def device(name, down=1.0, up=1.0):
    return {"name": name, "down": down, "up": up}
 
 
class TestFormatRate(unittest.TestCase):
    def test_none_renders_dash(self):
        self.assertEqual(format_rate(None), "—")
 
    def test_one_decimal(self):
        self.assertEqual(format_rate(12.44), "12.4")
 
 
class TestSelectTiles(unittest.TestCase):
    def test_alphabetical_order(self):
        tiles, extra = select_tiles([device("zeta"), device("alpha")])
        self.assertEqual([t["name"] for t in tiles], ["alpha", "zeta"])
        self.assertEqual(extra, 0)
 
    def test_no_overflow_at_capacity(self):
        tiles, extra = select_tiles([device("d%02d" % i) for i in range(MAX_TILES)])
        self.assertEqual(len(tiles), MAX_TILES)
        self.assertEqual(extra, 0)
 
    def test_overflow_keeps_busiest(self):
        devices = [device("d%02d" % i, down=0.0, up=0.0) for i in range(MAX_TILES + 3)]
        devices[-1] = device("zzz-busy", down=90.0, up=5.0)
        tiles, extra = select_tiles(devices)
        self.assertEqual(len(tiles), MAX_TILES - 1)
        self.assertEqual(extra, 4)
        self.assertIn("zzz-busy", [t["name"] for t in tiles])
 
    def test_overflow_still_alphabetical(self):
        devices = [device("d%02d" % i, down=float(i), up=0.0) for i in range(MAX_TILES + 2)]
        tiles, _ = select_tiles(devices)
        names = [t["name"] for t in tiles]
        self.assertEqual(names, sorted(names))
 
 
class TestRenderAss(unittest.TestCase):
    def test_includes_header_and_names(self):
        out = render_ass([device("Mark-iPhone-ja", 12.4, 0.8)], 28.6, 3.1)
        self.assertIn("WIFI ESZKÖZÖK", out)
        self.assertIn("Mark-iPhone-ja", out)
        self.assertIn("28.6", out)
        self.assertIn("12.4", out)
 
    def test_expired_stale_blanks_values(self):
        out = render_ass([device("Mark", 12.4, 0.8)], 28.6, 3.1, stale=STALE_LIMIT_SECONDS + 1)
        self.assertNotIn("12.4", out)
        self.assertIn("—", out)
 
    def test_overflow_tile_rendered(self):
        devices = [device("d%02d" % i) for i in range(MAX_TILES + 3)]
        out = render_ass(devices, 1.0, 1.0)
        self.assertIn("+4 további", out)
 
    def test_braces_in_name_are_escaped(self):
        out = render_ass([device("we{ird}")], 1.0, 1.0)
        self.assertIn("we\\{ird\\}", out)
 
    def test_no_devices_still_renders_header(self):
        out = render_ass([], 1.0, 1.0)
        self.assertIn("WIFI ESZKÖZÖK", out)
 
    def test_every_event_is_single_line(self):
        out = render_ass([device("Mark\nInjected")], 1.0, 1.0)
        for line in out.splitlines():
            self.assertNotIn("\n", line)
        self.assertNotIn("Mark\nInjected", out)
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_render -v'

Expected: FAIL with ModuleNotFoundError: No module named 'render'

  • Step 3: Write the implementation

Create /home/levander/wifi-usage/render.py:

SCREEN_W = 3840
SCREEN_H = 2160
MARGIN = 64
HEADER_Y = 1880
HEADER_FONT_SIZE = 46
TILE_Y = 1946
TILE_W = 509
TILE_H = 186
TILE_GAP = 24
MAX_TILES = 7
NAME_FONT_SIZE = 38
RATE_FONT_SIZE = 42
IDLE_THRESHOLD_MBPS = 0.1
STALE_WARN_SECONDS = 12.0
STALE_LIMIT_SECONDS = 60.0
ACTIVE_BG_ALPHA = 0x73
IDLE_BG_ALPHA = 0xA8
ACTIVE_TEXT_ALPHA = 0x00
IDLE_TEXT_ALPHA = 0x70
FONT = "DejaVu Sans"
 
 
def _escape(value):
    return (
        str(value)
        .replace("\\", "\\\\")
        .replace("{", "\\{")
        .replace("}", "\\}")
        .replace("\n", " ")
        .replace("\r", " ")
    )
 
 
def _rectangle(x, y, width, height, alpha):
    return (
        "{\\an7\\pos(%d,%d)\\p1\\bord0\\shad0\\c&H000000&\\1a&H%02X&}"
        "m 0 0 l %d 0 l %d %d l 0 %d{\\p0}" % (x, y, alpha, width, width, height, height)
    )
 
 
def _label(x, y, font_size, value, alpha, align=7):
    return "{\\an%d\\pos(%d,%d)\\fn%s\\b1\\fs%d\\bord3\\shad0\\c&HFFFFFF&\\1a&H%02X&}%s" % (
        align,
        x,
        y,
        FONT,
        font_size,
        alpha,
        _escape(value),
    )
 
 
def format_rate(value):
    if value is None:
        return "—"
    return "%.1f" % value
 
 
def select_tiles(devices):
    if len(devices) <= MAX_TILES:
        return sorted(devices, key=lambda item: item["name"].lower()), 0
    busiest = sorted(
        devices,
        key=lambda item: ((item["down"] or 0.0) + (item["up"] or 0.0)),
        reverse=True,
    )[: MAX_TILES - 1]
    kept = sorted(busiest, key=lambda item: item["name"].lower())
    return kept, len(devices) - len(kept)
 
 
def render_ass(devices, wan_down, wan_up, stale=0.0):
    expired = stale > STALE_LIMIT_SECONDS
    warning = " ⚠" if stale > STALE_WARN_SECONDS else ""
    tiles, overflow = select_tiles(devices)
    events = [_label(MARGIN, HEADER_Y, HEADER_FONT_SIZE, "WIFI ESZKÖZÖK", ACTIVE_TEXT_ALPHA)]
    header_down = None if expired else wan_down
    header_up = None if expired else wan_up
    events.append(
        _label(
            SCREEN_W - MARGIN,
            HEADER_Y,
            HEADER_FONT_SIZE,
            "INTERNET:  %s ↓ / %s ↑ Mbps%s" % (format_rate(header_down), format_rate(header_up), warning),
            ACTIVE_TEXT_ALPHA,
            align=9,
        )
    )
    x = MARGIN
    for item in tiles:
        down = None if expired else item["down"]
        up = None if expired else item["up"]
        idle = ((down or 0.0) + (up or 0.0)) < IDLE_THRESHOLD_MBPS
        background_alpha = IDLE_BG_ALPHA if idle else ACTIVE_BG_ALPHA
        text_alpha = IDLE_TEXT_ALPHA if idle else ACTIVE_TEXT_ALPHA
        events.append(_rectangle(x, TILE_Y, TILE_W, TILE_H, background_alpha))
        events.append(_label(x + 18, TILE_Y + 10, NAME_FONT_SIZE, item["name"], text_alpha))
        events.append(_label(x + 18, TILE_Y + 64, RATE_FONT_SIZE, format_rate(down) + " ↓", text_alpha))
        events.append(_label(x + 18, TILE_Y + 120, RATE_FONT_SIZE, format_rate(up) + " ↑", text_alpha))
        x += TILE_W + TILE_GAP
    if overflow:
        events.append(_rectangle(x, TILE_Y, TILE_W, TILE_H, IDLE_BG_ALPHA))
        events.append(
            _label(x + 18, TILE_Y + 64, RATE_FONT_SIZE, "+%d további" % overflow, IDLE_TEXT_ALPHA)
        )
    return "\n".join(events)
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest test_render -v'

Expected: OK — 12 tests pass.

Task 5: mpv IPC push

Files:

  • Create: /home/levander/wifi-usage/mpvipc.py

Interfaces:

  • Consumes: render.SCREEN_W, render.SCREEN_H
  • Produces: push_overlay(socket_path, ass_text) -> bool — returns False on any socket error, never raises

mpv destroys osd-overlay entries when the IPC client disconnects

A connect → send → close client does not work: mpv replies {"error":"success"} and then removes the overlay the moment the connection drops, so nothing appears on screen. This was proven empirically — the naive version was implemented first, produced no strip, and holding the same socket open with the identical payload made it appear.

The connection must therefore be persistent. That in turn means mpv’s event stream must be drained, or the socket buffer fills and the daemon eventually blocks. Reconnect-on-failure also gives self-healing when camwall restarts and recreates the socket.

  • Step 1: Write the implementation

Create /home/levander/wifi-usage/mpvipc.py:

import json
import socket
 
from render import SCREEN_H, SCREEN_W
 
OVERLAY_ID = 42
SOCKET_TIMEOUT = 3.0
CONNECT_ATTEMPTS = 2
 
_connection = None
 
 
def _close():
    global _connection
    if _connection is not None:
        try:
            _connection.close()
        except OSError:
            pass
        _connection = None
 
 
def _connect(socket_path):
    global _connection
    connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    connection.settimeout(SOCKET_TIMEOUT)
    connection.connect(socket_path)
    _connection = connection
    return connection
 
 
def _drain(connection):
    connection.setblocking(False)
    try:
        while True:
            chunk = connection.recv(65536)
            if not chunk:
                raise OSError("mpv closed the ipc connection")
    except BlockingIOError:
        pass
    finally:
        connection.setblocking(True)
        connection.settimeout(SOCKET_TIMEOUT)
 
 
def _send(connection, ass_text):
    command = {
        "command": [
            "osd-overlay",
            OVERLAY_ID,
            "ass-events",
            ass_text,
            SCREEN_W,
            SCREEN_H,
            0,
            False,
            False,
        ]
    }
    connection.sendall((json.dumps(command) + "\n").encode("utf-8"))
    _drain(connection)
 
 
def push_overlay(socket_path, ass_text):
    for _ in range(CONNECT_ATTEMPTS):
        try:
            connection = _connection
            if connection is None:
                connection = _connect(socket_path)
            _send(connection, ass_text)
            return True
        except (OSError, ValueError):
            _close()
    return False
  • Step 2: Smoke-test the overlay on the real wall

This is the first time anything appears on the TV. Run as root.

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && sudo python3 -c "
from mpvipc import push_overlay
from render import render_ass
devices = [
    {\"name\": \"Mark-iPhone-ja\", \"down\": 12.4, \"up\": 0.8},
    {\"name\": \"iPhone\", \"down\": 0.2, \"up\": 0.1},
    {\"name\": \"eszköz-be18\", \"down\": 0.0, \"up\": 0.0},
]
print(push_overlay(\"/run/camwall.sock\", render_ass(devices, 28.6, 3.1)))
"'

Expected: prints True.

  • Step 3: Confirm visually via framebuffer grab
/usr/bin/ssh levander@telep-mainframe 'sudo DISPLAY=:0 ffmpeg -hide_banner -loglevel error -f x11grab -video_size 3840x2160 -i :0.0 -frames:v 1 -y /tmp/strip_check.png'
/usr/bin/scp levander@telep-mainframe:/tmp/strip_check.png /tmp/strip_check.png

Then read /tmp/strip_check.png and confirm: the strip appears at the bottom, the camera image and TOP KÉPEK row are unchanged, and text is legible.

  • Step 4: Confirm the wall was not disturbed
/usr/bin/ssh levander@telep-mainframe 'systemctl show camwall.service -p NRestarts -p ActiveEnterTimestamp'

Expected: NRestarts unchanged and the start timestamp is still the original one — proving the overlay did not restart the wall.

  • Step 5: Clear the test overlay
/usr/bin/ssh levander@telep-mainframe 'sudo python3 -c "
import json, socket
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM); s.settimeout(3)
s.connect(\"/run/camwall.sock\")
s.sendall((json.dumps({\"command\":[\"osd-overlay\",42,\"none\",\"\",0,0,0,False,False]})+\"\n\").encode())
print(s.recv(400).decode().strip()); s.close()
"'

Expected: "error":"success" and the strip disappears from the TV.

Task 6: Collector daemon

Files:

  • Create: /home/levander/wifi-usage/collect.py

Interfaces:

  • Consumes: iwparse.parse_station_dump, iwparse.parse_dhcp_leases, iwparse.parse_wan_counters, rates.RateTracker, names.load_trusted, names.resolve_name, render.render_ass, mpvipc.push_overlay

  • Produces: main() entry point; module constants INTERVAL_SECONDS, MPV_SOCKET

  • Step 1: Write the implementation

Create /home/levander/wifi-usage/collect.py:

import os
import subprocess
import time
 
from iwparse import parse_dhcp_leases, parse_station_dump, parse_wan_counters
from mpvipc import push_overlay
from names import load_trusted, resolve_name
from rates import RateTracker
from render import render_ass
 
ROUTER = "root@192.168.1.1"
SSH_KEY = "/home/levander/.ssh/router_alarm"
CONTROL_PATH = "/run/wifi-usage/router.sock"
TRUSTED_PATH = "/home/levander/alarm/trusted.json"
MPV_SOCKET = "/run/camwall.sock"
INTERVAL_SECONDS = 5.0
POLL_TIMEOUT_SECONDS = 12
 
REMOTE_COMMAND = (
    "iw dev phy0-ap0 station dump; echo __AP2__; "
    "iw dev phy1-ap0 station dump; echo __WAN__; "
    "cat /sys/class/net/wan/statistics/rx_bytes /sys/class/net/wan/statistics/tx_bytes; "
    "echo __LEASES__; cat /tmp/dhcp.leases"
)
 
SSH_COMMAND = [
    "ssh",
    "-i", SSH_KEY,
    "-o", "BatchMode=yes",
    "-o", "ConnectTimeout=6",
    "-o", "StrictHostKeyChecking=no",
    "-o", "UserKnownHostsFile=/dev/null",
    "-o", "LogLevel=ERROR",
    "-o", "ControlMaster=auto",
    "-o", "ControlPath=" + CONTROL_PATH,
    "-o", "ControlPersist=60",
    ROUTER,
    REMOTE_COMMAND,
]
 
 
def poll_router():
    try:
        result = subprocess.run(
            SSH_COMMAND, capture_output=True, text=True, timeout=POLL_TIMEOUT_SECONDS
        )
    except (subprocess.SubprocessError, OSError):
        return None
    if result.returncode != 0:
        return None
    return result.stdout
 
 
def split_sections(raw):
    first_ap, _, remainder = raw.partition("__AP2__")
    second_ap, _, remainder = remainder.partition("__WAN__")
    wan, _, leases = remainder.partition("__LEASES__")
    return first_ap + "\n" + second_ap, wan, leases
 
 
def station_rates(tracker, mac, station, now):
    sample = tracker.update(mac, station["rx_bytes"], station["tx_bytes"], now)
    if sample is None:
        return None, None
    access_point_rx_mbps, access_point_tx_mbps = sample
    device_download = access_point_tx_mbps
    device_upload = access_point_rx_mbps
    return device_download, device_upload
 
 
def reload_trusted(cached, cached_mtime):
    try:
        mtime = os.path.getmtime(TRUSTED_PATH)
    except OSError:
        return cached, cached_mtime
    if mtime == cached_mtime:
        return cached, cached_mtime
    return load_trusted(TRUSTED_PATH), mtime
 
 
def main():
    station_tracker = RateTracker()
    wan_tracker = RateTracker()
    trusted = {}
    trusted_mtime = None
    devices = []
    wan_down = None
    wan_up = None
    last_success = time.monotonic()
    while True:
        cycle_start = time.monotonic()
        raw = poll_router()
        now = time.monotonic()
        if raw is not None:
            try:
                trusted, trusted_mtime = reload_trusted(trusted, trusted_mtime)
                station_text, wan_text, leases_text = split_sections(raw)
                stations = parse_station_dump(station_text)
                leases = parse_dhcp_leases(leases_text)
                collected = []
                for mac, station in stations.items():
                    down, up = station_rates(station_tracker, mac, station, now)
                    collected.append(
                        {"name": resolve_name(mac, trusted, leases), "down": down, "up": up}
                    )
                station_tracker.forget(set(stations))
                devices = collected
                counters = parse_wan_counters(wan_text)
                if counters is not None:
                    sample = wan_tracker.update("wan", counters[0], counters[1], now)
                    if sample is None:
                        wan_down = None
                        wan_up = None
                    else:
                        wan_down, wan_up = sample
                last_success = now
            except (ValueError, KeyError, TypeError):
                pass
        stale = time.monotonic() - last_success
        push_overlay(MPV_SOCKET, render_ass(devices, wan_down, wan_up, stale))
        remaining = INTERVAL_SECONDS - (time.monotonic() - cycle_start)
        time.sleep(max(0.5, remaining))
 
 
if __name__ == "__main__":
    main()
  • Step 2: Run one foreground cycle to verify the pipeline
/usr/bin/ssh levander@telep-mainframe 'sudo mkdir -p /run/wifi-usage && cd /home/levander/wifi-usage && sudo python3 -c "
import collect
raw = collect.poll_router()
print(\"poll ok:\", raw is not None)
stations, wan, leases = collect.split_sections(raw)
from iwparse import parse_station_dump, parse_wan_counters
print(\"stations:\", list(parse_station_dump(stations)))
print(\"wan:\", parse_wan_counters(wan))
"'

Expected: poll ok: True, a list of real MACs, and a (rx, tx) tuple. Verify no camera MAC appears — cross-check against the telep-cc camera MAC.

  • Step 3: Run the daemon in the foreground for ~20 seconds
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && sudo timeout 20 python3 collect.py; echo "exit=$?"'

Expected: exit=124 (killed by timeout, i.e. it ran without crashing), and the strip is live on the TV with real names and numbers.

Task 7: systemd service and end-to-end verification

Files:

  • Create: /etc/systemd/system/wifi-usage.service

Interfaces:

  • Consumes: collect.main()

  • Produces: a running wifi-usage.service

  • Step 1: Write the unit file

Create /etc/systemd/system/wifi-usage.service:

[Unit]
Description=WiFi device usage strip on the camwall
After=network-online.target camwall.service
Wants=network-online.target
 
[Service]
Type=simple
User=root
WorkingDirectory=/home/levander/wifi-usage
RuntimeDirectory=wifi-usage
ExecStart=/usr/bin/python3 /home/levander/wifi-usage/collect.py
Restart=always
RestartSec=5
 
[Install]
WantedBy=multi-user.target

Install it:

/usr/bin/ssh levander@telep-mainframe 'sudo tee /etc/systemd/system/wifi-usage.service > /dev/null <<UNIT
[Unit]
Description=WiFi device usage strip on the camwall
After=network-online.target camwall.service
Wants=network-online.target
 
[Service]
Type=simple
User=root
WorkingDirectory=/home/levander/wifi-usage
RuntimeDirectory=wifi-usage
ExecStart=/usr/bin/python3 /home/levander/wifi-usage/collect.py
Restart=always
RestartSec=5
 
[Install]
WantedBy=multi-user.target
UNIT
sudo systemctl daemon-reload && sudo systemctl enable --now wifi-usage.service && sleep 15 && systemctl is-active wifi-usage.service'

Expected: active

  • Step 2: Verify the strip is live and correct
/usr/bin/ssh levander@telep-mainframe 'sudo DISPLAY=:0 ffmpeg -hide_banner -loglevel error -f x11grab -video_size 3840x2160 -i :0.0 -frames:v 1 -y /tmp/strip_live.png'
/usr/bin/scp levander@telep-mainframe:/tmp/strip_live.png /tmp/strip_live.png

Read the PNG. Confirm: real device names, plausible Mbps, the INTERNET header, cameras and TOP KÉPEK undisturbed.

  • Step 3: Verify the numbers are real, not decorative

Generate load from a known WiFi device (e.g. start a large download on the phone), wait ~10s, grab the framebuffer again, and confirm that device’s ↓ figure and the INTERNET ↓ figure both rise.

/usr/bin/ssh levander@telep-mainframe 'sudo DISPLAY=:0 ffmpeg -hide_banner -loglevel error -f x11grab -video_size 3840x2160 -i :0.0 -frames:v 1 -y /tmp/strip_load.png'
/usr/bin/scp levander@telep-mainframe:/tmp/strip_load.png /tmp/strip_load.png

Expected: a visible increase. If the busy device’s ↑ rises instead of its ↓, the rx/tx mapping in station_rates is inverted — fix it there.

  • Step 4: Verify the wall was never restarted and video does not stutter
/usr/bin/ssh levander@telep-mainframe 'systemctl show camwall.service -p NRestarts -p ActiveEnterTimestamp; systemctl show camwall-watchdog.service -p NRestarts'

Expected: camwall NRestarts unchanged and the same ActiveEnterTimestamp as before Task 5 — the overlay never touched playback.

Watch the TV for 60 seconds and confirm the camera feeds are smooth (no 5-second stutter).

  • Step 5: Verify self-healing after a camwall restart
/usr/bin/ssh levander@telep-mainframe 'sudo systemctl restart camwall.service && sleep 25 && sudo DISPLAY=:0 ffmpeg -hide_banner -loglevel error -f x11grab -video_size 3840x2160 -i :0.0 -frames:v 1 -y /tmp/strip_healed.png'
/usr/bin/scp levander@telep-mainframe:/tmp/strip_healed.png /tmp/strip_healed.png

Expected: the strip reappears on its own within ~5s of the wall coming back, with no manual intervention.

  • Step 6: Verify graceful degradation when the router is unreachable
/usr/bin/ssh levander@telep-mainframe 'sudo mv /home/levander/.ssh/router_alarm /home/levander/.ssh/router_alarm.off && sleep 20 && sudo DISPLAY=:0 ffmpeg -hide_banner -loglevel error -f x11grab -video_size 3840x2160 -i :0.0 -frames:v 1 -y /tmp/strip_stale.png; sudo mv /home/levander/.ssh/router_alarm.off /home/levander/.ssh/router_alarm; systemctl is-active wifi-usage.service'
/usr/bin/scp levander@telep-mainframe:/tmp/strip_stale.png /tmp/strip_stale.png

Expected: the service stays active, the strip still shows the last values with a ⚠ in the header, and the wall is unaffected. After the key is restored, values resume within ~5s.

  • Step 7: Run the full test suite once more
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/wifi-usage && python3 -m unittest discover -p "test_*.py" -v 2>&1 | tail -5'

Expected: OK — all tests pass.

Self-review notes

Spec coverage: live Mbps (Task 2), all telep1 WiFi devices with camera VLAN excluded by construction (Task 6 remote command queries only the two APs), friendly names via the alarm’s chain (Task 3), WAN header (Tasks 2/4/6), 5s refresh (Task 6), horizontal tiles in stable name order (Task 4), overflow rule (Task 4), idle dimming (Task 4), staleness handling (Tasks 4/6, verified Task 7 Step 6), osd-overlay display with no stream reload (Task 5, verified Task 7 Step 4), self-healing (Task 7 Step 5), service (Task 7).

Deliberate deviation from the skill defaults: git commit steps are omitted because the user’s global rules forbid committing without an explicit request.

Graph View

Backlinks