jordan/stern-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Capture individual games and improve score dashboard

Browse Source
This commit is contained in:
MrARM
2026-08-01 18:05:56 -05:00
parent 67886ac023
commit 09b120a9fa
9 changed files with 883 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+5 -1
View File
@@ -2,7 +2,11 @@
API_BASE_URL=https://iop-dev.strndev.com
API_USERNAME=replace-me
API_PASSWORD=replace-me
POLL_SECONDS=300
# Individual games are captured by polling, so the interval bounds what can be
# caught: two games on the same title inside one interval yield only the later
# one. This deployment uses the requested three-minute interval; lower it toward
# 60s when capturing every quick back-to-back game matters more.
POLL_SECONDS=180
# Use distinct, long random values before exposing any service beyond localhost.
POSTGRES_DB=play_history
PLAY_HISTORY_STACK.md
+96 -11
View File
@@ -4,11 +4,65 @@
`docker compose` runs three services:
- **collector** logs in as the configured account, polls `/api/v1/portal/my_activity/`, follows same-origin pagination, and stores only new activity states.
- **PostgreSQL** stores normalized activity records and their original API payload.
- **Grafana** provisions a PostgreSQL datasource plus a dashboard for maximum score, total plays, and the latest activity table.
- **collector** logs in, polls the per-title score feeds, and records each
individual game it sees.
- **PostgreSQL** stores individual games (`game_event`) and per-visit rollups
(`play_session`).
- **Grafana** provisions a PostgreSQL datasource and a dashboard of individual
scores, personal bests over time, and capture coverage.
The collector uses the authenticated activity endpoint that returned 28 records in `play-history.jsonl`. It does not scrape unrelated accounts.
## How individual games are captured, and what that costs
**The Stern API has no per-game history endpoint.** This is the central
constraint, and it shapes everything else.
The endpoint this project originally used, `/api/v1/portal/my_activity/`,
returns a *daily rollup*: one row per (date, machine model, location) carrying
only `max_score` and `total_plays`. A day with 21 games on one machine collapses
to a single number, and the other 20 scores are simply not in the response.
Individual scores are exposed in exactly one place — as the *most recent* game:
| Endpoint | Gives |
| --- | --- |
| `/api/v1/portal/user_title_stats/?user_id=<pk>&title_id=<id>` | `most_recent_score` + `most_recent_date` for that title |
| `/api/v1/portal/user_highlights/?user_id=<username>` | the most recent game overall |
Those are real single-game scores, not maxima. A title's most recent score, its
all-time best, and a visit's best are three different values; the collector uses
the first one.
So the collector **polls and captures**, rather than fetching history. Each poll
reads every title you have played and stores any `(title_id, played_at)` pair it
has not seen. The play timestamp is the primary key, so re-observing the same
game is idempotent.
Two consequences follow directly, and the dashboard reports both rather than
hiding them:
- **Games played before the collector started cannot be recovered.** They are
not retrievable from any endpoint.
- **Two games on the same title within one poll interval yield only the later
one.** `POLL_SECONDS` therefore bounds fidelity; 60s against a 2-5 minute game
is comfortable, 300s is not.
The `capture_coverage` view and the dashboard's top panel compare games captured
against the play count the account reports, so the gap is always visible.
`play_session` keeps the per-visit rollup from `/api/v1/portal/user_activities/`
(location, timestamp, play count, visit high score). It is aggregate data, but it
is the only record of pre-collector history and it supplies the play counts that
make the coverage number meaningful.
### Endpoints that do not work
The `/api/v4/` tier (`user/activity/`, `recent/played_games/`,
`player/model_scores/`, `stats/model_scores/`, `user/info/`) returns HTTP 500 on
this server for every parameter combination tried, with both a `/api/v2/token/`
JWT and a `/api/v4/auth/login/` JWT. Only `/api/v4/stats/scores/` responds, and
it returns per-title bests, ignoring its parameters.
`/api/v1/portal/user_activities_from_session_details/` returns the same aggregate
shape as `user_activities` and ignores `score_key`.
## Start
@@ -18,22 +72,53 @@ cp .env.example .env
docker compose up --build -d
```
Open `http://localhost:3000`, log in with `GRAFANA_ADMIN_USER` and `GRAFANA_ADMIN_PASSWORD`, then open **Dashboards → Play History → Stern Play History**.
Open `http://localhost:3000`, log in with `GRAFANA_ADMIN_USER` and
`GRAFANA_ADMIN_PASSWORD`, then open **Dashboards → Play History → Stern Play
History**.
Expect the coverage panel to show a large uncaptured count at first — that is
accurate, and it shrinks only in the sense that newly played games are captured
from here on.
## Verify
```bash
python3 -m unittest discover -s tests -v # parsing tests, no DB or network needed
docker compose ps
docker compose logs collector --tail=50
docker compose exec postgres psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" \
-c 'SELECT count(*) AS score_states FROM activity_snapshot;'
-c 'SELECT * FROM capture_coverage;' \
-c 'SELECT played_at, title_name, score FROM game_event ORDER BY played_at DESC LIMIT 10;'
```
The first poll should report `collected 28 activities` and insert the current states. Later polls only insert new or changed score states, so the Grafana time series does not duplicate unchanged data.
The first poll captures one game per title you have played (each title's most
recent), plus the session rollup. Subsequent polls insert only genuinely new
games.
## Operations
- Change `POLL_SECONDS` in `.env`; the collector enforces a 30-second minimum.
- Grafana is bound to localhost by default. Do not expose it publicly without TLS and a stronger access-control layer.
- `API_PASSWORD`, JWTs, and refresh tokens are never stored in PostgreSQL or the collector logs.
- Use `docker compose down` to stop services. Add `-v` only if you intentionally want to delete stored history.
- `POLL_SECONDS` in `.env` controls capture fidelity; the collector enforces a
30-second minimum.
- Grafana is bound to localhost by default. Do not expose it publicly without TLS
and a stronger access-control layer.
- `API_PASSWORD` and JWTs are never stored in PostgreSQL or written to the logs.
- The collector authenticates once per poll cycle. Very short intervals mean
frequent logins; the server appears to throttle repeated authentication, so
staying at or above the 30-second floor matters.
- `docker compose down` stops the stack. Add `-v` only if you intentionally want
to delete stored history — captured games cannot be re-fetched.
## Schema
```
game_event(title_id, played_at, score, title_name, title_code, first_seen_at, source)
PRIMARY KEY (title_id, played_at) -- one row per game actually played
play_session(score_key, session_number, session_at, location_id, location_name,
title_name, model_type, num_plays, high_score, updated_at)
PRIMARY KEY (score_key) -- one row per title per visit
collection_run(id, observed_at, events_captured, sessions_upserted, reported_total_plays)
capture_coverage -- view: games_captured vs games_played_reported
```
app/collector.py
+306 -91
View File
@@ -1,24 +1,62 @@
#!/usr/bin/env python3
"""Poll a Stern Insider account's activity endpoint and persist score changes."""
"""Capture individual Stern Insider games by polling the per-title score feeds.
The API has no per-game history endpoint. `/api/v1/portal/my_activity/` returns a
daily rollup (one max_score and total_plays per day/model/location), which cannot
be decomposed back into individual games. What it does expose is the *most recent*
game per title, via `/api/v1/portal/user_title_stats/`, and the most recent game
overall, via `/api/v1/portal/user_highlights/`. Both report the exact play
timestamp and that game's own score, so polling faster than games are played
captures each game once.
Consequence worth knowing: games played before this collector ran are not
recoverable, and two games on the same title inside one poll interval yield only
the later one. `capture_coverage` reports the resulting gap.
"""
from __future__ import annotations
import hashlib
import importlib
import json
import logging
import os
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin, urlparse
from urllib.parse import urlencode, urljoin, urlparse
from urllib.request import Request, urlopen
psycopg = importlib.import_module("psycopg")
LOG = logging.getLogger("stern_collector")
USER_AGENT = "stern-play-history-collector/1.0"
USER_AGENT = "stern-play-history-collector/2.0"
@dataclass(frozen=True)
class GameEvent:
"""One game, identified by when it was played."""
title_id: int
title_name: str
title_code: str | None
played_at: datetime
score: int
source: str
@dataclass(frozen=True)
class SessionRow:
"""One title played during one visit to one location."""
score_key: str
session_number: int
session_at: datetime
location_id: int
location_name: str
title_name: str
model_type: str
num_plays: int
high_score: int
def setting(name: str) -> str:
@@ -31,12 +69,15 @@ def setting(name: str) -> str:
def request_json(
url: str,
method: str = "GET",
payload: dict[str, str] | None = None,
payload: dict[str, Any] | None = None,
token: str | None = None,
query: dict[str, Any] | None = None,
) -> tuple[int, Any]:
if query:
url += "?" + urlencode(query)
body = json.dumps(payload).encode() if payload is not None else None
headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
if body:
if body is not None:
headers["Content-Type"] = "application/json"
if token:
headers["Authorization"] = f"Bearer {token}"
@@ -55,15 +96,13 @@ def request_json(
def authenticate(base_url: str) -> str:
credential_fields = ("username", "password")
credential_values = (setting("API_USERNAME"), setting("API_PASSWORD"))
credentials: dict[str, str] = dict(
zip(credential_fields, credential_values, strict=True)
)
status, response = request_json(
urljoin(base_url, "api/v2/token/"),
method="POST",
payload=credentials,
payload={
"username": setting("API_USERNAME"),
"password": setting("API_PASSWORD"),
},
)
if (
status != 200
@@ -74,30 +113,188 @@ def authenticate(base_url: str) -> str:
return response["access"]
def activity_pages(base_url: str, token: str) -> list[dict[str, Any]]:
origin = urlparse(base_url).netloc
url = urljoin(base_url, "api/v1/portal/my_activity/")
entries: list[dict[str, Any]] = []
while url:
if urlparse(url).netloc != origin:
raise RuntimeError("API pagination pointed to a different origin")
status, payload = request_json(url, token=token)
if status != 200 or not isinstance(payload, dict):
raise RuntimeError(f"activity request failed (HTTP {status})")
page = payload.get("results")
if not isinstance(page, list):
raise RuntimeError("activity response did not contain a results list")
entries.extend(item for item in page if isinstance(item, dict))
next_page = payload.get("next")
url = (
urljoin(base_url, next_page)
if isinstance(next_page, str) and next_page
else ""
def get_json(base_url: str, path: str, token: str, **query: Any) -> Any:
status, payload = request_json(
urljoin(base_url, path), token=token, query=query or None
)
if status != 200:
raise RuntimeError(f"{path} failed (HTTP {status})")
return payload
def account_pk(base_url: str, token: str) -> int:
"""user_title_stats keys off the numeric account id, not the username."""
payload = get_json(base_url, "api/v2/portal/user_detail/", token)
pk = payload.get("user", {}).get("pk") if isinstance(payload, dict) else None
if not isinstance(pk, int):
raise RuntimeError("user_detail did not return a numeric account id")
return pk
def parse_timestamp(value: Any) -> datetime | None:
"""Accept both `...Z` and `...-05:00` forms the API mixes between endpoints."""
if not isinstance(value, str) or not value:
return None
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def parse_int(value: Any) -> int | None:
if isinstance(value, bool) or value in (None, ""):
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def titles_played(user_stats: Any) -> list[dict[str, Any]]:
stats = user_stats.get("stats") if isinstance(user_stats, dict) else None
entries = stats.get("titles") if isinstance(stats, dict) else None
if not isinstance(entries, list):
return []
return [
entry
for entry in entries
if isinstance(entry, dict) and isinstance(entry.get("id"), int)
]
def reported_total_plays(user_stats: Any) -> int | None:
stats = user_stats.get("stats") if isinstance(user_stats, dict) else None
return parse_int(stats.get("total_plays")) if isinstance(stats, dict) else None
def event_from_title_stats(title: dict[str, Any], payload: Any) -> GameEvent | None:
if not isinstance(payload, dict):
return None
played_at = parse_timestamp(payload.get("most_recent_date"))
score = parse_int(payload.get("most_recent_score"))
if played_at is None or score is None:
return None
return GameEvent(
title_id=title["id"],
title_name=str(title.get("name", "")),
title_code=title.get("code") or None,
played_at=played_at,
score=score,
source="user_title_stats",
)
def event_from_highlights(payload: Any) -> GameEvent | None:
"""Catches a game on a title that user_stats has not listed yet."""
highlights = payload.get("highlights") if isinstance(payload, dict) else None
recent = highlights.get("most_recent_game_played") if isinstance(highlights, dict) else None
if not isinstance(recent, dict):
return None
title = recent.get("title")
if not isinstance(title, dict) or not isinstance(title.get("id"), int):
return None
played_at = parse_timestamp(recent.get("last_played_date"))
score = parse_int(recent.get("last_played_score"))
if played_at is None or score is None:
return None
return GameEvent(
title_id=title["id"],
title_name=str(title.get("name", "")),
title_code=title.get("code") or None,
played_at=played_at,
score=score,
source="user_highlights",
)
def sessions_from_activities(payload: Any) -> list[SessionRow]:
user = payload.get("user") if isinstance(payload, dict) else None
raw = user.get("location_sessions") if isinstance(user, dict) else None
if not isinstance(raw, list):
return []
rows: list[SessionRow] = []
for session in raw:
if not isinstance(session, dict):
continue
session_at = parse_timestamp(session.get("date"))
number = parse_int(session.get("number"))
location = session.get("location")
if session_at is None or number is None or not isinstance(location, dict):
continue
location_id = parse_int(location.get("id"))
if location_id is None:
continue
for game in session.get("games_played") or []:
if not isinstance(game, dict):
continue
score_key = game.get("score_key")
num_plays = parse_int(game.get("num_plays"))
high_score = parse_int(game.get("high_score"))
if not isinstance(score_key, str) or num_plays is None or high_score is None:
continue
model = game.get("model")
rows.append(
SessionRow(
score_key=score_key,
session_number=number,
session_at=session_at,
location_id=location_id,
location_name=str(location.get("name", "")),
title_name=str(game.get("name", "")),
model_type=str(model.get("type", "")) if isinstance(model, dict) else "",
num_plays=num_plays,
high_score=high_score,
)
)
return rows
def collect(base_url: str, token: str) -> tuple[list[GameEvent], list[SessionRow], int | None]:
username = setting("API_USERNAME")
pk = account_pk(base_url, token)
user_stats = get_json(base_url, "api/v1/portal/user_stats/", token)
events: dict[tuple[int, datetime], GameEvent] = {}
for title in titles_played(user_stats):
try:
payload = get_json(
base_url,
"api/v1/portal/user_title_stats/",
token,
user_id=pk,
title_id=title["id"],
)
except RuntimeError as error:
LOG.warning("title %s stats unavailable: %s", title["id"], error)
continue
event = event_from_title_stats(title, payload)
if event:
events[(event.title_id, event.played_at)] = event
try:
highlight = event_from_highlights(
get_json(base_url, "api/v1/portal/user_highlights/", token, user_id=username)
)
return entries
if highlight:
events.setdefault((highlight.title_id, highlight.played_at), highlight)
except RuntimeError as error:
LOG.warning("highlights unavailable: %s", error)
sessions = sessions_from_activities(
get_json(base_url, "api/v1/portal/user_activities/", token, user_id=username)
)
return list(events.values()), sessions, reported_total_plays(user_stats)
def database_connection() -> Any:
# Imported here so the response-parsing helpers can be exercised without the
# driver installed.
psycopg = importlib.import_module("psycopg")
return psycopg.connect(
host=setting("POSTGRES_HOST"),
port=os.environ.get("POSTGRES_PORT", "5432"),
@@ -107,76 +304,94 @@ def database_connection() -> Any:
)
def required_text(entry: dict[str, Any], field: str) -> str:
value = entry.get(field)
if value is None:
raise ValueError(f"activity item omitted {field}")
return str(value)
def persist(
events: list[GameEvent], sessions: list[SessionRow], total_plays: int | None
) -> tuple[int, int]:
observed_at = datetime.now(timezone.utc)
captured = 0
upserted = 0
with database_connection() as connection, connection.cursor() as cursor:
for event in events:
cursor.execute(
"""
INSERT INTO game_event (
title_id, played_at, score, title_name, title_code,
first_seen_at, source
) VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (title_id, played_at) DO NOTHING
""",
(
event.title_id,
event.played_at,
event.score,
event.title_name,
event.title_code,
observed_at,
event.source,
),
)
captured += cursor.rowcount
for row in sessions:
# num_plays and high_score grow while a visit is still in progress.
cursor.execute(
"""
INSERT INTO play_session (
score_key, session_number, session_at, location_id,
location_name, title_name, model_type, num_plays,
high_score, updated_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (score_key) DO UPDATE SET
num_plays = EXCLUDED.num_plays,
high_score = EXCLUDED.high_score,
session_at = EXCLUDED.session_at,
updated_at = EXCLUDED.updated_at
WHERE play_session.num_plays IS DISTINCT FROM EXCLUDED.num_plays
OR play_session.high_score IS DISTINCT FROM EXCLUDED.high_score
""",
(
row.score_key,
row.session_number,
row.session_at,
row.location_id,
row.location_name,
row.title_name,
row.model_type,
row.num_plays,
row.high_score,
observed_at,
),
)
upserted += cursor.rowcount
def optional_int(entry: dict[str, Any], field: str) -> int | None:
value = entry.get(field)
if value in (None, ""):
return None
try:
return int(value)
except (TypeError, ValueError) as error:
raise ValueError(f"activity item has a non-integer {field}") from error
cursor.execute(
"""
INSERT INTO collection_run (
observed_at, events_captured, sessions_upserted, reported_total_plays
) VALUES (%s, %s, %s, %s)
""",
(observed_at, captured, upserted, total_plays),
)
return captured, upserted
def polling_interval() -> int:
try:
return max(30, int(os.environ.get("POLL_SECONDS", "300")))
return max(30, int(os.environ.get("POLL_SECONDS", "60")))
except ValueError as error:
raise RuntimeError("POLL_SECONDS must be an integer") from error
def persist(entries: list[dict[str, Any]]) -> int:
observed_at = datetime.now(timezone.utc)
inserted = 0
with database_connection() as connection, connection.cursor() as cursor:
for entry in entries:
canonical = json.dumps(entry, sort_keys=True, separators=(",", ":"))
source_hash = hashlib.sha256(canonical.encode()).hexdigest()
try:
cursor.execute(
"""
INSERT INTO activity_snapshot (
source_hash, observed_at, activity_date, game_model_id,
game_model_type, location_id, location_name, max_score,
total_plays, raw_activity
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb)
ON CONFLICT (source_hash) DO NOTHING
""",
(
source_hash,
observed_at,
required_text(entry, "date"),
required_text(entry, "game_model_id"),
required_text(entry, "game_model_type"),
required_text(entry, "location_id"),
required_text(entry, "location_name"),
optional_int(entry, "max_score"),
optional_int(entry, "total_plays"),
canonical,
),
)
inserted += cursor.rowcount
except (ValueError, psycopg.Error) as error:
LOG.warning("skipping malformed activity item: %s", error)
cursor.execute(
"INSERT INTO collection_run (observed_at, item_count, inserted_count) VALUES (%s, %s, %s)",
(observed_at, len(entries), inserted),
)
return inserted
def run_once(base_url: str) -> None:
token = authenticate(base_url)
entries = activity_pages(base_url, token)
inserted = persist(entries)
events, sessions, total_plays = collect(base_url, token)
captured, upserted = persist(events, sessions, total_plays)
LOG.info(
"collected %d activities; inserted %d new score states", len(entries), inserted
"polled %d titles; %d new games captured, %d sessions updated (account reports %s plays)",
len(events),
captured,
upserted,
total_plays if total_plays is not None else "unknown",
)
db/init.sql
+63 -19
View File
@@ -1,24 +1,68 @@
CREATE TABLE IF NOT EXISTS activity_snapshot (
source_hash TEXT PRIMARY KEY,
observed_at TIMESTAMPTZ NOT NULL,
activity_date DATE NOT NULL,
game_model_id TEXT NOT NULL,
game_model_type TEXT NOT NULL,
location_id TEXT NOT NULL,
location_name TEXT NOT NULL,
max_score BIGINT,
total_plays INTEGER,
raw_activity JSONB NOT NULL
-- Individual games, one row per game actually played.
--
-- The Stern API exposes no per-game history feed. The only per-game signal is
-- "most recent game" (per title, and overall), so the collector captures each
-- distinct (title, played_at) tuple as it appears. (title_id, played_at) is the
-- natural key: the API reports the exact play timestamp, so re-observing the
-- same game is idempotent.
CREATE TABLE IF NOT EXISTS game_event (
title_id INTEGER NOT NULL,
played_at TIMESTAMPTZ NOT NULL,
score BIGINT NOT NULL,
title_name TEXT NOT NULL,
title_code TEXT,
first_seen_at TIMESTAMPTZ NOT NULL,
source TEXT NOT NULL,
PRIMARY KEY (title_id, played_at)
);
CREATE INDEX IF NOT EXISTS activity_snapshot_observed_at_idx
ON activity_snapshot (observed_at DESC);
CREATE INDEX IF NOT EXISTS activity_snapshot_model_idx
ON activity_snapshot (game_model_type, game_model_id);
CREATE INDEX IF NOT EXISTS game_event_played_at_idx
ON game_event (played_at DESC);
CREATE INDEX IF NOT EXISTS game_event_title_idx
ON game_event (title_id, played_at DESC);
-- Per-visit rollup from /api/v1/portal/user_activities/. This is aggregate data
-- (a session high score and a play count), kept because it is the only record of
-- games played before this collector started, and because num_plays is what
-- makes missed-game accounting possible.
CREATE TABLE IF NOT EXISTS play_session (
score_key TEXT PRIMARY KEY,
session_number INTEGER NOT NULL,
session_at TIMESTAMPTZ NOT NULL,
location_id INTEGER NOT NULL,
location_name TEXT NOT NULL,
title_name TEXT NOT NULL,
model_type TEXT NOT NULL,
num_plays INTEGER NOT NULL,
high_score BIGINT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS play_session_session_at_idx
ON play_session (session_at DESC);
CREATE TABLE IF NOT EXISTS collection_run (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
observed_at TIMESTAMPTZ NOT NULL,
item_count INTEGER NOT NULL,
inserted_count INTEGER NOT NULL
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
observed_at TIMESTAMPTZ NOT NULL,
events_captured INTEGER NOT NULL,
sessions_upserted INTEGER NOT NULL,
reported_total_plays INTEGER
);
-- Honest accounting: polling can only ever catch one game per title per
-- interval, so this reports how much of the real play history was captured
-- rather than implying the event table is complete.
CREATE OR REPLACE VIEW capture_coverage AS
SELECT
(SELECT count(*) FROM game_event) AS games_captured,
coalesce(
(
SELECT reported_total_plays
FROM collection_run
WHERE reported_total_plays IS NOT NULL
ORDER BY observed_at DESC
LIMIT 1
),
(SELECT sum(num_plays) FROM play_session),
0
) AS games_played_reported;
docker-compose.yml
+1 -1
View File
@@ -22,7 +22,7 @@ services:
API_BASE_URL: ${API_BASE_URL}
API_USERNAME: ${API_USERNAME}
API_PASSWORD: ${API_PASSWORD}
POLL_SECONDS: ${POLL_SECONDS:-300}
POLL_SECONDS: ${POLL_SECONDS:-180}
POSTGRES_HOST: postgres
POSTGRES_PORT: 5432
POSTGRES_DB: ${POSTGRES_DB}
grafana/dashboards/play-history.json
+137 -43
View File
@@ -4,64 +4,158 @@
"tags": ["stern", "scores"],
"timezone": "browser",
"schemaVersion": 39,
"version": 1,
"refresh": "5m",
"version": 2,
"refresh": "1m",
"time": {"from": "now-1h", "to": "now"},
"timepicker": {
"quick_ranges": [
{"display": "Last 1 hour", "from": "now-1h", "to": "now"},
{"display": "Today", "from": "now/d", "to": "now"},
{"display": "Last 24 hours", "from": "now-24h", "to": "now"},
{"display": "Last 7 days", "from": "now-7d", "to": "now"}
]
},
"templating": {
"list": [
{
"name": "game_model_type",
"label": "Game model type",
"name": "title",
"label": "Title",
"type": "query",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"query": "SELECT DISTINCT game_model_type FROM activity_snapshot ORDER BY 1",
"definition": "SELECT DISTINCT game_model_type FROM activity_snapshot ORDER BY 1",
"includeAll": true,
"query": "SELECT 'All' AS __text, '%' AS __value UNION ALL SELECT DISTINCT title_name AS __text, title_name AS __value FROM game_event ORDER BY 1",
"definition": "SELECT 'All' AS __text, '%' AS __value UNION ALL SELECT DISTINCT title_name AS __text, title_name AS __value FROM game_event ORDER BY 1",
"includeAll": false,
"multi": true,
"current": {"text": "All", "value": ["$__all"]}
"current": {"text": "All", "value": ["%"]}
}
]
},
"panels": [
{
"id": 1,
"title": "Maximum score by observed activity",
"type": "timeseries",
"id": 10,
"title": "Capture coverage",
"type": "stat",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 0, "y": 0},
"targets": [
{
"format": "time_series",
"rawSql": "SELECT observed_at AS \"time\", max_score AS \"max score\", game_model_type || ' #' || game_model_id AS metric FROM activity_snapshot WHERE $__timeFilter(observed_at) AND game_model_type IN (${game_model_type:sqlstring}) ORDER BY 1",
"refId": "A"
}
],
"fieldConfig": {"defaults": {"unit": "none"}, "overrides": []}
},
{
"id": 2,
"title": "Total plays by observed activity",
"type": "timeseries",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 12, "y": 0},
"targets": [
{
"format": "time_series",
"rawSql": "SELECT observed_at AS \"time\", total_plays AS \"total plays\", game_model_type || ' #' || game_model_id AS metric FROM activity_snapshot WHERE $__timeFilter(observed_at) AND game_model_type IN (${game_model_type:sqlstring}) ORDER BY 1",
"refId": "A"
}
],
"fieldConfig": {"defaults": {"unit": "none"}, "overrides": []}
},
{
"id": 3,
"title": "Latest collected activity",
"type": "table",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 24, "x": 0, "y": 10},
"gridPos": {"h": 4, "w": 24, "x": 0, "y": 0},
"description": "The API exposes no per-game history, so games are captured by polling. Anything played before the collector started, or a second game on the same title inside one poll interval, is not recoverable.",
"targets": [
{
"format": "table",
"rawSql": "SELECT DISTINCT ON (game_model_type, game_model_id, location_id) observed_at AS \"observed at\", activity_date AS \"activity date\", game_model_type, game_model_id, location_name, max_score, total_plays FROM activity_snapshot WHERE game_model_type IN (${game_model_type:sqlstring}) ORDER BY game_model_type, game_model_id, location_id, observed_at DESC",
"rawSql": "SELECT games_captured AS \"games captured\", games_played_reported AS \"games played (reported)\", games_played_reported - games_captured AS \"not captured\" FROM capture_coverage",
"refId": "A"
}
],
"options": {
"reduceOptions": {"calcs": ["lastNotNull"], "fields": "", "values": false},
"textMode": "value_and_name",
"colorMode": "value"
},
"fieldConfig": {"defaults": {"unit": "none"}, "overrides": []}
},
{
"id": 1,
"title": "Pokémon scores over time",
"type": "timeseries",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 11, "w": 24, "x": 0, "y": 4},
"description": "Each Pokémon score captured by the collector, plotted at the time the game was played.",
"targets": [
{
"format": "time_series",
"rawSql": "SELECT played_at AS \"time\", score, title_name AS metric FROM game_event WHERE $__timeFilter(played_at) AND title_name = 'Pokémon' ORDER BY 1",
"refId": "A"
}
],
"fieldConfig": {
"defaults": {
"unit": "short",
"custom": {
"drawStyle": "line",
"lineInterpolation": "linear",
"pointSize": 6,
"showPoints": "always",
"lineWidth": 3
}
},
"overrides": []
}
},
{
"id": 2,
"title": "Personal best over time",
"type": "timeseries",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 0, "y": 15},
"description": "Running maximum per title, stepped at the moment each new best was set.",
"targets": [
{
"format": "time_series",
"rawSql": "SELECT played_at AS \"time\", max(score) OVER (PARTITION BY title_id ORDER BY played_at) AS \"personal best\", title_name AS metric FROM game_event WHERE $__timeFilter(played_at) AND ('%' = ANY(ARRAY[${title:sqlstring}]::text[]) OR title_name = ANY(ARRAY[${title:sqlstring}]::text[])) ORDER BY 1",
"refId": "A"
}
],
"fieldConfig": {
"defaults": {"unit": "short", "custom": {"drawStyle": "line", "lineInterpolation": "stepAfter", "lineWidth": 2}},
"overrides": []
}
},
{
"id": 3,
"title": "Games captured per day",
"type": "timeseries",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 12, "y": 15},
"targets": [
{
"format": "time_series",
"rawSql": "SELECT date_trunc('day', played_at) AS \"time\", count(*) AS \"games\", title_name AS metric FROM game_event WHERE $__timeFilter(played_at) AND ('%' = ANY(ARRAY[${title:sqlstring}]::text[]) OR title_name = ANY(ARRAY[${title:sqlstring}]::text[])) GROUP BY 1, title_name ORDER BY 1",
"refId": "A"
}
],
"fieldConfig": {
"defaults": {"unit": "none", "custom": {"drawStyle": "bars", "fillOpacity": 70, "stacking": {"mode": "normal"}}},
"overrides": []
}
},
{
"id": 4,
"title": "Per-title summary",
"type": "table",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 0, "y": 25},
"targets": [
{
"format": "table",
"rawSql": "SELECT title_name AS \"title\", count(*) AS \"games captured\", max(score) AS \"best\", round(avg(score)) AS \"average\", max(played_at) AS \"last played\" FROM game_event WHERE ('%' = ANY(ARRAY[${title:sqlstring}]::text[]) OR title_name = ANY(ARRAY[${title:sqlstring}]::text[])) GROUP BY title_name ORDER BY max(score) DESC",
"refId": "A"
}
]
},
{
"id": 5,
"title": "Recent games",
"type": "table",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 12, "x": 12, "y": 25},
"targets": [
{
"format": "table",
"rawSql": "SELECT played_at AS \"played at\", title_name AS \"title\", score, source AS \"captured via\" FROM game_event WHERE ('%' = ANY(ARRAY[${title:sqlstring}]::text[]) OR title_name = ANY(ARRAY[${title:sqlstring}]::text[])) ORDER BY played_at DESC LIMIT 100",
"refId": "A"
}
]
},
{
"id": 6,
"title": "Visits (session rollup, includes pre-collector history)",
"type": "table",
"datasource": {"type": "postgres", "uid": "play-history-postgres"},
"gridPos": {"h": 10, "w": 24, "x": 0, "y": 35},
"description": "Aggregate per-visit data from user_activities. Individual scores within a visit are not exposed by the API; only the visit's high score and play count are.",
"targets": [
{
"format": "table",
"rawSql": "SELECT session_at AS \"visit\", location_name AS \"location\", title_name AS \"title\", model_type AS \"model\", num_plays AS \"plays\", high_score AS \"visit best\" FROM play_session ORDER BY session_at DESC LIMIT 200",
"refId": "A"
}
]
tests/fixtures/api_responses.json
Vendored
+60
View File
@@ -0,0 +1,60 @@
{
"user_stats": {
"stats": {
"total_plays": 18,
"titles": [
{"id": 101, "name": "Example Game 1", "code": "game-1"},
{"id": 102, "name": "Example Game 2", "code": "game-2"},
{"id": 103, "name": "Example Game 3", "code": "game-3"},
{"id": 104, "name": "Example Game 4", "code": "game-4"},
{"id": 105, "name": "Example Game 5", "code": "game-5"},
{"id": 106, "name": "Example Game 6", "code": "game-6"},
{"id": 107, "name": "Example Game 7", "code": "game-7"},
{"id": 108, "name": "Example Game 8", "code": "game-8"},
{"id": 133, "name": "Pokémon", "code": "pokemon"}
]
}
},
"user_title_stats": {
"101": {"most_recent_date": "2026-01-01T12:00:00Z", "most_recent_score": 100, "max_score": 1000},
"102": {"most_recent_date": "2026-01-02T12:00:00Z", "most_recent_score": 200, "max_score": 1000},
"103": {"most_recent_date": "2026-01-03T12:00:00Z", "most_recent_score": 300, "max_score": 1000},
"104": {"most_recent_date": "2026-01-04T12:00:00Z", "most_recent_score": 400, "max_score": 1000},
"105": {"most_recent_date": "2026-01-05T12:00:00Z", "most_recent_score": 500, "max_score": 1000},
"106": {"most_recent_date": "2026-01-06T12:00:00Z", "most_recent_score": 600, "max_score": 1000},
"107": {"most_recent_date": "2026-01-07T12:00:00Z", "most_recent_score": 700, "max_score": 1000},
"108": {"most_recent_date": "2026-01-08T12:00:00Z", "most_recent_score": 800, "max_score": 1000},
"133": {"most_recent_date": "2026-01-09T12:00:00Z", "most_recent_score": 900, "max_score": 1000}
},
"user_highlights": {
"highlights": {
"most_recent_game_played": {
"title": {"id": 133, "name": "Pokémon", "code": "pokemon"},
"last_played_date": "2026-01-09T12:00:00Z",
"last_played_score": 900
}
}
},
"user_activities": {
"user": {
"location_sessions": [
{
"number": 1,
"location": {"id": 1, "name": "Example Arcade"},
"games_played": [
{"score_key": "example-1", "name": "Example Game 1", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 1000},
{"score_key": "example-2", "name": "Example Game 2", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 2000},
{"score_key": "example-3", "name": "Example Game 3", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 3000},
{"score_key": "example-4", "name": "Example Game 4", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 4000},
{"score_key": "example-5", "name": "Example Game 5", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 5000},
{"score_key": "example-6", "name": "Example Game 6", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 6000},
{"score_key": "example-7", "name": "Example Game 7", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 7000},
{"score_key": "example-8", "name": "Example Game 8", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 8000},
{"score_key": "example-9", "name": "Pokémon", "model": {"type": "Premium"}, "num_plays": 2, "high_score": 9000}
],
"date": "2026-01-09T13:00:00Z"
}
]
}
}
}
tests/test_collector.py
+158
View File
@@ -0,0 +1,158 @@
#!/usr/bin/env python3
"""Parsing tests for the collector, run against synthetic API-shaped responses.
Run: python3 -m unittest discover -s tests -v
"""
from __future__ import annotations
import json
import sys
import unittest
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.collector import (
event_from_highlights,
event_from_title_stats,
parse_int,
parse_timestamp,
reported_total_plays,
sessions_from_activities,
titles_played,
)
FIXTURES = json.loads(
(Path(__file__).parent / "fixtures" / "api_responses.json").read_text()
)
class TestTimestamps(unittest.TestCase):
def test_parses_utc_z_form(self):
parsed = parse_timestamp("2026-08-01T04:19:27.972205Z")
self.assertEqual(parsed, datetime(2026, 8, 1, 4, 19, 27, 972205, tzinfo=timezone.utc))
def test_parses_offset_form_and_normalizes_to_utc(self):
parsed = parse_timestamp("2026-07-31T23:19:27.972205-05:00")
self.assertEqual(parsed, datetime(2026, 8, 1, 4, 19, 27, 972205, tzinfo=timezone.utc))
def test_offset_and_z_forms_are_the_same_instant(self):
# The two endpoints report the same game in different notations; if these
# diverged, the same game would be stored twice under different keys.
self.assertEqual(
parse_timestamp("2026-08-01T04:19:27.972205Z"),
parse_timestamp("2026-07-31T23:19:27.972205-05:00"),
)
def test_rejects_junk(self):
for value in (None, "", "not-a-date", 17, {}):
self.assertIsNone(parse_timestamp(value))
class TestParseInt(unittest.TestCase):
def test_accepts_numeric_forms(self):
self.assertEqual(parse_int(5), 5)
self.assertEqual(parse_int("5"), 5)
def test_rejects_empty_and_bool(self):
self.assertIsNone(parse_int(None))
self.assertIsNone(parse_int(""))
self.assertIsNone(parse_int(True))
self.assertIsNone(parse_int("abc"))
class TestTitleStats(unittest.TestCase):
def test_lists_every_played_title(self):
titles = titles_played(FIXTURES["user_stats"])
self.assertEqual(len(titles), 9)
self.assertTrue(all(isinstance(t["id"], int) for t in titles))
def test_reports_total_plays(self):
expected = FIXTURES["user_stats"]["stats"]["total_plays"]
self.assertEqual(reported_total_plays(FIXTURES["user_stats"]), expected)
self.assertIsNone(reported_total_plays({}))
def test_builds_an_event_per_title(self):
titles = {t["id"]: t for t in titles_played(FIXTURES["user_stats"])}
events = []
for title_id, payload in FIXTURES["user_title_stats"].items():
event = event_from_title_stats(titles[int(title_id)], payload)
self.assertIsNotNone(event, f"title {title_id} produced no event")
events.append(event)
self.assertEqual(len(events), 9)
self.assertTrue(all(e.score > 0 for e in events))
self.assertEqual(len({(e.title_id, e.played_at) for e in events}), 9)
def test_captured_score_is_a_single_game_not_the_title_best(self):
# This is the whole reason the collector switched endpoints: most_recent_score
# is one game's score, whereas my_activity only ever exposed a max.
pokemon = FIXTURES["user_title_stats"]["133"]
self.assertLess(pokemon["most_recent_score"], pokemon["max_score"])
event = event_from_title_stats({"id": 133, "name": "Pokemon"}, pokemon)
self.assertEqual(event.score, pokemon["most_recent_score"])
def test_missing_fields_yield_no_event(self):
title = {"id": 1, "name": "X"}
self.assertIsNone(event_from_title_stats(title, {}))
self.assertIsNone(event_from_title_stats(title, {"most_recent_date": "2026-01-01T00:00:00Z"}))
self.assertIsNone(event_from_title_stats(title, {"most_recent_score": 10}))
class TestHighlights(unittest.TestCase):
def test_extracts_most_recent_game(self):
event = event_from_highlights(FIXTURES["user_highlights"])
self.assertIsNotNone(event)
self.assertEqual(event.title_id, 133)
self.assertEqual(event.source, "user_highlights")
def test_agrees_with_title_stats_for_the_same_game(self):
# Both endpoints describe the newest game; they must key identically or
# every poll would insert a duplicate row for it.
highlight = event_from_highlights(FIXTURES["user_highlights"])
titles = {t["id"]: t for t in titles_played(FIXTURES["user_stats"])}
from_stats = event_from_title_stats(
titles[highlight.title_id],
FIXTURES["user_title_stats"][str(highlight.title_id)],
)
self.assertEqual(highlight.played_at, from_stats.played_at)
self.assertEqual(highlight.score, from_stats.score)
def test_tolerates_missing_payload(self):
self.assertIsNone(event_from_highlights({}))
self.assertIsNone(event_from_highlights({"highlights": {}}))
self.assertIsNone(event_from_highlights({"highlights": {"most_recent_game_played": None}}))
class TestSessions(unittest.TestCase):
def test_flattens_sessions_into_rows(self):
rows = sessions_from_activities(FIXTURES["user_activities"])
self.assertGreater(len(rows), 0)
self.assertEqual(len({r.score_key for r in rows}), len(rows), "score_key must be unique")
def test_rows_carry_usable_values(self):
for row in sessions_from_activities(FIXTURES["user_activities"]):
self.assertGreaterEqual(row.num_plays, 1)
self.assertGreater(row.high_score, 0)
self.assertIsNotNone(row.session_at.tzinfo)
self.assertTrue(row.location_name)
def test_session_plays_exceed_capturable_events(self):
# Documents the known limitation: far more games were played than polling
# can ever recover, so the dashboard must report coverage honestly.
rows = sessions_from_activities(FIXTURES["user_activities"])
total_plays = sum(r.num_plays for r in rows)
self.assertGreater(total_plays, len(FIXTURES["user_title_stats"]))
def test_tolerates_malformed_input(self):
self.assertEqual(sessions_from_activities({}), [])
self.assertEqual(sessions_from_activities({"user": {}}), [])
self.assertEqual(sessions_from_activities({"user": {"location_sessions": "x"}}), [])
self.assertEqual(
sessions_from_activities({"user": {"location_sessions": [{"date": None}]}}), []
)
if __name__ == "__main__":
unittest.main()
tests/test_dashboard.py
+57
View File
@@ -0,0 +1,57 @@
#!/usr/bin/env python3
"""Static regression checks for empty-data-safe dashboard queries."""
from __future__ import annotations
import json
import unittest
from pathlib import Path
DASHBOARD = json.loads(
(Path(__file__).resolve().parent.parent / "grafana/dashboards/play-history.json").read_text()
)
class TestEmptyDashboard(unittest.TestCase):
def test_dashboard_defaults_to_last_hour_with_useful_quick_ranges(self):
self.assertEqual(DASHBOARD["time"], {"from": "now-1h", "to": "now"})
ranges = {
item["display"]: (item["from"], item["to"])
for item in DASHBOARD["timepicker"]["quick_ranges"]
}
self.assertEqual(ranges["Last 1 hour"], ("now-1h", "now"))
self.assertEqual(ranges["Today"], ("now/d", "now"))
def test_title_query_always_supplies_all_sentinel(self):
title = next(item for item in DASHBOARD["templating"]["list"] if item["name"] == "title")
self.assertIn("'%' AS __value", title["query"])
self.assertEqual(title["current"]["value"], ["%"])
def test_title_filtered_queries_accept_all_sentinel(self):
filtered = []
for panel in DASHBOARD["panels"]:
for target in panel.get("targets", []):
sql = target.get("rawSql", "")
if "${title:sqlstring}" in sql:
filtered.append(sql)
empty = sql.replace("${title:sqlstring}", "")
all_titles = sql.replace("${title:sqlstring}", "'%'")
self.assertIn("ARRAY[]::text[]", empty)
self.assertIn("'%' = ANY(ARRAY['%']::text[])", all_titles)
self.assertEqual(len(filtered), 4)
def test_pokemon_time_series_is_dedicated(self):
panel = next(panel for panel in DASHBOARD["panels"] if panel["id"] == 1)
self.assertEqual(panel["title"], "Pokémon scores over time")
self.assertEqual(panel["type"], "timeseries")
self.assertIn("title_name = 'Pokémon'", panel["targets"][0]["rawSql"])
style = panel["fieldConfig"]["defaults"]["custom"]
self.assertEqual(style["drawStyle"], "line")
self.assertEqual(style["lineInterpolation"], "linear")
self.assertGreater(style["lineWidth"], 0)
self.assertEqual(style["showPoints"], "always")
if __name__ == "__main__":
unittest.main()