Skip to content
Open
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
7 changes: 5 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ CLI tool to track and display [OpenCode](https://github.com/opencodeco/opencode)
## Features

- **Daily breakdown** — token usage and cost per day
- **Group by dimension** — model, agent, provider, or session
- **Group by dimension** — minute/hour/day/week/month time buckets, or model, agent, provider, session
- **Agent × Model view** — see which model each agent uses
- **Time filtering** — last N days, relative durations (`7d`, `2w`), or ISO dates
- **Period comparison** — compare current vs previous period with `--compare`
Expand Down Expand Up @@ -46,10 +46,13 @@ opencode-usage

# Time filtering
opencode-usage run --days 30
opencode-usage run --weeks 2
opencode-usage run --months 6
opencode-usage run --since 7d
opencode-usage run --since 2025-01-01

# Group by dimension
# Group by dimension (time buckets or entities)
opencode-usage run --by hour # minute, hour, day, week, month
opencode-usage run --by model
opencode-usage run --by agent # shows model per agent
opencode-usage run --by provider
Expand Down
56 changes: 43 additions & 13 deletions src/opencode_usage/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@

from . import __version__, render
from .db import OpenCodeDB, UsageRow
from .render import render_daily, render_grouped, render_summary
from .render import render_grouped, render_summary, render_time


def _parse_since(value: str) -> datetime:
Expand Down Expand Up @@ -41,20 +41,34 @@ def _parse_since(value: str) -> datetime:


def _add_time_args(parser: argparse.ArgumentParser) -> None:
"""Add --days and --since to *parser* (shared by run & insights)."""
"""Add time-window flags to *parser* (shared by run & insights)."""
parser.add_argument(
"--days",
type=int,
default=None,
metavar="N",
help="Show last N days (default: 7)",
help="Show the last N days (default: 7)",
)
parser.add_argument(
"--weeks",
type=int,
default=None,
metavar="N",
help="Show the last N weeks",
)
parser.add_argument(
"--months",
type=int,
default=None,
metavar="N",
help="Show the last N months (30 days each)",
)
parser.add_argument(
"--since",
type=_parse_since,
default=None,
metavar="SPEC",
help="Time filter: '7d', '2w', '30d', '3h', or ISO date",
help="Time filter: '7d', '2w', '30d', '3h', '12m', or ISO date",
)


Expand All @@ -72,9 +86,12 @@ def _build_parser() -> argparse.ArgumentParser:
_add_time_args(run_p)
run_p.add_argument(
"--by",
choices=["model", "agent", "provider", "session", "day"],
choices=["minute", "hour", "day", "week", "month", "model", "agent", "provider", "session"],
default=None,
help="Group results by dimension",
help=(
"Group by minute, hour, day, week, month, model, agent, provider, "
"or session"
),
)
run_p.add_argument(
"--limit",
Expand Down Expand Up @@ -126,16 +143,29 @@ def _build_parser() -> argparse.ArgumentParser:
return p


_TIME_GROUPINGS = ("minute", "hour", "day", "week", "month")


def _resolve_since(args: argparse.Namespace) -> tuple[datetime | None, str]:
"""Resolve the effective 'since' datetime and a human-readable period label."""
now = datetime.now().astimezone()

if args.since is not None:
return args.since, f"Since {args.since.strftime('%Y-%m-%d')}"

if args.days is not None:
since = now - timedelta(days=args.days)
return since, f"Last {args.days} days"
for value, label in (
(args.days, "days"),
(getattr(args, "weeks", None), "weeks"),
(getattr(args, "months", None), "months"),
):
if value is not None and value > 0:
if label == "weeks":
delta = timedelta(days=value * 7)
elif label == "months":
delta = timedelta(days=value * 30)
else:
delta = timedelta(days=value)
return now - delta, f"Last {value} {label}"

since = now - timedelta(days=7)
return since, "Last 7 days"
Expand All @@ -150,8 +180,8 @@ def _fetch_rows(
limit: int | None = None,
) -> list[UsageRow]:
"""Fetch rows based on group_by dimension."""
if group_by == "day":
return db.daily(since=since, until=until, limit=limit)
if group_by in _TIME_GROUPINGS:
return db.by_time(group_by, since=since, until=until, limit=limit)
if group_by == "model":
return db.by_model(since=since, until=until, limit=limit)
if group_by == "agent":
Expand Down Expand Up @@ -229,8 +259,8 @@ def _cmd_run(args: argparse.Namespace) -> None:

deltas = _compute_deltas(rows, prev_rows) if prev_rows else None

if group_by == "day":
render_daily(rows, period)
if group_by in _TIME_GROUPINGS:
render_time(rows, period, granularity=group_by)
else:
render_grouped(rows, group_by, period, deltas=deltas)

Expand Down
47 changes: 45 additions & 2 deletions src/opencode_usage/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,28 @@ class UsageRow:
detail: str | None = None


_TIME_FORMATS = {
"minute": "%Y-%m-%d %H:%M",
"hour": "%Y-%m-%d %H:00",
"day": "%Y-%m-%d",
"week": "%Y-W%W",
"month": "%Y-%m",
}


@dataclass
class Filters:
"""Row filters applied to every query (all fields are optional).

The row-filters feature uses this; it is inert until then.
"""

providers: list[str] | None = None
models: list[str] | None = None
agents: list[str] | None = None
exclude_providers: list[str] | None = None


@dataclass
class SessionMeta:
"""Metadata about a user session."""
Expand Down Expand Up @@ -151,22 +173,43 @@ def _base_query(

# ── public API ────────────────────────────────────────────────

def daily(
def by_time(
self,
granularity: str = "day",
since: datetime | None = None,
until: datetime | None = None,
limit: int | None = None,
filters: Filters | None = None,
) -> list[UsageRow]:
"""Group by a time bucket: minute, hour, day, week, or month.

``filters`` is forwarded to the query layer once row filters land
(tracked in the filters-and-output PR); it is inert here.
"""
fmt = _TIME_FORMATS.get(granularity, _TIME_FORMATS["day"])
kwargs: dict[str, object] = {}
if filters is not None:
kwargs["filters"] = filters
return self._base_query(
group_expr=(
"date(json_extract(data, '$.time.created') / 1000, 'unixepoch', 'localtime')"
f"strftime('{fmt}', json_extract(data, '$.time.created') / 1000, "
"'unixepoch', 'localtime')"
),
since=since,
until=until,
order="label DESC",
limit=limit,
**kwargs,
)

def daily(
self,
since: datetime | None = None,
until: datetime | None = None,
limit: int | None = None,
) -> list[UsageRow]:
return self.by_time("day", since=since, until=until, limit=limit)

def by_model(
self,
since: datetime | None = None,
Expand Down
23 changes: 19 additions & 4 deletions src/opencode_usage/render.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,19 +195,34 @@ def render_summary(
console.print(Panel(text, title=f"[bold]OpenCode Usage — {period}[/bold]", border_style="blue"))


def render_daily(rows: list[UsageRow], period: str) -> None:
"""Render the daily breakdown table."""
_TIME_LABELS = {
"minute": "Minute",
"hour": "Hour",
"day": "Date",
"week": "Week",
"month": "Month",
}


def render_time(rows: list[UsageRow], period: str, granularity: str = "day") -> None:
"""Render the time-bucketed breakdown table."""
label = _TIME_LABELS.get(granularity, "Bucket")
trend = [r.tokens.total for r in rows]
table = _make_table(
title=f"Daily Usage ({period})",
label_header="Date",
title=f"{label} Usage ({period})",
label_header=label,
rows=rows,
show_breakdown=True,
trend_values=trend,
)
console.print(table)


def render_daily(rows: list[UsageRow], period: str) -> None:
"""Render the daily breakdown table (legacy alias of render_time)."""
render_time(rows, period, granularity="day")


def render_grouped(
rows: list[UsageRow],
group_by: str,
Expand Down
46 changes: 46 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,3 +326,49 @@ def test_unknown_group_returns_empty(self, tmp_path):
db = OpenCodeDB(db_path=_make_cli_db(tmp_path))
rows = _fetch_rows(db, "unknown")
assert rows == []


# ── --weeks / --months and extended --by ─────────────────────


class TestWeeksMonthsFlags:
def test_weeks_flag(self):
args = _build_parser().parse_args(["run", "--weeks", "2"])
assert args.weeks == 2

def test_months_flag(self):
args = _build_parser().parse_args(["run", "--months", "6"])
assert args.months == 6

def test_insights_months_flag(self):
args = _build_parser().parse_args(["insights", "--months", "3"])
assert args.months == 3


class TestTimeGroupings:
def test_by_choices_time(self):
parser = _build_parser()
for choice in ("minute", "hour", "week", "month"):
args = parser.parse_args(["run", "--by", choice])
assert args.by == choice


class TestResolveSinceWeeksMonths:
def test_weeks_flag(self):
ns = argparse.Namespace(since=None, days=None, weeks=2, months=None)
since, period = _resolve_since(ns)
expected = datetime.now().astimezone() - timedelta(days=14)
assert abs((since - expected).total_seconds()) < 2
assert period == "Last 2 weeks"

def test_months_flag(self):
ns = argparse.Namespace(since=None, days=None, weeks=None, months=6)
since, period = _resolve_since(ns)
expected = datetime.now().astimezone() - timedelta(days=180)
assert abs((since - expected).total_seconds()) < 2
assert period == "Last 6 months"

def test_days_wins_over_months(self):
ns = argparse.Namespace(since=None, days=7, weeks=None, months=12)
_since, period = _resolve_since(ns)
assert period == "Last 7 days"
37 changes: 37 additions & 0 deletions tests/test_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,43 @@ def test_limit(self, db_path):
assert len(rows) == 1


# ── by_time ──────────────────────────────────────────────────


class TestByTime:
def test_hour_buckets(self, db_path):
db = OpenCodeDB(db_path=db_path)
rows = db.by_time("hour")
assert rows
for r in rows:
assert ":" in r.label # YYYY-MM-DD HH:00

def test_week_buckets(self, db_path):
db = OpenCodeDB(db_path=db_path)
rows = db.by_time("week")
assert rows
for r in rows:
assert "-W" in r.label # YYYY-WNN

def test_month_buckets(self, db_path):
db = OpenCodeDB(db_path=db_path)
rows = db.by_time("month")
assert rows
for r in rows:
assert len(r.label) == 7 # YYYY-MM

def test_unknown_granularity_falls_back_to_day(self, db_path):
db = OpenCodeDB(db_path=db_path)
rows = db.by_time("decade")
assert rows
for r in rows:
assert "-" in r.label

def test_daily_is_day_granularity(self, db_path):
db = OpenCodeDB(db_path=db_path)
assert db.daily() == db.by_time("day")


# ── by_model ─────────────────────────────────────────────────


Expand Down