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

switch daemon to /api/oauth/usage endpoint

Browse Source
This commit is contained in:
Brian Pugh
2026-05-24 15:26:43 -04:00
parent efc5641840
commit 525ebe8921
4 changed files with 100 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/claude-usage-daemon.sh
+47 -31
View File
@@ -1,8 +1,10 @@
#!/bin/bash
# Claude Usage Tracker Daemon (BLE)
# Reads Claude Code OAuth token, polls usage via API, sends to ESP32 over BLE GATT.
# Reads Claude Code OAuth token, polls usage via the OAuth usage endpoint
# (api.anthropic.com/api/oauth/usage — same one `claude /usage` uses, costs
# zero tokens), sends results to the ESP32 over BLE GATT.
# Auto-connects and reconnects to the Claude Controller BLE device.
# Dependencies: curl, awk, bluetoothctl
# Dependencies: curl, awk, python3, bluetoothctl
DEVICE_NAME="Claude Controller"
DEVICE_MAC="${DEVICE_MAC:-}" # auto-discovered if empty
@@ -194,42 +196,56 @@ write_gatt() {
poll() {
local token
token=$(read_token) || { log "Error: could not read token"; return 1; }
local now
now=$(date +%s)
local headers
headers=$(curl -s -D - -o /dev/null \
"https://api.anthropic.com/v1/messages" \
local body
body=$(curl -s \
"https://api.anthropic.com/api/oauth/usage" \
-H "Authorization: Bearer $token" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: oauth-2025-04-20" \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-H "User-Agent: claude-code/2.1.5" \
-d '{"model":"claude-haiku-4-5-20251001","max_tokens":1,"messages":[{"role":"user","content":"hi"}]}' \
2>/dev/null) || { log "Error: API call failed"; return 1; }
local s5h_util s5h_reset s7d_util s7d_reset status
s5h_util=$(echo "$headers" | grep -i "anthropic-ratelimit-unified-5h-utilization" | tr -d '\r' | awk '{print $2}')
s5h_reset=$(echo "$headers" | grep -i "anthropic-ratelimit-unified-5h-reset" | tr -d '\r' | awk '{print $2}')
s7d_util=$(echo "$headers" | grep -i "anthropic-ratelimit-unified-7d-utilization" | tr -d '\r' | awk '{print $2}')
s7d_reset=$(echo "$headers" | grep -i "anthropic-ratelimit-unified-7d-reset" | tr -d '\r' | awk '{print $2}')
status=$(echo "$headers" | grep -i "anthropic-ratelimit-unified-5h-status" | tr -d '\r' | awk '{print $2}')
s5h_util=${s5h_util:-0}
s5h_reset=${s5h_reset:-0}
s7d_util=${s7d_util:-0}
s7d_reset=${s7d_reset:-0}
status=${status:-unknown}
local payload
payload=$(awk -v u5="$s5h_util" -v r5="$s5h_reset" -v u7="$s7d_util" -v r7="$s7d_reset" -v st="$status" -v now="$now" \
'BEGIN {
sp = sprintf("%.0f", u5 * 100);
sr = (r5 - now) / 60; sr = sr > 0 ? sprintf("%.0f", sr) : 0;
wp = sprintf("%.0f", u7 * 100);
wr = (r7 - now) / 60; wr = wr > 0 ? sprintf("%.0f", wr) : 0;
printf "{\"s\":%s,\"sr\":%s,\"w\":%s,\"wr\":%s,\"st\":\"%s\",\"ok\":true}", sp, sr, wp, wr, st;
}')
payload=$(python3 -c '
import datetime, json, sys, time
try:
data = json.loads(sys.stdin.read())
except json.JSONDecodeError:
sys.exit(1)
def pct(w):
if not isinstance(w, dict):
return 0
u = w.get("utilization")
return int(round(u)) if isinstance(u, (int, float)) else 0
def reset_mins(w):
if not isinstance(w, dict):
return -1
s = w.get("resets_at")
if not isinstance(s, str):
return -1
try:
ts = datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
except ValueError:
return -1
m = (ts - time.time()) / 60.0
return int(round(m)) if m > 0 else 0
fh = data.get("five_hour")
sd = data.get("seven_day")
s = pct(fh)
print(json.dumps({
"s": s,
"sr": reset_mins(fh),
"w": pct(sd),
"wr": reset_mins(sd),
"st": "limited" if s >= 100 else "allowed",
"ok": True,
}, separators=(",", ":")))
' <<< "$body") || { log "Error: failed to parse usage JSON"; return 1; }
log "Sending: $payload"
write_gatt "$RX_CHAR_PATH" "$payload" || { log "Write failed"; return 1; }
daemon/claude_usage_daemon.py
+50 -36
View File
@@ -1,9 +1,10 @@
#!/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).
Polls the Claude OAuth usage endpoint (the same one Claude Code's `/usage`
command uses) and writes a JSON payload to the ESP32 "Claude Controller"
peripheral over a custom GATT service. Uses bleak (CoreBluetooth backend
on macOS).
"""
import asyncio
@@ -15,6 +16,7 @@ import signal
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
import httpx
@@ -36,18 +38,12 @@ 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_URL = "https://api.anthropic.com/api/oauth/usage"
API_HEADERS_TEMPLATE = {
"anthropic-version": "2023-06-01",
"anthropic-beta": "oauth-2025-04-20",
"Content-Type": "application/json",
"Accept": "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:
@@ -157,12 +153,40 @@ async def scan_for_device() -> str | None:
return None
def _parse_iso8601(s: object) -> float | None:
if not isinstance(s, str) or not s:
return None
try:
return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
except ValueError:
return None
def _window_pct(window: object) -> int:
if not isinstance(window, dict):
return 0
util = window.get("utilization")
if not isinstance(util, (int, float)):
return 0
return int(round(float(util)))
def _window_reset_mins(window: object, now: float) -> int:
if not isinstance(window, dict):
return -1
ts = _parse_iso8601(window.get("resets_at"))
if ts is None:
return -1
mins = (ts - now) / 60.0
return int(round(mins)) if mins > 0 else 0
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)
resp = await http.get(API_URL, headers=headers)
except httpx.HTTPError as e:
log(f"API call failed: {e}")
return None
@@ -170,34 +194,24 @@ async def poll_api(token: str) -> dict | None:
log(f"API HTTP {resp.status_code}: {resp.text[:200]}")
return None
def hdr(name: str, default: str = "0") -> str:
return resp.headers.get(name, default)
try:
body = resp.json()
except json.JSONDecodeError as e:
log(f"API returned non-JSON: {e}")
return None
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"),
five_hour = body.get("five_hour")
seven_day = body.get("seven_day")
s_pct = _window_pct(five_hour)
return {
"s": s_pct,
"sr": _window_reset_mins(five_hour, now),
"w": _window_pct(seven_day),
"wr": _window_reset_mins(seven_day, now),
"st": "limited" if s_pct >= 100 else "allowed",
"ok": True,
}
return payload
class Session: