dispel.processing.core module#

Core functionality to process Readings.

class dispel.processing.core.CoreProcessingStepGroup[source]#

Bases: ProcessingStep

A group of processing steps.

The CoreProcessingStepGroup allows to provide additional named arguments to the process() function of the grouped ProcessingSteps. The primary use case for this is to provide additional arguments to ExtractStep that use ValueDefinitionPrototypes.

Parameters:

steps (List[dispel.processing.core.ProcessingStep]) – The processing steps of the group
kwargs (Dict[str, Any]) – Additional arguments that are passed to the process() function of each step. This allows to provide additional values, such as placeholder values in value definitions to the actual processing function.

Examples

>>> from dispel.data.values import ValueDefinitionPrototype
>>> from dispel.processing.core import CoreProcessingStepGroup
>>> from dispel.processing.extract import ExtractStep
>>> steps = [
...     CoreProcessingStepGroup(
...         [
...             ExtractStep(
...                 'data-set',
...                 lambda data_set: 5,
...                 ValueDefinitionPrototype(
...                     id_='x',
...                     measure_name='{placeholder} x'
...                 )
...             ),
...         ],
...         placeholder='A name'
...     ),
...     ...
... ]

The above extract step will result in a measure value with the name A name x. For further applications of CoreProcessingStepGroup see also Measure extraction.

__init__(steps=None, **kwargs)[source]#

Parameters:: steps (List[ProcessingStep] | None) –

static __new__(cls, *args, **kwargs)[source]#: Instantiate a new ProcessingStepGroup.

get_kwargs()[source]#

Get keyword arguments to be added to the processing.

Return type:: Dict[str, Any]

get_steps()[source]#

Get processing steps within this group.

Return type:: List[ProcessingStep]

kwargs: Dict[str, Any] = {}#

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

See dispel.processing.core.ProcessingStep.process().

Parameters:: reading (Reading) –
Return type:: Generator[ProcessingResult | ProcessingControlResult, None, None]

set_kwargs(**kwargs)[source]#: Set the keyword arguments to be added to the processing.

set_steps(steps)[source]#

Set processing steps part of the group.

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

steps: List[ProcessingStep]#

class dispel.processing.core.ErrorHandling[source]#

Bases: Enum

Different ways of dealing with an exception.

IGNORE = 'ignore'#

RAISE = 'raise'#

classmethod from_bool(stop_processing)[source]#

Create the error handling class from a stop_processing boolean.

Parameters:: stop_processing (bool) –
Return type:: ErrorHandling

property should_raise: bool#: Return True if the handling is to overwrite.

exception dispel.processing.core.FlagError[source]#

Bases: ProcessingError

An error that flags the processing results from a reading.

__init__(flag, step)[source]#

Parameters:

flag (Flag) –
step (ProcessingStep) –

class dispel.processing.core.FlagReadingStep[source]#

Bases: FlagStepMixin, ProcessingStep

A reading flag class.

Parameters:

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.
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 applied to Reading. See FlagStepMixin.

Examples

Assuming you want to flag reading information such as whether the user has finished the evaluation properly, you can create the following flag step:

>>> from dispel.data.values import AbbreviatedValue as AV
>>> from dispel.processing.core import FlagReadingStep
>>> step = FlagReadingStep(
...     task_name = AV('Pinch test', 'pinch'),
...     flag_name = AV('unfinished evaluation', 'ua'),
...     reason = 'The evaluation has not been finished by the user.',
...     stop_processing=False,
...     flag_type=FlagType.TECHNICAL,
...     flag_severity=FlagSeverity.INVALIDATION,
...     flagging_function=lambda reading: reading.evaluation.finished,
... )

The flagging function will be called with the corresponding reading as argument.

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

