POS-Automation/docs/superpowers/plans/2026-06-07-ha-google-home-trigger.md

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

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:

"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:

    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

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:

"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:

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:

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):

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

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:

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

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:

#!/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:<msg>
"""
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

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:

[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:

# 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:

#!/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

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

ssh pve173 'pct exec 301 -- mosquitto_passwd -b /etc/mosquitto/passwd posbriefing "<NEW_PASSWORD>"'
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

cd /home/claude/bin/POS-Automation && ./install.sh
sudo sed -i 's/^MQTT_PASSWORD=.*/MQTT_PASSWORD=<NEW_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)

mosquitto_pub -h 192.168.0.154 -u posbriefing -P "<NEW_PASSWORD>" \
  -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:

# 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: <existing>
  service_account: !include <existing>
  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

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:

## 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:

- 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

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.