"""The dashboard builder: what a person writes to describe a dashboard.
A :class:`Dashboard` is a name and a list of :class:`Section` objects,
each holding the panels it shows. Panels take the values they plot as
:class:`Metric`, :class:`Param` and, for a :class:`RunDetailsPanel`,
:class:`Attribute`, :class:`Tag`, :class:`Link` and :class:`Artifact`.
Nothing here is the hub's JSON; :meth:`Dashboard.to_definition` produces
that from the schema models in :mod:`.schema`, and :meth:`Dashboard.load`
reads it back. ``AGENTS.md`` next to this module lists where the two
deliberately differ.
.. code-block:: python
from embedl_hub.tracking.dashboard import (
Attribute, Dashboard, LinePanel, Metric, Param, RunDetailsPanel,
Section,
)
dashboard = Dashboard(
"Training curves",
[
Section("Curves", [LinePanel("Loss", ["loss", "val_loss"])]),
Section(
"This run",
[
RunDetailsPanel(
"Run details",
[Attribute.STATUS, Param("lr"), Metric("val_acc", "max")],
)
],
collapsed=True,
),
],
)
client.log_dashboard(dashboard)
dashboard.save("training-dashboard.json")
"""
from __future__ import annotations
import functools
import json
from collections.abc import Mapping, Sequence
from dataclasses import KW_ONLY, fields
from enum import Enum
from pathlib import Path
from typing import Annotated, Any, Literal, TypeVar
from pydantic import ConfigDict, Field, ValidationError
from pydantic.dataclasses import dataclass
from embedl_hub._internal.tracking.dashboard import schema
from embedl_hub._internal.tracking.dashboard.schema import (
_IGNORE_UNKNOWN_FIELDS,
_OWN_VERSION,
DASHBOARD_SCHEMA_VERSION,
MAX_DASHBOARD_DESCRIPTION_LENGTH,
MAX_DASHBOARD_NAME_LENGTH,
MAX_DASHBOARD_PANELS,
MAX_EXPERIMENT_PANEL_ITEMS,
MAX_PANEL_TITLE_LENGTH,
MAX_SMOOTHING,
DashboardDefinition,
DashboardSchemaError,
_duplicates,
_fill_ids,
_utf16_length,
check_dashboard_schema_version,
)
Aggregation = Literal["latest", "final", "min", "max", "best"]
BestDirection = Literal["min", "max"]
PanelWidth = Literal["half", "full"]
_CONFIG = ConfigDict(validate_assignment=True, extra="forbid")
# The hub substitutes a fallback for a blank name and drops a blank id.
_Named = Annotated[str, Field(min_length=1)]
_STORED_ID = "_stored_id"
_READY = "_builder_ready"
C = TypeVar("C", bound=type)
def _describe(
name: str, exc: ValidationError, positional: Sequence[str] = ()
) -> str:
"""One line per offending field; a union names the types it accepts."""
by_field: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
for error in exc.errors():
loc = tuple(error["loc"])
if loc and isinstance(loc[0], int) and loc[0] < len(positional):
loc = (positional[loc[0]], *loc[1:])
field = loc[:-1] if _is_union_branch(loc) else loc
by_field.setdefault(field, []).append(error)
problems = []
for field, errors in by_field.items():
where = ".".join(str(part) for part in field) or "value"
if len(errors) > 1 and all(_is_union_branch(e["loc"]) for e in errors):
accepted = [_branch_name(e["loc"][-1]) for e in errors]
problems.append(
f"{where}: expected {', '.join(accepted[:-1])} or "
f"{accepted[-1]}, got {errors[0].get('input')!r}"
)
else:
problems.append(f"{where}: {errors[0]['msg']}")
return f"{name}: " + "; ".join(problems)
def _is_union_branch(loc: tuple[Any, ...]) -> bool:
return (
len(loc) > 1
and isinstance(loc[-1], str)
and (loc[-1][:1].isupper() or "[" in loc[-1])
)
def _branch_name(tag: str) -> str:
return tag.split("[")[-1].rstrip("]")
def _builder(cls: C) -> C:
"""A validated dataclass whose errors are :class:`DashboardSchemaError`."""
cls = dataclass(config=_CONFIG)(cls)
init, assign = cls.__init__, cls.__setattr__
names = [field.name for field in fields(cls)]
recheck = getattr(cls, "__post_init__", None)
@functools.wraps(init)
def __init__(self: Any, *args: Any, **kwargs: Any) -> None:
try:
init(self, *args, **kwargs)
except ValidationError as exc:
raise DashboardSchemaError(
_describe(cls.__name__, exc, names)
) from exc
object.__setattr__(self, _READY, True)
def __setattr__(self: Any, name: str, value: Any) -> None:
# Assignment revalidates the whole object, which drops what pydantic
# does not own: the stored id and the flag below.
kept = {
key: getattr(self, key)
for key in (_READY, _STORED_ID)
if hasattr(self, key)
}
try:
assign(self, name, value)
except ValidationError as exc:
raise DashboardSchemaError(_describe(cls.__name__, exc)) from exc
for key, value in kept.items():
object.__setattr__(self, key, value)
# The rules that held at construction hold for every later value too.
if recheck is not None and _READY in kept:
recheck(self)
cls.__init__ = __init__ # type: ignore[method-assign]
cls.__setattr__ = __setattr__ # type: ignore[method-assign]
return cls
class _Friendly:
"""Reports pydantic's errors as :class:`DashboardSchemaError`."""
def __init__(self, **data: Any) -> None:
try:
super().__init__(**data)
except ValidationError as exc:
raise DashboardSchemaError(
_describe(type(self).__name__, exc)
) from exc
def __setattr__(self, name: str, value: Any) -> None:
try:
super().__setattr__(name, value)
except ValidationError as exc:
raise DashboardSchemaError(
_describe(type(self).__name__, exc)
) from exc
[docs]
class Filters(_Friendly, schema.Filters):
"""Metrics, tags and parameters that narrow which runs are shown.
Pass it to a :class:`Dashboard` to narrow every panel, or to one
panel to narrow that panel further. A run selection is not a filter:
the runs a viewer compares are theirs to choose.
.. code-block:: python
Filters(metrics=["loss"], tags={"dataset": ["imagenet"]})
:param metrics: The metric names to keep, or empty for all of them.
:param tags: Each tag name mapped to the tag values to keep.
:param params: Each parameter name mapped to the values to keep.
"""
[docs]
class Chart(_Friendly, schema.Chart):
"""The smoothing the dashboard's chart control opens with.
Each :class:`LinePanel` carries its own smoothing, so this seeds the
control rather than the curves.
:param smoothing: Exponential smoothing, ``0`` to ``0.99``.
"""
[docs]
class Table(_Friendly, schema.Table):
"""What the run comparison table opens with.
:param columns: The columns to show, in order, or empty for the
hub's default set.
:param default_sort: The sort to open with, such as
``"startedAt.desc"``, or empty for the hub's default.
"""
def _require_title(owner: str, title: str) -> None:
if not title.strip():
raise DashboardSchemaError(f"{owner}: the title must not be blank.")
if _utf16_length(title) > MAX_PANEL_TITLE_LENGTH:
raise DashboardSchemaError(
f"{owner}: the title is longer than {MAX_PANEL_TITLE_LENGTH} "
"characters, which the hub refuses on a later edit."
)
def _stored_id(value: object) -> str | None:
return getattr(value, _STORED_ID, None)
def _remember_id(value: object, stored_id: str) -> None:
object.__setattr__(value, _STORED_ID, stored_id)
[docs]
class Attribute(str, Enum):
"""A fact about the run itself, shown in a :class:`RunDetailsPanel`.
.. code-block:: python
RunDetailsPanel("Run", [Attribute.STATUS, Attribute.DURATION])
"""
STATUS = "status"
TYPE = "type"
AUTHOR = "author"
STARTED_AT = "startedAt"
ENDED_AT = "endedAt"
DURATION = "duration"
[docs]
@_builder
class Param:
"""A logged parameter, by name.
Serves as a scatter or parallel axis and as a run details value.
:param name: The parameter name as it was logged.
"""
name: _Named
[docs]
@_builder
class Metric:
"""A logged metric, read as one aggregate over its steps.
Serves as a scatter or parallel axis, as the value of a
:class:`BarPanel`, and as a run details value. On an axis or in a
bar panel ``"best"`` picks the min or max according to
``best_direction``; a run details value shows latest, final, min or
max, so use ``"min"`` or ``"max"`` there.
:param name: The metric name as it was logged.
:param aggregation: Which value of the series to read: ``"latest"``,
``"final"``, ``"min"``, ``"max"`` or ``"best"``.
:param best_direction: Whether ``"best"`` means the ``"min"`` or the
``"max"`` value.
"""
name: _Named
aggregation: Aggregation = "latest"
best_direction: BestDirection = "max"
[docs]
@_builder
class Tag:
"""A run tag, by name, shown with its value in a run details panel.
:param name: The tag name as it was logged.
"""
name: _Named
[docs]
@_builder
class Link:
"""An external link the run logged, shown by its label.
:param label: The link label as it was logged.
"""
label: _Named
[docs]
@_builder
class Artifact:
"""An artifact the run logged, shown as a download.
:param file_name: The artifact's file name as it was logged.
"""
file_name: _Named
Item = Attribute | Param | Metric | Tag | Link | Artifact
AxisValue = Metric | Param
[docs]
@_builder
class LinePanel:
"""Metric curves over steps, one line per run and metric.
.. code-block:: python
LinePanel("Loss", ["loss", "val_loss"], smoothing=0.3)
:param title: The panel title.
:param metrics: Names of the metrics to plot.
:param width: ``"half"`` for one column of the section, ``"full"``
for both.
:param smoothing: Exponential smoothing of the curves, ``0`` to
``0.99``. ``None`` leaves the curves unsmoothed; the dashboard's
chart setting seeds the viewer's control, not this panel.
:param filters: Metrics, tags and parameters that narrow the runs this
panel shows below the dashboard's own filters.
:param excluded_run_ids: Runs this panel never shows.
"""
title: str
metrics: list[_Named]
_: KW_ONLY
width: PanelWidth = "half"
smoothing: float | None = Field(default=None, ge=0, le=MAX_SMOOTHING)
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[_Named] = Field(default_factory=list)
def __post_init__(self) -> None:
_require_title("LinePanel", self.title)
[docs]
@_builder
class BarPanel:
"""One aggregated metric value per run, as bars.
.. code-block:: python
BarPanel("Best accuracy", Metric("val_accuracy", "best", "max"))
:param title: The panel title.
:param metric: The metric and the aggregate each bar shows.
:param width: ``"half"`` for one column of the section, ``"full"``
for both.
:param filters: Metrics, tags and parameters that narrow the runs this
panel shows below the dashboard's own filters.
:param excluded_run_ids: Runs this panel never shows.
"""
title: str
metric: Metric
_: KW_ONLY
width: PanelWidth = "half"
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[_Named] = Field(default_factory=list)
def __post_init__(self) -> None:
_require_title("BarPanel", self.title)
[docs]
@_builder
class ScatterPanel:
"""Runs as points over two metrics or parameters.
.. code-block:: python
ScatterPanel("Size vs. accuracy", Param("batch_size"), Metric("top1", "max"))
:param title: The panel title.
:param x: The value along the x axis.
:param y: The value along the y axis.
:param width: ``"half"`` for one column of the section, ``"full"``
for both.
:param filters: Metrics, tags and parameters that narrow the runs this
panel shows below the dashboard's own filters.
:param excluded_run_ids: Runs this panel never shows.
"""
title: str
x: AxisValue
y: AxisValue
_: KW_ONLY
width: PanelWidth = "half"
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[_Named] = Field(default_factory=list)
def __post_init__(self) -> None:
_require_title("ScatterPanel", self.title)
[docs]
@_builder
class ParallelPanel:
"""Runs as lines across several metric or parameter axes.
.. code-block:: python
ParallelPanel("Sweep", [Param("lr"), Metric("val_loss", "min")])
:param title: The panel title.
:param axes: The values along the axes, left to right.
:param width: ``"half"`` for one column of the section, ``"full"``
for both.
:param filters: Metrics, tags and parameters that narrow the runs this
panel shows below the dashboard's own filters.
:param excluded_run_ids: Runs this panel never shows.
"""
title: str
axes: list[AxisValue]
_: KW_ONLY
width: PanelWidth = "half"
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[_Named] = Field(default_factory=list)
def __post_init__(self) -> None:
_require_title("ParallelPanel", self.title)
[docs]
@_builder
class TablePanel:
"""The run comparison table, one row per selected run.
:param title: The panel title.
:param width: ``"full"`` for both columns of the section, ``"half"``
for one.
:param columns: The columns to show, in order; empty for the hub's
default set.
:param default_sort: The sort the table opens with, such as
``"startedAt.desc"``; empty for the hub's default.
:param filters: Metrics, tags and parameters that narrow the runs this
panel shows below the dashboard's own filters.
:param excluded_run_ids: Runs this panel never shows.
"""
title: str
_: KW_ONLY
width: PanelWidth = "full"
columns: list[_Named] = Field(default_factory=list)
default_sort: str = ""
filters: Filters = Field(default_factory=Filters)
excluded_run_ids: list[_Named] = Field(default_factory=list)
def __post_init__(self) -> None:
_require_title("TablePanel", self.title)
[docs]
@_builder
class RunDetailsPanel:
"""Chosen values of one run, one row each, in the order given.
.. code-block:: python
RunDetailsPanel(
"Run details",
[Attribute.STATUS, Param("model"), Metric("val_acc", "max"), Link("Docs")],
)
:param title: The panel title.
:param items: The values to show, at most
:data:`MAX_EXPERIMENT_PANEL_ITEMS` and each at most once. A
:class:`Metric` here reads latest, final, min or max.
:param run_id: The run to show. ``None`` shows the first selected
run, which on a run's own dashboard tab is that run.
:param width: ``"half"`` for one column of the section, ``"full"``
for both.
"""
title: str
items: list[Item] = Field(
default_factory=list, max_length=MAX_EXPERIMENT_PANEL_ITEMS
)
_: KW_ONLY
run_id: _Named | None = None
width: PanelWidth = "half"
def __post_init__(self) -> None:
_require_title("RunDetailsPanel", self.title)
repeated = _duplicates(_item_key(item) for item in self.items)
if repeated:
listed = ", ".join(f"{kind} {name!r}" for kind, name in repeated)
raise DashboardSchemaError(
f"RunDetailsPanel: {listed} listed more than once. Each value "
"appears once."
)
best = [
item.name
for item in self.items
if isinstance(item, Metric) and item.aggregation == "best"
]
if best:
raise DashboardSchemaError(
f"RunDetailsPanel: {best} use aggregation 'best'. A run details "
"value shows latest, final, min or max; use 'min' or 'max'."
)
Panel = (
LinePanel
| BarPanel
| ScatterPanel
| ParallelPanel
| TablePanel
| RunDetailsPanel
)
_PANEL_TYPES = (
LinePanel,
BarPanel,
ScatterPanel,
ParallelPanel,
TablePanel,
RunDetailsPanel,
)
[docs]
@_builder
class Section:
"""A titled group of panels, laid out in two columns.
Panels fill the columns in order: a ``"half"`` panel takes one, a
``"full"`` panel takes both.
:param title: The section title.
:param panels: The panels the section shows, in order.
:param collapsed: Whether the section starts folded up.
"""
title: str
panels: list[Panel] = Field(default_factory=list)
_: KW_ONLY
collapsed: bool = False
def __post_init__(self) -> None:
_require_title("Section", self.title)
[docs]
@_builder
class Dashboard:
"""A dashboard to attach to a run: a name and its sections.
Hand it to :meth:`~embedl_hub.tracking.Client.log_dashboard`, or
write it with :meth:`save` for ``embedl-hub log dashboard``. Whatever
the hub would trim or drop is refused here instead, as a
:class:`DashboardSchemaError`.
:param name: The name the dashboard is listed under on the run.
:param sections: The sections, top to bottom, holding at most
:data:`MAX_DASHBOARD_PANELS` panels between them.
:param description: What the dashboard is for.
:param filters: Metrics, tags and parameters the dashboard narrows
its runs to.
:param chart: The smoothing the chart control opens with, which
does not change what a panel plots.
:param table: Columns and default sort of the run comparison table.
"""
name: str
sections: list[Section] = Field(default_factory=list)
_: KW_ONLY
description: str | None = None
filters: Filters = Field(default_factory=Filters)
chart: Chart = Field(default_factory=Chart)
table: Table = Field(default_factory=Table)
def __post_init__(self) -> None:
name = self.name.strip()
if not name:
raise DashboardSchemaError("The dashboard name must not be blank.")
if _utf16_length(name) > MAX_DASHBOARD_NAME_LENGTH:
raise DashboardSchemaError(
"The dashboard name is longer than "
f"{MAX_DASHBOARD_NAME_LENGTH} characters as the hub counts "
"them (UTF-16 code units)."
)
description = (self.description or "").strip() or None
if (
description is not None
and _utf16_length(description) > MAX_DASHBOARD_DESCRIPTION_LENGTH
):
raise DashboardSchemaError(
"The dashboard description is longer than "
f"{MAX_DASHBOARD_DESCRIPTION_LENGTH} characters as the hub "
"counts them (UTF-16 code units)."
)
object.__setattr__(self, "name", name)
object.__setattr__(self, "description", description)
self._check_panels()
def _revalidate(self) -> None:
"""Re-apply the construction rules to the whole tree."""
self.__post_init__()
for section in self.sections:
if not isinstance(section, Section):
raise DashboardSchemaError(
"Dashboard.sections holds a "
f"{type(section).__name__}; every entry is a Section."
)
section.__post_init__()
for panel in section.panels:
if not isinstance(panel, _PANEL_TYPES):
raise DashboardSchemaError(
f"Section {section.title!r} holds a "
f"{type(panel).__name__}; every entry is a panel."
)
panel.__post_init__()
def _check_panels(self) -> list[Panel]:
panels = [
panel for section in self.sections for panel in section.panels
]
if len(panels) > MAX_DASHBOARD_PANELS:
raise DashboardSchemaError(
f"A dashboard holds at most {MAX_DASHBOARD_PANELS} panels; "
f"this one has {len(panels)}."
)
if len({id(panel) for panel in panels}) != len(panels):
raise DashboardSchemaError(
"The same panel is in more than one place. Build one panel "
"per place."
)
return panels
[docs]
def to_definition(self) -> DashboardDefinition:
"""The definition the hub stores, with ids filled in.
Panels get ``panel-1``, ``panel-2``, ... and sections
``section-1``, ... in order, except those read from a saved
dashboard, which keep the ids they had.
:return: The definition, as the hub's API accepts it.
:raises DashboardSchemaError: If the dashboard exceeds what the
hub stores.
"""
self._revalidate()
panels = self._check_panels()
panel_ids = _fill_ids([_stored_id(panel) for panel in panels], "panel")
section_ids = _fill_ids(
[_stored_id(section) for section in self.sections], "section"
)
for repeated, what in (
(_duplicates(panel_ids), "panel"),
(_duplicates(section_ids), "section"),
):
if repeated:
raise DashboardSchemaError(
f"Saved {what} ids {repeated} appear more than once."
)
by_panel = dict(zip(map(id, panels), panel_ids, strict=True))
try:
return DashboardDefinition(
filters=self.filters,
chart=self.chart,
table=self.table,
sections=[
schema.Section(
id=section_id,
title=section.title,
collapsed=section.collapsed,
panel_ids=[by_panel[id(p)] for p in section.panels],
)
for section, section_id in zip(
self.sections, section_ids, strict=True
)
],
panels=[
_panel_to_schema(panel, panel_id)
for panel, panel_id in zip(panels, panel_ids, strict=True)
],
)
except ValidationError as exc:
raise DashboardSchemaError(_describe("Dashboard", exc)) from exc
[docs]
def to_dict(self) -> dict[str, Any]:
"""The upload body: ``name``, ``description`` and ``config``.
:return: JSON-ready data, the definition using the hub's key names.
:raises DashboardSchemaError: As :meth:`to_definition`.
"""
return {
"name": self.name,
"description": self.description,
"config": self.to_definition().model_dump(
mode="json", by_alias=True, exclude_none=True
),
}
[docs]
def to_json(self, *, indent: int | None = 2) -> str:
"""Serialize :meth:`to_dict` as JSON.
:param indent: Indentation passed to :func:`json.dumps`.
:return: The JSON text.
:raises DashboardSchemaError: As :meth:`to_definition`.
"""
return json.dumps(self.to_dict(), indent=indent)
[docs]
def save(self, path: Path | str) -> Path:
"""Write :meth:`to_json` to ``path`` for ``embedl-hub log dashboard``.
:param path: Where to write the file.
:return: The path written.
:raises DashboardSchemaError: As :meth:`to_definition`.
"""
target = Path(path)
target.write_text(self.to_json() + "\n", encoding="utf-8")
return target
[docs]
@classmethod
def from_dict(
cls,
data: Mapping[str, Any],
*,
name: str | None = None,
description: str | None = None,
) -> Dashboard:
"""Read a dashboard from :meth:`to_dict` output or a bare config.
:param data: The upload body written by :meth:`save`, or a bare
definition as the hub stores it.
:param name: Overrides, or supplies for a bare config, the name.
:param description: Overrides the description.
:return: The dashboard the data describes, with the ids it had.
:raises DashboardSchemaError: If the data is not a dashboard, its
schema version is not one this SDK reads, its fields do not
match the schema, or it ends up without a name. Fields that
a newer non-breaking schema version added are ignored.
"""
if not isinstance(data, Mapping):
raise DashboardSchemaError(
"Expected a dashboard as a JSON object, written by "
"Dashboard.save()."
)
enveloped = "config" in data
raw_config = data["config"] if enveloped else data
if not isinstance(raw_config, Mapping):
raise DashboardSchemaError(
"The dashboard config must be a JSON object."
)
newer_patch = check_dashboard_schema_version(raw_config) > _OWN_VERSION
if enveloped and not newer_patch:
unknown = sorted(set(data) - {"name", "description", "config"})
if unknown:
raise DashboardSchemaError(
f"A dashboard file holds name, description and config; "
f"found {unknown} as well."
)
stored_name = data.get("name") if enveloped else None
resolved_name = name if name is not None else stored_name
if not isinstance(resolved_name, str) or not resolved_name.strip():
raise DashboardSchemaError(
"The dashboard has no name. Pass name=..., or export it "
"with Dashboard.save(), which records the name."
)
if description is None and enveloped:
stored = data.get("description")
description = stored if isinstance(stored, str) else None
try:
definition = DashboardDefinition.model_validate(
raw_config,
context={_IGNORE_UNKNOWN_FIELDS: True}
if newer_patch
else None,
)
except ValidationError as exc:
raise DashboardSchemaError(
"The dashboard does not match schema "
f"{DASHBOARD_SCHEMA_VERSION}: {exc}"
) from exc
return cls(
resolved_name,
_sections_from_schema(definition),
description=description,
filters=Filters(**definition.filters.model_dump()),
chart=Chart(**definition.chart.model_dump()),
table=Table(**definition.table.model_dump()),
)
[docs]
@classmethod
def from_json(
cls,
text: str,
*,
name: str | None = None,
description: str | None = None,
) -> Dashboard:
"""Read a dashboard from the JSON written by :meth:`to_json`.
:param text: The JSON text.
:param name: Overrides, or supplies for a bare config, the name.
:param description: Overrides the description.
:return: The dashboard the text describes.
:raises DashboardSchemaError: If the text is not JSON, or as
:meth:`from_dict`.
"""
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
raise DashboardSchemaError(
f"The dashboard is not valid JSON: {exc}"
) from exc
return cls.from_dict(data, name=name, description=description)
[docs]
@classmethod
def load(
cls,
path: Path | str,
*,
name: str | None = None,
description: str | None = None,
) -> Dashboard:
"""Read a dashboard file written by :meth:`save`.
:param path: The file to read.
:param name: Overrides, or supplies for a bare config, the name.
:param description: Overrides the description.
:return: The dashboard the file describes.
:raises DashboardSchemaError: If the file is not a dashboard this
SDK reads; see :meth:`from_dict`.
:raises FileNotFoundError: If there is no file at ``path``.
"""
source = Path(path)
try:
return cls.from_json(
source.read_text(encoding="utf-8"),
name=name,
description=description,
)
except DashboardSchemaError as exc:
raise DashboardSchemaError(f"{source}: {exc}") from exc
def _item_key(item: Item) -> tuple[str, str]:
schema_item = _item_to_schema(item)
return schema_item.kind, schema_item.name
def _item_to_schema(item: Item) -> schema.PanelItem:
if isinstance(item, Attribute):
return schema.PanelItem(kind="attribute", name=item.value)
if isinstance(item, Param):
return schema.PanelItem(kind="param", name=item.name)
if isinstance(item, Metric):
aggregation = (
"latest" if item.aggregation == "best" else item.aggregation
)
return schema.PanelItem(
kind="metric", name=item.name, aggregation=aggregation
)
if isinstance(item, Tag):
return schema.PanelItem(kind="tag", name=item.name)
if isinstance(item, Link):
return schema.PanelItem(kind="externalLink", name=item.label)
return schema.PanelItem(kind="artifact", name=item.file_name)
def _item_from_schema(item: schema.PanelItem) -> Item:
if item.kind == "attribute":
return Attribute(item.name)
if item.kind == "param":
return Param(item.name)
if item.kind == "metric":
return Metric(item.name, item.aggregation or "latest")
if item.kind == "tag":
return Tag(item.name)
if item.kind == "externalLink":
return Link(item.name)
return Artifact(item.name)
def _axis_to_schema(value: AxisValue) -> schema.Axis:
if isinstance(value, Param):
return schema.Axis(source="param", name=value.name)
return schema.Axis(
source="metric",
name=value.name,
aggregation=value.aggregation,
best_direction=value.best_direction,
)
def _axis_from_schema(axis: schema.Axis) -> AxisValue:
if axis.source == "param":
return Param(axis.name)
return Metric(
axis.name, axis.aggregation or "latest", axis.best_direction or "max"
)
def _panel_to_schema(panel: Panel, panel_id: str) -> schema.Panel:
common: dict[str, Any] = {
"id": panel_id,
"title": panel.title,
"width": panel.width,
}
if not isinstance(panel, RunDetailsPanel):
common.update(
filters=panel.filters,
excluded_run_ids=list(panel.excluded_run_ids),
)
if isinstance(panel, LinePanel):
return schema.LinePanel(
**common,
metric_names=list(panel.metrics),
smoothing=panel.smoothing,
)
if isinstance(panel, BarPanel):
return schema.BarPanel(
**common,
metric_name=panel.metric.name,
aggregation=panel.metric.aggregation,
best_direction=panel.metric.best_direction,
)
if isinstance(panel, ScatterPanel):
return schema.ScatterPanel(
**common,
x_axis=_axis_to_schema(panel.x),
y_axis=_axis_to_schema(panel.y),
)
if isinstance(panel, ParallelPanel):
return schema.ParallelPanel(
**common, axes=[_axis_to_schema(axis) for axis in panel.axes]
)
if isinstance(panel, TablePanel):
return schema.TablePanel(
**common,
table=schema.Table(
columns=list(panel.columns), default_sort=panel.default_sort
),
)
return schema.RunDetailsPanel(
**common,
run_id=panel.run_id,
items=[_item_to_schema(item) for item in panel.items],
)
def _panel_from_schema(panel: schema.Panel) -> Panel:
built: Panel
narrowing: dict[str, Any] = {
"filters": Filters(**panel.filters.model_dump()),
"excluded_run_ids": list(panel.excluded_run_ids),
}
if panel.type == "line":
built = LinePanel(
panel.title,
list(panel.metric_names or []),
width=panel.width,
smoothing=panel.smoothing,
**narrowing,
)
elif panel.type == "bar":
built = BarPanel(
panel.title,
Metric(
panel.metric_name or "",
panel.aggregation or "latest",
panel.best_direction or "max",
),
width=panel.width,
**narrowing,
)
elif panel.type == "scatter":
built = ScatterPanel(
panel.title,
_axis_from_schema(panel.x_axis or schema.Axis(name="")),
_axis_from_schema(panel.y_axis or schema.Axis(name="")),
width=panel.width,
**narrowing,
)
elif panel.type == "parallel":
built = ParallelPanel(
panel.title,
[_axis_from_schema(axis) for axis in panel.axes or []],
width=panel.width,
**narrowing,
)
elif panel.type == "table":
built = TablePanel(
panel.title,
width=panel.width,
columns=list(panel.table.columns),
default_sort=panel.table.default_sort,
**narrowing,
)
else:
built = RunDetailsPanel(
panel.title,
[_item_from_schema(item) for item in panel.items or []],
run_id=panel.run_id,
width=panel.width,
)
if panel.id is not None:
_remember_id(built, panel.id)
return built
def _sections_from_schema(
definition: schema.DashboardDefinition,
) -> list[Section]:
by_id: dict[str, schema.Panel] = {}
for panel in definition.panels:
if panel.id is None or panel.id in by_id:
raise DashboardSchemaError(
"Every stored panel needs its own id; found "
f"{panel.id!r} more than once or missing."
)
by_id[panel.id] = panel
placed: set[str] = set()
sections: list[Section] = []
for stored in definition.sections:
panels: list[Panel] = []
for panel_id in stored.panel_ids:
if panel_id not in by_id or panel_id in placed:
raise DashboardSchemaError(
f"Section {stored.title!r} refers to panel {panel_id!r}, "
"which is missing or already placed."
)
placed.add(panel_id)
panels.append(_panel_from_schema(by_id[panel_id]))
section = Section(stored.title, panels, collapsed=stored.collapsed)
if stored.id is not None:
_remember_id(section, stored.id)
sections.append(section)
left_out = [panel_id for panel_id in by_id if panel_id not in placed]
if left_out and sections:
raise DashboardSchemaError(
f"Panels {left_out} are in no section. Every panel belongs to "
"one section."
)
if left_out:
sections.append(
Section("Charts", [_panel_from_schema(by_id[i]) for i in left_out])
)
return sections
