"""Frontend page configuration loader and schema."""
from __future__ import annotations
import logging
from functools import lru_cache
from pathlib import Path
from typing import Annotated, Any, Literal
import yaml
from pydantic import (
BaseModel,
ConfigDict,
Field,
TypeAdapter,
ValidationError,
model_validator,
)
logger = logging.getLogger(__name__)
[docs]
class LayoutConfig(BaseModel):
"""Layout configuration for a page."""
rows: int = 1
cols: int = 1
max_per_row: int = 1
[docs]
class PageConfig(BaseModel):
"""Configuration for a single page."""
name: str
label: str
plots: list[dict[str, Any]] = Field(default_factory=list)
layout: LayoutConfig = Field(
default_factory=lambda: LayoutConfig(rows=2, cols=2, max_per_row=2)
)
css: list[str] = Field(default_factory=list)
js: list[str] = Field(default_factory=list)
template: str = "pages/array_element_type/generic_page.html"
@model_validator(mode="after")
def _validate_plots(self) -> PageConfig:
if not self.plots:
raise ValueError("Page config must define plots.")
for plot in self.plots:
title = plot.get("title")
if not title:
raise ValueError("Plot title is required for every plot entry.")
return self
[docs]
def plot_titles(self) -> list[str]:
"""Return plot titles in the order defined by the page config."""
return [plot.get("title") or "" for plot in self.plots]
[docs]
def plot_keys(self) -> list[str]:
"""Return generated internal plot keys in the order defined by the page config."""
keys: list[str] = []
seen: dict[str, int] = {}
for idx, plot in enumerate(self.plots):
metric = plot.get("metric")
plot_type = plot.get("plotType")
base_seed = (
f"{metric}_{plot_type}" if metric and plot_type else plot.get("title")
)
base = self._slugify(base_seed or f"plot_{idx + 1}")
if not base:
base = f"plot_{idx + 1}"
count = seen.get(base, 0) + 1
seen[base] = count
suffix = "" if count == 1 else f"_{count}"
keys.append(f"{base}{suffix}")
return keys
[docs]
def plot_metrics(self) -> list[str]:
"""Return plot metric names in the order defined by the page config."""
return [plot.get("metric") or "" for plot in self.plots]
def _slugify(self, value: str) -> str:
return "".join(ch if ch.isalnum() else "_" for ch in value.lower()).strip("_")
[docs]
class AxisConfigBase(BaseModel):
"""Base axis configuration for plots."""
model_config = ConfigDict(extra="forbid", populate_by_name=True)
label: str
domain: list[float] | None = None
unit: str | None = None
field: str | None = None
[docs]
class ScatterAxisConfig(AxisConfigBase):
"""Axis configuration for scatter plots."""
scale: Literal["linear", "log", "time", "ordinal", "band", "point"] | None = None
[docs]
class HistogramAxisConfig(AxisConfigBase):
"""Axis configuration for histogram plots."""
scale: Literal["linear", "log"] | None = None
[docs]
class CameraAxisConfig(AxisConfigBase):
"""Axis configuration for camera views."""
label: str | None = None
scale: Literal["linear"] | None = None
[docs]
class ScatterMarksConfig(BaseModel):
"""Marks configuration for scatter plots."""
model_config = ConfigDict(extra="forbid", populate_by_name=True)
type: (
Literal[
"circle",
"triangle",
"cross",
"wye",
"star",
"square",
"diamond",
]
| None
) = None
fill: str | None = None
stroke: str | None = None
stroke_width: float | None = Field(default=None, alias="strokeWidth")
opacity: float | None = None
size: float | None = None
[docs]
class HistogramBarConfig(BaseModel):
"""Bar style configuration for histogram plots."""
model_config = ConfigDict(extra="forbid", populate_by_name=True)
fill: str | None = None
stroke: str | None = None
stroke_width: float | None = Field(default=None, alias="strokeWidth")
opacity: float | None = None
[docs]
class BasePlotConfig(BaseModel):
"""Shared configuration for plot types."""
model_config = ConfigDict(extra="forbid", populate_by_name=True)
metric: str
plot_type: str = Field(alias="plotType")
title: str
subtitle: str | None = None
[docs]
class ScatterPlotConfig(BasePlotConfig):
"""Configuration for scatter plots."""
plot_type: Literal["scatterplot"] = Field(alias="plotType")
x: ScatterAxisConfig
y: ScatterAxisConfig
marks: ScatterMarksConfig | None = None
line: (
Literal[
"none",
"solid",
"dashed",
"dotted",
"dot-dashed",
"dot-dashed-dashed",
]
| None
) = None
[docs]
class CameraViewConfig(BasePlotConfig):
"""Configuration for camera views."""
plot_type: Literal["cameraview"] = Field(alias="plotType")
x: CameraAxisConfig | None = None
y: CameraAxisConfig | None = None
[docs]
class Histogram1DConfig(BasePlotConfig):
"""Configuration for 1D histogram plots."""
plot_type: Literal["histogram1d"] = Field(alias="plotType")
x: HistogramAxisConfig
y: HistogramAxisConfig
bins: int | None = None
marks: HistogramBarConfig | None = None
PlotConfig = Annotated[
ScatterPlotConfig | CameraViewConfig | Histogram1DConfig,
Field(discriminator="plot_type"),
]
PLOT_CONFIG_ADAPTER = TypeAdapter(PlotConfig)
[docs]
class PageIndexEntry(BaseModel):
"""Reference to a page config file."""
name: str
source: str
[docs]
class BaseFrontendConfig(BaseModel):
"""Shared frontend configuration fields."""
array_element_types: list[str] = Field(
default_factory=lambda: ["LSTs", "MSTs", "SSTs"]
)
auxiliary_subitems: list[str] = Field(
default_factory=lambda: ["Lidar", "FRAM", "Weather Station"]
)
[docs]
class FrontendIndexConfig(BaseFrontendConfig):
"""Index configuration listing available pages."""
pages: list[PageIndexEntry]
[docs]
class UserIndexConfig(BaseModel):
"""Optional user overrides for pages and auxiliary subitems."""
pages: list[PageIndexEntry] | None = None
auxiliary_subitems: list[str] | None = None
def _merge_page_entries(
default_entries: list[PageIndexEntry], user_entries: list[PageIndexEntry]
) -> list[PageIndexEntry]:
"""Merge user page entries into defaults without allowing name collisions.
User entries are appended preserving user order. If a user entry has the same
name as a default entry, the user index is considered invalid.
"""
merged = list(default_entries)
default_names = {entry.name for entry in default_entries}
for user_entry in user_entries:
if user_entry.name in default_names:
raise ValueError(
f"User page '{user_entry.name}' conflicts with an existing default page."
)
merged.append(user_entry)
return merged
[docs]
class FrontendConfig(BaseFrontendConfig):
"""Resolved frontend configuration."""
pages: list[PageConfig]
page_entries: list[PageIndexEntry] = Field(default_factory=list)
page_errors: dict[str, str] = Field(default_factory=dict)
user_index_error: str | None = None
[docs]
def page_by_name(self, name: str) -> PageConfig | None:
"""Return a page configuration by name."""
return next((page for page in self.pages if page.name == name), None)
[docs]
def page_error(self, name: str) -> str | None:
"""Return a page validation error by name, if any."""
return self.page_errors.get(name)
[docs]
def page_label(self, name: str) -> str:
"""Return a label for a page, falling back to the name."""
page = self.page_by_name(name)
if page:
return page.label
return name.replace("_", " ")
def _default_config_path() -> Path:
return Path(__file__).resolve().parent / "view" / "pages" / "pages_index.yaml"
def _user_index_path() -> Path:
return (
Path(__file__).resolve().parent
/ "view"
/ "pages"
/ "user_pages"
/ "pages_index.yaml"
)
def _load_page_config(config_path: Path) -> PageConfig:
raw: Any = yaml.safe_load(config_path.read_text())
page = PageConfig.model_validate(raw)
return page
[docs]
@lru_cache(maxsize=1)
def load_frontend_config(config_path: Path | None = None) -> FrontendConfig:
"""Load and validate the frontend configuration file."""
path = config_path or _default_config_path()
try:
raw: Any = yaml.safe_load(path.read_text())
except FileNotFoundError as exc:
logger.error("Frontend config not found at %s", path)
raise FileNotFoundError(f"Frontend config not found at {path}.") from exc
index = FrontendIndexConfig.model_validate(raw)
user_index_error: str | None = None
user_index_path = _user_index_path()
if user_index_path.exists():
try:
user_raw: Any = yaml.safe_load(user_index_path.read_text())
user_index = UserIndexConfig.model_validate(user_raw)
if user_index.pages is not None:
index.pages = _merge_page_entries(index.pages, user_index.pages)
if user_index.auxiliary_subitems is not None:
index.auxiliary_subitems = user_index.auxiliary_subitems
except Exception as exc:
logger.error("Invalid user index config: %s", exc)
user_index_error = str(exc)
base_dir = path.parent
pages: list[PageConfig] = []
page_errors: dict[str, str] = {}
for entry in index.pages:
page_path = (base_dir / entry.source).resolve()
try:
page = _load_page_config(page_path)
if page.name != entry.name:
raise ValueError(
"Page name mismatch for "
f"{entry.source}: expected {entry.name}, got {page.name}"
)
pages.append(page)
except Exception as exc:
logger.error("Invalid page config for %s: %s", entry.name, exc)
page_errors[entry.name] = str(exc)
return FrontendConfig(
array_element_types=index.array_element_types,
auxiliary_subitems=index.auxiliary_subitems,
pages=pages,
page_entries=index.pages,
page_errors=page_errors,
user_index_error=user_index_error,
)
[docs]
def build_nav_pages(config: FrontendConfig) -> list[dict[str, str]]:
"""Build navigation entries from config pages."""
if config.page_entries:
return [
{"name": entry.name, "label": config.page_label(entry.name)}
for entry in config.page_entries
]
return [{"name": page.name, "label": page.label} for page in config.pages]
[docs]
def build_plot_state(
page: PageConfig,
) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
"""Build plot configurations and collect validation errors per plot."""
plot_configs: dict[str, dict[str, Any]] = {}
plot_errors: dict[str, str] = {}
for plot_key, plot in zip(page.plot_keys(), page.plots):
if not plot.get("plotType"):
# Title-only placeholder plots are allowed without config validation.
continue
try:
validated = PLOT_CONFIG_ADAPTER.validate_python(plot)
except ValidationError as exc:
logger.error("Invalid plot configuration for %s: %s", plot_key, exc)
plot_errors[plot_key] = str(exc)
continue
plot_configs[plot_key] = validated.model_dump(exclude_none=True, by_alias=True)
return plot_configs, plot_errors
