nannyml.runner module

Used as an access point to start using NannyML in its most simple form.

class nannyml.runner.CalculatorFactory[source]

Bases: object

A factory class that produces Metric instances based on a given magic string or a metric specification.

classmethod register(name: str, calculator_type: Union[Type[nannyml._typing.Calculator], Type[nannyml._typing.Estimator]])[source]

registry: Dict[str, Type] = {'cbpe': <class 'nannyml.performance_estimation.confidence_based.cbpe.CBPE'>, 'dle': <class 'nannyml.performance_estimation.direct_loss_estimation.dle.DLE'>, 'missing_values': <class 'nannyml.data_quality.missing.calculator.MissingValuesCalculator'>, 'multivariate_drift': <class 'nannyml.drift.multivariate.data_reconstruction.calculator.DataReconstructionDriftCalculator'>, 'performance': <class 'nannyml.performance_calculation.calculator.PerformanceCalculator'>, 'univariate_drift': <class 'nannyml.drift.univariate.calculator.UnivariateDriftCalculator'>, 'unseen_values': <class 'nannyml.data_quality.unseen.calculator.UnseenValuesCalculator'>}

class nannyml.runner.RunContext(current_step: int, total_steps: int, current_calculator: str, current_calculator_config: Optional[Dict[str, Any]] = None, current_calculator_success: bool = True, run_success: bool = True, result: Optional[nannyml._Result] = None)[source]

Bases: object

STEPS_PER_CALCULATOR = 3

current_calculator: str

current_calculator_config: Optional[Dict[str, Any]] = None

current_calculator_success: bool = True

current_step: int

increase_step()[source]

result: Optional[nannyml._typing.Result] = None

run_success: bool = True

total_steps: int

class nannyml.runner.RunInput(reference_data: pandas.core.frame.DataFrame, analysis_data: pandas.core.frame.DataFrame, target_data: Optional[pandas.core.frame.DataFrame] = None, target_join_column: Optional[str] = None)[source]

Bases: object

analysis_data: pandas.core.frame.DataFrame

reference_data: pandas.core.frame.DataFrame

target_data: Optional[pandas.core.frame.DataFrame] = None

target_join_column: Optional[str] = None

class nannyml.runner.RunnerLogger(logger: logging.Logger, console: Optional[rich.console.Console] = None)[source]

Bases: object

log(message: object, log_level: int = 20)[source]

nannyml.runner.get_output_writers(outputs_config: Optional[List[nannyml.config.WriterConfig]], logger: Optional[nannyml.runner.RunnerLogger] = None) → List[Tuple[nannyml.io.base.Writer, Dict[str, Any]]][source]

nannyml.runner.get_store(store_config: Optional[nannyml.config.StoreConfig], logger: Optional[nannyml.runner.RunnerLogger] = None) → Optional[nannyml.io.store.base.Store][source]

nannyml.runner.read_data(input_config: nannyml.config.InputDataConfig, logger: Optional[nannyml.runner.RunnerLogger] = None) → pandas.core.frame.DataFrame[source]

nannyml.runner.run(config: nannyml.config.Config, input: typing.Optional[nannyml.runner.RunInput] = None, logger: logging.Logger = <Logger nannyml.runner (WARNING)>, console: typing.Optional[rich.console.Console] = None, on_fit: typing.Optional[typing.Callable[[nannyml.runner.RunContext], typing.Any]] = None, on_calculate: typing.Optional[typing.Callable[[nannyml.runner.RunContext], typing.Any]] = None, on_write: typing.Optional[typing.Callable[[nannyml.runner.RunContext], typing.Any]] = None, on_calculator_complete: typing.Optional[typing.Callable[[nannyml.runner.RunContext], typing.Any]] = None, on_run_complete: typing.Optional[typing.Callable[[nannyml.runner.RunContext], typing.Any]] = None, on_fail: typing.Optional[typing.Callable[[nannyml.runner.RunContext, typing.Optional[Exception]], typing.Any]] = None)[source]

nannyml.runner.run_context(config: nannyml.config.Config)[source]