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) <noreply@anthropic.com>
This commit is contained in:
Mortdecai
2026-06-07 13:20:34 -04:00
parent fa580ec936
commit 7bfe862f53
4 changed files with 109 additions and 1 deletions
+3 -1
View File
@@ -12,5 +12,7 @@
"remote_host": "192.168.1.x", "remote_host": "192.168.1.x",
"remote_pass": "ssh-password", "remote_pass": "ssh-password",
"printer_ip": "192.168.1.x", "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"
} }
+1
View File
@@ -0,0 +1 @@
collect_ignore_glob = ["CREATE_PROJECT.md", "GITEA_API.md"]
+66
View File
@@ -7,6 +7,7 @@ import os
import re import re
import json import json
import socket import socket
import subprocess
import requests import requests
import yfinance as yf import yfinance as yf
from datetime import datetime from datetime import datetime
@@ -31,6 +32,14 @@ REMOTE_HOST = config["remote_host"]
REMOTE_PASS = config["remote_pass"] REMOTE_PASS = config["remote_pass"]
PRINTER_IP = config["printer_ip"] PRINTER_IP = config["printer_ip"]
PRINTER_PORT = int(config.get("printer_port", 9100)) 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") EASTERN_TZ = ZoneInfo("America/New_York")
# TM-m30 80mm paper, Font B — column width measured from test print # 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 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): def print_receipt(raw_bytes):
try: try:
with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=10) as sock: with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=10) as sock:
+39
View File
@@ -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)