Add Content Blocks (#7)

Author: ramppdev

ablate/__init__.py

@@ -1,6 +1,6 @@
-from . import queries, sources
+from . import blocks, queries, sources
 
 
-__all__ = ["queries", "sources"]
+__all__ = ["blocks", "queries", "sources"]
 
 __version__ = "0.1.0"

ablate/blocks/__init__.py

@@ -0,0 +1,21 @@
+from .abstract_block import AbstractBlock
+from .figure_blocks import AbstractFigureBlock, MetricPlot
+from .table_blocks import AbstractTableBlock, Table
+from .text_blocks import H1, H2, H3, H4, H5, H6, AbstractTextBlock, Text
+
+
+__all__ = [
+    "AbstractBlock",
+    "AbstractFigureBlock",
+    "AbstractTableBlock",
+    "AbstractTextBlock",
+    "H1",
+    "H2",
+    "H3",
+    "H4",
+    "H5",
+    "H6",
+    "MetricPlot",
+    "Table",
+    "Text",
+]

ablate/blocks/abstract_block.py

@@ -0,0 +1,26 @@
+from abc import ABC, abstractmethod
+from typing import Any, List
+
+from ablate.core.types import Run
+
+
+class AbstractBlock(ABC):
+    def __init__(self, runs: List[Run] | None = None) -> None:
+        """Abstract content block for a report.
+
+        Args:
+            runs: Optional list of runs to be used for the block instead of the default
+                runs from the report. Defaults to None.
+        """
+        self.runs = runs
+
+    @abstractmethod
+    def build(self, runs: List[Run]) -> Any:
+        """Build the intermediate representation of the block, ready for rendering.
+
+        Args:
+            runs: List of runs to be used for the block.
+
+        Returns:
+            The intermediate representation of the block.
+        """

ablate/blocks/figure_blocks.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, List
+
+import pandas as pd
+
+from ablate.queries import AbstractMetric, Id, Param
+
+from .abstract_block import AbstractBlock
+
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ablate.core.types import Run
+
+
+class AbstractFigureBlock(AbstractBlock, ABC):
+    @abstractmethod
+    def build(self, runs: List[Run]) -> pd.DataFrame: ...
+
+
+class MetricPlot(AbstractBlock):
+    def __init__(
+        self,
+        metric: AbstractMetric | List[AbstractMetric],
+        identifier: Param | None = None,
+        runs: List[Run] | None = None,
+    ) -> None:
+        """Block for plotting metrics over time.
+
+        Args:
+            metric: Metric or list of metrics to be plotted over time.
+            identifier: Optional identifier for the runs. If None, the run ID is used.
+                Defaults to None.
+            runs: Optional list of runs to be used for the block instead of the default
+                runs from the report. Defaults to None.
+        """
+        super().__init__(runs)
+        self.metrics = metric if isinstance(metric, list) else [metric]
+        self.identifier = identifier or Id()
+
+    def build(self, runs: List[Run]) -> pd.DataFrame:
+        data = []
+        for run in runs:
+            for metric in self.metrics:
+                series = run.temporal.get(metric.name, [])
+                for step, value in series:
+                    data.append(
+                        {
+                            "step": step,
+                            "value": value,
+                            "metric": metric.label,
+                            "run": self.identifier(run),
+                            "run_id": run.id,
+                        }
+                    )
+        return pd.DataFrame(data)

ablate/blocks/table_blocks.py

@@ -0,0 +1,41 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+import pandas as pd
+
+from ablate.core.types import Run
+from ablate.queries import AbstractSelector
+
+from .abstract_block import AbstractBlock
+
+
+class AbstractTableBlock(AbstractBlock, ABC):
+    def __init__(
+        self,
+        columns: List[AbstractSelector],
+        runs: List[Run] | None = None,
+    ) -> None:
+        """Table block for a report.
+
+        Args:
+            columns: Columns to be included in the table. Each column is defined by a
+                selector that extracts the data from the runs.
+            runs: Optional list of runs to be used for the block instead of the default
+                runs from the report. Defaults to None.
+        """
+        super().__init__(runs)
+        self.columns = columns
+
+    @abstractmethod
+    def build(self, runs: List[Run]) -> pd.DataFrame: ...
+
+
+class Table(AbstractTableBlock):
+    def build(self, runs: List[Run]) -> pd.DataFrame:
+        rows = []
+        for run in runs:
+            row = {}
+            for column in self.columns:
+                row[column.label] = column(run)
+            rows.append(row)
+        return pd.DataFrame(rows, columns=[column.label for column in self.columns])

ablate/blocks/text_blocks.py

@@ -0,0 +1,43 @@
+from abc import ABC
+from typing import List
+
+from ablate.core.types import Run
+
+from .abstract_block import AbstractBlock
+
+
+class AbstractTextBlock(AbstractBlock, ABC):
+    def __init__(self, text: str, runs: List[Run] | None = None) -> None:
+        """Block containing styled text for a report.
+
+        Args:
+            text: The text content of the block.
+            runs: Optional list of runs to be used for the block instead of the default
+                runs from the report. Defaults to None.
+        """
+        super().__init__(runs)
+        self.text = text
+
+    def build(self, runs: List[Run]) -> str:
+        return self.text.strip()
+
+
+class H1(AbstractTextBlock): ...
+
+
+class H2(AbstractTextBlock): ...
+
+
+class H3(AbstractTextBlock): ...
+
+
+class H4(AbstractTextBlock): ...
+
+
+class H5(AbstractTextBlock): ...
+
+
+class H6(AbstractTextBlock): ...
+
+
+class Text(AbstractTextBlock): ...

ablate/queries/selectors.py

@@ -6,13 +6,16 @@
 
 
 class AbstractSelector(ABC):
-    def __init__(self, name: str) -> None:
+    def __init__(self, name: str, label: str | None = None) -> None:
         """Abstract class for selecting runs based on a specific attribute.
 
         Args:
             name: Name of the attribute to select on.
+            label: Optional label for displaying purposes. If None, defaults to `name`.
+                Defaults to None.
         """
         self.name = name
+        self.label = label or name
 
     @abstractmethod
     def __call__(self, run: Run) -> Any: ...
@@ -43,9 +46,14 @@ class AbstractParam(AbstractSelector, ABC): ...
 
 
 class Id(AbstractParam):
-    def __init__(self) -> None:
-        """Selector for the ID of the run."""
-        super().__init__("id")
+    def __init__(self, label: str | None = None) -> None:
+        """Selector for the ID of the run.
+
+        Args:
+            label: Optional label for displaying purposes. If None, defaults to `name`.
+                Defaults to None.
+        """
+        super().__init__("id", label)
 
     def __call__(self, run: Run) -> str:
         return run.id
@@ -63,8 +71,9 @@ def __init__(
         self,
         name: str,
         direction: Literal["min", "max"],
+        label: str | None = None,
     ) -> None:
-        super().__init__(name)
+        super().__init__(name, label)
         if direction not in ("min", "max"):
             raise ValueError(
                 f"Invalid direction: '{direction}'. Must be 'min' or 'max'."
@@ -79,6 +88,8 @@ class Metric(AbstractMetric):
         name: Name of the metric to select on.
         direction: Direction of the metric. "min" for minimization, "max" for
             maximization.
+        label: Optional label for displaying purposes. If None, defaults to `name`.
+            Defaults to None.
     """
 
     def __call__(self, run: Run) -> float:
@@ -94,6 +105,7 @@ def __init__(
         name: str,
         direction: Literal["min", "max"],
         reduction: Literal["min", "max", "first", "last"] | None = None,
+        label: str | None = None,
     ) -> None:
         """Selector for a specific temporal metric of the run.
 
@@ -105,8 +117,10 @@ def __init__(
                 minimum, "max" for maximum, "first" for the first value, and "last"
                 for the last value. If None, the direction is used as the reduction.
                 Defaults to None.
+            label: Optional label for displaying purposes. If None, defaults to `name`.
+                Defaults to None.
         """
-        super().__init__(name, direction)
+        super().__init__(name, direction, label)
         if reduction is not None and reduction not in ("min", "max", "first", "last"):
             raise ValueError(
                 f"Invalid reduction method: '{reduction}'. Must be 'min', 'max', "

tests/blocks/test_blocks.py

@@ -0,0 +1,56 @@
+from typing import List
+
+import pandas as pd
+
+from ablate.blocks import H1, MetricPlot, Table, Text
+from ablate.core.types import Run
+from ablate.queries.selectors import Metric, Param
+
+
+def make_runs() -> List[Run]:
+    return [
+        Run(
+            id="a",
+            params={"model": "resnet", "seed": 1},
+            metrics={"accuracy": 0.7},
+            temporal={"accuracy": [(0, 0.6), (1, 0.7)]},
+        ),
+        Run(
+            id="b",
+            params={"model": "resnet", "seed": 2},
+            metrics={"accuracy": 0.8},
+            temporal={"accuracy": [(0, 0.7), (1, 0.8)]},
+        ),
+    ]
+
+
+def test_text_blocks() -> None:
+    assert Text(" simple ").build(make_runs()) == "simple"
+    assert H1("# Title").build(make_runs()) == "# Title"
+
+
+def test_table_block() -> None:
+    table = Table(columns=[Param("model"), Param("seed")])
+    df = table.build(make_runs())
+    assert isinstance(df, pd.DataFrame)
+    assert list(df.columns) == ["model", "seed"]
+    assert df.iloc[0]["model"] == "resnet"
+
+
+def test_metric_plot_single() -> None:
+    plot = MetricPlot(
+        metric=Metric("accuracy", direction="max"), identifier=Param("seed")
+    )
+    df = plot.build(make_runs())
+    assert isinstance(df, pd.DataFrame)
+    assert set(df.columns) >= {"step", "value", "metric", "run", "run_id"}
+    assert df["metric"].unique().tolist() == ["accuracy"]
+
+
+def test_metric_plot_multi() -> None:
+    plot = MetricPlot(
+        metric=[Metric("accuracy", direction="max")], identifier=Param("seed")
+    )
+    df = plot.build(make_runs())
+    assert isinstance(df, pd.DataFrame)
+    assert all(k in df.columns for k in ["step", "value", "metric", "run"])