>>> from dispel.data.flags import FlagType
>>> from dispel.processing.core import FlagReadingStep
>>> class UnfinishedEvaluation(FlagReadingStep):
...     task_name = AV('Pinch test', 'pinch')
...     flag_name = AV('unfinished evaluation', 'ua')
...     flag_type = FlagType.BEHAVIORAL
...     flag_severity = FlagSeverity.INVALIDATION
...     reason = 'The evaluation has not been finished by the user.'
...     stop_processing = False
...     flagging_function = lambda reading: reading.evaluation.finished

Another convenient way to provide the flagging function is to use the @flag decorator:

>>> from dispel.data.core import Reading
>>> from dispel.processing.core import FlagReadingStep
>>> from dispel.processing.flags import flag
>>> class UnfinishedEvaluation(FlagReadingStep):
...     task_name = AV('Pinch test', 'pinch')
...     flag_name = AV('unfinished evaluation', 'ua')
...     flag_type = 'behavioral'
...     flag_severity = FlagSeverity.INVALIDATION
...     reason = 'The evaluation has not been finished by the user.'
...     stop_processing = False
...
...     @flag
...     def _unfinished_evaluation(self, reading: Reading) -> bool:
...         return reading.evaluation.finished

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. Also, one can use multiple flag decorators in the same flag class.

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

Parameters:

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_reading(reading, **kwargs)[source]#

Flag the provided reading.

Parameters:: reading (Reading) –
Return type:: Generator[Flag, None, None]

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

Get flag targets for reading flag.

Parameters:

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

Return type:

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

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

Get flag targets for reading flag.

Parameters:: reading (Reading) –
Return type:: Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch]

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

Process the provided reading.

Parameters:: reading (Reading) –
Return type:: Generator[ProcessingResult | ProcessingControlResult, None, None]

exception dispel.processing.core.InvalidDataError[source]#

Bases: ProcessingError

An error that flags the processing input.

class dispel.processing.core.Parameter[source]#

Bases: Generic[ParameterType]

A parameter defining aspects of processing.

Parameters:

id – The name of the parameter used to identify the configurable entity.
default_value – A default value to be used if the user does not specify otherwise.
validator – A callable that accepts values and raises an exception for any unexpected value.
description – A description of the parameter explaining its usage and influence on the affected processing steps.

__init__(id_, default_value=None, validator=None, description=None)[source]#

Parameters:

id_ (str) –
default_value (ParameterType | None) –
validator (Callable[[Any], None] | None) –
description (str | None) –

static __new__(cls, id_, *_args, **_kwargs)[source]#

Create a new Parameter object.

Parameters:: id_ (str) –

classmethod has_parameter(full_id)[source]#

Check if a parameter was set.

Parameters:: full_id (str) –
Return type:: bool

property id#

Get the ID of the parameter.

The id will be set automatically based on the context of the creation of the parameter.

Returns:: The ID of the parameter. It is comprised by the name of the frame in which it was defined and the specified id_.
Return type:: str

classmethod set_value(full_id, value)[source]#

Set the value of a parameter.

Parameters:

full_id (str) –
value (Any) –

property value: ParameterType#: Get the value set for the parameter.

class dispel.processing.core.ProcessingControlResult[source]#

Bases: ProcessingResultBase

The processing result of a processing error.

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

Parameters:

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

error: ProcessingError | StopProcessingError | FlagError | InvalidDataError#: The captured processing error

error_handling: ErrorHandling#: The type of handling for the captured error

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 processing control result object.

Return type:

ProcessingControlResult

Raises:

ValueError – If the level is passed.
ValueError – If the caught exception does not contain a message ot if the message is inconsistent.

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 processing control result object.

Return type:

ProcessingControlResult

get_targets()[source]#

Get the targets of the flag.

Return type:: Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch]

targets: Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch | None = None#: The flag targets (if the error is an flag error)

exception dispel.processing.core.ProcessingError[source]#

Bases: Exception

A base error class for errors that were caused during the processing.

Parameters:

step – The processing step that raised the exception
message – The error message

__init__(message, step)[source]#

Parameters:

message (str) –
step (ProcessingStep) –

class dispel.processing.core.ProcessingResult[source]#

Bases: ProcessingResultBase

The processing result of a processing step.

