Configuration (`dqa.configuration`)#

class dqa.configuration.configuration.Algorithm(the_class: type, parameters: Dict[str, Any], meta_parameters: Dict[str, Any])#

Represents a (machine learning) algorithm in the system.

There can be one global object or one object for each machine.

Parameters:

the_class (type) – The class that performs the algorithm. This must implement a fit(X, y) and a predict(X) method in the scikit-learn style.
parameters (dict) – The constructor parameters to create the object of the_class.
meta_parameters (dict) – Meta parameters for the algorithm object. Most importantly, a ‘Filename’ entry indicates that the algorithm should be loaded from a pickled file. ‘PerMachine’ indicates that one instance of the algorithm should be created for each machine.

Methods

fit
fit2
generate_object
get_object_for_machine
load_objects_from_file
predict
save
setup_new_objects

fit(machine_name: str, X, y, sample_weights=None, **kwargs)#

fit2(algo_object, X, y, sample_weights=None, **kwargs)#

generate_object() → Any#

get_object_for_machine(machine_name: str) → Any#

load_objects_from_file()#

predict(machine_name: str, X, **kwargs)#

save(filename)#

setup_new_objects()#

class dqa.configuration.configuration.ConfigClass(the_class: type, parameters: Dict)#

Represents a class that is used in a configuration list, for example a task class.

Parameters:

the_class (type) – The type (class) to be referred to, of which objects are generated.
parameters (dict) – General settings to configure this class. This will be assigned as a static variable of the class.

Methods

`find_key_for_class`(class_dict, the_class)	Given a class, find the corresponding identifier :param class_dict: The dictionary mapping identifiers to classes.
`load_from_file`(file_path)	Loads config classes from a json definition file.

static find_key_for_class(class_dict: Dict[str, Any], the_class: type) → str#: Given a class, find the corresponding identifier :param class_dict: The dictionary mapping identifiers to classes. :param the_class: The class to search for. :return: The identifier of the class.

static load_from_file(file_path: str) → Dict[str, Any]#: Loads config classes from a json definition file. :param file_path: The path to the file. :return: A dictionary mapping the identifiers to the corresponding ConfigClass objects.

parameters: Dict[str, Any]#

the_class: type#

class dqa.configuration.configuration.ConfigCollection(class_dict: Dict[str, ConfigClass], meta_parameter_names: List[str])#

Implements a collection (list or dictionary) of ConfigClass objects. Such collections are usually loaded by a (part of a ) JSON configuration file.

Methods

`init_entry`(config)	Initializes a single entry in the collection from a JSON configuration.
`init_entry_collection`(config)	Initializes the collection from a config object.
`init_object_collection`()	When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

add_resolving_parameters
copy_resolving_parameters

add_resolving_parameters(additional_parameters)#

class_dict: Dict[str, ConfigClass]#

abstract copy_resolving_parameters(additional_parameters: Dict[str, Any])#

entry_classes: Any#

entry_parameters: Any#

init_entry(config: Dict[str, Any])#: Initializes a single entry in the collection from a JSON configuration. :param config: The configuration, usually taken from a JSON file. :return: Tuple, containing the ConfigClass object, the constructor parameters, the meta parameters.

abstract init_entry_collection(config)#: Initializes the collection from a config object. This generates the collection of the ConfigClass objects for the individual entries. However, it does not generate the final objects yet. :param config: The configuration of the collection, usually taken from a JSON file.

abstract init_object_collection()#: When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

meta_parameter_names: List[str]#

meta_parameters: Any#

objects: Any#

resolving_parameters: Dict[str, Any]#

class dqa.configuration.configuration.ConfigDict(class_dict: Dict[str, ConfigClass], meta_parameter_names: List[str])#

A specific implementation of the ConfigCollection class where the entries are stored in a dictionary. (For example the dictionary of algorithms).

Methods

`init_entry`(config)	Initializes a single entry in the collection from a JSON configuration.
`init_entry_collection`(config)	Initializes the collection from a config object.
`init_object_collection`()	When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

add_resolving_parameters
copy_resolving_parameters

copy_resolving_parameters(additional_parameters: Dict[str, Any])#

entry_classes: Dict[str, ConfigClass | None] | None#

entry_parameters: Dict[str, Dict[str, Any]] | None#

init_entry_collection(config: Dict[str, Dict])#: Initializes the collection from a config object. This generates the collection of the ConfigClass objects for the individual entries. However, it does not generate the final objects yet. :param config: The configuration of the collection, usually taken from a JSON file.

init_object_collection()#: When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

meta_parameters: Dict[str, Dict[str, Any]] | None#

objects: Dict[str, T] | None#

class dqa.configuration.configuration.ConfigList(class_dict: Dict[str, ConfigClass], meta_parameter_names: List[str])#

A specific implementation of the ConfigCollection class where the entries are stored in a list. (For example Task lists).

Methods

`init_entry`(config)	Initializes a single entry in the collection from a JSON configuration.
`init_entry_collection`(config)	Initializes the collection from a config object.
`init_object_collection`()	When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

add_resolving_parameters
copy_resolving_parameters

copy_resolving_parameters(additional_parameters: Dict[str, Any])#

entry_classes: List[ConfigClass | None] | None#

entry_parameters: List[Dict[str, Any]] | None#

