arkas.runner¶

arkas.runner ¶

Contain runners.

arkas.runner.AnalysisRunner ¶

Bases: BaseRunner

Implement a runner to analyze data.

Parameters:

Name	Type	Description	Default
`ingestor`	`BaseIngestor \| dict`	The data ingestor or its configuration.	required
`transformer`	`BaseTransformer \| dict`	The data transformer or its configuration.	required
`analyzer`	`BaseAnalyzer \| dict`	The analyzer or its configuration.	required
`exporter`	`BaseExporter \| dict`	The output exporter or its configuration.	required
`lazy`	`bool`	If `True`, the analyzer computation is done lazily.	`True`

Example usage:

>>> import tempfile
>>> import polars as pl
>>> from pathlib import Path
>>> from grizz.ingestor import Ingestor
>>> from grizz.transformer import SequentialTransformer
>>> from arkas.analyzer import AccuracyAnalyzer
>>> from arkas.exporter import MetricExporter
>>> from arkas.runner import AnalysisRunner
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     runner = AnalysisRunner(
...         ingestor=Ingestor(
...             pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             )
...         ),
...         transformer=SequentialTransformer(transformers=[]),
...         analyzer=AccuracyAnalyzer(y_true="target", y_pred="pred"),
...         exporter=MetricExporter(Path(tmpdir).joinpath("metrics.pkl")),
...     )
...     print(runner)
...     runner.run()
...
AnalysisRunner(
  (ingestor): Ingestor(shape=(5, 2))
  (transformer): SequentialTransformer()
  (analyzer): AccuracyAnalyzer(y_true='target', y_pred='pred', drop_nulls=True, missing_policy='raise', nan_policy='propagate')
  (exporter): MetricExporter(
      (path): .../metrics.pkl
      (saver): PickleSaver()
      (exist_ok): False
      (show_metrics): False
    )
  (lazy): True
)

arkas.runner.BaseRunner ¶

Bases: ABC

Define the base class to implement a runner.

Example usage:

>>> import tempfile
>>> import polars as pl
>>> from pathlib import Path
>>> from grizz.ingestor import Ingestor
>>> from grizz.transformer import SequentialTransformer
>>> from arkas.analyzer import AccuracyAnalyzer
>>> from arkas.exporter import MetricExporter
>>> from arkas.runner import AnalysisRunner
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     runner = AnalysisRunner(
...         ingestor=Ingestor(
...             pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             )
...         ),
...         transformer=SequentialTransformer(transformers=[]),
...         analyzer=AccuracyAnalyzer(y_true="target", y_pred="pred"),
...         exporter=MetricExporter(Path(tmpdir).joinpath("metrics.pkl")),
...     )
...     print(runner)
...     runner.run()
...
AnalysisRunner(
  (ingestor): Ingestor(shape=(5, 2))
  (transformer): SequentialTransformer()
  (analyzer): AccuracyAnalyzer(y_true='target', y_pred='pred', drop_nulls=True, missing_policy='raise', nan_policy='propagate')
  (exporter): MetricExporter(
      (path): .../metrics.pkl
      (saver): PickleSaver()
      (exist_ok): False
      (show_metrics): False
    )
  (lazy): True
)

arkas.runner.BaseRunner.run `abstractmethod` ¶

run() -> Any

Execute the logic of the runner.

Returns:

Type	Description
`Any`	Any artifact of the runner

Example usage:

>>> import tempfile
>>> import polars as pl
>>> from pathlib import Path
>>> from grizz.ingestor import Ingestor
>>> from grizz.transformer import SequentialTransformer
>>> from arkas.analyzer import AccuracyAnalyzer
>>> from arkas.exporter import MetricExporter
>>> from arkas.runner import AnalysisRunner
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     runner = AnalysisRunner(
...         ingestor=Ingestor(
...             pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             )
...         ),
...         transformer=SequentialTransformer(transformers=[]),
...         analyzer=AccuracyAnalyzer(y_true="target", y_pred="pred"),
...         exporter=MetricExporter(Path(tmpdir).joinpath("metrics.pkl")),
...     )
...     runner.run()
...

arkas.runner.EvaluationRunner ¶

Bases: BaseRunner

Implement a simple evaluation runner.

Parameters:

Name	Type	Description	Default
`ingestor`	`BaseIngestor \| dict`	The data ingestor or its configuration.	required
`transformer`	`BaseTransformer \| dict`	The data transformer or its configuration.	required
`evaluator`	`BaseEvaluator \| dict`	The evaluator or its configuration.	required
`saver`	`BaseSaver \| dict`	The metric saver or its configuration.	required
`path`	`Path \| str`	The path where to save the metrics.	required
`show_metrics`	`bool`	If `True`, the metrics are shown in the logging output.	`True`

