From 7bfe862f53865fa300c21ca0ec980a21360af8af Mon Sep 17 00:00:00 2001 From: Mortdecai Date: Sun, 7 Jun 2026 13:20:34 -0400 Subject: [PATCH] feat: discover printer by MAC (survives DHCP IP changes) Add discover_printer(), _mac_of(), _port_open(), _sweep_arp(), _cache_ip() with PRINTER_MAC/PRINTER_SUBNET/CACHE_FILE config + unit tests. Also adds conftest.py to work around broken symlink that blocked pytest collection. Co-Authored-By: Claude Opus 4.8 (1M context) --- config.example.json | 4 +- conftest.py | 1 + pos_briefing.py | 66 +++++++++++++++++++++++++++++++++ tests/test_printer_discovery.py | 39 +++++++++++++++++++ 4 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 conftest.py create mode 100644 tests/test_printer_discovery.py diff --git a/config.example.json b/config.example.json index 8e47f69..ced6b0c 100644 --- a/config.example.json +++ b/config.example.json @@ -12,5 +12,7 @@ "remote_host": "192.168.1.x", "remote_pass": "ssh-password", "printer_ip": "192.168.1.x", - "printer_port": 9100 + "printer_port": 9100, + "printer_mac": "aa:bb:cc:dd:ee:ff", + "printer_subnet": "192.168.1" } diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..dcc9a40 --- /dev/null +++ b/conftest.py @@ -0,0 +1 @@ +collect_ignore_glob = ["CREATE_PROJECT.md", "GITEA_API.md"] diff --git a/pos_briefing.py b/pos_briefing.py index f165007..7ce7294 100644 --- a/pos_briefing.py +++ b/pos_briefing.py @@ -7,6 +7,7 @@ import os import re import json import socket +import subprocess import requests import yfinance as yf from datetime import datetime @@ -31,6 +32,14 @@ REMOTE_HOST = config["remote_host"] REMOTE_PASS = config["remote_pass"] PRINTER_IP = config["printer_ip"] PRINTER_PORT = int(config.get("printer_port", 9100)) + +from pathlib import Path + +PRINTER_MAC = config.get("printer_mac", "").lower() +PRINTER_SUBNET = config.get("printer_subnet", "192.168.0") +CACHE_DIR = Path.home() / ".cache" / "pos-briefing" +CACHE_FILE = CACHE_DIR / "last_ip" + EASTERN_TZ = ZoneInfo("America/New_York") # TM-m30 80mm paper, Font B — column width measured from test print @@ -437,6 +446,63 @@ def build_receipt(articles, zfs, reddit, grafana, weather, finance): return p.output +def _mac_of(ip): + subprocess.run(["ping", "-c1", "-W1", ip], capture_output=True) + out = subprocess.run(["ip", "neigh", "show", ip], + capture_output=True, text=True).stdout + m = re.search(r"lladdr ([0-9a-f:]{17})", out) + return m.group(1).lower() if m else None + + +def _port_open(ip, port, timeout=3): + try: + with socket.create_connection((ip, port), timeout=timeout): + return True + except OSError: + return False + + +def _sweep_arp(): + """Concurrent ping-prime the /24, then return {mac: ip} from the neigh table.""" + procs = [subprocess.Popen(["ping", "-c1", "-W1", f"{PRINTER_SUBNET}.{i}"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + for i in range(1, 255)] + for p in procs: + p.wait() + out = subprocess.run(["ip", "neigh", "show"], + capture_output=True, text=True).stdout + found = {} + for line in out.splitlines(): + m = re.match(r"([\d.]+).*lladdr ([0-9a-f:]{17})", line) + if m: + found[m.group(2).lower()] = m.group(1) + return found + + +def _cache_ip(ip): + CACHE_DIR.mkdir(parents=True, exist_ok=True) + CACHE_FILE.write_text(ip) + + +def discover_printer(): + """Return the printer's current IP, resolved by MAC. Raise if not found.""" + seeds = [] + if CACHE_FILE.exists(): + seeds.append(CACHE_FILE.read_text().strip()) + if PRINTER_IP: + seeds.append(PRINTER_IP) + for ip in seeds: + if ip and _mac_of(ip) == PRINTER_MAC and _port_open(ip, PRINTER_PORT): + _cache_ip(ip) + return ip + if PRINTER_MAC: + ip = _sweep_arp().get(PRINTER_MAC) + if ip and _port_open(ip, PRINTER_PORT): + _cache_ip(ip) + return ip + raise RuntimeError(f"printer-not-found mac={PRINTER_MAC}") + + def print_receipt(raw_bytes): try: with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=10) as sock: diff --git a/tests/test_printer_discovery.py b/tests/test_printer_discovery.py new file mode 100644 index 0000000..f6bd25d --- /dev/null +++ b/tests/test_printer_discovery.py @@ -0,0 +1,39 @@ +import os, sys +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import pos_briefing as pb + + +def test_cache_hit_returns_cached_ip(monkeypatch, tmp_path): + cache = tmp_path / "last_ip" + cache.write_text("192.168.0.50") + monkeypatch.setattr(pb, "CACHE_FILE", cache) + monkeypatch.setattr(pb, "PRINTER_MAC", "50:57:9c:eb:0f:07") + monkeypatch.setattr(pb, "_mac_of", lambda ip: "50:57:9c:eb:0f:07" if ip == "192.168.0.50" else None) + monkeypatch.setattr(pb, "_port_open", lambda ip, port, timeout=3: True) + assert pb.discover_printer() == "192.168.0.50" + + +def test_sweep_finds_printer_on_cache_miss(monkeypatch, tmp_path): + cache = tmp_path / "last_ip" + monkeypatch.setattr(pb, "CACHE_FILE", cache) + monkeypatch.setattr(pb, "PRINTER_IP", "192.168.0.99") # stale seed + monkeypatch.setattr(pb, "PRINTER_MAC", "50:57:9c:eb:0f:07") + monkeypatch.setattr(pb, "PRINTER_SUBNET", "192.168.0") + monkeypatch.setattr(pb, "_mac_of", lambda ip: None) # seed/cache miss + monkeypatch.setattr(pb, "_sweep_arp", lambda: {"50:57:9c:eb:0f:07": "192.168.0.184"}) + monkeypatch.setattr(pb, "_port_open", lambda ip, port, timeout=3: True) + assert pb.discover_printer() == "192.168.0.184" + assert cache.read_text().strip() == "192.168.0.184" + + +def test_not_found_raises(monkeypatch, tmp_path): + monkeypatch.setattr(pb, "CACHE_FILE", tmp_path / "last_ip") + monkeypatch.setattr(pb, "PRINTER_IP", "") + monkeypatch.setattr(pb, "PRINTER_MAC", "50:57:9c:eb:0f:07") + monkeypatch.setattr(pb, "_mac_of", lambda ip: None) + monkeypatch.setattr(pb, "_sweep_arp", lambda: {}) + try: + pb.discover_printer() + assert False, "expected RuntimeError" + except RuntimeError as e: + assert "printer-not-found" in str(e)