Small Models, Real Switches: A Local AI Home Assistant on a Raspberry Pi 4

# 1. Check the OS and GPIO access
uname -m                 # must print aarch64
ls /dev/gpiochip*
pinctrl set 17 op dh && pinctrl get 17   # expect "hi"
pinctrl set 17 op dl && pinctrl get 17   # expect "lo"

# 2. GPIO library (system packages)
sudo apt install -y python3-gpiozero python3-lgpio

# 3. Python environment for Needle, able to see system packages
python3 -m venv ~/needle-env
sed -i 's/^include-system-site-packages.*/include-system-site-packages = true/' ~/needle-env/pyvenv.cfg
source ~/needle-env/bin/activate
pip install cactus-needle
python -c "import gpiozero, needle; print('OK')"

# 4. Ollama and the writer model
curl -fsSL https://ollama.com/install.sh | sh
ollama pull qwen2.5:0.5b

# 5. Run
mkdir -p ~/home-assistant && cd ~/home-assistant
# save the scripts from Appendix B and C here, then:
python home_assistant.py              # v1, interactive
python home_assistant_v2.py --test    # v2, scored test
python home_assistant_v2.py           # v2, interactive

#!/usr/bin/env python3
"""Pi home assistant POC: Needle routes commands, Qwen (Ollama) writes summaries."""
import os
os.environ.setdefault("NEEDLE_TELEMETRY", "0")   # Needle's documented opt-out
os.environ.setdefault("DO_NOT_TRACK", "1")

import json
import readline  # noqa: F401  (arrow-key history at the prompt)
import subprocess
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Literal

import needle
from gpiozero import OutputDevice

BASE = Path(__file__).resolve().parent
MEMORY = BASE / "memory.md"
LOG = BASE / "log.jsonl"
OLLAMA = "http://localhost:11434"
ACT_THRESHOLD = 0.7
QWEN = None   # set at startup from the model installed in Ollama

# ---- Devices: the allowlist. Only these pins can ever be switched. ----
DEVICES = {"living room light": 17, "bedroom light": 27, "fan": 22}
ALIASES = {"living room": "living room light", "lounge light": "living room light",
           "bedroom": "bedroom light", "bedroom lamp": "bedroom light",
           "ceiling fan": "fan"}
outputs = {name: OutputDevice(pin, initial_value=False) for name, pin in DEVICES.items()}


def resolve(device):
    d = device.strip().lower()
    if d.startswith("the "):
        d = d[4:]
    d = ALIASES.get(d, d)
    return d if d in DEVICES else None


def level(pin):
    out = subprocess.run(["pinctrl", "get", str(pin)], capture_output=True, text=True).stdout
    return "hi" if "| hi" in out else ("lo" if "| lo" in out else "?")


def status_line(name):
    pin = DEVICES[name]
    return f"{name}: {'ON' if outputs[name].value else 'OFF'} (GPIO{pin} reads {level(pin)})"


# ---- Memory file ----
def ensure_memory():
    if not MEMORY.exists():
        MEMORY.write_text("# Home memory\n\n## Notes\n")


def read_notes():
    ensure_memory()
    return [line[2:] for line in MEMORY.read_text().splitlines() if line.startswith("- ")]


# ---- Qwen via Ollama ----
def pick_model():
    with urllib.request.urlopen(OLLAMA + "/api/tags", timeout=10) as resp:
        models = json.load(resp).get("models", [])
    return models[0]["name"] if models else None


def ask_qwen(prompt):
    payload = {"model": QWEN, "prompt": prompt, "stream": False,
               "options": {"num_predict": 160, "num_ctx": 2048, "temperature": 0.2}}
    req = urllib.request.Request(OLLAMA + "/api/generate", data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=300) as resp:
        d = json.load(resp)
    return d["response"].strip(), d["eval_count"] / (d["eval_duration"] / 1e9)


