dispel.processing.level module#

Level processing functionalities.

class dispel.processing.level.DefaultLevelFilter[source]#

Bases: LevelFilter

A default level filter that considers all levels.

filter(levels)[source]#

Filter method returns a set of levels.

Parameters:: levels (Iterable[Level]) –
Return type:: Set[Level]

repr()[source]#

Get representation of the filter.

Return type:: str

class dispel.processing.level.FlagLevelStep[source]#

Bases: FlagStepMixin, LevelProcessingStep

A level flag class.

Parameters:

level_filter (dispel.processing.level.LevelFilter) – An optional filter to limit the levels being processed. See LevelProcessingStep.
task_name (dispel.data.values.AbbreviatedValue | str) – An optional abbreviated name value of the task used for the flag. See FlagStepMixin.
flag_name (dispel.data.values.AbbreviatedValue | str) – An optional abbreviated name value of the considered flag. See FlagStepMixin.
flag_type (dispel.data.flags.FlagType | str) – An optional flag type. See FlagType.
flag_severity (dispel.data.flags.FlagSeverity | str) – An optional flag severity. See FlagSeverity.
reason (str) – An optional string reason of the considered flag. See FlagStepMixin.
stop_processing (bool) – An optional boolean that specifies whether the flag is stop_processing, i.e., raises an error or not. See FlagStepMixin.
flagging_function (Callable[[...], bool] | None) – An optional flagging function to be applied to Level. See FlagStepMixin.

Examples

Assuming you want to flag the time frame duration of a given level, you can use the following flag step:

>>> from dispel.data.values import AbbreviatedValue as AV
>>> from dispel.processing.level import FlagLevelStep
>>> step = FlagLevelStep(
...     'utt',
...     AV('U-Turn test', 'utt'),
...     AV('u-turn duration', 'utt_dur'),
...     'technical',
...     'The U-Turn test duration exceeds 5 minutes.',
...     stop_processing=False,
...     flagging_function=lambda level:                 level.effective_time_frame.duration.total_seconds() < 5 * 60,
... )

The flagging function will be called with the level 'utt' as specified in the level_filter argument. If the function has a named parameter matching reading, the reading will be passed to the flagging function.

Another common scenario is to define a class that can be reused.

>>> from dispel.data.flags import FlagType
>>> from dispel.processing.level import FlagLevelStep
>>> class UTTDuration(FlagLevelStep):
...     task_name = AV('U-Turn test', 'utt')
...     flag_name = AV('u-turn duration', 'utt_dur')
...     flag_type = FlagType.TECHNICAL
...     flag_severity = FlagSeverity.DEVIATION
...     reason = 'The U-Turn test duration exceeds 5 minutes.'
...     stop_processing = False
...     flagging_function = lambda level:                 level.effective_time_frame.duration.total_seconds() < 5 * 60

Another convenient way to provide the flagging function is to use the @flag decorator, one can also use multiple flags for the same class as follows:

>>> from dispel.data.levels import Level
>>> from dispel.data.core import  Reading
>>> from dispel.processing.flags import flag
>>> from dispel.processing.level import FlagLevelStep
>>> class UTTDuration(FlagLevelStep):
...     task_name = AV('U-Turn test', 'utt')
...     flag_name = AV('u-turn duration', 'utt_dur')
...     flag_type = 'technical'
...     flag_severity = FlagSeverity.DEVIATION
...     reason = 'The U-Turn test duration exceeds {duration} minutes.'
...     stop_processing = False
...
...     @flag(duration=5)
...     def _duration_5(self, level: Level) -> bool:
...         duration = level.duration
...         return duration.total_seconds() < 5 * 60
...
...     @flag(duration=4)
...     def _duration_4(self, level: Level, reading: Reading) -> bool:
...         evaluation_start = reading.evaluation.start
...         level_start = level.start
...         assert evaluation_start <= level_start
...
...         duration = level.duration
...         return duration.total_seconds() < 4 * 60

Note that the @flag decorator can take keyword arguments. These kwargs are merged with any keyword arguments that come from processing step groups in order to format the flag reason.

__init__(level_filter=None, task_name=None, flag_name=None, flag_type=None, flag_severity=None, reason=None, stop_processing=False, flagging_function=None)[source]#

Parameters:

level_filter (str | LevelId | List[str] | List[LevelId] | LevelFilter | None) –
task_name (str | AbbreviatedValue | None) –
flag_name (str | AbbreviatedValue | None) –
flag_type (FlagType | str | None) –
flag_severity (FlagSeverity | str | None) –
reason (str | AbbreviatedValue | None) –
stop_processing (bool) –
flagging_function (Callable[[...], bool] | None) –

flag_level(level, reading, **kwargs)[source]#

Flag the provided level.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Generator[Flag, None, None]

get_flag_targets(reading, level=None, **kwargs)[source]#

Get flag targets for level flag.

Parameters:

reading (Reading) –
level (Level | None) –

Return type:

Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch]

get_level_flag_targets(level, reading, **kwargs)[source]#

Get flag targets for reading flag.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch]

