MetricMeter

danling.metrics.metric_meter

MetricMeter

Bases: AverageMeter

A memory-efficient metric tracker that computes and averages metrics across batches.

MetricMeter applies a metric function to each batch and maintains running averages without storing the complete history of predictions and labels. This makes it ideal for metrics that can be meaningfully averaged across batches (like accuracy or loss).

Attributes:

Name	Type	Description
metric	Callable	The metric function to compute on each batch
preprocess		Optional preprocessing function applied before the metric
val	float	Result from the most recent batch
bat	float	Result from the most recent batch, synchronized across devices
avg	float	Weighted average of all results so far
sum	float	Running sum of (metric × batch_size) values
count	int	Running sum of batch sizes

Parameters:

Name	Type	Description	Default
metric	Callable	Function that computes a metric given input and target tensors	required
preprocess	Callable | None	Optional preprocessing function to apply before computing the metric	None

Examples:

>>> import torch
>>> from danling.metrics.functional import accuracy
>>> meter = MetricMeter(accuracy)
>>> meter.update(torch.tensor([0.1, 0.8, 0.6, 0.2]), torch.tensor([0, 1, 0, 0]))
>>> meter.val
0.75
>>> meter.avg
0.75
>>> meter.update(torch.tensor([0.1, 0.7, 0.3, 0.2, 0.8, 0.4]), torch.tensor([0, 1, 1, 0, 0, 1]))
>>> meter.val
0.5
>>> meter.avg
0.6
>>> meter.sum
6.0
>>> meter.count
10
>>> meter.reset()
MetricMeter(accuracy)
>>> meter.val
nan
>>> meter.avg
nan

Notes

• MetricMeter is more memory-efficient than Metrics because it only stores running statistics
• Only suitable for metrics that can be meaningfully averaged batch-by-batch
• Not suitable for metrics like AUROC that need the entire dataset
• The metric function should accept input and target tensors and return a scalar value
• For multiple metrics, use MetricMeters

See Also

• AverageMeter: A lightweight utility to compute and store running averages of values.

Source code in danling/metrics/metric_meter.py

class MetricMeter(AverageMeter):
    r"""
    A memory-efficient metric tracker that computes and averages metrics across batches.

    MetricMeter applies a metric function to each batch and maintains running averages
    without storing the complete history of predictions and labels. This makes it ideal for
    metrics that can be meaningfully averaged across batches (like accuracy or loss).

    Attributes:
        metric: The metric function to compute on each batch
        preprocess: Optional preprocessing function applied before the metric
        val: Result from the most recent batch
        bat: Result from the most recent batch, synchronized across devices
        avg: Weighted average of all results so far
        sum: Running sum of (metric × batch_size) values
        count: Running sum of batch sizes

    Args:
        metric: Function that computes a metric given input and target tensors
        preprocess: Optional preprocessing function to apply before computing the metric

    Examples:
        >>> import torch
        >>> from danling.metrics.functional import accuracy
        >>> meter = MetricMeter(accuracy)
        >>> meter.update(torch.tensor([0.1, 0.8, 0.6, 0.2]), torch.tensor([0, 1, 0, 0]))
        >>> meter.val
        0.75
        >>> meter.avg
        0.75
        >>> meter.update(torch.tensor([0.1, 0.7, 0.3, 0.2, 0.8, 0.4]), torch.tensor([0, 1, 1, 0, 0, 1]))
        >>> meter.val
        0.5
        >>> meter.avg
        0.6
        >>> meter.sum
        6.0
        >>> meter.count
        10
        >>> meter.reset()
        MetricMeter(accuracy)
        >>> meter.val
        nan
        >>> meter.avg
        nan

    Notes:
        - MetricMeter is more memory-efficient than [`Metrics`][danling.metrics.metrics.Metrics]
          because it only stores running statistics
        - Only suitable for metrics that can be meaningfully averaged batch-by-batch
        - Not suitable for metrics like AUROC that need the entire dataset
        - The metric function should accept input and target tensors and return a scalar value
        - For multiple metrics, use [`MetricMeters`][danling.metrics.metric_meter.MetricMeters]

    See Also:
        - [`AverageMeter`][danling.metrics.average_meter.AverageMeter]:
            A lightweight utility to compute and store running averages of values.
    """

    metric: Callable

    def __init__(
        self,
        metric: Callable,
        *,
        preprocess: Callable | None = None,
    ) -> None:
        super().__init__()
        if not callable(metric):
            raise ValueError(f"Expected metric to be callable, but got {type(metric)}")
        self.metric = metric
        self.preprocess = preprocess

    def update(  # type: ignore[override] # pylint: disable=W0237
        self,
        input: Tensor | NestedTensor,  # pylint: disable=W0622
        target: Tensor | NestedTensor,
        *,
        n: int | None = None,
    ) -> None:
        r"""
        Updates the average and current value in the meter.

        Args:
            input: Prediction tensor or nested tensor.
            target: Ground-truth tensor or nested tensor.
            n: Optional number of samples represented by this update. When omitted,
                the batch size is inferred from the inputs.
        """

        if self.preprocess is not None:
            input, target = self.preprocess(input, target)

        self._update(input, target, n=n)

    def _update(
        self,
        input: Tensor | NestedTensor | Sequence,
        target: Tensor | NestedTensor | Sequence,
        *,
        n: int | None = None,
    ) -> None:
        if n is None:
            try:
                n = len(input)
            except TypeError:
                n = 1
        elif not isinstance(n, int):
            raise ValueError(f"n must be a number, but got {n!r}")
        elif n <= 0:
            raise ValueError(f"n must be a positive integer, but got {n!r}")

        value = self.metric(input, target)
        if isinstance(value, Tensor):
            if value.numel() != 1:
                raise ValueError(
                    "Metric functions used with MetricMeter must return a scalar tensor, "
                    f"but got shape {tuple(value.shape)}"
                )
            value = value.item()
        super().update(value=value, n=n)

    def __repr__(self):
        metric = self.metric
        if isinstance(metric, partial):
            metric = metric.func
        return f"{self.__class__.__name__}({metric.__name__})"