Example usage:

>>> import tempfile
>>> import polars as pl
>>> from pathlib import Path
>>> from iden.io import PickleSaver
>>> from grizz.ingestor import Ingestor
>>> from grizz.transformer import SequentialTransformer
>>> from arkas.evaluator import AccuracyEvaluator
>>> from arkas.runner import EvaluationRunner
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     path = Path(tmpdir).joinpath("metrics.pkl")
...     runner = EvaluationRunner(
...         ingestor=Ingestor(
...             pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             )
...         ),
...         transformer=SequentialTransformer(transformers=[]),
...         evaluator=AccuracyEvaluator(y_true="target", y_pred="pred"),
...         saver=PickleSaver(),
...         path=path,
...     )
...     print(runner)
...     runner.run()
...
EvaluationRunner(
  (ingestor): Ingestor(shape=(5, 2))
  (transformer): SequentialTransformer()
  (evaluator): AccuracyEvaluator(y_true='target', y_pred='pred', drop_nulls=True, nan_policy='propagate')
  (saver): PickleSaver()
  (path): .../metrics.pkl
  (show_metrics): True
)

arkas.runner.is_runner_config ¶

is_runner_config(config: dict) -> bool

Indicate if the input configuration is a configuration for a BaseRunner.

This function only checks if the value of the key _target_ is valid. It does not check the other values. If _target_ indicates a function, the returned type hint is used to check the class.

Parameters:

Name	Type	Description	Default
`config`	`dict`	The configuration to check.	required

Returns:

Type	Description
`bool`	`True` if the input configuration is a configuration for a `BaseRunner` object.

Example usage:

>>> import polars as pl
>>> from arkas.runner import is_runner_config
>>> is_runner_config(
...     {
...         "_target_": "arkas.runner.AnalysisRunner",
...         "ingestor": {
...             "_target_": "grizz.ingestor.Ingestor",
...             "frame": pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             ),
...         },
...         "transformer": {"_target_": "grizz.transformer.DropDuplicate"},
...         "analyzer": {
...             "_target_": "arkas.analyzer.AccuracyAnalyzer",
...             "y_true": "target",
...             "y_pred": "pred",
...         },
...         "exporter": {
...             "_target_": "arkas.exporter.MetricExporter",
...             "path": "/path/to/data.csv",
...         },
...     }
... )
True

arkas.runner.setup_runner ¶

setup_runner(runner: BaseRunner | dict) -> BaseRunner

Set up a runner.

The runner is instantiated from its configuration by using the BaseRunner factory function.

Parameters:

Name	Type	Description	Default
`runner`	`BaseRunner \| dict`	Specifies a runner or its configuration.	required

Returns:

Type	Description
`BaseRunner`	An instantiated runner.

Example usage:

>>> import polars as pl
>>> from arkas.runner import setup_runner
>>> runner = setup_runner(
...     {
...         "_target_": "arkas.runner.AnalysisRunner",
...         "ingestor": {
...             "_target_": "grizz.ingestor.Ingestor",
...             "frame": pl.DataFrame(
...                 {
...                     "pred": [3, 2, 0, 1, 0],
...                     "target": [3, 2, 0, 1, 0],
...                 }
...             ),
...         },
...         "transformer": {"_target_": "grizz.transformer.DropDuplicate"},
...         "analyzer": {
...             "_target_": "arkas.analyzer.AccuracyAnalyzer",
...             "y_true": "target",
...             "y_pred": "pred",
...         },
...         "exporter": {
...             "_target_": "arkas.exporter.MetricExporter",
...             "path": "/path/to/data.csv",
...         },
...     }
... )
>>> runner
AnalysisRunner(
  (ingestor): Ingestor(shape=(5, 2))
  (transformer): DropDuplicateTransformer(columns=None, exclude_columns=(), missing_policy='raise')
  (analyzer): AccuracyAnalyzer(y_true='target', y_pred='pred', drop_nulls=True, missing_policy='raise', nan_policy='propagate')
  (exporter): MetricExporter(
      (path): /path/to/data.csv
      (saver): PickleSaver()
      (exist_ok): False
      (show_metrics): False
    )
  (lazy): True
)

arkas.runner¶

arkas.runner ¶

arkas.runner.AnalysisRunner ¶

arkas.runner.BaseRunner ¶

arkas.runner.BaseRunner.run abstractmethod ¶

arkas.runner.EvaluationRunner ¶

arkas.runner.is_runner_config ¶

arkas.runner.setup_runner ¶

arkas.runner.BaseRunner.run `abstractmethod` ¶