Skip to content

README

liblaf.cherries.core.metrics

Type Aliases:

  • MetricsLike –

    Nested mapping accepted by batch metric logging.

Classes:

  • Metric –

    In-memory samples for one scalar metric.

  • MetricPluginProtocol –

    Hook surface for plugins that receive scalar metrics.

  • MetricsManager –

    Store scalar metrics and mirror them to plugins.

MetricsLike 

MetricsLike = Mapping[str, SupportsFloat | MetricsLike]

Nested mapping accepted by batch metric logging.

Metric

In-memory samples for one scalar metric.

Values, steps, and timestamps are stored in compact arrays and materialized as a Polars dataframe only when requested.

Parameters:

  • name (str) –

    Metric name.

  • value (array[float], default: array('d') ) –

    Metric values as doubles.

  • step (array[int], default: array('L') ) –

    Metric steps as unsigned integers.

  • timestamp_ms (array[int], default: array('Q') ) –

    Sample timestamps as milliseconds since the Unix epoch.

Methods:

  • append –

    Append one metric sample.

  • to_polars –

    Convert samples to a dataframe with name, value, step, and time.

Attributes:

name instance-attribute 

name: str

Metric name.

step class-attribute instance-attribute 

step: array[int] = field(factory=lambda: array('L'))

Metric steps as unsigned integers.

timestamp_ms class-attribute instance-attribute 

timestamp_ms: array[int] = field(factory=lambda: array("Q"))

Sample timestamps as milliseconds since the Unix epoch.

value class-attribute instance-attribute 

value: array[float] = field(factory=lambda: array('d'))

Metric values as doubles.

append 

append(value: float, step: int, time: datetime) -> None

Append one metric sample.

Parameters:

  • value (float) –

    Scalar value.

  • step (int) –

    Metric step.

  • time (datetime) –

    Sample timestamp.

Source code in src/liblaf/cherries/core/metrics/_struct.py 
28
29
30
31
32
33
34
35
36
37
38
39
def append(self, value: float, step: int, time: datetime.datetime) -> None:
    """Append one metric sample.

    Args:
        value: Scalar value.
        step: Metric step.
        time: Sample timestamp.
    """
    timestamp_ms: int = int(time.timestamp() * 1000)
    self.value.append(value)
    self.step.append(step)
    self.timestamp_ms.append(timestamp_ms)

to_polars 

to_polars() -> DataFrame

Convert samples to a dataframe with name, value, step, and time.

Source code in src/liblaf/cherries/core/metrics/_struct.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def to_polars(self) -> pl.DataFrame:
    """Convert samples to a dataframe with `name`, `value`, `step`, and `time`."""
    df: pl.DataFrame = (
        pl.from_dict(
            {
                "value": self.value,
                "step": self.step,
                "timestamp_ms": self.timestamp_ms,
            },
            {"value": pl.Float64, "step": pl.UInt64, "timestamp_ms": pl.UInt64},
        )
        .with_columns(
            pl.lit(self.name).alias("name"),
            pl.col("timestamp_ms").cast(pl.Datetime("ms")).alias("time"),
        )
        .select("name", "value", "step", "time")
    )
    return df

MetricPluginProtocol

Bases: Protocol


              flowchart TD
              liblaf.cherries.core.metrics.MetricPluginProtocol[MetricPluginProtocol]

              

              click liblaf.cherries.core.metrics.MetricPluginProtocol href "" "liblaf.cherries.core.metrics.MetricPluginProtocol"

Hook surface for plugins that receive scalar metrics.

Methods:

  • log_metric –

    Record one metric sample.

  • log_metrics –

    Record a batch of already-flattened metric samples.

log_metric 

log_metric(
    name: str, value: float, *, step: int, time: datetime
) -> None

Record one metric sample.

Source code in src/liblaf/cherries/core/metrics/_protocol.py 
12
13
14
15
16
def log_metric(
    self, name: str, value: float, *, step: int, time: datetime.datetime
) -> None:
    """Record one metric sample."""
    ...

log_metrics 

log_metrics(
    metrics: dict[str, float], *, step: int, time: datetime
) -> None

Record a batch of already-flattened metric samples.

Source code in src/liblaf/cherries/core/metrics/_protocol.py 
18
19
20
21
22
def log_metrics(
    self, metrics: dict[str, float], *, step: int, time: datetime.datetime
) -> None:
    """Record a batch of already-flattened metric samples."""
    ...

MetricsManager

Store scalar metrics and mirror them to plugins.

Metrics are kept in memory as named series and can be exported as Polars dataframes. Batch logging accepts nested mappings and stores them with slash-delimited names.

