lir.transform package

class lir.transform.BinaryClassifierTransformer(estimator: SKLearnPipelineModule)[source]

Bases: Transformer

Implementation of a binary class classifier as scikit-learn Pipeline step.

Parameters:: estimator (SKLearnPipelineModule) – Estimator used to produce transformed or scored outputs.

apply(instances: InstanceData) → FeatureData[source]

Convert instances by applying the fitted model.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: FeatureData

fit(instances: InstanceData) → Self[source]

Fit the model on the provided instances.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: This transformer instance after fitting.
Return type:: Self

class lir.transform.DataWriter(*args, **kwargs)[source]

Bases: Protocol

Representation of a data writer and necessary methods.

writerow(row: Any) → None[source]

Write row to output.

Parameters:: row (Any) – CSV row dictionary to parse.

class lir.transform.FunctionTransformer(func: Callable)[source]

Bases: Transformer

Implementation of a transformer function as scikit-learn Pipeline step.

Parameters:: func (Callable) – Callable used to transform input instances.

apply(instances: InstanceData) → FeatureData[source]

Call the custom defined function on the feature data instances and use output as features.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: FeatureData

class lir.transform.Identity[source]

Bases: Transformer

Represent the Identity function of a transformer.

When apply() is called on such a transformer, it simply returns the instances.

apply(instances: InstanceDataType) → InstanceDataType[source]

Simply provide the instances.

Parameters:: instances (InstanceDataType) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceDataType

class lir.transform.SKLearnPipelineModule(*args, **kwargs)[source]

Bases: Protocol

Representation of the interface required for estimators by the scikit-learn Pipeline.

fit(X: ndarray, y: ndarray | None) → Self[source]

predict_proba(X: ndarray) → Any[source]

transform(X: ndarray) → Any[source]

class lir.transform.SklearnTransformer(transformer: SklearnTransformerType)[source]

Bases: Transformer

Implementation of a binary class classifier as scikit-learn Pipeline step.

Parameters:: transformer (SklearnTransformerType) – Transformer instance wrapped by this adapter.

apply(instances: InstanceData) → InstanceData[source]

Convert instances by applying the fitted model.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Fit the model on the provided instances.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: This transformer instance after fitting.
Return type:: Self

fit_apply(instances: InstanceData) → FeatureData[source]

Combine call to .fit() followed by .apply().

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: FeatureData

class lir.transform.SklearnTransformerType(*args, **kwargs)[source]

Bases: Protocol

Representation of the interface required for transformers by the scikit-learn Pipeline.

fit(features: ndarray, labels: ndarray | None) → Self[source]

fit_transform(features: ndarray, labels: ndarray | None) → ndarray[source]

transform(features: ndarray) → Any[source]

class lir.transform.Tee(transformers: list[Transformer])[source]

Bases: Transformer

Implementation of a custom transformer allowing to perform two separate tasks on a given input.

Parameters:: transformers (list[Transformer]) – Collection of transformers applied in sequence or parallel.

apply(instances: InstanceData) → InstanceData[source]

Delegate apply() to all specified transformers.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Delegate fit() to all specified transformers.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: This tee transformer instance after delegating fit.
Return type:: Self

class lir.transform.TransformerWrapper(wrapped_transformer: Transformer)[source]

Bases: Transformer

Base class for a transformer wrapper.

This class is derived from AdvancedTransformer and has a default implementation of all functions by forwarding the call to the wrapped transformer. A subclass may add or change functionality by overriding functions.

Parameters:: wrapped_transformer (Transformer) – Value passed via wrapped_transformer.

apply(instances: InstanceData) → InstanceData[source]

Delegate calls to underlying wrapped transformer but return the Wrapper instance.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Delegate calls to underlying wrapped transformer but return the Wrapper instance.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: This wrapper instance after delegating fit.
Return type:: Self

lir.transform.as_transformer(transformer_like: Any) → Transformer[source]

Provide a Transformer instance of the provided transformer like input.

For any transformer-like object, wrap if necessary, and return a Transformer.

The transformer-like object may be one of the following: - an instance of Transformer, which is returned as-is; - a scikit-learn style transformer which implements transform() and optionally fit() and/or fit_transform(); - a scikit-learn style estimator, which implements fit() and predict_proba(); or - a callable which takes an np.ndarray argument and returns another np.ndarray.

Parameters:: transformer_like (Any) – Object to convert to the internal Transformer interface.
Returns:: Equivalent object adapted to the internal Transformer interface.
Return type:: Transformer

Submodules

lir.transform.composite module

class lir.transform.composite.CategoricalCompositeTransformer(factory: Callable[[], Transformer], category_field: str)[source]

