linumpy.metrics.core#

Core metrics primitives: JSON encoder, PipelineMetrics, and IO.

Attributes#

logger

Classes#

`MetricsEncoder`	Custom JSON encoder to handle numpy types.
`PipelineMetrics`	Class for collecting and managing metrics from pipeline steps.

Functions#

load_metrics(filepath)

Load metrics from a JSON file.

Module Contents#

linumpy.metrics.core.logger[source]#

class linumpy.metrics.core.MetricsEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: json.JSONEncoder

Custom JSON encoder to handle numpy types.

default(o)[source]#

Serialize numpy integer and float types to Python builtins.

Parameters:: o (Any)
Return type:: Any

class linumpy.metrics.core.PipelineMetrics(step_name, output_dir=None)[source]#

Class for collecting and managing metrics from pipeline steps.

Each step can record multiple metrics with associated quality indicators. Metrics are saved as JSON files for later aggregation and report generation.

Parameters:

step_name (str)
output_dir (str | None)

DEFAULT_THRESHOLDS: ClassVar[dict][source]#

step_name[source]#

output_dir[source]#

metrics: dict[str, Any][source]#

warnings: list[str] = [][source]#

errors: list[str] = [][source]#

timestamp[source]#

add_metric(name, value, unit=None, threshold_name=None, custom_thresholds=None, description=None)[source]#

Add a metric with optional quality assessment.

Parameters:

name (str) – Name of the metric.
value (Any) – Value of the metric.
unit (str, optional) – Unit of measurement.
threshold_name (str, optional) – Name of threshold to use from DEFAULT_THRESHOLDS.
custom_thresholds (dict, optional) – Custom thresholds {‘warning’: val, ‘error’: val, ‘higher_is_better’: bool}
description (str, optional) – Human-readable description of the metric.

Return type:

None

add_info(name, value, description=None)[source]#

Add informational data (not quality-assessed).

Parameters:

name (str) – Name of the info field.
value (Any) – Value of the info field.
description (str, optional) – Human-readable description.

Return type:

None

add_params(params)[source]#

Record each entry of params as an info field. No-op if params is falsy.

Parameters:: params (dict | None)
Return type:: None

finalize(filename=None)[source]#

Save metrics, log warnings/errors, and return self.

Parameters:: filename (str | None)
Return type:: PipelineMetrics

get_overall_status()[source]#

Get overall status based on all metrics.

Returns:: ‘error’, ‘warning’, or ‘ok’
Return type:: str

to_dict()[source]#

Convert metrics to dictionary format.

Returns:: Dictionary containing all metrics and metadata.
Return type:: dict

save(filename=None)[source]#

Save metrics to JSON file.

Parameters:: filename (str, optional) – Filename for metrics file. Defaults to ‘{step_name}_metrics.json’
Returns:: Path to the saved metrics file.
Return type:: Path

log_issues()[source]#

Log any warnings or errors to the logger.

Return type:: None

linumpy.metrics.core.load_metrics(filepath)[source]#

Load metrics from a JSON file.

Parameters:: filepath (str or Path) – Path to the metrics JSON file.
Returns:: Loaded metrics dictionary.
Return type:: dict