process_level(level, reading, **kwargs)[source]#

Process the provided level.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

class dispel.processing.level.LevelFilter[source]#

Bases: ABC

A base class to filter levels during processing.

LevelFilter provide a central mechanism to differentiate processing steps in combination with LevelProcessingStep and ConcatenateLevels. Each filter implementation must provide a filter() function that consumes a container of levels and returns a set of levels containing those that should be processed. Furthermore, the method repr() provides a hook to create a readable representation of the filter.

Filters can also be combined by using logical operators & and |.

Examples

Each level filter has to implement the methods filter and repr. Assuming we want to create a filter that inspects some variables in the context of each level, we can do the following:

>>> from typing import Any, Iterable, Set
>>> from dispel.data.levels import Level
>>> from dispel.processing.level import LevelFilter
>>> class MyContextLevelFilter(LevelFilter):
...     def __init__(self, variable: str, value: Any):
...         self.variable = variable
...         self.value = value
...     def filter(self, levels: Iterable[Level]) -> Set[Level]:
...         return set(filter(
...             lambda level: level.context.get_raw_value(
...                 self.variable) == self.value, levels))
...     def repr(self) -> str:
...         return f'{self.variable} equals "{self.value}"'

Given this filter one can now process levels with a specific context value by creating a filter like MyContextLevelFilter('usedHand', 'right').

Since LevelFilter can be used with logical operators one can create more complex filters by simply combining them as such:

>>> MyContextLevelFilter('foo', 'bar') & MyContextLevelFilter('baz', 'bam')
<LevelFilter: (foo equals "bar" and baz equals "bam")>

This filter will now only consider levels where the context variables foo equal bar and baz equals bam. This also works with or logic (|).

One can also use the inverse logic by applying the ~ operator to a level filter in order to obtain its inverse.

filter(levels)[source]#

Filter level.

Parameters:: levels (Iterable[Level]) – The levels to be inspected for filtering
Raises:: NotImplementedError – This method is not implemented since there is no unambiguous definition of filters.
Return type:: Set[Level]

repr()[source]#

Get representation of the filter.

Raises:: NotImplementedError – This method is not implemented since there is no unambiguous representation of filters.
Return type:: str

class dispel.processing.level.LevelFilterProcessingStepMixin[source]#

Bases: object

A mixin class for all processing steps using level filters.

Parameters:: level_filter (dispel.processing.level.LevelFilter) – The filter to be used to select the levels to be processed.

__init__(*args, **kwargs)[source]#

get_level_filter()[source]#

Get the level filter to sub-select levels to be processed.

Return type:: LevelFilter

inject_level_filter_from_step(step)[source]#

Inject the level filter from a step into the filter of this step.

This function allows to have this processing step depend on the level filter of another step.

Parameters:: step (LevelFilterProcessingStepMixin) – A level filter processing step of which the level filter is used in this step too.

level_filter: LevelFilter = <DefaultLevelFilter: *>#

Parameters:: levels (Iterable[Level]) –
Return type:: Set[Level]

set_level_filter(level_filter)[source]#

Set a level filter to sub-select levels to be processed.

Parameters:: level_filter (str | LevelId | List[str] | List[LevelId] | LevelFilter) – The level filter to be used.

class dispel.processing.level.LevelIdFilter[source]#

Bases: LevelFilter

A level filter based on level ids.

Parameters:: level_ids – The level id(s) to be filtered for processing. The level id can be provided as str, LevelId or lists of either. Levels with the respective provided ids will be considered during processing.

__init__(level_ids)[source]#

Parameters:: level_ids (str | LevelId | List[str] | List[LevelId]) –

filter(levels)[source]#

Filter levels being part of a predefined level id set.

Parameters:: levels (Iterable[Level]) –
Return type:: Set[Level]

repr()[source]#

Get representation of the filter.

Return type:: str

class dispel.processing.level.LevelProcessingControlResult[source]#

Bases: ProcessingControlResult, LevelProcessingResultBase

The processing result of an error in a level.

__init__(level, step, error, error_handling, targets=None)#

Parameters:

level (Level) –
step (ProcessingStep) –
error (ProcessingError | StopProcessingError | FlagError | InvalidDataError) –
error_handling (ErrorHandling) –
targets (Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch | None) –

Return type:

None

classmethod from_assertion_error(step, error, level=None)[source]#

Initialize object from a caught assertion error.

Parameters:

step (ProcessingStep) – The processing step from where the assertion error has been caught.
error (AssertionError) – The assertion error that has been caught.
level (Level | None) – The level corresponding to the LevelProcessingControlResult.

Returns:

The initialized level processing control result object.

Return type:

LevelProcessingControlResult

classmethod from_flag(flag, step, targets, level=None)[source]#

Initialize processing control result from an flag.

Parameters:

flag (Flag) – The flag from which the control processing result is to be created.
step (ProcessingStep) – The associated processing step.
targets (Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch]) – The flag target entities.
level (Level | None) – The level corresponding to the LevelProcessingControlResult.

Returns:

The initialized level processing control result object.

Return type:

LevelProcessingControlResult

class dispel.processing.level.LevelProcessingResult[source]#

Bases: ProcessingResult, LevelProcessingResultBase

The processing result of a level processing step.

__init__(level, step, sources, result)#

Parameters:

level (Level) –
step (ProcessingStep) –
sources (Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch) –
result (Level | MeasureValue | MeasureSet | LevelEpoch | RawDataSet) –

Return type:

None

class dispel.processing.level.LevelProcessingResultBase[source]#

Bases: object

The processing result base of a level processing step.

__init__(level)#

Parameters:: level (Level) –
Return type:: None

level: Level#: The level where the processing result is stored.

class dispel.processing.level.LevelProcessingStep[source]#

Bases: LevelProcessingStepProtocol, LevelFilterProcessingStepMixin, ProcessingStep

A level processing step is a processing step specific on levels.

The level processing steps’ LevelProcessingStepProtocol.process_level() method is called with the level, reading and additional arguments passed to process(). Results from the process step are expected to be an instance of ProcessingResult.

The process_level() is only called with levels that pass the provided level_filter (see LevelFilter). Each level will be processed if no level filter is provided. The level_filter also accepts str, LevelIds and lists of either and passes them to a LevelIdFilter for convenience.

Examples

processing-step

>>> from dispel.data.measures import MeasureValue
>>> from dispel.data.values import ValueDefinition
>>> from dispel.processing import process
>>> from dispel.processing.level import (LevelProcessingStep,
...                                   LevelProcessingResult)
>>> class MyLevelStep(LevelProcessingStep):
...     def process_level(self, level, reading, **kwargs):
...         raw_data_set = level.get_raw_data_set('my-data-set')
...         yield LevelProcessingResult(
...             step=self,
...             sources=raw_data_set,
...             level=level,
...             result=MeasureValue(
...                 ValueDefinition('my-measure-id', 'max value'),
...                 raw_data_set.data.max().max()
...             )
...         )
>>> _ = process(reading, MyLevelStep())
>>> reading.get_measure_set('my-level').get_raw_value('my-measure-id')
5

assert_valid_level(level, reading, **kwargs)[source]#

Perform assertions that a given level can be processed.

Parameters:

level (Level) –
reading (Reading) –

flag_level(level, reading, **kwargs)[source]#

Flag the provided level.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Generator[Flag, None, None]

abstract process_level(level, reading, **kwargs)[source]#

Process the provided Level.

Parameters:

level (Level) – The level to be processed
reading (Reading) – The reading to be processed
kwargs – Additional arguments passed by process_level().

Yields:

ProcessResultType – Results from processing the level.

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

process_reading(reading, **kwargs)[source]#

Process all levels in reading.

Parameters:

reading (Reading) –

The reading to be processed. Each level of the reading will be passed to the
level_filter and if it returns True process_level() is called.
kwargs – Additional named arguments. This allows to provide additional values, such as placeholder values in value definitions to the actual processing function.

Yields:

ProcessResultType – Passes through anything that is yielded from process_level().

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

class dispel.processing.level.LevelProcessingStepProtocol[source]#

Bases: object

Protocol for level processing steps.

abstract assert_valid_level(level, reading, **kwargs)[source]#

Perform assertions that a given level can be processed.

Parameters:

level (Level) – The level to be tested for validity
reading (Reading) – The parent reading of the level
kwargs – Additional keyword arguments passed to the processing function.

abstract flag_level(level, reading, **kwargs)[source]#

Flag the provided level.

Parameters:

level (Level) – The level to be flagged.
reading (Reading) – The reading associated to the provided level.
kwargs – Additional arguments passed by process().

Yields:

Flag – The resulted flags.

Return type:

Generator[Flag, None, None]

get_level_flag_targets(level, reading, **kwargs)[source]#

Get the level flag targets.

Parameters:

level (Level) – The level to be flagged.
reading (Reading) – The reading associated with the level flag.
kwargs – Additional keyword arguments eventually used for flag targets extraction.

Returns:

An iterable of entities that are flagged.

Return type:

Iterable[EntityType]

abstract process_level(level, reading, **kwargs)[source]#

Process the provided Level.

Parameters:

level (Level) – The level to be processed
reading (Reading) – The reading to be processed
kwargs – Additional arguments passed by process_level().

Yields:

ProcessResultType – Results from processing the level.

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

class dispel.processing.level.ProcessingStepGroup[source]#

Bases: LevelFilterProcessingStepMixin, CoreProcessingStepGroup

A group of processing steps with an optional level filter.

For examples see dispel.processing.core.CoreProcessingStepGroup. This class ensures that level filters are injected to the steps of this group.

inject_level_filter_from_step(step)[source]#

Inject level filter into group and steps in group.

Parameters:: step (LevelFilterProcessingStepMixin) –

set_steps(steps)[source]#

Set processing steps part of the group.

This method ensures that steps added to the group inherit the level filter of the group.

Parameters:: steps (List[ProcessingStep]) – The steps contained in the processing group.