# ---- What each tool actually does ----
def do_set_device(device, state):
    name = resolve(device)
    if name is None:
        return f"REFUSED: unknown device '{device}'. Known: {', '.join(DEVICES)}"
    if state == "on":
        outputs[name].on()
    else:
        outputs[name].off()
    return status_line(name)


def do_device_status(device):
    if device.strip().lower() in ("all", "everything", "all devices", "home", "house"):
        return "\n    ".join(status_line(n) for n in DEVICES)
    name = resolve(device)
    if name is None:
        return f"REFUSED: unknown device '{device}'. Known: {', '.join(DEVICES)}"
    return status_line(name)


def do_add_note(text):
    text = text.strip()
    if not text:
        return "REFUSED: empty note"
    ensure_memory()
    stamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    with MEMORY.open("a") as f:
        f.write(f"- {stamp} | {text}\n")
    return f"Saved note: {text}"


def do_summarize_notes(period):
    notes = read_notes()
    if period == "today":
        today = datetime.now().strftime("%Y-%m-%d")
        notes = [n for n in notes if n.startswith(today)]
    if not notes:
        return "No notes to summarize."
    if QWEN is None:
        return "ERROR: no Ollama model available. Start it with: sudo systemctl start ollama"
    print(f"  (asking {QWEN} to summarize {len(notes)} notes...)")
    prompt = ("These are personal notes, one per line, each with a date. "
              "Summarize them in at most 3 short bullet points. "
              "Use only what is written in the notes.\n\n" + "\n".join(notes[-30:]))
    try:
        text, tps = ask_qwen(prompt)
    except urllib.error.URLError:
        return "ERROR: Ollama not reachable. Start it with: sudo systemctl start ollama"
    return f"Summary ({QWEN}, {tps:.1f} tok/s):\n{text}"


IMPL = {"set_device": do_set_device, "device_status": do_device_status,
        "add_note": do_add_note, "summarize_notes": do_summarize_notes}


# ---- What Needle sees (the docstring is the description) ----
@needle.tool
def set_device(device: str, state: Literal["on", "off"]):
    "Turn a home device on or off. Devices: living room light, bedroom light, fan. on = switch on, start. off = switch off, stop."
    return do_set_device(device, state)


@needle.tool
def device_status(device: str):
    "Check whether a home device is on or off. Use 'all' to check every device."
    return do_device_status(device)


@needle.tool
def add_note(text: str):
    "Save a personal note or reminder to memory. text is the note itself, e.g. 'buy milk'."
    return do_add_note(text)


@needle.tool
def summarize_notes(period: Literal["today", "all"]):
    "Summarize the saved notes, either today's notes or all notes."
    return do_summarize_notes(period)


TOOLS = [set_device, device_status, add_note, summarize_notes]
try:
    agent = needle.Needle(tools=TOOLS, stateless=True)
    STATELESS = True
except TypeError:              # older package without stateless: reset every turn instead
    agent = needle.Needle(tools=TOOLS)
    STATELESS = False


# ---- Main loop ----
def fmt(c):
    args = ", ".join(f"{k}={v!r}" for k, v in c.get("arguments", {}).items())
    return f"{c['name']}({args})"


def run_calls(calls):
    out = []
    for c in calls:
        fn = IMPL.get(c["name"])
        if fn is None:
            out.append(f"REFUSED: unknown tool '{c['name']}'")
            continue
        try:
            out.append(fn(**c.get("arguments", {})))
        except Exception as e:
            out.append(f"ERROR in {c['name']}: {e}")
    return out


