jc3248 bufo desk-pet — BLE host feeds: broadcast_time.py (connectionless time), usage_bridge.py (Claude usage over NUS connection), poke_time.py (one-off time sync)

broadcast_time.py

# /// script
# requires-python = ">=3.11"
# dependencies = ["bless>=0.2.6"]
# ///
"""broadcast_time.py — CONNECTIONLESS time broadcast to the jc3248 pet over BLE.

The pet is BLE-only and now SCANS (observer role) for a time beacon. This advertises
the host's NTP-synced time in the BLE *local name* — "clk:<epoch>:<tzoff>" — with no
connection. The pet reads it off the advertisement and sets its clock; it then ticks
locally, so a refresh every couple of minutes is plenty (corrects drift).

Why the name (not manufacturer data): macOS CoreBluetooth only lets you advertise a
LocalName + ServiceUUIDs, so the payload rides the name — same trick as
lab/beacon/ble-beacon.py.

Run alongside usage_bridge.py (which keeps the connection for the usage windows):
    uv run tools/broadcast_time.py
macOS: the terminal needs Bluetooth permission. Ctrl-C / SIGTERM to stop cleanly.
"""
import asyncio
import signal
import time

from bless import BlessServer

REFRESH = 120.0   # seconds between re-advertising with a fresh epoch


def time_name() -> str:
    now = int(time.time())
    lt = time.localtime(now)
    tzoff = -time.timezone + (3600 if lt.tm_isdst else 0)   # local UTC offset, secs
    return f"clk:{now}:{int(tzoff)}"


async def main() -> None:
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        try:
            loop.add_signal_handler(sig, stop.set)
        except NotImplementedError:
            pass

    print("broadcasting time (connectionless) — Ctrl-C to stop")
    while not stop.is_set():
        nm = time_name()
        server = BlessServer(name=nm)
        await server.start()
        print("→", nm)
        try:
            await asyncio.wait_for(stop.wait(), timeout=REFRESH)
        except asyncio.TimeoutError:
            pass            # refresh cycle: restart with a new epoch
        await server.stop()  # stop cleanly so macOS doesn't replay a stale advert
    print("stopped")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nbye")

poke_time.py

#!/usr/bin/env python3
"""poke_time.py — push the host's NTP-synced clock to the jc3248 pet over BLE.

The pet is BLE-only (no Wi-Fi), so it can't reach NTP itself. This connects to
the pet's Nordic UART Service (NUS) and writes a heartbeat line carrying the
current Unix epoch + local timezone offset; the pet stores it and ticks locally
off its own timer (so it keeps running between syncs). Re-sends every 60s to
correct drift.

Usage (no install needed):
    uv run --with bleak tools/poke_time.py              # auto-find Claude-*
    uv run --with bleak tools/poke_time.py Claude-ECC8  # by exact name
Or:  pip install bleak && python3 tools/poke_time.py

macOS: the terminal needs Bluetooth permission (System Settings > Privacy).
"""
import asyncio
import json
import sys
import time

from bleak import BleakScanner, BleakClient

NUS_RX = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"   # client -> device, WRITE
NAME_PREFIX = sys.argv[1] if len(sys.argv) > 1 else "Claude-"
PERIOD_S = 60


def heartbeat_line() -> bytes:
    """One newline-terminated JSON line the pet's ble.c parseLine() understands."""
    now = time.time()
    lt = time.localtime(now)
    # local offset from UTC in seconds (handles DST). GMT+7 -> 25200.
    tzoff = -time.timezone + (3600 if lt.tm_isdst else 0)
    payload = {"epoch": int(now), "tzoff": int(tzoff)}
    return (json.dumps(payload) + "\n").encode()


async def main() -> None:
    print(f"scanning for '{NAME_PREFIX}*' (10s)…")
    dev = await BleakScanner.find_device_by_filter(
        lambda d, ad: bool(d.name) and d.name.startswith(NAME_PREFIX),
        timeout=10.0,
    )
    if not dev:
        print(f"!! no device whose name starts with '{NAME_PREFIX}' — is the pet advertising?")
        return
    print(f"connecting {dev.name} ({dev.address})…")
    async with BleakClient(dev) as client:
        print("connected — sending time every", PERIOD_S, "s (Ctrl-C to stop)")
        while True:
            data = heartbeat_line()
            await client.write_gatt_char(NUS_RX, data, response=False)
            print("→", data.decode().strip())
            await asyncio.sleep(PERIOD_S)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nbye")

