"""The dashboard definition, as the hub stores and accepts it.
These models mirror the web app's ``experiment-dashboard-definition.ts``
one-to-one, with the panel fields it takes from ``experiments.ts``; what
a person writes is the builder in :mod:`.builder`, which converts to and
from them. The ``AGENTS.md`` next to this module says how the two are
kept in step.
"""
from __future__ import annotations
import re
from collections.abc import Iterable, Mapping
from typing import Any, Literal, TypeVar
from pydantic import (
BaseModel,
ConfigDict,
Field,
ModelWrapValidatorHandler,
ValidationInfo,
field_validator,
model_validator,
)
from pydantic.alias_generators import to_camel
from embedl_hub._internal.tracking.errors import TrackingUsageError
DASHBOARD_SCHEMA_VERSION = "0.0.1"
MAX_DASHBOARD_PANELS = 36
MAX_EXPERIMENT_PANEL_ITEMS = 50
MAX_SMOOTHING = 0.99
MAX_DASHBOARD_NAME_LENGTH = 100
MAX_DASHBOARD_DESCRIPTION_LENGTH = 500
MAX_PANEL_TITLE_LENGTH = 200
Aggregation = Literal["latest", "final", "min", "max", "best"]
ItemAggregation = Literal["latest", "final", "min", "max"]
BestDirection = Literal["min", "max"]
PanelWidth = Literal["half", "full"]
PanelItemKind = Literal[
"attribute", "param", "metric", "tag", "externalLink", "artifact"
]
PanelType = Literal[
"line", "bar", "scatter", "parallel", "table", "runDetails"
]
RunAttributeName = Literal[
"status", "type", "author", "startedAt", "endedAt", "duration"
]
_VERSION_PATTERN = re.compile(r"([0-9]+)\.([0-9]+)\.([0-9]+)")
_OWN_VERSION = tuple(
int(part)
for part in _VERSION_PATTERN.fullmatch(DASHBOARD_SCHEMA_VERSION).groups() # type: ignore[union-attr]
)
_ACCEPTED_VERSIONS = f"{_OWN_VERSION[0]}.{_OWN_VERSION[1]}.x"
_IGNORE_UNKNOWN_FIELDS = "ignore_unknown_fields"
T = TypeVar("T")
[docs]
class DashboardSchemaError(TrackingUsageError):
"""Raised when a dashboard does not fit this SDK's schema.
Covers a file the SDK cannot read as well as a :class:`Dashboard`
whose sections and panels do not form a valid layout. Like every
:class:`TrackingUsageError` it escapes a safe scope: the dashboard is
the caller's to fix, not a Hub failure to tolerate.
"""
def check_dashboard_schema_version(
config: Mapping[str, Any],
) -> tuple[int, int, int]:
"""Refuse a dashboard definition this SDK cannot read.
A definition is readable when its ``schemaVersion`` shares the first
two components with :data:`DASHBOARD_SCHEMA_VERSION`; the third may
differ. The integer ``version`` markers the hub still reads from
older stored dashboards are refused.
:param config: The parsed dashboard definition.
:return: The version as its three components.
:raises DashboardSchemaError: If the version is missing, malformed,
obsolete, or differs in a paradigm or breaking component.
"""
version = config.get("schemaVersion")
match = (
_VERSION_PATTERN.fullmatch(version)
if isinstance(version, str)
else None
)
if match is None:
obsolete = config.get("version")
if version is None and isinstance(obsolete, int):
raise DashboardSchemaError(
"The dashboard config carries the obsolete integer version "
f"{obsolete}. This embedl-hub writes {DASHBOARD_SCHEMA_VERSION} "
f"and reads {_ACCEPTED_VERSIONS}; rebuild the dashboard and "
"export it again with Dashboard.save()."
)
raise DashboardSchemaError(
"The dashboard config has no schemaVersion of the form X.Y.Z "
f"(found {version!r}). This embedl-hub writes "
f"{DASHBOARD_SCHEMA_VERSION} and reads {_ACCEPTED_VERSIONS}; "
"export the dashboard again with Dashboard.save()."
)
major, minor, patch = (int(part) for part in match.groups())
if (major, minor) != _OWN_VERSION[:2]:
raise DashboardSchemaError(
f"Dashboard schema {version} is not supported by this "
f"embedl-hub, which uses {DASHBOARD_SCHEMA_VERSION} and reads "
f"{_ACCEPTED_VERSIONS}. Upgrade embedl-hub or rebuild the "
"dashboard with this version."
)
return major, minor, patch
def _utf16_length(value: str) -> int:
return len(value.encode("utf-16-le")) // 2
class _Model(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
validate_by_name=True,
validate_by_alias=True,
serialize_by_alias=True,
validate_assignment=True,
extra="forbid",
)
@model_validator(mode="wrap")
@classmethod
def _ignore_fields_of_a_newer_patch(
cls,
data: Any,
handler: ModelWrapValidatorHandler[Any],
info: ValidationInfo,
) -> Any:
if (
isinstance(data, Mapping)
and info.context
and info.context.get(_IGNORE_UNKNOWN_FIELDS)
):
known = set(cls.model_fields)
known.update(
field.alias
for field in cls.model_fields.values()
if field.alias is not None
)
data = {key: value for key, value in data.items() if key in known}
return handler(data)
class Axis(_Model):
"""A metric or parameter read along one axis of a panel."""
source: Literal["metric", "param"] = "metric"
name: str
aggregation: Aggregation | None = None
best_direction: BestDirection | None = None
class PanelItem(_Model):
"""One row of a run details panel.
``kind`` says where ``name`` is looked up on the run: a run
attribute (``status``, ``type``, ``author``, ``startedAt``,
``endedAt`` or ``duration``), a parameter, a metric, a tag, an
external link or an artifact.
"""
kind: PanelItemKind
name: str = Field(min_length=1)
aggregation: ItemAggregation | None = None
@model_validator(mode="after")
def _attribute_names_are_known(self) -> PanelItem:
if (
self.kind == "attribute"
and self.name not in RunAttributeName.__args__
):
raise ValueError(
f"{self.name!r} is not a run attribute. Use one of "
f"{', '.join(RunAttributeName.__args__)}."
)
return self
class Filters(_Model):
"""Metrics, tags and parameters a dashboard or panel narrows its runs to."""
metrics: list[str] = Field(default_factory=list)
tags: dict[str, list[str]] = Field(default_factory=dict)
params: dict[str, list[str]] = Field(default_factory=dict)
class Table(_Model):
"""Columns and default sort of the run comparison table."""
columns: list[str] = Field(default_factory=list)
default_sort: str = ""
class Chart(_Model):
"""The smoothing the chart controls open with.
Panels carry their own smoothing; this only seeds the dashboard's
control.
"""
smoothing: float = Field(default=0, ge=0, le=MAX_SMOOTHING)
class Panel(_Model):
"""Any dashboard panel.
The subclasses fix ``type`` and name the fields that type reads;
this base carries every field the hub stores, so a dashboard read
back from the hub or a file keeps all of them. ``revision`` is the
hub's edit counter: read from a stored dashboard, never written.
"""
id: str | None = Field(default=None, min_length=1)
revision: int | None = Field(default=None, gt=0, exclude=True)
type: PanelType
title: str
width: PanelWidth = "half"
run_id: str | None = None
items: list[PanelItem] | None = Field(
default=None, max_length=MAX_EXPERIMENT_PANEL_ITEMS
)
metric_names: list[str] | None = None
metric_name: str | None = None
aggregation: Aggregation | None = None
best_direction: BestDirection | None = None
x_axis: Axis | None = None
y_axis: Axis | None = None
axes: list[Axis] | None = None
smoothing: float | None = Field(default=None, ge=0, le=MAX_SMOOTHING)
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[str] = Field(default_factory=list)
table: Table = Field(default_factory=Table)
@field_validator("items")
@classmethod
def _items_are_distinct(
cls, items: list[PanelItem] | None
) -> list[PanelItem] | None:
repeated = _duplicates((item.kind, item.name) for item in items or [])
if repeated:
listed = ", ".join(f"{kind} {name!r}" for kind, name in repeated)
raise ValueError(
f"Items are listed more than once: {listed}. The hub keeps "
"the first item of each kind and name."
)
return items
class LinePanel(Panel):
"""Metric curves over steps, one line per run and metric."""
type: Literal["line"] = "line"
metric_names: list[str]
class BarPanel(Panel):
"""One aggregated metric value per run."""
type: Literal["bar"] = "bar"
metric_name: str
aggregation: Aggregation = "latest"
best_direction: BestDirection = "max"
class ScatterPanel(Panel):
"""Runs as points over two metrics or parameters."""
type: Literal["scatter"] = "scatter"
x_axis: Axis
y_axis: Axis
class ParallelPanel(Panel):
"""Runs as lines across several metric or parameter axes."""
type: Literal["parallel"] = "parallel"
axes: list[Axis]
class TablePanel(Panel):
"""The run comparison table, one row per selected run."""
type: Literal["table"] = "table"
width: PanelWidth = "full"
class RunDetailsPanel(Panel):
"""Chosen values of one run, one row each, in the order given.
``run_id`` names the run; the hub falls back to the first selected
run when it is unset or no longer selected.
"""
type: Literal["runDetails"] = "runDetails"
run_id: str | None = None
items: list[PanelItem] = Field(
default_factory=list, max_length=MAX_EXPERIMENT_PANEL_ITEMS
)
class Section(_Model):
"""A titled group of panels, referenced by their ids.
``revision`` is the hub's edit counter: read, never written.
"""
id: str | None = Field(default=None, min_length=1)
revision: int | None = Field(default=None, gt=0, exclude=True)
title: str = "Charts"
collapsed: bool = False
panel_ids: list[str] = Field(default_factory=list)
class DashboardDefinition(_Model):
"""The dashboard definition the hub stores and the API accepts as ``config``."""
schema_version: str = DASHBOARD_SCHEMA_VERSION
filters: Filters = Field(default_factory=Filters)
chart: Chart = Field(default_factory=Chart)
table: Table = Field(default_factory=Table)
sections: list[Section] = Field(default_factory=list)
panels: list[Panel] = Field(
default_factory=list, max_length=MAX_DASHBOARD_PANELS
)
def _duplicates(values: Iterable[T]) -> list[T]:
seen: set[T] = set()
repeated: set[T] = set()
for value in values:
if value in seen:
repeated.add(value)
seen.add(value)
return sorted(repeated)
def _fill_ids(given: list[str | None], prefix: str) -> list[str]:
taken = {value for value in given if value is not None}
filled: list[str] = []
for position, value in enumerate(given, start=1):
if value is None:
number = position
while f"{prefix}-{number}" in taken:
number += 1
value = f"{prefix}-{number}"
taken.add(value)
filled.append(value)
return filled