def handle(text):
    if not STATELESS:
        agent.reset()
    t0 = time.time()
    r = agent.complete(text)
    ms = (time.time() - t0) * 1000
    calls = r.get("function_calls") or []
    held = r.get("suppressed_calls") or []
    conf = r.get("confidence", 0.0)
    shown = calls or held
    print(f"  [needle {ms:.0f} ms | conf {conf:.2f}] {', '.join(fmt(c) for c in shown) or 'no tool'}")
    print(f"  reasoning: {r.get('reasoning')}")

    decision, results = "none", []
    if calls and conf >= ACT_THRESHOLD:
        results, decision = run_calls(calls), "executed"
    elif shown:
        if input("  Not sure. Run it? [y/N] ").strip().lower() == "y":
            results, decision = run_calls(shown), "confirmed"
        else:
            decision = "declined"
    else:
        print("  I can't do that yet.")

    for line in results:
        print("  → " + line)
    with LOG.open("a") as f:
        f.write(json.dumps({"ts": datetime.now().isoformat(timespec="seconds"), "input": text,
                            "calls": calls, "suppressed": held, "confidence": conf,
                            "decision": decision, "latency_ms": round(ms),
                            "prefill_tps": r.get("prefill_tps"),
                            "decode_tps": r.get("decode_tps")}) + "\n")


def main():
    global QWEN
    ensure_memory()
    try:
        QWEN = pick_model()
    except urllib.error.URLError:
        QWEN = None
    print(f"Home assistant POC | Needle stateless={STATELESS} | writer: {QWEN or 'not available'}")
    print("Devices: " + ", ".join(f"{n} (GPIO{p})" for n, p in DEVICES.items()))
    print(f"Memory: {MEMORY}")
    print("Commands: /status  /notes  /help  quit")
    try:
        while True:
            text = input("\nyou> ").strip()
            if not text:
                continue
            if text.lower() in ("quit", "exit"):
                break
            if text == "/status":
                for n in DEVICES:
                    print("  " + status_line(n))
            elif text == "/notes":
                print(MEMORY.read_text())
            elif text == "/help":
                print("  Try: turn on the living room light | is the fan on | "
                      "note that I need to buy milk | summarize today's notes")
            else:
                handle(text)
    except (KeyboardInterrupt, EOFError):
        print()
    finally:
        for o in outputs.values():
            o.off()
            o.close()
        print("All devices off (safe state), bye")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Pi home assistant POC v2: Needle routes commands, Qwen (Ollama) writes summaries.

v2 changes vs v1:
  1. fixed device list in the tool schema (grammar-constrained names)
  2. one tool per action: turn_on, turn_off, check_device
  3. risk-based confidence thresholds (physical vs read-only)
  4. plain-language confirmation prompts
Run:  python home_assistant_v2.py          interactive
      python home_assistant_v2.py --test   scored routing test, no actions taken