update

Python

update(input: Tensor | NestedTensor, target: Tensor | NestedTensor, *, n: int | None = None) -> None

Updates the average and current value in the meter.

Parameters:

Name	Type	Description	Default
input	Tensor | NestedTensor	Prediction tensor or nested tensor.	required
target	Tensor | NestedTensor	Ground-truth tensor or nested tensor.	required
n	int | None	Optional number of samples represented by this update. When omitted, the batch size is inferred from the inputs.	None

Source code in danling/metrics/metric_meter.py

def update(  # type: ignore[override] # pylint: disable=W0237
    self,
    input: Tensor | NestedTensor,  # pylint: disable=W0622
    target: Tensor | NestedTensor,
    *,
    n: int | None = None,
) -> None:
    r"""
    Updates the average and current value in the meter.

    Args:
        input: Prediction tensor or nested tensor.
        target: Ground-truth tensor or nested tensor.
        n: Optional number of samples represented by this update. When omitted,
            the batch size is inferred from the inputs.
    """

    if self.preprocess is not None:
        input, target = self.preprocess(input, target)

    self._update(input, target, n=n)

MetricMeters

Bases: AverageMeters

A container for managing multiple MetricMeter instances with shared preprocessing.

MetricMeters allows you to organize and track multiple metrics in a unified interface, with consistent preprocessing applied to all inputs before computing each metric. This is particularly useful when you want to track several metrics that can be meaningfully averaged across batches.

Attributes:

Name	Type	Description
preprocess		Shared preprocessing function for all meters
val	RoundDict[str, float]	Dictionary of current values from all meters
avg	RoundDict[str, float]	Dictionary of running averages from all meters
sum	RoundDict[str, float]	Dictionary of sums from all meters
count	RoundDict[str, int]	Dictionary of counts from all meters

Parameters:

Name	Type	Description	Default
*args		Either metric functions or a Metrics instance to extract metrics from	()
preprocess	Callable	Preprocessing function to apply to inputs before computing metrics	base_preprocess
**meters		Named MetricMeter instances or metric functions	{}

Examples:

>>> import torch
>>> from danling.metrics.functional import accuracy, auroc, auprc
>>> meters = MetricMeters(acc=accuracy, auroc=auroc, auprc=auprc)
>>> meters.update([0.1, 0.8, 0.6, 0.2], [0, 1, 0, 0])
>>> meters.sum.dict()
{'acc': 3.0, 'auroc': 4.0, 'auprc': 4.0}
>>> meters.count.dict()
{'acc': 4, 'auroc': 4, 'auprc': 4}
>>> meters['auroc'].update(torch.tensor([0.2, 0.8]), torch.tensor([0, 1]))
>>> meters.sum.dict()
{'acc': 3.0, 'auroc': 6.0, 'auprc': 4.0}
>>> meters.count.dict()
{'acc': 4, 'auroc': 6, 'auprc': 4}
>>> meters.update([[0.1, 0.7, 0.3, 0.2], [0.8, 0.4]], [[0, 0, 1, 0], [0, 0]])
>>> meters.sum.round(2).dict()
{'acc': 6.0, 'auroc': 8.4, 'auprc': 5.5}
>>> meters.count.dict()
{'acc': 10, 'auroc': 12, 'auprc': 10}
>>> meters['auroc'].update(torch.tensor([0.4, 0.8, 0.6, 0.2]), torch.tensor([0, 1, 1, 0]))
>>> meters.avg.round(4).dict()
{'acc': 0.6, 'auroc': 0.775, 'auprc': 0.55}
>>> meters.update(dict(loss=""))
Traceback (most recent call last):
TypeError: ...update() missing 1 required positional argument: 'target'

Notes

• MetricMeters manages multiple MetricMeter instances with shared preprocessing
• Each metric is computed independently but uses the same inputs
• All meters are updated simultaneously when you call update()
• Individual meters can be accessed like dictionary items or attributes

See Also

• AverageMeters: A container for managing multiple average meters in one object.
• Metrics: Metric tracker that stores the complete prediction and target history.

Source code in danling/metrics/metric_meter.py

class MetricMeters(AverageMeters):
    r"""
    A container for managing multiple MetricMeter instances with shared preprocessing.

    MetricMeters allows you to organize and track multiple metrics in a unified interface,
    with consistent preprocessing applied to all inputs before computing each metric.
    This is particularly useful when you want to track several metrics that can be
    meaningfully averaged across batches.

    Attributes:
        preprocess: Shared preprocessing function for all meters
        val: Dictionary of current values from all meters
        avg: Dictionary of running averages from all meters
        sum: Dictionary of sums from all meters
        count: Dictionary of counts from all meters

    Args:
        *args: Either metric functions or a Metrics instance to extract metrics from
        preprocess: Preprocessing function to apply to inputs before computing metrics
        **meters: Named MetricMeter instances or metric functions

    Examples:
        >>> import torch
        >>> from danling.metrics.functional import accuracy, auroc, auprc
        >>> meters = MetricMeters(acc=accuracy, auroc=auroc, auprc=auprc)
        >>> meters.update([0.1, 0.8, 0.6, 0.2], [0, 1, 0, 0])
        >>> meters.sum.dict()
        {'acc': 3.0, 'auroc': 4.0, 'auprc': 4.0}
        >>> meters.count.dict()
        {'acc': 4, 'auroc': 4, 'auprc': 4}
        >>> meters['auroc'].update(torch.tensor([0.2, 0.8]), torch.tensor([0, 1]))
        >>> meters.sum.dict()
        {'acc': 3.0, 'auroc': 6.0, 'auprc': 4.0}
        >>> meters.count.dict()
        {'acc': 4, 'auroc': 6, 'auprc': 4}
        >>> meters.update([[0.1, 0.7, 0.3, 0.2], [0.8, 0.4]], [[0, 0, 1, 0], [0, 0]])
        >>> meters.sum.round(2).dict()
        {'acc': 6.0, 'auroc': 8.4, 'auprc': 5.5}
        >>> meters.count.dict()
        {'acc': 10, 'auroc': 12, 'auprc': 10}
        >>> meters['auroc'].update(torch.tensor([0.4, 0.8, 0.6, 0.2]), torch.tensor([0, 1, 1, 0]))
        >>> meters.avg.round(4).dict()
        {'acc': 0.6, 'auroc': 0.775, 'auprc': 0.55}
        >>> meters.update(dict(loss=""))  # doctest: +ELLIPSIS
        Traceback (most recent call last):
        TypeError: ...update() missing 1 required positional argument: 'target'

    Notes:
        - `MetricMeters` manages multiple `MetricMeter` instances with shared preprocessing
        - Each metric is computed independently but uses the same inputs
        - All meters are updated simultaneously when you call `update()`
        - Individual meters can be accessed like dictionary items or attributes

    See Also:
        - [`AverageMeters`][danling.metrics.average_meter.AverageMeters]:
            A container for managing multiple average meters in one object.
        - [`Metrics`][danling.metrics.metrics.Metrics]:
            Metric tracker that stores the complete prediction and target history.
    """

    preprocess = base_preprocess
    meter_cls = MetricMeter  # type: ignore[assignment]

    def __init__(
        self,
        *args,
        preprocess: Callable = base_preprocess,
        **meters,
    ) -> None:
        if args:
            from .metrics import Metrics

            if len(args) == 1 and isinstance(args[0], Metrics):
                metrics = args[0]
                for name, metric in metrics.metrics.items():
                    meters.setdefault(name, metric)
                if preprocess is base_preprocess:
                    preprocess = metrics.preprocess
            else:
                for metric in args:
                    if not callable(metric):
                        raise ValueError(f"Expected metric to be callable, but got {type(metric)}")
                    meters.setdefault(metric.__name__, metric)
        self.setattr("preprocess", preprocess)
        super().__init__(**meters)

    def _coerce_meter(self, value):  # type: ignore[override]
        meter_cls = self.getattr("meter_cls", MetricMeter)
        if isinstance(value, meter_cls):
            value.preprocess = None
            return value
        if callable(value):
            return meter_cls(value, preprocess=None)
        raise ValueError(f"Expected meter to be an instance of {meter_cls.__name__}, but got {type(value)}")

    def update(  # type: ignore[override] # pylint: disable=W0221
        self,
        input: Tensor | NestedTensor | Sequence,  # pylint: disable=W0622
        target: Tensor | NestedTensor | Sequence,
        *,
        n: int | None = None,
    ) -> None:
        r"""
        Updates the average and current value in all meters.

        Args:
            input: Input values to compute the metrics.
            target: Target values to compute the metrics.
            n: Optional number of samples represented by this update. Defaults to
                the inferred batch size.
        """

        input, target = self.preprocess(input, target)  # type: ignore[arg-type]
        if (
            isinstance(input, (Tensor, NestedTensor))
            and isinstance(target, (Tensor, NestedTensor))
            and input.ndim == target.ndim + 1
        ):
            input = input.squeeze(-1)
        if isinstance(input, (Tensor, NestedTensor)):
            input = input.detach().to("cpu")
        if isinstance(target, (Tensor, NestedTensor)):
            target = target.detach().to("cpu")
        for meter in self.values():
            if isinstance(meter, MetricMeter):
                meter._update(input, target, n=n)
            else:
                meter.update(input, target, n=n)

    def __repr__(self):
        keys = tuple(i for i in self.keys())
        return f"{self.__class__.__name__}{keys}"

