wenil/clawdmeter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

v2 host: portable desktop app (tray + FastAPI + WebView2 panel), unified config, HA control

Browse Source 
Fold the Windows tray/daemon into one self-contained Clawdmeter.exe with a
settings UI, and wire the host side of the watch features:

- config.py: single %LOCALAPPDATA%\Clawdmeter\config.json (ha/buttons/settings),
  atomic writes, auto-migration from the old ha_config.json.
- server.py: local FastAPI (127.0.0.1:8723) — GET/PUT /api/config (token masked),
  POST /api/ha/test, GET /api/ha/entities, GET /api/status.
- web/index.html: brand-styled settings panel (Status/HA/Buttons/Settings tabs).
- panel.py: pywebview/WebView2 window, launched as its own process (pywebview and
  pystray both want the main thread); tray "Settings" opens it via --panel.
- daemon: HA command dispatch (toggle/bri/ct), dynamic button labels + index→
  action mapping, watch-battery low warning toast, and the dimmer "dim" snapshot
  (dimreq → light_snapshot of the first entity) so the watch dial seeds from HA.
- clawdmeter.spec / requirements: bundle fastapi+uvicorn+pywebview+webview backend.
- build-exe.ps1: ASCII-only (Windows PowerShell 5.1 mangles em-dashes under cp1251).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wenil
2026-07-09 20:20:44 +03:00
co-authored by Claude Opus 4.8
parent 578bc04248
commit 1c64386996
10 changed files with 977 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/claude_usage_daemon_windows.py
+195 -6
View File
@@ -32,6 +32,14 @@ DEVICE_ADDRESS = os.environ.get("CLAWDMETER_ADDRESS")
SERVICE_UUID = "4c41555a-4465-7669-6365-000000000001"
RX_CHAR_UUID = "4c41555a-4465-7669-6365-000000000002"
REQ_CHAR_UUID = "4c41555a-4465-7669-6365-000000000004"
CMD_CHAR_UUID = "4c41555a-4465-7669-6365-000000000005" # watch → host commands (Phase 6 HA)
# Phase 7 M2 — dynamic watch buttons. The daemon pushes up to this many button
# labels (each truncated) in the RX payload's "btns" array on the ~60s heartbeat;
# the watch renders a grid and reports only the pressed index. Bounds keep the
# merged BLE payload comfortably under the firmware's 512-byte RX buffer.
WATCH_MAX_BUTTONS = 6
WATCH_LABEL_MAX = 16
POLL_INTERVAL = 60 # Anthropic rate-limit / usage poll cadence (seconds)
NOWPLAYING_INTERVAL = 3 # Windows media-session read cadence (seconds). The
@@ -296,9 +304,58 @@ async def scan_for_device():
class Session:
def __init__(self, client: BleakClient) -> None:
def __init__(self, client: BleakClient, tray_state=None) -> None:
self.client = client
self.tray_state = tray_state
self.refresh_requested = asyncio.Event()
self.dim_requested = asyncio.Event() # watch opened the Dimmer screen → push a fresh light snapshot
self.ha = None # HAClient once a valid HA config is loaded
self.ha_entities: list = [] # controllable entity_ids from the config
self._loop = None # event loop captured for thread-safe BLE dispatch
self.low_bat_pct = 15 # warn at/below this watch battery %
self._battery_warned = False
self._last_logged_bat = None
self.buttons = [] # config buttons; list index == watch button index
def reload_buttons(self) -> None:
"""Re-read the config's button list so panel edits propagate to the watch
(called each poll). The list index is the index the watch reports back."""
try:
try:
from daemon.config import load_config
except ImportError:
from config import load_config
self.buttons = load_config().get("buttons") or []
except Exception as e:
log(f"Button reload failed: {e!r}")
def button_labels(self) -> list:
"""Compact label list for the RX 'btns' payload (count + length capped so
the merged BLE write stays under the firmware's 512-byte buffer)."""
out = []
for b in self.buttons[:WATCH_MAX_BUTTONS]:
lbl = str(b.get("label") or b.get("entity") or "Button").strip()
out.append(lbl[:WATCH_LABEL_MAX])
return out
async def dim_snapshot(self) -> dict | None:
"""Compact live state of the tilt-dimmer's light for the RX 'dim' object,
so the watch dial seeds from reality. The dimmer targets the first HA
entity (same default the bri/ct commands fall back to). Returns None when
HA is off or the read fails — the watch then just shows its last value."""
if not self.ha or not self.ha_entities:
return None
snap = await self.ha.light_snapshot(self.ha_entities[0])
if not snap:
return None
out = {"on": 1 if snap["on"] else 0,
"mink": int(snap["min_kelvin"]),
"maxk": int(snap["max_kelvin"])}
if snap["brightness_pct"] is not None:
out["bri"] = int(snap["brightness_pct"])
if snap["color_temp_kelvin"] is not None:
out["ct"] = int(snap["color_temp_kelvin"])
return out
def _on_refresh(self, _char, _data: bytearray) -> None:
log("Refresh requested by device")
@@ -316,6 +373,112 @@ class Session:
except (BleakError, ValueError, OSError) as e:
log(f"Refresh subscription unavailable: {e}")
async def setup_command_subscription(self) -> None:
# Watch → host channel (…0005): battery telemetry (always) plus HA commands
# (when HA is configured). We subscribe regardless of HA so the low-battery
# warning works even with Home Assistant unset. Imports are lazy so a
# missing config/dep can never break daemon startup, and so they resolve
# after `log` is defined (ha_client imports it).
self._loop = asyncio.get_running_loop() # thread-safe dispatch from the BLE callback
try:
try:
from daemon.ha_client import HAClient
from daemon.config import load_config, ha_settings
except ImportError:
from ha_client import HAClient
from config import load_config, ha_settings
cfg = load_config()
self.low_bat_pct = int(cfg.get("settings", {}).get("low_battery_pct", 15))
self.buttons = cfg.get("buttons") or []
ha = ha_settings(cfg)
if ha:
self.ha = HAClient(ha["url"], ha["token"])
self.ha_entities = ha["entities"]
except Exception as e:
log(f"Config/HA init failed ({e!r}); HA control disabled")
try:
await self.client.start_notify(CMD_CHAR_UUID, self._on_command)
log(f"Command channel ready (HA={'on' if self.ha else 'off'}, "
f"low-batt={self.low_bat_pct}%, entities={self.ha_entities})")
except (BleakError, ValueError, OSError) as e:
log(f"Command subscription unavailable: {e}")
def _on_command(self, _char, data: bytearray) -> None:
try:
payload = json.loads(bytes(data).decode("utf-8", "replace"))
except (ValueError, UnicodeDecodeError):
log(f"Watch msg: bad payload {bytes(data)!r}")
return
# Battery telemetry ({"bat":pct,"mv":..,"chg":0/1}) is handled inline (quick,
# no await). HA commands ({"cmd":...}) are dispatched onto the loop because
# bleak may deliver this callback on a non-loop thread.
if "bat" in payload:
self._handle_battery(payload)
return
# Dimmer screen opened (M3): ask the poll loop to push a fresh light
# snapshot on its next tick (~3s). asyncio.Event isn't thread-safe and
# this callback may run off-loop, so flip it via the loop.
if payload.get("cmd") == "dimreq":
loop = self._loop
if loop is not None:
loop.call_soon_threadsafe(self.dim_requested.set)
return
# Phase 7 M2: a watch button press carries only its index — map it to the
# configured action/entity/value here (the watch stays dumb).
if payload.get("cmd") == "btn":
i = payload.get("i")
if not isinstance(i, int) or not (0 <= i < len(self.buttons)):
log(f"Watch button {i}: out of range (have {len(self.buttons)})")
return
b = self.buttons[i]
payload = {"cmd": b.get("action") or "toggle", "e": b.get("entity")}
if b.get("value") is not None:
payload["v"] = b["value"]
log(f"Watch button {i} -> {payload}")
loop = self._loop
if loop is None:
return
asyncio.run_coroutine_threadsafe(self._dispatch_command(payload), loop)
def _handle_battery(self, payload: dict) -> None:
pct = payload.get("bat")
charging = bool(payload.get("chg", 0))
if not isinstance(pct, int):
return
if pct != self._last_logged_bat:
self._last_logged_bat = pct
log(f"Watch battery: {pct}%{' (charging)' if charging else ''}")
if self.tray_state is not None:
self.tray_state.battery_pct = pct
# Warn once on the way down; re-arm when charging or comfortably recovered.
if charging or pct > self.low_bat_pct + 10:
self._battery_warned = False
if not charging and pct <= self.low_bat_pct and not self._battery_warned:
self._battery_warned = True
log(f"Watch battery low: {pct}% (<= {self.low_bat_pct}%)")
if self.tray_state is not None:
self.tray_state.toasts.put(
("Clawdmeter", f"Watch battery low - {pct}%. Time to charge it."))
async def _dispatch_command(self, payload: dict) -> None:
if not self.ha:
return
cmd = payload.get("cmd")
entity = payload.get("e") or (self.ha_entities[0] if self.ha_entities else None)
if not entity:
log("HA cmd: no entity to target")
return
if cmd == "toggle":
ok = await self.ha.toggle(entity)
elif cmd == "bri": # step 3: brightness %, joystick
ok = await self.ha.set_brightness(entity, float(payload.get("v", 0)))
elif cmd == "ct": # step 3: color temperature, kelvin
ok = await self.ha.set_color_temp(entity, float(payload.get("v", 0)))
else:
log(f"HA cmd: unknown cmd {cmd!r}")
return
log(f"HA cmd {cmd} -> {'ok' if ok else 'FAIL'}")
async def write_payload(self, payload: dict) -> bool:
# ensure_ascii=False keeps non-Latin titles as compact UTF-8 (~2 bytes/char)
# instead of \uXXXX (6 bytes), which roughly thirds the size of a Cyrillic
@@ -661,8 +824,9 @@ async def connect_and_run(device, stop_event: asyncio.Event, tray_state=None) ->
return False
log("Connected")
session = Session(client)
session = Session(client, tray_state)
await session.setup_refresh_subscription()
await session.setup_command_subscription()
# Two cadences share one connection: the Anthropic usage / rate-limit poll runs
# every POLL_INTERVAL (60s) while the Windows media session is read every
@@ -680,10 +844,14 @@ async def connect_and_run(device, stop_event: asyncio.Event, tray_state=None) ->
now = time.time()
claude_due = (session.refresh_requested.is_set()
or (now - last_claude_poll) >= POLL_INTERVAL)
dim_due = session.dim_requested.is_set() # watch opened the Dimmer screen
if dim_due:
session.dim_requested.clear()
auth_problem = False
if claude_due:
session.refresh_requested.clear()
session.reload_buttons() # pick up desktop-panel edits to the button set
# Local token usage (Session screen) needs no network — compute it
# every cycle so the watch keeps updating even with no/expired token.
@@ -744,13 +912,29 @@ async def connect_and_run(device, stop_event: asyncio.Event, tray_state=None) ->
log(f"Now-playing skipped: {e!r}")
np = {"np": 0}
# Light snapshot for the tilt-dimmer dial — fetched only when the watch
# opens the Dimmer screen (dim_due), to seed the dial from reality. No
# per-heartbeat fetch: during a session the watch is authoritative (it
# streams absolute values), and it re-requests on every re-open. Best-
# effort — a slow/dead HA just omits the field, never stalls the loop.
dim_snap = None
if dim_due:
try:
dim_snap = await session.dim_snapshot()
except Exception as e:
log(f"Dim snapshot skipped: {e!r}")
# Write when the usage data was just refreshed (this is also the ~60s
# heartbeat) or when the track / playback state changed since the last
# write — skip otherwise so the link, and the field log, stay quiet
# between changes instead of repeating an identical payload every 3s.
if claude_due or np != last_np_sent:
# heartbeat), when the track / playback state changed since the last
# write, or when the watch asked for a dimmer snapshot — skip otherwise
# so the link, and the field log, stay quiet between changes.
if claude_due or dim_due or np != last_np_sent:
payload = dict(cached)
payload.update(np)
if claude_due:
payload["btns"] = session.button_labels() # ~60s heartbeat only
if dim_snap is not None:
payload["dim"] = dim_snap
if await session.write_payload(payload):
used_successfully = True
consecutive_failures = 0 # D-03: reset on success
@@ -790,6 +974,11 @@ async def connect_and_run(device, stop_event: asyncio.Event, tray_state=None) ->
await client.disconnect()
except (BleakError, OSError):
pass
if session.ha:
try:
await session.ha.aclose()
except Exception:
pass
log("Device disconnected" if not stop_event.is_set() else "Stopping")
return used_successfully