usage_bridge.py

#!/usr/bin/env python3
"""usage_bridge.py — feed the jc3248-pet REAL Claude usage over BLE.

No hardcoding: the numbers come straight from Claude Code. Claude Code writes a
statusline JSON every render that includes the live rate limits:

    "rate_limits": {
      "five_hour": { "used_percentage": 6,  "resets_at": <epoch> },
      "seven_day": { "used_percentage": 25, "resets_at": <epoch> }
    }

Our statusline command (~/.claude/statusline-command.sh) dumps that JSON to
/tmp/claude-statusline/raw.json. This bridge reads it, turns resets_at into
seconds-until-reset, and streams {pct5h,reset5h,pct7d,reset7d} to the pet's
Nordic UART every few seconds — so the pet HUD shows LIVE usage and the link
dot stays green. Reconnects automatically if the pet drops.

    /tmp/blevenv/bin/python tools/usage_bridge.py            # stream forever
    /tmp/blevenv/bin/python tools/usage_bridge.py --once     # send one snapshot and exit

macOS: the terminal running this needs Bluetooth permission.
"""
import asyncio, json, time, sys
from bleak import BleakScanner, BleakClient

RX     = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"   # NUS RX (write: us -> pet)
RAW    = "/tmp/claude-statusline/raw.json"        # Claude Code's statusline JSON dump
PERIOD = 8.0                                       # seconds between heartbeats (pet stale = 30s)
PREFIX = "Claude"


def build_snapshot():
    """Read Claude Code's live rate_limits → a pet heartbeat. None if unavailable."""
    try:
        d = json.load(open(RAW))
    except Exception:
        return None
    rl = d.get("rate_limits", {}) or {}
    fh = rl.get("five_hour", {}) or {}
    sd = rl.get("seven_day", {}) or {}
    now = int(time.time())
    lt  = time.localtime(now)
    tzoff = -time.timezone + (3600 if lt.tm_isdst else 0)   # local UTC offset (secs); GMT+7 → 25200
    # epoch/tzoff feed the IDF pet's top-strip clock (it's BLE-only, can't NTP itself)
    snap = {"total": 1, "running": 0, "waiting": 0, "epoch": now, "tzoff": int(tzoff)}
    if "used_percentage" in fh:
        snap["pct5h"] = int(fh["used_percentage"])
        if fh.get("resets_at"):
            snap["reset5h"] = max(0, int(fh["resets_at"]) - now)
    if "used_percentage" in sd:
        snap["pct7d"] = int(sd["used_percentage"])
        if sd.get("resets_at"):
            snap["reset7d"] = max(0, int(sd["resets_at"]) - now)
    cw = d.get("context_window", {}) or {}
    if "used_percentage" in cw:
        snap["msg"] = f"ctx {int(cw['used_percentage'])}%"
    # Always send: the heartbeat now carries the clock (epoch/tzoff) even before
    # rate-limit data is available, so the pet's time stays synced regardless.
    return snap


async def main():
    once = "--once" in sys.argv
    print("usage_bridge — streaming REAL Claude rate_limits to the pet")
    while True:
        print(f"scanning for '{PREFIX}*' ...")
        dev = await BleakScanner.find_device_by_filter(
            lambda d, a: (d.name or "").startswith(PREFIX), timeout=15.0)
        if not dev:
            print("no Claude-* device — pet powered & advertising? retrying...")
            if once: return
            await asyncio.sleep(3); continue
        try:
            async with BleakClient(dev) as c:
                print(f"connected to {dev.name} — link should go GREEN")
                while True:
                    snap = build_snapshot()
                    if snap:
                        await c.write_gatt_char(RX, (json.dumps(snap) + "\n").encode(), response=False)
                        print(f"  -> 5h {snap.get('pct5h','?')}%  7d {snap.get('pct7d','?')}%"
                              f"  reset5h={snap.get('reset5h','?')}s reset7d={snap.get('reset7d','?')}s")
                    else:
                        print(f"  (no rate_limits in {RAW} yet — render the statusline once)")
                    if once:
                        await asyncio.sleep(0.8); return
                    await asyncio.sleep(PERIOD)
        except Exception as e:
            print("link dropped:", e, "— reconnecting")
            if once: return
            await asyncio.sleep(2)


try:
    asyncio.run(main())
except KeyboardInterrupt:
    print("\nstopped.")