update

Python

update(input: Tensor | NestedTensor | Sequence, target: Tensor | NestedTensor | Sequence, *, n: int | None = None) -> None

Updates the average and current value in all meters.

Parameters:

Name	Type	Description	Default
input	Tensor | NestedTensor | Sequence	Input values to compute the metrics.	required
target	Tensor | NestedTensor | Sequence	Target values to compute the metrics.	required
n	int | None	Optional number of samples represented by this update. Defaults to the inferred batch size.	None

Source code in danling/metrics/metric_meter.py

def update(  # type: ignore[override] # pylint: disable=W0221
    self,
    input: Tensor | NestedTensor | Sequence,  # pylint: disable=W0622
    target: Tensor | NestedTensor | Sequence,
    *,
    n: int | None = None,
) -> None:
    r"""
    Updates the average and current value in all meters.

    Args:
        input: Input values to compute the metrics.
        target: Target values to compute the metrics.
        n: Optional number of samples represented by this update. Defaults to
            the inferred batch size.
    """

    input, target = self.preprocess(input, target)  # type: ignore[arg-type]
    if (
        isinstance(input, (Tensor, NestedTensor))
        and isinstance(target, (Tensor, NestedTensor))
        and input.ndim == target.ndim + 1
    ):
        input = input.squeeze(-1)
    if isinstance(input, (Tensor, NestedTensor)):
        input = input.detach().to("cpu")
    if isinstance(target, (Tensor, NestedTensor)):
        target = target.detach().to("cpu")
    for meter in self.values():
        if isinstance(meter, MetricMeter):
            meter._update(input, target, n=n)
        else:
            meter.update(input, target, n=n)