Bases: Transformer

Composite transformer.

Incoming data is categorized by a category field. For each category, a separate transformer is used.

Parameters:

factory (Callable[[], Transformer]) – Value passed via factory.
category_field (str) – Value passed via category_field.

apply(instances: InstanceData) → InstanceData[source]

Apply the specialized transformers for all instances in instances, based on their categories.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Fit the transformer for all categories found in instances.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: This composite transformer instance after fitting category models.
Return type:: Self

lir.transform.csv_writer module

class lir.transform.csv_writer.CsvWriter(path: Path | str, include_batch_number: bool = True, include_fields: list[str] | None = None, exclude_fields: list[str] | None = None)[source]

Bases: Transformer

Implementation of a transformation step in a Pipeline that writes to CSV.

This might be used to obtain temporary or intermediate results for logging or debugging purposes.

Parameters:

path (Path) – Filesystem path used by this operation.
include_batch_number (bool) – The batch number is the sequence number of the call to apply(). Iff include_batch, this value is included as a column in the CSV file (default: True).
include_fields (list[str] | None) – Fields to be included, or include all fields if not specified.
exclude_fields (list[str] | None) – Optional list of fields to be excluded.

apply(instances: DataType) → DataType[source]

Write numpy feature vector to CSV output file.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: InstanceData

fit_apply(instances: DataType) → DataType[source]

Provide required fit_apply() and return all instances.

Since the CsvWriter is implemented as a step (Transformer) in the pipeline, it should support the fit_apply method which is called on all transformers of the pipeline.

This fit_apply method is typically called during training, and output is supposed to be generated for the test set only, so no output is generated at this stage. We also don’t need to actually fit or transform anything, so we simply return the instances (as is).

Parameters:: instances (InstanceDataType) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceDataType

lir.transform.data_validator module

class lir.transform.data_validator.ValidateFeatureDataType[source]

Bases: Transformer

Module that validates the data types of the features in the instances.

This transformer is useful for ensuring that the data types of the features in the instances are consistent with the data types determined during fitting. This can help prevent errors during the application of a model/pipeline. Especially useful when the model or the applied data is read from file.

In short, it checks: 1. That the number of features in the instances matches the number of data types determined during fitting. 2. That the data types of the features can be cast to the data types determined during fitting.

apply(instances: InstanceData) → FeatureData[source]

Apply the transformer to the data by enforcing the data types of each feature.

Will raise an error if the data types of the features in the instances do not match the data types determined during fitting.

Parameters:

instances (InstanceData) – The data to apply the transformer on.

Returns:

The transformed data with validated data types.

Return type:

FeatureData

Raises:

ValueError – If the number of features in the instances does not match the number of data types determined during fitting
TypeError – If the data types of the features in the instances cannot be cast to the data type determined during fitting

fit(instances: InstanceData) → Self[source]

Fit the transformer to the data by determining the data types of each feature in the first instance.

It assumes that all instances have the same data types for each feature.

Parameters:: instances (InstanceData) – The data to fit the transformer on.
Returns:: The fitted transformer.
Return type:: Self

lir.transform.distance module

class lir.transform.distance.ElementWiseDifference[source]

Bases: Transformer

Calculate the element-wise absolute difference between pairs.

Takes an array of sample pairs and returns the element-wise absolute difference.

Expects: - a PairedFeatureData object with n_trace_instances=1 and n_ref_instances=1;

Return type:

a copy of the FeatureData object with features of shape (n, f)

apply(instances: InstanceData) → FeatureData[source]

Calculate the absolute difference between all elements in the instance data (pairs).

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: FeatureData

class lir.transform.distance.EuclideanDistance[source]

Bases: Transformer

Calculate the Euclidean distance between pairs.

Takes a PairedFeatureData object or a FeatureData object and returns the euclidean distance.

If the input is a PairedFeatureData object, the distance is computed as the euclidean distance, i.e. the square root of the sum of the squared element-wise difference between both sides of the pairs, for all features.

If the input is a FeatureData object, it is assumed that it contains the element-wise differences, and the square root of the sum over these differences is calculated.

In yaml configurations, it can be used by specifying euclidean_distance, e.g.: scoring: euclidean_distance

apply(instances: InstanceData) → FeatureData[source]

Calculate the Euclidean distance between all elements in the instance data (pairs).

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: FeatureData

class lir.transform.distance.ManhattanDistance[source]

Bases: Transformer

Calculate the Manhattan distance between pairs.

Takes a PairedFeatureData object or a FeatureData object and returns the manhattan distance.