"""
import os
os.environ.setdefault("NEEDLE_TELEMETRY", "0")   # Needle's documented opt-out
os.environ.setdefault("DO_NOT_TRACK", "1")

import json
import readline  # noqa: F401  (arrow-key history at the prompt)
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Literal, get_args

import needle
from gpiozero import OutputDevice

VERSION = "v2"
BASE = Path(__file__).resolve().parent
MEMORY = BASE / "memory.md"
LOG = BASE / "log.jsonl"
OLLAMA = "http://localhost:11434"
QWEN = None   # set at startup from the model installed in Ollama

# ---- Devices: names are for Needle, pin numbers are for the code only ----
DEVICES = {"living room light": 17, "bedroom light": 27, "fan": 22}
Device = Literal["living room light", "bedroom light", "fan"]
DeviceOrAll = Literal["living room light", "bedroom light", "fan", "all"]
assert set(get_args(Device)) == set(DEVICES), "Device list and DEVICES must match"
outputs = {name: OutputDevice(pin, initial_value=False) for name, pin in DEVICES.items()}

# ---- Risk-based thresholds: stricter when something physical happens ----
ACT_AT = {"turn_on": 0.7, "turn_off": 0.7,
          "check_device": 0.4, "add_note": 0.4, "summarize_notes": 0.4}
FLOOR = 0.3   # below this: don't act and don't ask


def level(pin):
    out = subprocess.run(["pinctrl", "get", str(pin)], capture_output=True, text=True).stdout
    return "hi" if "| hi" in out else ("lo" if "| lo" in out else "?")


def status_line(name):
    pin = DEVICES[name]
    return f"{name}: {'ON' if outputs[name].value else 'OFF'} (GPIO{pin} reads {level(pin)})"


# ---- Memory file ----
def ensure_memory():
    if not MEMORY.exists():
        MEMORY.write_text("# Home memory\n\n## Notes\n")


def read_notes():
    ensure_memory()
    return [line[2:] for line in MEMORY.read_text().splitlines() if line.startswith("- ")]


# ---- Qwen via Ollama ----
def pick_model():
    with urllib.request.urlopen(OLLAMA + "/api/tags", timeout=10) as resp:
        models = json.load(resp).get("models", [])
    return models[0]["name"] if models else None
def ask_qwen(prompt):
    payload = {"model": QWEN, "prompt": prompt, "stream": False,
               "options": {"num_predict": 160, "num_ctx": 2048, "temperature": 0.2}}
    req = urllib.request.Request(OLLAMA + "/api/generate", data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=300) as resp:
        d = json.load(resp)
    return d["response"].strip(), d["eval_count"] / (d["eval_duration"] / 1e9)


# ---- What each tool actually does ----
def switch(device, on):
    if device not in DEVICES:                       # second safety check after the grammar
        return f"REFUSED: unknown device '{device}'"
    if on:
        outputs[device].on()
    else:
        outputs[device].off()
    return status_line(device)


def do_turn_on(device):
    return switch(device, True)


def do_turn_off(device):
    return switch(device, False)


def do_check_device(device):
    if device == "all":
        return "\n    ".join(status_line(n) for n in DEVICES)
    if device not in DEVICES:
        return f"REFUSED: unknown device '{device}'"
    return status_line(device)
def do_add_note(text):
    text = text.strip()
    if not text:
        return "REFUSED: empty note"
    ensure_memory()
    stamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    with MEMORY.open("a") as f:
        f.write(f"- {stamp} | {text}\n")
    return f"Saved note: {text}"


def do_summarize_notes(period):
    notes = read_notes()
    if period == "today":
        today = datetime.now().strftime("%Y-%m-%d")
        notes = [n for n in notes if n.startswith(today)]
    if not notes:
        return "No notes to summarize."
    if QWEN is None:
        return "ERROR: no Ollama model available. Start it with: sudo systemctl start ollama"
    texts = [n.split(" | ", 1)[-1] for n in notes[-30:]]     # Qwen sees the notes, not timestamps
    print(f"  (asking {QWEN} to summarize {len(texts)} notes...)")
    prompt = ("Summarize these personal notes in at most 3 short bullet points. "
              "Use only the facts in the notes. Do not add dates, names or details.\n\n"
              + "\n".join(f"- {t}" for t in texts))
    try:
        text, tps = ask_qwen(prompt)
    except urllib.error.URLError:
        return "ERROR: Ollama not reachable. Start it with: sudo systemctl start ollama"
    return f"Summary ({QWEN}, {tps:.1f} tok/s):\n{text}"


IMPL = {"turn_on": do_turn_on, "turn_off": do_turn_off, "check_device": do_check_device,
        "add_note": do_add_note, "summarize_notes": do_summarize_notes}


# ---- What Needle sees (the docstring is the description) ----
@needle.tool
def turn_on(device: Device):
    "Switch a home device on. For commands like 'turn on the fan', 'start the fan', 'bedroom light on'."
    return do_turn_on(device)


@needle.tool
def turn_off(device: Device):
    "Switch a home device off. For commands like 'turn off the fan', 'stop the fan', 'bedroom light off'."
    return do_turn_off(device)


@needle.tool
def check_device(device: DeviceOrAll):
    "Report whether devices are on or off; never changes anything. For questions like 'is the fan on?' or 'what's on in the house?'. Use 'all' for every device."
    return do_check_device(device)


@needle.tool
def add_note(text: str):
    "Save a personal note or reminder to memory. text is the whole note including any day or time, e.g. 'buy milk tomorrow'."
    return do_add_note(text)


@needle.tool
def summarize_notes(period: Literal["today", "all"]):
    "Summarize the saved notes, either today's notes or all notes."
    return do_summarize_notes(period)


TOOLS = [turn_on, turn_off, check_device, add_note, summarize_notes]
try:
    agent = needle.Needle(tools=TOOLS, stateless=True)
    STATELESS = True
except TypeError:              # older package without stateless: reset every turn instead
    agent = needle.Needle(tools=TOOLS)
    STATELESS = False
# ---- Decision logic ----
def fmt(c):
    args = ", ".join(f"{k}={v!r}" for k, v in c.get("arguments", {}).items())
    return f"{c['name']}({args})"


def describe(c):
    a = c.get("arguments", {})
    n = c["name"]
    if n == "turn_on":
        return f"Turn ON the {a.get('device')}?"
    if n == "turn_off":
        return f"Turn OFF the {a.get('device')}?"
    if n == "check_device":
        return "Check all devices?" if a.get("device") == "all" else f"Check the {a.get('device')}?"
    if n == "add_note":
        return f"Save note \"{a.get('text')}\"?"
    if n == "summarize_notes":
        return f"Summarize {a.get('period')} notes?"
    return f"Run {fmt(c)}?"


def ask_needle(text):
    if not STATELESS:
        agent.reset()
    t0 = time.time()
    r = agent.complete(text)
    return r, (time.time() - t0) * 1000


def decide(r):
    calls = r.get("function_calls") or []
    held = r.get("suppressed_calls") or []
    conf = r.get("confidence", 0.0)
    cands = calls or held
    if not cands or conf < FLOOR:
        return [], [], "refuse", cands
    auto = [c for c in calls if conf >= ACT_AT.get(c["name"], 0.7)]
    ask = [c for c in cands if c not in auto]
    label = "act" if not ask else ("ask" if not auto else "mixed")
    return auto, ask, label, cands


def run_call(c):
    fn = IMPL.get(c["name"])
    if fn is None:
        return f"REFUSED: unknown tool '{c['name']}'"
    try:
        return fn(**c.get("arguments", {}))
    except Exception as e:
        return f"ERROR in {c['name']}: {e}"


def log_turn(text, r, ms, label, mode, executed=(), result=None):
    entry = {"ts": datetime.now().isoformat(timespec="seconds"), "version": VERSION,
             "mode": mode, "input": text, "calls": r.get("function_calls") or [],
             "suppressed": r.get("suppressed_calls") or [], "confidence": r.get("confidence", 0.0),
             "decision": label, "executed": [fmt(c) for c in executed], "test_result": result,
             "latency_ms": round(ms), "prefill_tps": r.get("prefill_tps"),
             "decode_tps": r.get("decode_tps")}
    with LOG.open("a") as f:
        f.write(json.dumps(entry, default=str) + "\n")


# ---- Interactive mode ----
def handle(text):
    r, ms = ask_needle(text)
    auto, ask, label, cands = decide(r)
    print(f"  [needle {ms:.0f} ms | conf {r.get('confidence', 0.0):.2f}] "
          f"{', '.join(fmt(c) for c in cands) or 'no tool'}")
    print(f"  reasoning: {r.get('reasoning')}")
    done = []
    if label == "refuse":
        print("  I can't do that yet." if not cands else "  Sorry, I'm not sure what you mean.")
    for c in auto:
        done.append(c)
        print("  → " + run_call(c))
    for c in ask:
        if input(f"  {describe(c)} [y/N] ").strip().lower() == "y":
            done.append(c)
            print("  → " + run_call(c))
        else:
            print("  → skipped")
    log_turn(text, r, ms, label, "live", executed=done)


def repl():
    print("Devices: " + ", ".join(f"{n} (GPIO{p})" for n, p in DEVICES.items()))
    print(f"Memory: {MEMORY}")
    print("Commands: /status  /notes  /help  quit")
    while True:
        text = input("\nyou> ").strip()
        if not text:
            continue
        if text.lower() in ("quit", "exit"):
            break
        if text == "/status":
            for n in DEVICES:
                print("  " + status_line(n))
        elif text == "/notes":
            print(MEMORY.read_text())
        elif text == "/help":
            print("  Try: turn on the fan | is the bedroom light on | "
                  "note that I need to buy milk tomorrow | summarize today's notes")
        else:
            handle(text)


# ---- Test mode: same 11 commands as the v1 run; nothing is executed ----
TESTS = [
    ("turn on the living room light", "turn_on", {"device": "living room light"}),
    ("switch off the living room light", "turn_off", {"device": "living room light"}),
    ("turn the fan on", "turn_on", {"device": "fan"}),
    ("is the bedroom light on", "check_device", {"device": "bedroom light"}),
    ("what's on in the house", "check_device", {"device": "all"}),
    ("note that I need to buy milk tomorrow", "add_note", {"text": ["milk", "tomorrow"]}),
    ("remember the plumber comes on Friday at 10", "add_note", {"text": ["plumber", "friday"]}),
    ("add a note: renew the car insurance", "add_note", {"text": ["insurance"]}),
    ("summarize today's notes", "summarize_notes", {"period": "today"}),
    ("turn on the kitchen light", None, {}),    # unknown device: must NOT act
    ("what's the weather", None, {}),           # no tool yet: must NOT act
]


def matches(c, tool, exp):
    if c["name"] != tool:
        return False
    args = c.get("arguments", {})
    for k, v in exp.items():
        got = str(args.get(k, "")).lower()
        if isinstance(v, list):
            if not all(w in got for w in v):
                return False
        elif got != v:
            return False
    return True


def run_tests():
    print(f"\n{'command':<38} {'needle call':<44} {'conf':>5} {'decide':<6} result")
    print("-" * 104)
    counts, lat = {"PASS": 0, "ASK": 0, "FAIL": 0}, []
    for text, tool, exp in TESTS:
        r, ms = ask_needle(text)
        lat.append(ms)
        auto, ask, label, cands = decide(r)
        if tool is None:
            result = "FAIL" if auto else ("ASK" if ask else "PASS")
        else:
            correct = bool(cands) and all(matches(c, tool, exp) for c in cands)
            result = "FAIL" if not correct or label == "refuse" else ("PASS" if label == "act" else "ASK")
        counts[result] += 1
        shown = ", ".join(fmt(c) for c in cands) or "-"
        print(f"{text[:38]:<38} {shown[:44]:<44} {r.get('confidence', 0.0):>5.2f} {label:<6} {result}")
        log_turn(text, r, ms, label, "test", result=result)
    n = len(TESTS)
    print("-" * 104)
    print(f"PASS {counts['PASS']}/{n} | ASK {counts['ASK']}/{n} | FAIL {counts['FAIL']}/{n} | "
          f"avg Needle latency {sum(lat) / n:.0f} ms")
    print("PASS = right action, no prompt | ASK = right action, asks first | "
          "FAIL = wrong tool or arguments, or acted when it should not")


def main():
    global QWEN
    ensure_memory()
    try:
        QWEN = pick_model()
    except urllib.error.URLError:
        QWEN = None
    print(f"Home assistant POC {VERSION} | Needle stateless={STATELESS} | writer: {QWEN or 'not available'}")
    try:
        if "--test" in sys.argv:
            run_tests()
        else:
            repl()
    except (KeyboardInterrupt, EOFError):
        print()
    finally:
        for o in outputs.values():
            o.off()
            o.close()
        print("All devices off (safe state), bye")


if __name__ == "__main__":
    main()