Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 11 additions & 30 deletions daemon/codex-usage-daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -934,44 +934,25 @@ def claude_plan_usage_snapshot() -> UsageSnapshot | None:
except (OSError, json.JSONDecodeError):
return None

session = data.get("session")
weekly = data.get("weekly")
if not isinstance(session, dict) and not isinstance(weekly, dict):
return None
session = data.get("session") if isinstance(data.get("session"), dict) else None
weekly = data.get("weekly") if isinstance(data.get("weekly"), dict) else None

session_used = (
float(session.get("used_pct") or 0)
if isinstance(session, dict)
else 0.0
)
weekly_used = (
float(weekly.get("used_pct") or 0)
if isinstance(weekly, dict)
else 0.0
)
session_reset = (
int(session.get("reset_mins") or -1)
if isinstance(session, dict)
else -1
)
weekly_reset = (
int(weekly.get("reset_mins") or -1)
if isinstance(weekly, dict)
else -1
)
# Require both windows so a partial payload can't override another source
# with a phantom 100%-remaining for the missing one.
if session is None or weekly is None:
return None

status_parts = []
if isinstance(session, dict):
status_parts.append(f"5h {round(session_used)}% used")
if isinstance(weekly, dict):
status_parts.append(f"7d {round(weekly_used)}% used")
session_used = float(session.get("used_pct") or 0)
weekly_used = float(weekly.get("used_pct") or 0)
session_reset = int(session.get("reset_mins") or -1)
weekly_reset = int(weekly.get("reset_mins") or -1)

return UsageSnapshot(
session_pct=remaining_pct_from_used(session_used),
session_reset_mins=session_reset,
weekly_pct=remaining_pct_from_used(weekly_used),
weekly_reset_mins=weekly_reset,
status=" · ".join(status_parts) or "Claude plan usage",
status=f"5h {round(session_used)}% used · 7d {round(weekly_used)}% used",
ok=True,
)

Expand Down
94 changes: 68 additions & 26 deletions scripts/claude-plan-usage-cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,45 +31,87 @@ def minutes_from_duration(hours: int, minutes: int) -> int:
return max(0, hours * 60 + minutes)


def _parse_duration_reset(text: str) -> int | None:
"""Parse 'Resets in 3 hr 17 min', 'Resets in 32 min', or 'Resets in 2 hr'."""
match = re.search(
r"Resets in\s+(?:(\d+)\s*hr)?\s*(?:(\d+)\s*min)?",
text,
re.IGNORECASE,
)
if not match or (match.group(1) is None and match.group(2) is None):
return None
hours = int(match.group(1) or 0)
minutes = int(match.group(2) or 0)
return minutes_from_duration(hours, minutes)


def _parse_weekday_reset(text: str) -> int | None:
"""Parse 'Resets Fri 7:00 PM' (legacy format)."""
match = re.search(
r"Resets\s+([A-Za-z]{3})\s+(\d{1,2}):(\d{2})\s*([AP]M)",
text,
re.IGNORECASE,
)
if not match:
return None
hour = int(match.group(2))
minute = int(match.group(3))
meridiem = match.group(4).lower()
if meridiem == "pm" and hour != 12:
hour += 12
elif meridiem == "am" and hour == 12:
hour = 0
return minutes_until_weekday_time(match.group(1), hour, minute)


def parse_usage_text(text: str) -> dict:
compact = "\n".join(line.strip() for line in text.splitlines() if line.strip())
session_match = re.search(
r"Current session\s+Resets in\s+(\d+)\s+hr\s+(\d+)\s+min.*?(\d+)%\s+used",

# Find each section's text block: the label up to the next blank line / next
# known section. We use lookahead-free non-greedy match against the other
# known section headers and a sentinel "% used" boundary.
sentinels = r"(?:Weekly limits|All models|Claude Design|Additional features|Last updated|$)"
session_block = re.search(
rf"Current session\s+(.*?{sentinels})",
compact,
re.IGNORECASE | re.DOTALL,
)
weekly_match = re.search(
r"All models\s+Resets\s+([A-Za-z]{3})\s+(\d{1,2}):(\d{2})\s*([AP]M).*?(\d+)%\s+used",
weekly_block = re.search(
rf"All models\s+(.*?{sentinels})",
compact,
re.IGNORECASE | re.DOTALL,
)

if not session_match and not weekly_match:
raise ValueError("Claude usage text did not include recognizable plan usage")

usage: dict = {
"source": "claude_usage_page_text",
"captured_at": datetime.now().astimezone().isoformat(),
}
if session_match:
usage["session"] = {
"used_pct": int(session_match.group(3)),
"reset_mins": minutes_from_duration(
int(session_match.group(1)), int(session_match.group(2))
),
}
if weekly_match:
hour = int(weekly_match.group(2))
minute = int(weekly_match.group(3))
meridiem = weekly_match.group(4).lower()
if meridiem == "pm" and hour != 12:
hour += 12
elif meridiem == "am" and hour == 12:
hour = 0
usage["weekly"] = {
"used_pct": int(weekly_match.group(5)),
"reset_mins": minutes_until_weekday_time(weekly_match.group(1), hour, minute),
}

def percent_used(block_text: str) -> int | None:
m = re.search(r"(\d+)\s*%\s*used", block_text, re.IGNORECASE)
return int(m.group(1)) if m else None

if session_block:
block = session_block.group(1)
used = percent_used(block)
reset = _parse_duration_reset(block)
if reset is None:
reset = _parse_weekday_reset(block)
if used is not None and reset is not None:
usage["session"] = {"used_pct": used, "reset_mins": reset}

if weekly_block:
block = weekly_block.group(1)
used = percent_used(block)
reset = _parse_duration_reset(block)
if reset is None:
reset = _parse_weekday_reset(block)
if used is not None and reset is not None:
usage["weekly"] = {"used_pct": used, "reset_mins": reset}

if "session" not in usage and "weekly" not in usage:
raise ValueError("Claude usage text did not include recognizable plan usage")

return usage


Expand Down
Loading