512039e740
Cherry-picks the macOS-specific pieces from PR #5: Python daemon using bleak (CoreBluetooth backend), LaunchAgent template, flash-mac.sh, and install-mac.sh. Token is read from the macOS Keychain ("Claude Code-credentials" service); Linux behavior is unchanged. README split into "macOS installation" and "Linux installation" sections with parallel Flash / Pair / Install subsections. Co-Authored-By: Chris Davidson <36679917+lorddavidson@users.noreply.github.com> Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
332 lines
10 KiB
Python
Executable File
332 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Claude Usage Tracker Daemon (BLE) — macOS port of claude-usage-daemon.sh.
|
|
|
|
Polls Claude API rate-limit headers and writes a JSON payload to the
|
|
ESP32 "Claude Controller" peripheral over a custom GATT service. Uses
|
|
bleak (CoreBluetooth backend on macOS).
|
|
"""
|
|
|
|
import asyncio
|
|
import getpass
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from bleak import BleakClient, BleakScanner
|
|
from bleak.exc import BleakError
|
|
|
|
DEVICE_NAME = "Claude Controller"
|
|
SERVICE_UUID = "4c41555a-4465-7669-6365-000000000001"
|
|
RX_CHAR_UUID = "4c41555a-4465-7669-6365-000000000002"
|
|
REQ_CHAR_UUID = "4c41555a-4465-7669-6365-000000000004"
|
|
|
|
POLL_INTERVAL = 60
|
|
TICK = 5
|
|
SCAN_TIMEOUT = 8.0
|
|
|
|
# macOS: token lives in Keychain (service "Claude Code-credentials").
|
|
# Linux: token lives in ~/.claude/.credentials.json.
|
|
KEYCHAIN_SERVICE = "Claude Code-credentials"
|
|
CREDENTIALS_PATH = Path.home() / ".claude" / ".credentials.json"
|
|
SAVED_ADDR_FILE = Path.home() / ".config" / "claude-usage-monitor" / "ble-address"
|
|
|
|
API_URL = "https://api.anthropic.com/v1/messages"
|
|
API_HEADERS_TEMPLATE = {
|
|
"anthropic-version": "2023-06-01",
|
|
"anthropic-beta": "oauth-2025-04-20",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "claude-code/2.1.5",
|
|
}
|
|
API_BODY = {
|
|
"model": "claude-haiku-4-5-20251001",
|
|
"max_tokens": 1,
|
|
"messages": [{"role": "user", "content": "hi"}],
|
|
}
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
|
|
|
|
|
|
def _extract_access_token(blob: str) -> str | None:
|
|
"""Pull the accessToken out of a credentials blob.
|
|
|
|
Claude Code stores credentials as a JSON object; the blob may also be
|
|
nested ({"claudeAiOauth": {"accessToken": "..."}}). Fall back to a
|
|
regex match so unexpected shapes still work, and finally treat the
|
|
blob as a raw token if nothing else matches.
|
|
"""
|
|
blob = blob.strip()
|
|
if not blob:
|
|
return None
|
|
try:
|
|
data = json.loads(blob)
|
|
except json.JSONDecodeError:
|
|
data = None
|
|
if isinstance(data, dict):
|
|
# direct: {"accessToken": "..."}
|
|
if isinstance(data.get("accessToken"), str):
|
|
return data["accessToken"]
|
|
# nested: {"claudeAiOauth": {"accessToken": "..."}}
|
|
for v in data.values():
|
|
if isinstance(v, dict) and isinstance(v.get("accessToken"), str):
|
|
return v["accessToken"]
|
|
m = re.search(r'"accessToken"\s*:\s*"([^"]+)"', blob)
|
|
if m:
|
|
return m.group(1)
|
|
# Raw token (no JSON wrapper) — must look plausible (sk-ant-... etc.)
|
|
if re.fullmatch(r"[A-Za-z0-9_\-.~+/=]{20,}", blob):
|
|
return blob
|
|
return None
|
|
|
|
|
|
def _read_token_keychain() -> str | None:
|
|
try:
|
|
out = subprocess.run(
|
|
[
|
|
"security",
|
|
"find-generic-password",
|
|
"-s",
|
|
KEYCHAIN_SERVICE,
|
|
"-a",
|
|
getpass.getuser(),
|
|
"-w",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
log(f"Keychain read failed (rc={e.returncode}): {e.stderr.strip()}")
|
|
return None
|
|
except (FileNotFoundError, subprocess.TimeoutExpired) as e:
|
|
log(f"Keychain access error: {e}")
|
|
return None
|
|
return _extract_access_token(out.stdout)
|
|
|
|
|
|
def _read_token_file() -> str | None:
|
|
try:
|
|
raw = CREDENTIALS_PATH.read_text()
|
|
except OSError as e:
|
|
log(f"Error reading credentials: {e}")
|
|
return None
|
|
return _extract_access_token(raw)
|
|
|
|
|
|
def read_token() -> str | None:
|
|
if sys.platform == "darwin":
|
|
return _read_token_keychain()
|
|
return _read_token_file()
|
|
|
|
|
|
def load_cached_address() -> str | None:
|
|
if not SAVED_ADDR_FILE.exists():
|
|
return None
|
|
addr = SAVED_ADDR_FILE.read_text().strip()
|
|
# Accept both Linux MAC (AA:BB:CC:DD:EE:FF) and macOS CoreBluetooth UUID
|
|
# (E621E1F8-C36C-495A-93FC-0C247A3E6E5F).
|
|
if re.fullmatch(r"(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}", addr) or re.fullmatch(
|
|
r"[0-9A-Fa-f]{8}-(?:[0-9A-Fa-f]{4}-){3}[0-9A-Fa-f]{12}", addr
|
|
):
|
|
return addr
|
|
log("Cached address malformed, discarding")
|
|
SAVED_ADDR_FILE.unlink(missing_ok=True)
|
|
return None
|
|
|
|
|
|
def save_address(addr: str) -> None:
|
|
SAVED_ADDR_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
SAVED_ADDR_FILE.write_text(addr)
|
|
|
|
|
|
async def scan_for_device() -> str | None:
|
|
log(f"Scanning for '{DEVICE_NAME}' ({SCAN_TIMEOUT}s)...")
|
|
devices = await BleakScanner.discover(timeout=SCAN_TIMEOUT)
|
|
for d in devices:
|
|
if d.name == DEVICE_NAME:
|
|
log(f"Found: {d.address}")
|
|
return d.address
|
|
return None
|
|
|
|
|
|
async def poll_api(token: str) -> dict | None:
|
|
headers = dict(API_HEADERS_TEMPLATE)
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=20.0) as http:
|
|
resp = await http.post(API_URL, headers=headers, json=API_BODY)
|
|
except httpx.HTTPError as e:
|
|
log(f"API call failed: {e}")
|
|
return None
|
|
|
|
def hdr(name: str, default: str = "0") -> str:
|
|
return resp.headers.get(name, default)
|
|
|
|
now = time.time()
|
|
|
|
def reset_minutes(reset_ts: str) -> int:
|
|
try:
|
|
r = float(reset_ts)
|
|
except ValueError:
|
|
return 0
|
|
mins = (r - now) / 60.0
|
|
return int(round(mins)) if mins > 0 else 0
|
|
|
|
def pct(util: str) -> int:
|
|
try:
|
|
return int(round(float(util) * 100))
|
|
except ValueError:
|
|
return 0
|
|
|
|
payload = {
|
|
"s": pct(hdr("anthropic-ratelimit-unified-5h-utilization")),
|
|
"sr": reset_minutes(hdr("anthropic-ratelimit-unified-5h-reset")),
|
|
"w": pct(hdr("anthropic-ratelimit-unified-7d-utilization")),
|
|
"wr": reset_minutes(hdr("anthropic-ratelimit-unified-7d-reset")),
|
|
"st": hdr("anthropic-ratelimit-unified-5h-status", "unknown"),
|
|
"ok": True,
|
|
}
|
|
return payload
|
|
|
|
|
|
class Session:
|
|
def __init__(self, client: BleakClient) -> None:
|
|
self.client = client
|
|
self.refresh_requested = asyncio.Event()
|
|
|
|
def _on_refresh(self, _char, _data: bytearray) -> None:
|
|
log("Refresh requested by device")
|
|
self.refresh_requested.set()
|
|
|
|
async def setup_refresh_subscription(self) -> None:
|
|
try:
|
|
await self.client.start_notify(REQ_CHAR_UUID, self._on_refresh)
|
|
except (BleakError, ValueError) as e:
|
|
log(f"Refresh subscription unavailable: {e}")
|
|
|
|
async def write_payload(self, payload: dict) -> bool:
|
|
data = json.dumps(payload, separators=(",", ":")).encode()
|
|
log(f"Sending: {data.decode()}")
|
|
try:
|
|
await self.client.write_gatt_char(RX_CHAR_UUID, data, response=False)
|
|
return True
|
|
except BleakError as e:
|
|
log(f"Write failed: {e}")
|
|
return False
|
|
|
|
|
|
async def connect_and_run(address: str, stop_event: asyncio.Event) -> bool:
|
|
"""Connect to a known address and poll until disconnected or stopped.
|
|
|
|
Returns True if the connection was used successfully (so the caller
|
|
keeps the cached address), False if the connection failed and the
|
|
cache should be invalidated.
|
|
"""
|
|
log(f"Connecting to {address}...")
|
|
client = BleakClient(address)
|
|
try:
|
|
await client.connect()
|
|
except (BleakError, asyncio.TimeoutError) as e:
|
|
log(f"Connection failed: {e}")
|
|
return False
|
|
|
|
if not client.is_connected:
|
|
log("Connection failed (no error but not connected)")
|
|
return False
|
|
|
|
log("Connected")
|
|
session = Session(client)
|
|
await session.setup_refresh_subscription()
|
|
|
|
last_poll = 0.0
|
|
used_successfully = False
|
|
try:
|
|
while client.is_connected and not stop_event.is_set():
|
|
now = time.time()
|
|
elapsed = now - last_poll
|
|
if session.refresh_requested.is_set() or elapsed >= POLL_INTERVAL:
|
|
session.refresh_requested.clear()
|
|
token = read_token()
|
|
if not token:
|
|
log("No token; skipping poll")
|
|
else:
|
|
payload = await poll_api(token)
|
|
if payload is not None:
|
|
if await session.write_payload(payload):
|
|
last_poll = time.time()
|
|
used_successfully = True
|
|
|
|
try:
|
|
await asyncio.wait_for(session.refresh_requested.wait(), timeout=TICK)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
finally:
|
|
try:
|
|
await client.disconnect()
|
|
except BleakError:
|
|
pass
|
|
|
|
log("Device disconnected" if not stop_event.is_set() else "Stopping")
|
|
return used_successfully
|
|
|
|
|
|
async def main() -> None:
|
|
stop_event = asyncio.Event()
|
|
loop = asyncio.get_running_loop()
|
|
|
|
def _stop(*_args: object) -> None:
|
|
log("Daemon stopping")
|
|
stop_event.set()
|
|
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
try:
|
|
loop.add_signal_handler(sig, _stop)
|
|
except NotImplementedError:
|
|
signal.signal(sig, _stop)
|
|
|
|
log("=== Claude Usage Tracker Daemon (BLE, macOS) ===")
|
|
log(f"Poll interval: {POLL_INTERVAL}s")
|
|
|
|
backoff = 1
|
|
while not stop_event.is_set():
|
|
address = load_cached_address()
|
|
if not address:
|
|
address = await scan_for_device()
|
|
if address:
|
|
save_address(address)
|
|
else:
|
|
log(f"Device not found, retrying in {backoff}s...")
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=backoff)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
backoff = min(backoff * 2, 60)
|
|
continue
|
|
|
|
ok = await connect_and_run(address, stop_event)
|
|
if not ok:
|
|
log("Invalidating cached address")
|
|
SAVED_ADDR_FILE.unlink(missing_ok=True)
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=backoff)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
backoff = min(backoff * 2, 60)
|
|
else:
|
|
backoff = 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
sys.exit(0)
|