init_entry_collection(config: List[Dict])#: Initializes the collection from a config object. This generates the collection of the ConfigClass objects for the individual entries. However, it does not generate the final objects yet. :param config: The configuration of the collection, usually taken from a JSON file.

init_object_collection()#: When init_entry_collection() has been called, this method creates the final objects, i.e. it creates instances of the Python classes associated with the ConfigClass objects loaded in the configuration.

meta_parameters: List[Dict[str, Any]] | None#

objects: List[T] | None#

class dqa.configuration.configuration.ConfigSystem#

Representation of an entire DQA config.

The config contains different aspects (e.g. task lists and algorithms) that are represented by ConfigDict/ConfigList objects.

Attributes:

algorithms_classes
tasks_classes

Methods

load_algorithm_class
load_algorithm_classes
load_algorithms
load_classes
load_task_class
load_task_classes
load_task_lists
load_task_module
process_config
process_config_file
process_includes
register_parameter_objects
remove_comments

algorithms: Dict[str, Algorithm]#

algorithms_classes = None#

includes: Dict[str, Any]#

static load_algorithm_class(the_class: type, parameters: dict | None = None)#

static load_algorithm_classes(filename: str)#

load_algorithms(config_algos: Dict[str, Dict])#

static load_classes()#

static load_task_class(the_class: type, parameters: dict | None = None)#

static load_task_classes(filename: str)#

load_task_lists(config_task_lists: Dict[str, List[Dict[str, Any]]])#

static load_task_module(module, parameters: dict | None = None)#

model_export: Dict[str, Any] | None#

preprocessor: ConfigPreprocessor#

process_config(config: Dict[str, Any], config_dir: str, parameters: Dict[str, Any] | None = None)#

process_config_file(config_filename: str, config_dir: str | None = None, parameters: Dict[str, Any] | None = None)#

process_includes(includes_config: List[Dict[str, Any]], config_dir: str)#

register_parameter_objects(prefix: str, objects: Dict[str, Any])#

static remove_comments(config: Any) → Any#

task_lists: Dict[str, ConfigList[Task]]#

tasks_classes = None#

class dqa.configuration.configuration.NestedTask(task_list: ConfigList[Task], parameters: Dict[str, Any] | None = None, batch_processing: bool = False, **kwargs)#

A task that consists of a nested list of more tasks.

task_listConfigList[Task]: A loaded configuration of a task list to be executed inside the task.
parametersdict, default=None: A dictionary of parameters passed to the configuration of the task list.

Methods

`finish`()	Can perform actions that are required to clean up after the task has finished, e.g. close network connections etc.
`push_parameters_to_sub_tasks`(parameters)	Copies the specified parameters to all entries in self.task_list of the class NestedTask.

in_out_default
input_output_dataset
input_output_machine
input_output_mode
input_output_name
log
modify_data_row
modify_dataset
modify_dataset_dict
modify_machine
modify_measurement
set_logging_level
transfer_metadata

finish()#: Can perform actions that are required to clean up after the task has finished, e.g. close network connections etc.

modify_dataset_dict(datasets: Dict[str, Dataset]) → Dict[str, Dataset]#

push_parameters_to_sub_tasks(parameters: Dict[str, Any])#

Copies the specified parameters to all entries in self.task_list of the class NestedTask.

Parameters:: parameters – Dictionary of parameters.

set_logging_level(level)#

class dqa.configuration.configuration.PickledClassifier(filename: str)#

algorithm: Any#

class dqa.configuration.preprocessor.ConfigPreprocessor(parameters: Dict[str, Any] | None = None)#

Methods

load_standard_functions
register_function
resolve
resolve_and_check
resolve_list_and_check
resolve_string_and_check
set_variable

functions: Dict#

load_standard_functions()#

register_function(name: str, function_obj, unresolved_arguments: List[int] | None = None)#

resolve(input_config: Any) → Any#

resolve_and_check(input_config: ~typing.Any) -> (typing.Any, <class 'bool'>, <class 'bool'>)#

resolve_list_and_check(input_list: ~typing.List) -> (typing.Any, typing.List[bool], <class 'bool'>)#

resolve_string_and_check(input_str: str) -> (typing.Any, <class 'bool'>, <class 'bool'>)#

set_variable(name: str, value: Any)#

variables: Dict[str, Any]#

exception dqa.configuration.preprocessor.RemoveEntry#

dqa.configuration.preprocessor.preprocessor_add(pp, *args) → Any#

dqa.configuration.preprocessor.preprocessor_append(pp, base_list: List[Any], *args) → List[Any]#

dqa.configuration.preprocessor.preprocessor_datetime(pp, time_string: str, format: str) → datetime64#

dqa.configuration.preprocessor.preprocessor_for(pp, expression: Any, iteration_list: List[Any], variable: str = 'X') → List[str]#

dqa.configuration.preprocessor.preprocessor_if(pp, condition: bool, expression: Any) → Any#

dqa.configuration.preprocessor.preprocessor_range(pp, p1: int, p2: int | None = None, p3: int | None = None) → List[int]#

dqa.configuration.preprocessor.preprocessor_timestamp(pp, time_string: str, format: str) → float#

Configuration (dqa.configuration)#

Configuration (`dqa.configuration`)#