If the input is a PairedFeatureData object, the distance is computed as the manhattan distance, i.e. the sum of the element-wise difference between both sides of the pairs, for all features.

If the input is a FeatureData object, it is assumed that it contains the element-wise differences, and the sum over these differences is calculated.

apply(instances: InstanceData) → FeatureData[source]

Calculate the Manhattan distance between all elements in the instance data (pairs).

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: FeatureData object parsed from the source.
Return type:: FeatureData

lir.transform.pairing module

class lir.transform.pairing.InstancePairing(same_source_limit: int | None = None, different_source_limit: int | None = None, ratio_limit: float | None = None, seed: int | None = None)[source]

Bases: PairingMethod

Construct pairs from a set of instances.

Note that this pairing method may cause performance problems with large datasets, even if the number of instances in the output is limited.

The ratio is ds pairs / ss pairs. The number of ds pairs will not exceed ratio_limit * ss pairs. If both ratio_limit and same_source_limit/different_source_limit are specified, the number of pairs is chosen such that the ratio_limit is preserved and the limit(s) are not exceeded, while taking as many pairs as possible within these constraints.

Parameters:

same_source_limit (int | None) – Limit for the number or fraction of same-source pairs.
different_source_limit (int | None) – Limit for the number or fraction of different-source pairs.
ratio_limit (float | None) – Maximum allowed ratio between same-source and different-source pairs.
seed (int | None) – Random seed controlling stochastic behaviour for reproducible results.

pair(instances: InstanceData, n_trace_instances: int = 1, n_ref_instances: int = 1) → PairedFeatureData[source]

Construct pairs.

Parameters:

instances (InstanceData) – Input instances to be processed by this method.
n_trace_instances (int) – Number of trace instances to include in each pairing.
n_ref_instances (int) – Number of reference instances to include in each pairing.

Returns:

FeatureData object parsed from the source.

Return type:

PairedFeatureData

property rng: Generator

Obtain a random number generator using a provided seed.

Returns:: Random number generator initialized from the configured seed.
Return type:: np.random.Generator

class lir.transform.pairing.PairingMethod[source]

Bases: ABC

Base class for pairing methods.

A pairing method should implement the pair() function.

abstractmethod pair(instances: InstanceData, n_trace_instances: int = 1, n_ref_instances: int = 1) → PairedFeatureData[source]

Take instances as input, and return pairs.

A pair may be a pair of sources, with multiple instances per source.

