a291f3cb6e
Conversations that never edit a `projects/<slug>` file — service work, skills, scripts, docs, plain questions — used to fall out of every project-shaped view as "(unattributed)". They now land on a root project card backed by the repo root itself. - projects.py: ROOT_SLUG (`lab`), attributed_projects() (derived at read time, never written to the sidecar), project_dir()/repo_rel() so callers resolve the slug to REPO_DIR instead of assuming projects/<dir>, plus the gallery record and detail view (repo CLAUDE.md, GOAL.md, git history). A real ~/projects/lab dir would win over the synthesised record. - main.py: _conv_meta applies the fallback (so tags, project pages, graph and dashboards all see it), and the goals board + goal-work resolve the unit's dir through project_dir()/repo_rel(). - project_costs.py: same fallback, so the lab's spend rolls up on its card. - gitdiff.py: `project:lab` resolves to the superproject. - count_loc: the lab's LoC comes from `git ls-files` (tracked source only) — a directory walk would count services/media and the untracked data/ tree. - Frontend: house icon, dashed tinted frame and a `root` badge on the card and detail page; no .env editor; the tag's repo path reads `.`.
109 lines
4.2 KiB
Python
109 lines
4.2 KiB
Python
"""Aggregate Claude conversation spend per homelab project.
|
||
|
||
A conversation is attributed to a project when the project's directory name
|
||
appears in its inferred ``projectsAuto`` list (projects it created/edited files
|
||
in) *or* the manual ``projects`` list in the action-metadata sidecar — the exact
|
||
same union the viewer uses to decide which conversations to list on a project
|
||
page (``meta.projects``). Every
|
||
attributed conversation's real per-turn usage is rolled up into a per-project
|
||
total: dollar/token totals, mean/median per conversation, and a per-token-type
|
||
($ + tokens) breakdown. The project *detail* view additionally divides the total
|
||
by the project's lines of code (see ``projects.count_loc``).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from statistics import median
|
||
|
||
import projects as projects_mod
|
||
from conversations import price_for, usage_cost, usage_tokens, zero_usage
|
||
|
||
# Token types we break spend down by, in display order. Each carries its own
|
||
# token count and dollar cost (priced with the conversation's model).
|
||
TYPES = ("input", "output", "cacheRead", "cacheWrite")
|
||
|
||
|
||
def _project_slugs(summary: dict, meta_all: dict) -> set[str]:
|
||
"""The set of project dir-names a conversation is attributed to.
|
||
|
||
Conversations with no project of their own roll up into the root project
|
||
(see ``projects.attributed_projects``), exactly as in ``main._conv_meta`` —
|
||
so the lab's own spend is a card in the gallery instead of vanishing."""
|
||
slugs = set(summary.get("projectsAuto") or [])
|
||
sid = summary.get("sessionId") or ""
|
||
m = meta_all.get(sid) or {}
|
||
for p in (m.get("projects") or []):
|
||
if p:
|
||
slugs.add(p)
|
||
return set(projects_mod.attributed_projects(sorted(slugs)))
|
||
|
||
|
||
def _type_costs(u: dict, model: str) -> dict[str, dict]:
|
||
"""Split one usage block into per-token-type {tokens, cost}.
|
||
|
||
Mirrors ``conversations.usage_cost``: input at the input rate, output at the
|
||
output rate, cache reads at 0.1× input, cache writes priced off their
|
||
weighted *units* (5m ≈ 1.25×, 1h ≈ 2×) but counted in raw tokens.
|
||
"""
|
||
pi, po = price_for(model)
|
||
pi /= 1e6
|
||
po /= 1e6
|
||
return {
|
||
"input": {"tokens": u["input"], "cost": u["input"] * pi},
|
||
"output": {"tokens": u["output"], "cost": u["output"] * po},
|
||
"cacheRead": {"tokens": u["cacheRead"], "cost": u["cacheRead"] * 0.1 * pi},
|
||
"cacheWrite": {"tokens": u["cacheWriteTokens"], "cost": u["cacheWriteUnits"] * pi},
|
||
}
|
||
|
||
|
||
def _aggregate(convs: list[dict]) -> dict:
|
||
"""Roll a project's attributed conversations into a spend summary."""
|
||
costs: list[float] = []
|
||
tokens: list[int] = []
|
||
total_cost = 0.0
|
||
total_tokens = 0
|
||
by_type = {t: {"tokens": 0, "cost": 0.0} for t in TYPES}
|
||
|
||
for s in convs:
|
||
model = s.get("model") or ""
|
||
u = {**zero_usage(), **(s.get("usage") or {})}
|
||
c = s.get("cost")
|
||
if c is None:
|
||
c = usage_cost(u, model)
|
||
tk = s.get("tokens")
|
||
if tk is None:
|
||
tk = usage_tokens(u)
|
||
costs.append(c)
|
||
tokens.append(tk)
|
||
total_cost += c
|
||
total_tokens += tk
|
||
for t, v in _type_costs(u, model).items():
|
||
by_type[t]["tokens"] += int(v["tokens"])
|
||
by_type[t]["cost"] += v["cost"]
|
||
|
||
n = len(convs)
|
||
return {
|
||
"conversations": n,
|
||
"cost": total_cost,
|
||
"tokens": total_tokens,
|
||
"meanCost": total_cost / n if n else 0.0,
|
||
"medianCost": float(median(costs)) if costs else 0.0,
|
||
"meanTokens": total_tokens / n if n else 0.0,
|
||
"medianTokens": float(median(tokens)) if tokens else 0.0,
|
||
"byType": by_type,
|
||
}
|
||
|
||
|
||
def costs_by_project(summaries: list[tuple[str, dict]], meta_all: dict) -> dict[str, dict]:
|
||
"""{project dir-name -> spend aggregate} across all conversations.
|
||
|
||
``summaries`` is ``store.all_summaries()`` (``[(path, summary), …]``);
|
||
``meta_all`` is ``meta_store.all()``.
|
||
"""
|
||
buckets: dict[str, list[dict]] = {}
|
||
for _path, s in summaries:
|
||
if not s.get("messages"):
|
||
continue
|
||
for slug in _project_slugs(s, meta_all):
|
||
buckets.setdefault(slug, []).append(s)
|
||
return {slug: _aggregate(convs) for slug, convs in buckets.items()}
|