__init__(step, sources, result)#

Parameters:

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

Return type:

None

get_kwargs()[source]#

Get key word arguments for the additional class attributes.

The additional arguments are passed to the setter function in Reading while assigning processing results and creating the data trace. It allows to handle the result to be set depending on the arguments passed.

Returns:: Returns a dictionary of attributes of the instance omitting the step, sources, and result.
Return type:: Dict[str, Any]

get_sources()[source]#

Get the sources of the processing result.

Return type:: Iterable[Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch]

result: Level | MeasureValue | MeasureSet | LevelEpoch | RawDataSet#: The result of the processing function

sources: Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch#: The sources used for the processing function

class dispel.processing.core.ProcessingResultBase[source]#

Bases: object

The processing result base of a processing step.

__init__(step)#

Parameters:: step (ProcessingStep) –
Return type:: None

get_kwargs()[source]#

Get key word arguments for the additional class attributes.

Returns:: Returns a dictionary of attributes of the instance omitting the step.
Return type:: Dict[str, Any]

step: ProcessingStep#: The processing step associated with the processing

class dispel.processing.core.ProcessingStep[source]#

Bases: object

A processing step in a processing sequence.

ProcessingStep is the basic entity through which Readings are processed. The processing step’s process_reading() function is called with the reading and additional arguments passed to process(). Results from the process step are expected to be an instance of ProcessingResult. For a comprehensive description see Measure extraction.

The method flag_reading() can be overwritten to ensure that the reading about to be processed is valid, and return Flags if that is not the case.

Examples

processing-step

>>> from dispel.data.measures import MeasureValue
>>> from dispel.data.values import ValueDefinition
>>> from dispel.processing import process
>>> from dispel.processing.core import ProcessingResult, ProcessingStep
>>> class MyStep(ProcessingStep):
...     def process_reading(self, reading, **kwargs):
...         level = reading.get_level('my-level')
...         raw_data_set = level.get_raw_data_set('my-data-set')
...         data = raw_data_set.data
...         yield ProcessingResult(
...             step=self,
...             sources=raw_data_set,
...             result=MeasureValue(
...                 ValueDefinition('my-measure-id','max value'),
...                 data.max().max()
...             )
...         )
>>> _ = process(reading, MyStep())
>>> reading.measure_set.get_raw_value('my-measure-id')
5

__init__()[source]#

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

Assert that reading is valid.

Parameters:: reading (Reading) –

chain(successor)[source]#

Chain this step with the successor step.

Parameters:: successor (ProcessingStep) –
Return type:: ProcessingStep

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

Flag the provided reading.

Parameters:

reading (Reading) – The reading to be flagged.
kwargs – Additional arguments passed by process().

Yields:

Flag – The resulted flags.

Return type:

Generator[Flag, None, None]

get_parameters()[source]#

Get all parameters defined by the processing step.

Returns:: A list of tuples of parameter name and Parameter objects defined by the processing step.
Return type:: List[Tuple[str, Parameter]]

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

Get the reading flag targets.

Parameters:

reading (Reading) – The reading that is concerned with flagging.
kwargs – Additional keyword arguments eventually used for flag targets extraction.

Returns:

An iterable of entities that are flagged.

Return type:

Iterable[EntityType]

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

Check reading for validity and process it.

Parameters:

reading (Reading) – The reading to be processed
kwargs – Additional arguments passed by process().

Yields:

ProcessResultType – The results from processing readings.

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

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

Process the provided reading.

Parameters:

reading (Reading) – The reading to be processed
kwargs – Additional arguments passed by process().

Yields:

ProcessResultType – The results from processing readings.

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

set_next(step)[source]#

Set the next step in a processing chain of this step.

Parameters:: step (ProcessingStep) –

set_previous(step)[source]#

Set the previous step in a processing chain of this step.

Parameters:: step (ProcessingStep) –

exception dispel.processing.core.StopProcessingError[source]#

Bases: ProcessingError

An error that halts the processing of a reading.