The returned features have dimensions (p, i, …)` where the first dimension is the pairs, the second dimension is the instances, and subsequent dimensions are the features. If the input has labels, the returned labels are an array of source labels, one label per pair, where the labels are 0=different source, 1=same source. Any other attributes are combined into tuples.

Parameters:

instances (InstanceData) – Input instances to be processed by this method.
n_trace_instances (int) – Number of trace instances to include in each pairing.
n_ref_instances (int) – Number of reference instances to include in each pairing.

Returns:

FeatureData object parsed from the source.

Return type:

PairedFeatureData

class lir.transform.pairing.SourcePairing(same_source_limit: int | None = None, different_source_limit: int | None = None, ratio_limit: int | None = None, seed: Any | int = None)[source]

Bases: PairingMethod

Construct pairs of sources (i.e. classes) from an array of instances.

While pairing at instance level results in pairs of instances, some same-source and some different-source, pairing at source level results in pairing of multiple instances of source A against multiple instances of source B, where A and B can be same-source or different-source.

Parameters:

same_source_limit (int | None) – Limit for the number or fraction of same-source pairs.
different_source_limit (int | None) – Limit for the number or fraction of different-source pairs.
ratio_limit (int | None) – Maximum allowed ratio between same-source and different-source pairs.
seed (Any | int) – Random seed controlling stochastic behaviour for reproducible results.

pair(instances: InstanceData, n_trace_instances: int = 1, n_ref_instances: int = 1) → PairedFeatureData[source]

Pair sources.

Takes a FeatureData object that contains instances. Returns pairs as a PairedFeatureData object.

The input is expected to have source_ids, that govern how pairs are compiled. The input instances may be used in a pair, either as a trace instance or as a reference instance.

Parameters:

instances (InstanceData) – Input instances to be processed by this method.
n_trace_instances (int) – Number of trace instances to include in each pairing.
n_ref_instances (int) – Number of reference instances to include in each pairing.

Returns:

FeatureData object parsed from the source.

Return type:

PairedFeatureData

lir.transform.pipeline module

class lir.transform.pipeline.LoggingPipeline(steps: list[tuple[str, Transformer | Any]], output_file: PathLike, include_batch_number: bool = True, include_labels: bool = True, include_fields: list[str] | None = None, include_steps: list[str] | None = None, include_input: bool = True)[source]

Bases: Pipeline

A pipeline that writes debugging output to a CSV file.

This pipeline act like a normal Pipeline, but has a CSV file as a side effect. Depending on the settings and the data, the CSV file may have the following columns:

batch: if the data strategy yields multiple train/test splits, the batch value is the sequence number of the test set
label: the hypothesis label

In addition, there may be columns for the input features and output of individual steps. These columns are named featuresI for input features or stepnameI for step output, where stepname is replaced by the name of the step, and I refers to the index of the feature value.

Parameters:

steps (list[tuple[str, Transformer | Any]]) – Ordered transformer steps executed by this pipeline.
output_file (PathLike) – Destination file used to log intermediate pipeline output.
include_batch_number (bool) – Whether to include the batch number in logged output.
include_labels (bool) – Whether to include labels in logged output.
include_fields (list[str] | None) – Additional instance fields to include in logged output.
include_steps (list[str] | None) – Whether to include step names in logged output.
include_input (bool) – Whether to include original inputs in logged output.

apply(instances: InstanceData) → InstanceData[source]

Apply the pipeline to the incoming instances.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

class lir.transform.pipeline.Pipeline(steps: list[tuple[str, Transformer | Any]])[source]

Bases: Transformer

A pipeline of processing modules.

Each step in the pipeline may be a

a scikit-learn style transformer (with fit() and transform() functions),
a scikit-learn style estimator (with fit() and predict_proba()), or
a LiR Transformer object.

Example:

from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from lir.transform.pipeline import Pipeline
from lir.algorithms.logistic_regression import LogitCalibrator
from lir.util import probability_to_odds

pipeline = Pipeline(steps=[
        ('scaler', StandardScaler()),       # a scikit-learn transformer, for scaling the data
        ('clf', RandomForestClassifier()),  # a scikit-learn estimator, to calculate pseudo-probabilities
        ('to_odds', probability_to_odds),   # a plain function, to convert probabilities to pseudo-LLRs
        ('calibrator', LogitCalibrator()),  # a LiR transformer, to calibrate the LLRs
])

Parameters:: steps (list[tuple[str, Transformer | Any]]) – Ordered transformer steps executed by this pipeline.

apply(instances: InstanceData) → InstanceData[source]

Apply the fitted model on the instance data.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Fit the model on the instance data.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: The fitted pipeline instance.
Return type:: Self

fit_apply(instances: InstanceData) → InstanceData[source]

Combine fitting the transformer/estimator and applying the model to the instances.

Parameters:: instances (InstanceData) – Input instances to be processed by this method.
Returns:: Instance data object produced by this operation.
Return type:: InstanceData

lir.transform.pipeline.logging_pipeline: alias of ConfigParserFunction

lir.transform.save_features module

class lir.transform.save_features.SaveFeatureTransformer(save_as: str)[source]

Bases: Transformer

Transformer to save the features of the instances in a new field.

The idea of this transformer is to capture the features of the instances at a specific point in the pipeline, and save them in a new field in the instance data. This can be useful for later analysis, plotting, or debugging.

Parameters:: save_as (str) – The name of the field to save the features in.

apply(instances: InstanceData) → InstanceData[source]

Save the features of the instances in a new field, and return the instances.

Parameters:: instances (InstanceData) – The instances to transform.
Returns:: The instances with the saved features.
Return type:: InstanceData

fit(instances: InstanceData) → Self[source]

Fit the transformer to the instances. This transformer does not require fitting, so this method returns self.

Parameters:: instances (InstanceData) – The instances to fit.
Returns:: The fitted transformer.
Return type:: SaveFeatureTransformer

lir.transform.select_instances module

class lir.transform.select_instances.SelectInstances(select_element_fn: Callable[[int], bool])[source]

Bases: Transformer

Select elements in a dataset from their indices.

Parameters:: select_element_fn (Callable[[int], bool]) – A function that takes a line number and returns True if it should be included, or False if it should be discarded.

Examples

This filter can be used in a YAML configuration:

data:
  provider: [...]
  strategy: [...]
  filter:
    method: select_instances # drop instances unless their indices matches any of the following patterns
    indices:
      - 2                    # select element 2 -- the third record
      - 0-99                 # select element 0-99 (inclusive) -- the first 100 records
      - 0-99,800-899         # select element 0-99 and 800-899
      - /5                   # select every fifth element: 0, 5, 10, ...

apply(instances: DataType) → DataType[source]

Apply the selection to a dataset.

Parameters:: instances (InstanceData) – The dataset to select instances from.
Returns:: A dataset with only the selected instances.
Return type:: InstanceData