tpcp.misc.BaseTypedIterator#

class tpcp.misc.BaseTypedIterator(data_type: type[DataclassT], aggregations: Sequence[tuple[str, Callable[[list[TypedIteratorResultTuple[InputTypeT, DataclassT]]], Any]]] = cf([]))[source]#

A Base class to implement custom typed iterators.

This class is missing the iterate method, which needs to be implemented by the child class. It has a _iterate method though that does most of the heavy lifting. The actual iterate method should handle turning the inputs into an iterable and then call _iterate.

TypedIterator provides a “dummy” implementation that expects any type of iterator for the iterate method. Custom base classes could provide more elaborate preprocessing of the inputs before iteration. For example, cutting sections out of a dataframe based on a list of start and end indices, and then iterating over the cut sections.

Parameters:

data_type: A dataclass that defines the result type you expect from each iteration.
aggregations: An optional list of aggregations to apply to the results. This has the form [(result_name, aggregation_function), ...]. Each aggregation function gets raw_results_ provided as input and can return an arbitrary object. Note, that the aggregation function needs to handle the case where no result was set for a specific attribute. If a result-name is in the list, the aggregation will be applied to it, when accessing the results_ (i.e. results_.{result_name}). If no aggregation is defined for a result, a simple list of all results will be returned. .. note:: It is possible to define aggregations with names not present in the result dataclass.

This allows to provide multiple aggregations from the same data. The results can still be accessed via the additional_results_ attribute.
NULL_VALUE: (Class attribute) The value that is used to initialize the result dataclass and will remain in the results, if no result was for a specific attribute in one or more iterations.
IteratorResult: (Class attribute) Type alias for the result-type of the iterator. raw_results_ will be a list of these. Note, that when using this outside of the class, this type will be a generic without a type for the input and result field.

Attributes:

results_: The aggregated results.
additional_results_: A dictionary with additional results that were created by aggregators with names not present in the result dataclass.
raw_results_: List of all results as TypedIteratorResultTuple instances. This is the input to the aggregation functions. The attribute of the result dataclass instance will have the value of _NOT_SET if no result was set. To check for this, you can use isinstance(val, BaseTypedIterator.NULL_VALUE) or the BaseTypedIterator.filter_iterator_results method to remove all results with a NULL_VALUE.
done_: A dictionary indicating of a specific iterator is done. This usually only has the key __main__ for the main iteration triggered by iterate. However, subclasses can define nested iterations with more complex logic. The value will be True if the respective iteration is done, False if it is currently running and missing if it was never started. If the main iterator is not done, but you try to access the results, an error will be raised.

Notes

Under the hood, the iterator supports having multiple iterations at the same time by providing an iteration_name to the _iterate method. This information is stored in the results and can be used to separate the results of different iterations. Together with the iteration_context parameter, this allows for more complex iteration structures. One example, would be the use of nested iterations, that are aware of the parent iteration.

In the mobilise-d/mobgap library this is used to support the iteration of multiple levels of interests within data. For example, the outer level iterates Gait-Tests, the inner level iterates gait sequences within each gait test that are dynamically detected within the outer iteration. The iterator can then still patch all results from the inner iteration together to provide a single result object with times that are relative to the start of the entire recording.

Methods

`clone`()	Create a new instance of the class with all parameters copied over.
`get_params`([deep])	Get parameters for this algorithm.
`set_params`(**params)	Set the parameters of this Algorithm.

IteratorResult
filter_iterator_results

clone() → Self[source]#

Create a new instance of the class with all parameters copied over.

This will create a new instance of the class itself and all nested objects

get_params(deep: bool = True) → dict[str, Any][source]#

Get parameters for this algorithm.

Parameters:

deep: Only relevant if object contains nested algorithm objects. If this is the case and deep is True, the params of these nested objects are included in the output using a prefix like nested_object_name__ (Note the two “_” at the end)

Returns:

params: Parameter names mapped to their values.

property results_: DataclassT#

The aggregated results.

Note, that this returns an instance of the result object, even-though the datatypes of the attributes might be different depending on the aggregation function. We still decided it makes sense to return an instance of the result object, as it will allow to autocomplete the attributes, even-though the associated times might not be correct.

set_params(**params: Any) → Self[source]#

Set the parameters of this Algorithm.

To set parameters of nested objects use nested_object_name__para_name=.