# POS Briefing — HA/Google Home Trigger 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:** Print the daily briefing on-demand from a Google Home button (no cron), via an MQTT bridge on steel141 that mirrors the lgtv pattern, with MAC-based printer discovery and gemma4 summaries. **Architecture:** Google Home → HA `script` (exposed as a Scene) → MQTT broker CT 301 → `pos-briefing-mqtt.service` (systemd, User=claude, steel141) → `pos_briefing.py` → printer discovered by MAC → ESC/POS over TCP. Status/availability published back to HA. **Tech Stack:** Python 3.13, `paho-mqtt`, `python-escpos`, `requests`, `yfinance`, systemd, Mosquitto, Home Assistant MQTT discovery, gemma4:26b on steel141 Ollama. **Spec:** `docs/superpowers/specs/2026-06-07-ha-google-home-trigger-design.md` --- ## File Structure - `pos_briefing.py` (modify) — printer discovery by MAC; wire into `print_receipt`; gemma4 model+prompt. - `config.json` / `config.example.json` (modify) — add `printer_mac`, `printer_subnet`; set `ollama_model`. - `pos_briefing_mqtt.py` (create) — MQTT bridge. - `pos-briefing-mqtt.service` (create) — systemd unit. - `pos-briefing.env.example` (create) — env template (`MQTT_PASSWORD`). - `install.sh` (create) — copy bridge/unit/env, daemon-reload, enable. - `tests/test_printer_discovery.py` (create) — unit tests for discovery. - `docs/ha-setup.md` (create) — Seth's HA script + Google expose steps. --- ## Task 1: Commit the in-tree weather rewrite (clean baseline) The working tree already has a verified hourly-weather rewrite in `pos_briefing.py` plus CONTEXT.md/SESSION.md edits. Lock it in before layering new work. **Files:** - Modify (already changed): `pos_briefing.py`, `CONTEXT.md`, `SESSION.md` - [ ] **Step 1: Confirm the pipeline still runs (print suppressed)** Run: `cd /home/claude/bin/POS-Automation && python3 scripts/dryrun.py` Expected: `OK weather: [...]`, `articles summarized: 5`, `print_receipt() suppressed — NNNN bytes`. - [ ] **Step 2: Commit** ```bash git add pos_briefing.py CONTEXT.md SESSION.md git commit -m "feat: hourly weather forecast + steel141 migration notes" gitea push ``` --- ## Task 2: Switch summaries to gemma4:26b (converged prompt) **Files:** - Modify: `config.json`, `config.example.json` - Modify: `pos_briefing.py:171-177` (the `prompt_summary` block) - [ ] **Step 1: Point the model at gemma4:26b** In `config.json` and `config.example.json`, set: ```json "ollama_model": "gemma4:26b" ``` (Both phase-1 selection and phase-2 summary use `OLLAMA_MODEL`.) - [ ] **Step 2: Replace the summary prompt with the converged one** In `pos_briefing.py`, replace the `prompt_summary = (...)` assignment (lines ~171-177) with: ```python prompt_summary = ( "You are a professional news editor writing a daily briefing for a " "receipt printer.\n" f"For EACH article, write a concise factual summary ({length}).\n" "No preamble, no opinion, do not write \"this article\".\n" "LANGUAGE RULE: Use English. Translate non-English/Spanish articles.\n" "Return STRICTLY a JSON list: [{\"id\": 0, \"summary\": \"...\"}, ...]\n\n" f"CONTENT:\n{content_input}" ) ``` (Converged via the `ask_gemma4` MCP against real FreshRSS articles 2026-06-07: gemma4:26b @ temp 0.3 produced clean 30-50-word factual summaries; the existing `re.search(r'\[.*\]', raw, re.DOTALL)` parser already tolerates ```json fences.) - [ ] **Step 3: Verify end-to-end (print suppressed)** Run: `python3 scripts/dryrun.py` Expected: `articles summarized: 5`; inspect printed summaries are factual, ~30-50 words. - [ ] **Step 4: Commit** ```bash git add config.json config.example.json pos_briefing.py git commit -m "feat: summarize with gemma4:26b using converged prompt" gitea push ``` --- ## Task 3: Printer discovery by MAC (unit-tested) Printer IP is DHCP-unstable. Resolve it by MAC `50:57:9c:eb:0f:07`, cache last-known IP, sweep on miss. **Files:** - Modify: `config.json`, `config.example.json` (add `printer_mac`, `printer_subnet`) - Modify: `pos_briefing.py` (new discovery functions + constants near `PRINTER_IP`) - Create: `tests/test_printer_discovery.py` - [ ] **Step 1: Add config keys** In `config.json` and `config.example.json` add: ```json "printer_mac": "50:57:9c:eb:0f:07", "printer_subnet": "192.168.0" ``` (`printer_ip` stays as a seed/fallback.) - [ ] **Step 2: Write the failing test** Create `tests/test_printer_discovery.py`: ```python 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) ``` - [ ] **Step 3: Run it to verify it fails** Run: `cd /home/claude/bin/POS-Automation && python3 -m pytest tests/test_printer_discovery.py -v` Expected: FAIL (`AttributeError: module 'pos_briefing' has no attribute 'discover_printer'`). (If pytest missing: `pip install --user --break-system-packages pytest`.) - [ ] **Step 4: Implement discovery** In `pos_briefing.py`, after the `PRINTER_PORT` line (~33) add: ```python 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" ``` Then add these functions (near `print_receipt`): ```python 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}") ``` - [ ] **Step 5: Run tests to verify they pass** Run: `python3 -m pytest tests/test_printer_discovery.py -v` Expected: 3 passed. - [ ] **Step 6: Live sanity check** Run: `python3 -c "import pos_briefing as pb; print(pb.discover_printer())"` Expected: prints `192.168.0.184` (current printer IP). - [ ] **Step 7: Commit** ```bash git add config.json config.example.json pos_briefing.py tests/test_printer_discovery.py git commit -m "feat: discover printer by MAC (survives DHCP IP changes)" gitea push ``` --- ## Task 4: Wire discovery into print_receipt **Files:** - Modify: `pos_briefing.py:438-446` (`print_receipt`) - [ ] **Step 1: Use discovered IP in print_receipt** Replace `print_receipt` with: ```python def print_receipt(raw_bytes): try: ip = discover_printer() except Exception as e: print(f"[!] Printer discovery failed: {e}") return False try: with socket.create_connection((ip, PRINTER_PORT), timeout=10) as sock: sock.sendall(raw_bytes) print(f"✓ Receipt sent to {ip}:{PRINTER_PORT}") return True except Exception as e: print(f"[!] Print failed: {e}") return False ``` - [ ] **Step 2: Real print test (printer is on at .184)** Run: `python3 pos_briefing.py` Expected: `✓ Receipt sent to 192.168.0.184:9100` and a physical receipt. (This is the first real print — confirm paper output looks right: weather block, gemma4 summaries.) - [ ] **Step 3: Commit** ```bash git add pos_briefing.py git commit -m "feat: print to discovered printer IP" gitea push ``` --- ## Task 5: MQTT bridge script **Files:** - Create: `pos_briefing_mqtt.py` - [ ] **Step 1: Write the bridge** Create `pos_briefing_mqtt.py`: ```python #!/usr/bin/env python3 """POS Briefing MQTT bridge for Home Assistant. Runs on steel141 as user `claude` via systemd. Subscribes to a command topic; on PRESS, runs pos_briefing.py as a subprocess and prints the daily briefing. Registers an HA button via MQTT discovery and reports status/availability. Topics: homeassistant/button/pos_briefing/config -- discovery (retained, on connect) posbriefing/availability -- online / offline (LWT) posbriefing/button/press -- receives: PRESS posbriefing/status -- idle / printing / ok / error: """ import os import json import logging import subprocess import threading import paho.mqtt.client as mqtt MQTT_HOST = os.environ.get("MQTT_HOST", "192.168.0.154") MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883")) MQTT_USER = os.environ.get("MQTT_USER", "posbriefing") MQTT_PASS = os.environ.get("MQTT_PASSWORD", "") BRIEFING = "/home/claude/bin/POS-Automation/pos_briefing.py" PRINT_TIMEOUT = 180 DISCOVERY_TOPIC = "homeassistant/button/pos_briefing/config" AVAIL_TOPIC = "posbriefing/availability" COMMAND_TOPIC = "posbriefing/button/press" STATUS_TOPIC = "posbriefing/status" logging.basicConfig(level=logging.INFO, format="%(asctime)s [pos-mqtt] %(levelname)s %(message)s") log = logging.getLogger(__name__) _busy = threading.Lock() def publish_discovery(client): payload = { "name": "Print Daily Briefing", "unique_id": "pos_briefing_button", "command_topic": COMMAND_TOPIC, "payload_press": "PRESS", "availability_topic": AVAIL_TOPIC, "payload_available": "online", "payload_not_available": "offline", "device": { "identifiers": ["pos_briefing"], "name": "POS Briefing", "manufacturer": "Seth", "model": "TM-m30 briefing", }, "icon": "mdi:printer-pos", } client.publish(DISCOVERY_TOPIC, json.dumps(payload), retain=True) client.publish(AVAIL_TOPIC, "online", retain=True) client.publish(STATUS_TOPIC, "idle", retain=True) log.info("Published MQTT discovery payload") def run_briefing(client): if not _busy.acquire(blocking=False): log.warning("Briefing already running; ignoring press") return try: client.publish(STATUS_TOPIC, "printing", retain=True) log.info("Running briefing...") r = subprocess.run(["/usr/bin/python3", BRIEFING], capture_output=True, text=True, timeout=PRINT_TIMEOUT) if r.returncode == 0 and "Receipt sent" in r.stdout: client.publish(STATUS_TOPIC, "ok", retain=True) log.info("Briefing printed OK") else: msg = (r.stdout + r.stderr).strip().splitlines()[-1:] or ["nonzero"] client.publish(STATUS_TOPIC, f"error:{msg[0][:80]}", retain=True) log.error(f"Briefing failed rc={r.returncode}: {msg}") except subprocess.TimeoutExpired: client.publish(STATUS_TOPIC, "error:timeout", retain=True) log.error("Briefing timed out") finally: _busy.release() def on_connect(client, userdata, flags, rc): if rc == 0: log.info("Connected to MQTT broker") publish_discovery(client) client.subscribe(COMMAND_TOPIC) log.info(f"Subscribed to {COMMAND_TOPIC}") else: log.error(f"MQTT connect failed: rc={rc}") def on_message(client, userdata, msg): cmd = msg.payload.decode().strip().upper() log.info(f"Received: {cmd}") if cmd == "PRESS": threading.Thread(target=run_briefing, args=(client,), daemon=True).start() else: log.warning(f"Unknown command: {cmd}") def main(): client = mqtt.Client(client_id="pos-briefing-bridge") client.username_pw_set(MQTT_USER, MQTT_PASS) client.on_connect = on_connect client.on_message = on_message client.will_set(AVAIL_TOPIC, "offline", retain=True) log.info(f"Connecting to {MQTT_HOST}:{MQTT_PORT}...") client.connect(MQTT_HOST, MQTT_PORT, keepalive=60) try: client.loop_forever() finally: client.publish(AVAIL_TOPIC, "offline", retain=True) client.disconnect() if __name__ == "__main__": main() ``` - [ ] **Step 2: Syntax check** Run: `python3 -m py_compile pos_briefing_mqtt.py && echo OK` Expected: `OK` - [ ] **Step 3: Commit** ```bash git add pos_briefing_mqtt.py git commit -m "feat: MQTT bridge exposing an HA button to print the briefing" gitea push ``` --- ## Task 6: systemd unit, env template, installer **Files:** - Create: `pos-briefing-mqtt.service`, `pos-briefing.env.example`, `install.sh` - Modify: `.gitignore` (ignore real `pos-briefing.env` if ever placed locally) - [ ] **Step 1: Create the unit** `pos-briefing-mqtt.service`: ```ini [Unit] Description=POS Briefing MQTT bridge for Home Assistant Documentation=https://git.sethpc.xyz/Seth/POS-Automation After=network-online.target Wants=network-online.target [Service] Type=simple User=claude # MQTT_PASSWORD lives here, root-owned 0600, never committed. EnvironmentFile=/etc/pos-briefing/pos-briefing.env ExecStart=/usr/bin/python3 /usr/local/bin/pos_briefing_mqtt.py Restart=on-failure RestartSec=10 [Install] WantedBy=multi-user.target ``` - [ ] **Step 2: Create the env template** `pos-briefing.env.example`: ```bash # Copy to /etc/pos-briefing/pos-briefing.env (root:root 0600). Never commit the real file. MQTT_HOST=192.168.0.154 MQTT_PORT=1883 MQTT_USER=posbriefing MQTT_PASSWORD=CHANGEME ``` - [ ] **Step 3: Create the installer** `install.sh`: ```bash #!/usr/bin/env bash set -euo pipefail HERE="$(cd "$(dirname "$0")" && pwd)" sudo install -m 0755 "$HERE/pos_briefing_mqtt.py" /usr/local/bin/pos_briefing_mqtt.py sudo install -d -m 0755 /etc/pos-briefing if [ ! -f /etc/pos-briefing/pos-briefing.env ]; then sudo install -m 0600 "$HERE/pos-briefing.env.example" /etc/pos-briefing/pos-briefing.env echo "!! Edit /etc/pos-briefing/pos-briefing.env and set MQTT_PASSWORD" fi sudo install -m 0644 "$HERE/pos-briefing-mqtt.service" /etc/systemd/system/pos-briefing-mqtt.service sudo systemctl daemon-reload sudo systemctl enable pos-briefing-mqtt.service echo "Installed. Start with: sudo systemctl start pos-briefing-mqtt.service" ``` - [ ] **Step 4: Make executable + commit** ```bash chmod +x install.sh git add pos-briefing-mqtt.service pos-briefing.env.example install.sh git commit -m "feat: systemd unit + env template + installer for MQTT bridge" gitea push ``` --- ## Task 7: Create broker user, install, and trigger over MQTT **Files:** none (operational) - [ ] **Step 1: Create the `posbriefing` Mosquitto user on CT 301** ```bash ssh pve173 'pct exec 301 -- mosquitto_passwd -b /etc/mosquitto/passwd posbriefing ""' ssh pve173 'pct exec 301 -- systemctl reload mosquitto' ``` (Confirm broker host/CT: 192.168.0.154 is CT 301. If passwd path differs, check `/etc/mosquitto/mosquitto.conf` `password_file`. Use a generated password.) - [ ] **Step 2: Put the password in the env file and install** ```bash cd /home/claude/bin/POS-Automation && ./install.sh sudo sed -i 's/^MQTT_PASSWORD=.*/MQTT_PASSWORD=/' /etc/pos-briefing/pos-briefing.env sudo systemctl start pos-briefing-mqtt.service sudo systemctl status pos-briefing-mqtt.service --no-pager ``` Expected: `active (running)`, log line `Published MQTT discovery payload`. - [ ] **Step 3: Trigger a print over MQTT (no Google needed yet)** ```bash mosquitto_pub -h 192.168.0.154 -u posbriefing -P "" \ -t posbriefing/button/press -m PRESS journalctl -u pos-briefing-mqtt.service -n 20 --no-pager ``` Expected: bridge logs `Running briefing...` → `Briefing printed OK`; a physical receipt prints; `posbriefing/status` goes `printing` → `ok`. - [ ] **Step 4: Confirm the HA entity appeared** In HA → Settings → Devices → MQTT, confirm device **POS Briefing** with button **Print Daily Briefing**. (No commit — operational.) --- ## Task 8: HA script + Google Home exposure (Seth's 2-minute paste) **Files:** - Create: `docs/ha-setup.md` - [ ] **Step 1: Write the HA setup doc** Create `docs/ha-setup.md`: ````markdown # HA + Google Home setup — POS Briefing button The MQTT bridge auto-creates `button.print_daily_briefing` via discovery. Google's `google_assistant` integration does not support `button` entities, so wrap it in a script (scripts expose as a Google **Scene** = a tappable button). ## 1. Add the script (Settings → Automations & Scenes → Scripts → edit in YAML), or paste into `scripts.yaml`: ```yaml print_daily_briefing: alias: Print Daily Briefing sequence: - service: button.press target: entity_id: button.print_daily_briefing mode: single ``` ## 2. Expose it to Google (same `google_assistant:` block that exposes the LG TV): ```yaml google_assistant: project_id: service_account: !include report_state: true exposed_domains: - switch entity_config: script.print_daily_briefing: name: Print Daily Briefing expose: true ``` (Per lgtv DECISIONS 2026-06-06: manual integration, account-linking scopes must include `email`+`name`, "HTTP basic auth header" OFF, Test activated by slingshooter08. The GCP OAuth consent screen is unrelated — don't touch it.) ## 3. Re-sync: say "Hey Google, sync my devices" or in the Home app re-run setup. The button appears as **Print Daily Briefing**; tapping it prints. Optionally show `sensor`-style status by adding an MQTT sensor on `posbriefing/status`. ```` - [ ] **Step 2: Commit** ```bash git add docs/ha-setup.md git commit -m "docs: HA script + Google Home exposure steps" gitea push ``` - [ ] **Step 3: Hand to Seth** Tell Seth: paste the script + `entity_config` from `docs/ha-setup.md`, reload, and say "Hey Google, sync my devices." Then tap the button to verify end-to-end. --- ## Task 9: Cleanup + persistence **Files:** - Modify: `CONTEXT.md`, `SESSION.md`, `DECISIONS.md` (create if absent) - [ ] **Step 1: Remove the stale cron section from CONTEXT.md** Delete the `## Cron` block (the `30 4 * * *` line) and replace with: ```markdown ## Trigger On-demand only. Google Home button → HA script → MQTT (CT 301) → pos-briefing-mqtt.service on steel141 → pos_briefing.py. No cron. See docs/superpowers/specs/2026-06-07-ha-google-home-trigger-design.md. ``` - [ ] **Step 2: Record decisions** Append to (or create) `DECISIONS.md`: ```markdown - 2026-06-07: On-demand trigger via MQTT bridge (lgtv pattern), not cron — Seth wants a Google Home button. Bridge: pos-briefing-mqtt.service, User=claude. - 2026-06-07: Printer discovered by MAC 50:57:9c:eb:0f:07 — DHCP IP is unstable (no reservation). Cache last IP, sweep /24 on miss. - 2026-06-07: Summaries via gemma4:26b (faster than 31b; quality verified via ask_gemma4 MCP convergence). - 2026-06-07: yfinance installed on steel141 (pip --user --break-system-packages). ``` - [ ] **Step 3: Commit** ```bash git add CONTEXT.md SESSION.md DECISIONS.md git commit -m "docs: retire cron, record MQTT/discovery/gemma4 decisions" gitea push ``` --- ## Self-Review - **Spec coverage:** MQTT bridge (T5-7), printer discovery (T3-4), gemma4 (T2), HA/Google (T8), deps (yfinance done; pytest/paho noted), weather rewrite (T1), cleanup (T9). All spec sections mapped. - **Open items resolved:** broker user = dedicated `posbriefing` (T7); model = `gemma4:26b` (T2). - **Type/name consistency:** `discover_printer`, `_mac_of`, `_port_open`, `_sweep_arp`, `_cache_ip`, `CACHE_FILE`, `PRINTER_MAC`, `PRINTER_SUBNET` used consistently across T3/T4 and the tests. Topics consistent across T5/T7/T8. - **Out of scope (unchanged):** zfs/reddit/grafana = None on steel141.