Parameters:

  • plugins (MetricPluginProtocol) –

    Plugin delegate that receives metric events.

  • metrics (dict[str, Metric], default: <class 'dict'> ) –

    dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

  • step (int, default: 0 ) –

    Default step used when a log call does not provide one.

Methods:

  • get_metric –

    Return one metric as a Polars dataframe.

  • get_metrics –

    Return selected metrics concatenated into one dataframe.

  • log_metric –

    Log one scalar metric.

  • log_metrics –

    Log multiple scalar metrics.

Attributes:

metrics class-attribute instance-attribute 

metrics: dict[str, Metric] = field(factory=dict)

Metric series by name.

plugins instance-attribute 

plugins: MetricPluginProtocol

Plugin delegate that receives metric events.

step class-attribute instance-attribute 

step: int = 0

Default step used when a log call does not provide one.

get_metric 

get_metric(name: str) -> DataFrame

Return one metric as a Polars dataframe.

Raises:

  • KeyError –

    If name has not been logged.

Source code in src/liblaf/cherries/core/metrics/_manager.py 
32
33
34
35
36
37
38
def get_metric(self, name: str) -> pl.DataFrame:
    """Return one metric as a Polars dataframe.

    Raises:
        KeyError: If `name` has not been logged.
    """
    return self.metrics[name].to_polars()

get_metrics 

get_metrics(
    names: Iterable[str] | None = None,
) -> DataFrame

Return selected metrics concatenated into one dataframe.

Parameters:

  • names (Iterable[str] | None, default: None ) –

    Metric names to include. When omitted, all known metrics are returned.

Raises:

  • KeyError –

    If any requested name has not been logged.

Source code in src/liblaf/cherries/core/metrics/_manager.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_metrics(self, names: Iterable[str] | None = None) -> pl.DataFrame:
    """Return selected metrics concatenated into one dataframe.

    Args:
        names: Metric names to include. When omitted, all known metrics are
            returned.

    Raises:
        KeyError: If any requested name has not been logged.
    """
    if names is None:
        names: Iterable[str] = self.metrics.keys()
    return pl.concat(
        [self.metrics[name].to_polars() for name in names], how="vertical"
    )

log_metric 

log_metric(
    name: str,
    value: SupportsFloat,
    *,
    step: SupportsInt | None = None,
    time: datetime | None = None,
) -> None

Log one scalar metric.

Parameters:

  • name (str) –

    Metric name.

  • value (SupportsFloat) –

    Numeric value convertible with float().

  • step (SupportsInt | None, default: None ) –

    Optional step override. Defaults to step.

  • time (datetime | None, default: None ) –

    Optional timestamp. Defaults to the current local time.

Source code in src/liblaf/cherries/core/metrics/_manager.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def log_metric(
    self,
    name: str,
    value: SupportsFloat,
    *,
    step: SupportsInt | None = None,
    time: datetime | None = None,
) -> None:
    """Log one scalar metric.

    Args:
        name: Metric name.
        value: Numeric value convertible with `float()`.
        step: Optional step override. Defaults to [`step`][liblaf.cherries.core.metrics.MetricsManager.step].
        time: Optional timestamp. Defaults to the current local time.
    """
    step, time = self._parse_inputs(step, time)
    value: float = float(value)
    self._append_metric(name, value, step=step, time=time)
    self.plugins.log_metric(name, value, step=step, time=time)

log_metrics 

log_metrics(
    metrics: MetricsLike,
    *,
    step: SupportsInt | None = None,
    time: datetime | None = None,
) -> None

Log multiple scalar metrics.

Nested mappings are flattened with /, so {"train": {"loss": 0.5}} is stored as train/loss.

Parameters:

  • metrics (MetricsLike) –

    Metric values to log.

  • step (SupportsInt | None, default: None ) –

    Optional step override shared by all values.

  • time (datetime | None, default: None ) –

    Optional timestamp shared by all values.

Source code in src/liblaf/cherries/core/metrics/_manager.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def log_metrics(
    self,
    metrics: MetricsLike,
    *,
    step: SupportsInt | None = None,
    time: datetime | None = None,
) -> None:
    """Log multiple scalar metrics.

    Nested mappings are flattened with `/`, so `{"train": {"loss": 0.5}}`
    is stored as `train/loss`.

    Args:
        metrics: Metric values to log.
        step: Optional step override shared by all values.
        time: Optional timestamp shared by all values.
    """
    step, time = self._parse_inputs(step, time)
    flat: dict[str, SupportsFloat] = flatten_dict(metrics)
    flat: dict[str, float] = {name: float(value) for name, value in flat.items()}
    for name, value in flat.items():
        self._append_metric(name, value, step=step, time=time)
    self.plugins.log_metrics(flat, step=step, time=time)