|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Mirror the kernel ARP/neighbour state for the peer node onto the OLED. |
| 3 | +
|
| 4 | +This is a read-only window onto Linux's neighbour cache (the ARP table). It |
| 5 | +polls `ip neigh` for one peer IP and paints that entry's NUD state on the |
| 6 | +SSD1306 — REACHABLE, STALE, DELAY, PROBE, INCOMPLETE, FAILED, or ABSENT when |
| 7 | +there's no entry at all. It never changes the cache; you drive the state |
| 8 | +machine yourself from another shell and watch the panel follow along: |
| 9 | +
|
| 10 | + curl http://10.10.0.2 # traffic to the peer: ABSENT/STALE -> REACHABLE |
| 11 | + ip neigh show 10.10.0.2 # the same thing this screen is reading |
| 12 | + sudo ip neigh del 10.10.0.2 dev eth0 # -> ABSENT |
| 13 | + sudo ip neigh replace 10.10.0.2 dev eth0 \ |
| 14 | + lladdr <mac> nud stale # force it STALE |
| 15 | + # then leave it idle ~30s and REACHABLE decays to STALE on its own. |
| 16 | +
|
| 17 | +The peer defaults to the other half of the 10.10.0.1 <-> 10.10.0.2 lab pair |
| 18 | +(auto-picked from this node's own address), or pass --peer. |
| 19 | +
|
| 20 | + /opt/little-internet/venv/bin/python3 arp_oled.py |
| 21 | + arp_oled.py --peer 10.10.0.2 --interval 0.5 |
| 22 | + arp_oled.py --address 0x3d --controller sh1106 |
| 23 | + arp_oled.py --font /usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf |
| 24 | +""" |
| 25 | +import argparse |
| 26 | +import json |
| 27 | +import os |
| 28 | +import subprocess |
| 29 | +import sys |
| 30 | +import time |
| 31 | + |
| 32 | +from PIL import Image, ImageDraw, ImageFont |
| 33 | + |
| 34 | +from luma.core.interface.serial import i2c |
| 35 | +from luma.oled.device import sh1106, ssd1306 |
| 36 | + |
| 37 | +# The Phase 1 lab pair. With no --peer we watch the *other* node, so the same |
| 38 | +# image and service work unchanged on both cards. |
| 39 | +PAIR = {"10.10.0.1": "10.10.0.2", "10.10.0.2": "10.10.0.1"} |
| 40 | +DEFAULT_PEER = "10.10.0.2" |
| 41 | + |
| 42 | +# NUD states worth a one-glance read. Anything else is shown verbatim. |
| 43 | +KNOWN_STATES = { |
| 44 | + "REACHABLE", "STALE", "DELAY", "PROBE", |
| 45 | + "INCOMPLETE", "FAILED", "NOARP", "PERMANENT", "NONE", |
| 46 | +} |
| 47 | + |
| 48 | +# The Phase 1 BOM panel is a dual-colour 0.96" SSD1306: its top 16 pixel rows |
| 49 | +# emit yellow and the bottom 48 emit blue — fixed in the glass, not settable in |
| 50 | +# software. So we treat it as two bands: a yellow header strip (the peer label) |
| 51 | +# and the blue body (the live MAC line + big state). There's an unlit ~2px gap |
| 52 | +# at the seam, so nothing is drawn across it. Set to 0 for a single-colour panel. |
| 53 | +YELLOW_H = 16 |
| 54 | +# Top of the blue body's big-state area (a MAC line sits just above it). |
| 55 | +BODY_TOP = YELLOW_H + 14 |
| 56 | + |
| 57 | +# Monospace TrueType for the state readout, in preference order: JetBrains Mono |
| 58 | +# (ngrok's mono, from fonts-jetbrains-mono on the image), then DejaVu Sans Mono |
| 59 | +# as a fallback on stock systems. Bold first — heavier strokes survive 1-bit |
| 60 | +# rendering better. Without any of these we fall back to PIL's bitmap font. |
| 61 | +STATE_FONTS = ( |
| 62 | + "/usr/share/fonts/truetype/jetbrains-mono/JetBrainsMono-Bold.ttf", |
| 63 | + "/usr/share/fonts/truetype/jetbrains-mono/JetBrainsMono-Regular.ttf", |
| 64 | + "/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Bold.ttf", |
| 65 | + "/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf", |
| 66 | +) |
| 67 | +# Longest state we render; the auto-fit sizes to this so every state shares one |
| 68 | +# size (and a fixed-width font keeps them column-aligned too). |
| 69 | +WIDEST_STATE = "INCOMPLETE" |
| 70 | + |
| 71 | + |
| 72 | +def load_state_font(width, max_h, path=None, size=None): |
| 73 | + """Largest monospace TTF that fits the widest state, or None for fallback. |
| 74 | +
|
| 75 | + Tries `path` (else the bundled DejaVu Sans Mono). With an explicit `size`, |
| 76 | + loads at that size verbatim; otherwise picks the biggest size at which |
| 77 | + WIDEST_STATE still fits the panel. Returns None when no usable TTF exists, |
| 78 | + so the caller drops back to PIL's scaled bitmap font. |
| 79 | + """ |
| 80 | + for fp in ([path] if path else STATE_FONTS): |
| 81 | + if not fp or not os.path.exists(fp): |
| 82 | + continue |
| 83 | + if size: |
| 84 | + return ImageFont.truetype(fp, size) |
| 85 | + for s in range(max_h * 2, 6, -1): |
| 86 | + font = ImageFont.truetype(fp, s) |
| 87 | + b = font.getbbox(WIDEST_STATE) |
| 88 | + if b[2] - b[0] <= width - 4 and b[3] - b[1] <= max_h: |
| 89 | + return font |
| 90 | + return None |
| 91 | + |
| 92 | + |
| 93 | +def local_ipv4s(): |
| 94 | + """Set of this node's IPv4 addresses, for auto-picking the peer.""" |
| 95 | + try: |
| 96 | + out = subprocess.run( |
| 97 | + ["ip", "-json", "-4", "addr"], |
| 98 | + capture_output=True, text=True, check=True, |
| 99 | + ).stdout |
| 100 | + return { |
| 101 | + ai["local"] |
| 102 | + for iface in json.loads(out) |
| 103 | + for ai in iface.get("addr_info", []) |
| 104 | + if ai.get("family") == "inet" and "local" in ai |
| 105 | + } |
| 106 | + except (subprocess.SubprocessError, json.JSONDecodeError, OSError): |
| 107 | + return set() |
| 108 | + |
| 109 | + |
| 110 | +def pick_peer(): |
| 111 | + """Default peer: the other node in the pair, else DEFAULT_PEER.""" |
| 112 | + for ip in local_ipv4s(): |
| 113 | + if ip in PAIR: |
| 114 | + return PAIR[ip] |
| 115 | + return DEFAULT_PEER |
| 116 | + |
| 117 | + |
| 118 | +def read_neighbour(peer): |
| 119 | + """Return (state, mac, dev) for `peer` from the kernel neighbour cache. |
| 120 | +
|
| 121 | + state is "ABSENT" when the kernel holds no entry for the peer. mac/dev are |
| 122 | + None when the cache has no link-layer address yet (INCOMPLETE/FAILED). |
| 123 | + """ |
| 124 | + try: |
| 125 | + out = subprocess.run( |
| 126 | + ["ip", "-json", "neigh", "show", peer], |
| 127 | + capture_output=True, text=True, check=True, |
| 128 | + ).stdout |
| 129 | + entries = json.loads(out or "[]") |
| 130 | + except (subprocess.SubprocessError, json.JSONDecodeError, OSError): |
| 131 | + return "ERROR", None, None |
| 132 | + |
| 133 | + if not entries: |
| 134 | + return "ABSENT", None, None |
| 135 | + |
| 136 | + # Prefer an entry that actually has a link-layer address (a resolved one) |
| 137 | + # so a stray INCOMPLETE on another device doesn't mask the real state. |
| 138 | + entry = next((e for e in entries if e.get("lladdr")), entries[0]) |
| 139 | + states = entry.get("state") or ["NONE"] |
| 140 | + return " ".join(states), entry.get("lladdr"), entry.get("dev") |
| 141 | + |
| 142 | + |
| 143 | +def render(device, peer, state, mac, dev, beat, state_font=None): |
| 144 | + """Draw one frame across the panel's two colour bands (see YELLOW_H). |
| 145 | +
|
| 146 | + Yellow strip: the peer label. Blue body: the live MAC/dev line and the |
| 147 | + state in big type. Nothing crosses the seam at YELLOW_H. The state is drawn |
| 148 | + in `state_font` (a monospace TTF) when one is available, otherwise PIL's |
| 149 | + bitmap font scaled 2x. |
| 150 | + """ |
| 151 | + small = ImageFont.load_default() |
| 152 | + frame = Image.new("1", (device.width, device.height)) |
| 153 | + draw = ImageDraw.Draw(frame) |
| 154 | + draw.fontmode = "1" # no antialiasing — crisp edges on a 1-bit panel |
| 155 | + |
| 156 | + # Yellow band: who we're watching, plus a heartbeat so a steady screen |
| 157 | + # still reads as "running". |
| 158 | + draw.text((2, 2), f"PEER {peer}", fill=1, font=small) |
| 159 | + if beat: |
| 160 | + draw.rectangle((device.width - 3, 1, device.width - 1, 3), fill=1) |
| 161 | + |
| 162 | + # Blue band: live link-layer line, then the state centred in the body. |
| 163 | + draw.text((2, YELLOW_H + 2), |
| 164 | + f"{mac} {dev or ''}".rstrip() if mac else "(no MAC yet)", |
| 165 | + fill=1, font=small) |
| 166 | + |
| 167 | + avail = device.height - BODY_TOP |
| 168 | + if state_font is not None: |
| 169 | + # TrueType: draw at native size, centring the inked bbox. |
| 170 | + b = state_font.getbbox(state) |
| 171 | + w, h = b[2] - b[0], b[3] - b[1] |
| 172 | + x = max(0, (device.width - w) // 2) - b[0] |
| 173 | + y = BODY_TOP + (avail - h) // 2 - b[1] |
| 174 | + draw.text((x, y), state, fill=1, font=state_font) |
| 175 | + else: |
| 176 | + # Fallback: blow the bitmap font up 2x with nearest-neighbour. The |
| 177 | + # longest state (INCOMPLETE) is exactly 2x the panel width, so every |
| 178 | + # state lands at the same 2x — no odd-one-out sizing. |
| 179 | + bbox = small.getbbox(state) |
| 180 | + tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] |
| 181 | + if tw and th: |
| 182 | + glyph = Image.new("1", (tw, th)) |
| 183 | + ImageDraw.Draw(glyph).text((-bbox[0], -bbox[1]), state, fill=1, font=small) |
| 184 | + scale = 2 if tw * 2 <= device.width else 1 |
| 185 | + glyph = glyph.resize((tw * scale, th * scale), Image.NEAREST) |
| 186 | + gx = max(0, (device.width - glyph.width) // 2) |
| 187 | + gy = BODY_TOP + (avail - glyph.height) // 2 |
| 188 | + frame.paste(glyph, (gx, gy)) |
| 189 | + |
| 190 | + device.display(frame) |
| 191 | + |
| 192 | + |
| 193 | +def main(): |
| 194 | + p = argparse.ArgumentParser( |
| 195 | + description="Mirror the kernel ARP state for the peer node on the OLED.") |
| 196 | + p.add_argument("--peer", default=None, |
| 197 | + help="peer IP to watch (default: the other 10.10.0.x node)") |
| 198 | + p.add_argument("--interval", type=float, default=1.0, |
| 199 | + help="seconds between cache polls (default 1.0)") |
| 200 | + p.add_argument("--port", type=int, default=1, |
| 201 | + help="I2C bus (default 1 / /dev/i2c-1)") |
| 202 | + p.add_argument("--address", type=lambda x: int(x, 0), default=0x3C, |
| 203 | + help="I2C address (default 0x3C; some modules use 0x3D)") |
| 204 | + p.add_argument("--controller", choices=("ssd1306", "sh1106"), default="ssd1306", |
| 205 | + help="display controller (default ssd1306; try sh1106 if garbled)") |
| 206 | + p.add_argument("--font", default=None, |
| 207 | + help="path to a .ttf for the state text " |
| 208 | + "(default: JetBrains Mono, then DejaVu Sans Mono, else bitmap font)") |
| 209 | + p.add_argument("--font-size", type=int, default=None, |
| 210 | + help="state font size in px (default: auto-fit the longest state)") |
| 211 | + args = p.parse_args() |
| 212 | + |
| 213 | + peer = args.peer or pick_peer() |
| 214 | + |
| 215 | + try: |
| 216 | + serial = i2c(port=args.port, address=args.address) |
| 217 | + controller = sh1106 if args.controller == "sh1106" else ssd1306 |
| 218 | + device = controller(serial, width=128, height=64) |
| 219 | + except Exception as e: |
| 220 | + print(f"Could not open the display: {e}") |
| 221 | + print(f"Check `i2cdetect -y {args.port}` for the address, the wiring, " |
| 222 | + "and that I2C is enabled.") |
| 223 | + sys.exit(1) |
| 224 | + |
| 225 | + state_font = load_state_font(device.width, device.height - BODY_TOP, |
| 226 | + args.font, args.font_size) |
| 227 | + if state_font is None and (args.font or args.font_size): |
| 228 | + print("Requested font unavailable; using the built-in bitmap font.") |
| 229 | + |
| 230 | + print(f"Watching ARP state for {peer} (poll {args.interval}s). Ctrl-C to stop.") |
| 231 | + beat = False |
| 232 | + try: |
| 233 | + while True: |
| 234 | + state, mac, dev = read_neighbour(peer) |
| 235 | + render(device, peer, state, mac, dev, beat, state_font) |
| 236 | + beat = not beat |
| 237 | + time.sleep(args.interval) |
| 238 | + except KeyboardInterrupt: |
| 239 | + device.clear() |
| 240 | + print("\nDone.") |
| 241 | + |
| 242 | + |
| 243 | +if __name__ == "__main__": |
| 244 | + main() |
0 commit comments