"""Design system for the agent Space — tokens, CSS, altair theme, components.
Everything visual that is shared between pages lives here so a page only
composes: ``page_header``, ``status_pill``, ``kpi_row``, ``empty_state``,
``chip``, ``feed`` and the altair theme. Nothing in this module talks to the
database or changes behavior; it is presentation only.
Palette / chrome follow the project design system (colorblind-validated
categorical order, status colors reserved for state, one custom light theme
with a serviceable dark variant picked from ``st.context.theme``).
"""
from __future__ import annotations
import html as _html
from datetime import datetime, timezone
from typing import Iterable, Optional
import streamlit as st
# ---------------------------------------------------------------------------
# tokens
# ---------------------------------------------------------------------------
CATEGORICAL_LIGHT = ["#2a78d6", "#eb6834", "#1baf7a", "#eda100",
"#e87ba4", "#008300", "#4a3aa7", "#e34948"]
CATEGORICAL_DARK = ["#3987e5", "#d95926", "#199e70", "#c98500",
"#d55181", "#008300", "#9085e9", "#e66767"]
SEQUENTIAL = ["#cde2fb", "#9ec5f4", "#6da7ec", "#3987e5",
"#256abf", "#184f95", "#0d366b"]
STATUS = { # reserved for state — never used as series colors
"good": "#0ca30c",
"warning": "#fab219",
"serious": "#ec835a",
"critical": "#d03b3b",
}
LIGHT = dict(
ink="#0b0b0b", ink2="#52514e", muted="#898781",
grid="#e1e0d9", baseline="#c3c2b7", hairline="rgba(11,11,11,0.10)",
surface="#fcfcfb", plane="#f9f9f7", success_text="#006300",
tint="rgba(11,11,11,0.035)", primary="#2a78d6",
)
DARK = dict(
ink="#f3f3f1", ink2="#c2c1bb", muted="#8f8e88",
grid="#33332f", baseline="#4a4a45", hairline="rgba(255,255,255,0.12)",
surface="#151515", plane="#0e0e0e", success_text="#5fd35f",
tint="rgba(255,255,255,0.05)", primary="#3987e5",
)
FONT = 'system-ui, -apple-system, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif'
MONO = 'ui-monospace, SFMono-Regular, Menlo, Consolas, "Liberation Mono", monospace'
def is_dark() -> bool:
"""True only when the *configured* Streamlit theme is dark.
Deliberately NOT st.context.theme.type: with a single custom light theme
that call reports the viewer's OS preference ("dark" on a dark desktop)
while Streamlit still renders the light theme — injecting dark tokens on
a light page. The config is the one source of truth for the whole app.
"""
try:
return (st.get_option("theme.base") or "light") == "dark"
except Exception:
return False
def tokens() -> dict:
return DARK if is_dark() else LIGHT
def series() -> list[str]:
"""Categorical series colors in fixed order (color follows the entity)."""
return CATEGORICAL_DARK if is_dark() else CATEGORICAL_LIGHT
def esc(x) -> str:
"""HTML-escape anything (None/NaN-proof)."""
if x is None:
return ""
try:
import pandas as pd
if isinstance(x, float) and pd.isna(x):
return ""
except Exception:
pass
return _html.escape(str(x), quote=True)
def fmt_int(n) -> str:
"""Thousands-separated integer, '—' for None/NaN."""
try:
if n is None:
return "—"
import pandas as pd
if pd.isna(n):
return "—"
return f"{int(n):,}"
except Exception:
return "—"
def rel_time(ts) -> str:
"""'3 min ago' style relative time; '' if unparsable."""
if ts is None:
return ""
try:
import pandas as pd
if pd.isna(ts):
return ""
t = pd.Timestamp(ts)
if t.tzinfo is None:
t = t.tz_localize("UTC")
now = pd.Timestamp.now(tz="UTC")
s = int((now - t).total_seconds())
except Exception:
return ""
if s < 0:
s = 0
if s < 45:
return "just now"
if s < 3600:
return f"{s // 60} min ago"
if s < 86400:
h = s // 3600
return f"{h} h ago"
d = s // 86400
return f"{d} day{'s' if d != 1 else ''} ago"
def fmt_ts(ts, fmt: str = "%b %d, %H:%M") -> str:
if ts is None:
return ""
try:
import pandas as pd
if pd.isna(ts):
return ""
return pd.Timestamp(ts).strftime(fmt)
except Exception:
return str(ts)
# ---------------------------------------------------------------------------
# CSS (single place)
# ---------------------------------------------------------------------------
def _css(t: dict) -> str:
good, warn, serious, crit = (STATUS["good"], STATUS["warning"],
STATUS["serious"], STATUS["critical"])
return f"""
"""
def inject_css() -> None:
"""Inject the design-system CSS once per run (call from app.py)."""
st.html(_css(tokens()))
# ---------------------------------------------------------------------------
# altair theme
# ---------------------------------------------------------------------------
def _altair_config(t: dict, cats: list[str]) -> dict:
axis = dict(labelColor=t["muted"], titleColor=t["ink2"], gridColor=t["grid"],
domainColor=t["baseline"], tickColor=t["baseline"], labelFont=FONT,
titleFont=FONT, labelFontSize=11, titleFontSize=11, titleFontWeight=600,
labelPadding=6, titlePadding=8, gridDash=[1, 0], gridWidth=1, ticks=False,
labelFlush=True)
return {
"config": {
"background": t["surface"],
"view": {"stroke": None, "continuousWidth": 400, "continuousHeight": 220},
"font": FONT,
"padding": {"left": 4, "top": 6, "right": 8, "bottom": 4},
"title": {"font": FONT, "fontSize": 13, "fontWeight": 650, "color": t["ink"],
"anchor": "start", "offset": 10, "subtitleColor": t["muted"],
"subtitleFontSize": 11, "subtitleFont": FONT},
"axis": axis,
"axisX": {**axis, "grid": False, "domain": True},
"axisY": {**axis, "grid": True, "domain": False},
"legend": {"labelColor": t["ink2"], "titleColor": t["ink2"], "labelFont": FONT,
"titleFont": FONT, "labelFontSize": 11, "titleFontSize": 11,
"symbolSize": 70, "symbolType": "circle", "orient": "top",
"direction": "horizontal", "offset": 6, "padding": 0},
"range": {"category": cats, "ordinal": SEQUENTIAL, "ramp": SEQUENTIAL,
"heatmap": SEQUENTIAL},
"bar": {"color": cats[0], "cornerRadiusTopLeft": 4, "cornerRadiusTopRight": 4,
"binSpacing": 2, "stroke": None},
"line": {"color": cats[0], "strokeWidth": 2, "interpolate": "linear"},
"area": {"color": cats[0], "opacity": 0.18, "line": True},
"point": {"color": cats[0], "size": 46, "filled": True},
"rule": {"color": t["baseline"]},
"text": {"color": t["ink2"], "font": FONT, "fontSize": 11},
"mark": {"tooltip": True},
}
}
def register_altair_theme() -> None:
"""Register + enable the 'aim' altair theme (works with altair 4/5/6)."""
try:
import altair as alt
except Exception:
return
t = tokens()
cats = series()
cfg = _altair_config(t, cats)
def _aim_theme():
return cfg
try: # altair >= 5.5 / 6
alt.theme.register("aim", enable=True)(_aim_theme)
return
except Exception:
pass
try: # altair 4 / 5.0-5.4
alt.themes.register("aim", _aim_theme)
alt.themes.enable("aim")
except Exception:
pass
def chart(c, key: Optional[str] = None, height: Optional[int] = None) -> None:
"""Render an altair chart with the AIM theme (not Streamlit's built-in one)."""
kwargs = {"theme": None}
if key:
kwargs["key"] = key
try:
st.altair_chart(c, width="stretch", **kwargs)
except TypeError: # older Streamlit: use_container_width
st.altair_chart(c, use_container_width=True, **kwargs)
# ---------------------------------------------------------------------------
# components
# ---------------------------------------------------------------------------
_PILL_ICON = {"ok": "●", "running": None, "stale": "▲", "warn": "▲", "crashed": "■",
"error": "■", "idle": "○", "off": "○", "info": "○", "new": "✦"}
def pill_html(kind: str, text: str) -> str:
kind = (kind or "idle").lower()
if kind not in _PILL_ICON:
kind = {"ok": "ok", "success": "ok", "warning": "warn", "critical": "crashed",
"failed": "crashed", "on": "ok"}.get(kind, "idle")
icon = _PILL_ICON.get(kind)
lead = '' if icon is None else f'{icon}'
return f'{lead}{esc(text)}'
def status_pill(kind: str, text: str) -> None:
st.html(pill_html(kind, text))
def chip_html(text: str, kind: str = "muted", title: str = "") -> str:
kind = (kind or "muted").lower()
t = f' title="{esc(title)}"' if title else ""
return f'{esc(text)}'
def chips_html(items: Iterable[tuple], gap_label: bool = True) -> str:
"""items: (text, kind) or (label, text, kind) → one flex row of chips."""
parts = []
for it in items:
if len(it) == 2:
text, kind = it
parts.append(chip_html(text, kind))
else:
label, text, kind = it
parts.append(f'{esc(label)}{esc(text)}')
return '' + "".join(parts) + "
"
def level_chip_html(level: str) -> str:
lv = (level or "info").lower()
kind = {"info": "info", "warn": "warn", "warning": "warn", "error": "error",
"critical": "error"}.get(lv, "info")
return chip_html(lv, kind)
def page_header(title: str, subtitle: str = "", right: Optional[str] = None) -> None:
"""H1 + one-line muted subtitle + optional right-aligned pill HTML."""
right_html = f'{right}
' if right else ""
st.html(
f'{esc(title)}
'
f'
{esc(subtitle)}
{right_html}
{esc(delta)}
' if delta else ""
cls = "aim-kpi-value small" if small else "aim-kpi-value"
tiles.append(f'{label}{help_html}
'
f'
{val}
{delta_html}
' + "".join(tiles) + "
")
def empty_state(icon: str, title: str, hint: str = "") -> None:
st.html(f'{esc(icon)}
'
f'
{esc(title)}
{esc(hint)}
{esc(title)} {sub_html}{right_html}
')
def _is_traceback(msg: str) -> bool:
return ("Traceback" in msg) or ("\n" in msg.strip())
def feed_html(rows: Iterable[dict], empty: str = "No events yet.") -> str:
"""rows: dicts with ts, level, node, message → compact event feed."""
out = []
for r in rows:
msg = str(r.get("message") or "")
if _is_traceback(msg):
head, _, tail = msg.partition("\n")
body = esc(head) + (f"{esc(tail)}" if tail.strip() else "")
else:
body = esc(msg)
ts = r.get("ts")
out.append(
f'{level_chip_html(r.get("level"))}
'
f'
{esc(r.get("node") or "")}
'
f'
{body}
'
f'
{esc(rel_time(ts))}'
f'{esc(fmt_ts(ts, "%b %d, %H:%M"))}
' + "".join(out) + "
"
def feed(rows: Iterable[dict], empty: str = "No events yet.") -> None:
st.html(feed_html(rows, empty))
def run_status_kind(status) -> str:
s = str(status or "").lower()
return {"ok": "ok", "running": "running", "stale": "stale", "crashed": "crashed",
"error": "error", "timeout": "crashed"}.get(s, "idle")
def status_cell_style(status) -> str:
"""pandas Styler CSS for a run/level status cell (word + color, never color alone)."""
s = str(status or "").lower()
if s in ("ok", "ingested", "info", "good"):
return "background-color: rgba(12,163,12,.14); color: #0a6a0a; font-weight: 600;"
if s in ("running",):
return "background-color: rgba(42,120,214,.14); color: #1b5cab; font-weight: 600;"
if s in ("stale", "warn", "warning") or s.startswith("skipped"):
return "background-color: rgba(250,178,25,.22); color: #7a5200; font-weight: 600;"
if s in ("crashed", "error", "failed", "critical", "timeout"):
return "background-color: rgba(208,59,59,.14); color: #9c1f1f; font-weight: 600;"
if s.startswith("flag"):
return "background-color: rgba(250,178,25,.22); color: #7a5200; font-weight: 600;"
return ""
def brand_row(db_name: str, extra: str = "") -> None:
st.html(
'AIM for Composites '
'· Autonomous Ingestion Agent
'
f'
database: {esc(db_name or "?")} · read-only for visitors{esc(extra)}