dispel.data.flags module#

A module containing models for flags.

Flags are basically a structured way to mark the data contained in a certain entity as deviating or invalidating assumptions while processing.

class dispel.data.flags.Flag[source]#

Bases: object

A class for entity flag.

__init__(id_, reason, stop_processing=False)#

Parameters:

id_ (dataclasses.InitVar[Union[str, dispel.data.flags.FlagId]]) –
reason (str) –
stop_processing (bool) –

Return type:

None

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

Format an flag.

Return type:: Flag

id: FlagId#: The flag identifier

id_: dataclasses.InitVar[Union[str, dispel.data.flags.FlagId]]#: The flag identifier (string or id format)

reason: str#: The detailed reason for the flag

stop_processing: bool = False#: Stop processing

exception dispel.data.flags.FlagAlreadyExists[source]#

Bases: Exception

Exception raised when the same flag is added twice to an entity.

Parameters:: flag – The duplicated flag.

__init__(flag)[source]#

Parameters:: flag (Flag) –

class dispel.data.flags.FlagId[source]#

Bases: DefinitionId

The identifier of an entity flag for a task.

Parameters:

task_name – The name and abbreviation of the task. Note that if no abbreviation is provided the name is used directly in the id.
flag_name – The name of the flag and its abbreviation.
flag_type – The type of the flag. See FlagType.
flag_severity – The severity of the flag. See FlagSeverity.

Notes

The abbreviations of values are passed using AbbreviatedValue. To generate the actual id the .abbr accessor is used. If one passes only strings, the class actually wraps those into AbbreviatedValue instances.

Examples

>>> from dispel.data.values import AbbreviatedValue as AV
>>> from dispel.data.flags import FlagId, FlagType
>>> FlagId(
...     task_name=AV('Cognitive Processing Speed', 'CPS'),
...     flag_name=AV('tilt angle', 'ta'),
...     flag_type=FlagType.BEHAVIORAL,
...     flag_severity=FlagSeverity.DEVIATION,
... )
cps-behavioral-deviation-ta

__init__(task_name, flag_name, flag_type, flag_severity)[source]#

Parameters:

task_name (str | AbbreviatedValue) –
flag_name (str | AbbreviatedValue) –
flag_type (str | FlagType) –
flag_severity (str | FlagSeverity) –

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

Format an flag identifier.

Return type:: FlagId

classmethod from_str(value)[source]#

Create a flag id from a string representation.

Parameters:: value (str) – The string from which the flag id is to be constructed. It ought to respect the following format <task_name>-<flag_type>-<flag_name> where the flag type should one of the enumerations defined in FlagType.
Returns:: The initialised flag identifier.
Return type:: FlagId
Raises:: ValueError – If the flag string representation does not respect the required format.

class dispel.data.flags.FlagMixIn[source]#

Bases: ABC

Flag Mix-In class that groups multiple flags.

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

add_flag(flag, ignore_duplicates=False)[source]#

Add a flag to the flag mix in.

Parameters:

flag (Flag) – The flag to be added to the mix in.
ignore_duplicates (bool) – Set to True to ignore if a flag with the same id is already present. False will raise an FlagAlreadyExists

Raises:

FlagAlreadyExists – If ignore_duplicates is False and the flag already exists.

add_flags(flags, ignore_duplicates=False)[source]#

Add multiple flags to mix in.

Parameters:

flags (Iterable[Flag] | FlagMixIn) – The flags to be added. It can both an iterable of flags or an instance of FlagMixIn.
ignore_duplicates (bool) – Set to True to ignore if a flag with the same id is already present. False will raise an FlagAlreadyExists

property flag_count: int#: Get the number of flags in the mix in.

property flag_count_repr: str#: Get the string representation of the flag count.

property flag_ids: Set[FlagId]#: Get the unique ids of the flags in the mix in.

get_flags(flag_id=None)[source]#

Retrieve flags from the mix in.

Parameters:: flag_id (str | FlagId | None) – The id corresponding to the flags that are to be retrieved. If None is provided, all flags will be returned.
Returns:: The flags corresponding to the given id.
Return type:: List[Flag]
Raises:: FlagNotFound – If the given flag id does not correspond to any flag in the mix-in.

has_flag(value)[source]#
has_flag(flag)
has_flag(flag_id)
has_flag(flag_id): Return whether the flag is inside the mix in.

property is_valid: bool#: Return whether the flag mix in contains no flags.

exception dispel.data.flags.FlagNotFound[source]#

Bases: Exception

Exception raised when a flag is not found in the mix in.

Parameters:

flag_id – The flag id.
entity – The flag Mix-In entity where the flag was not found.
message – An optional error message to be added at the end.

__init__(flag_id, entity, message='')[source]#

Parameters:

flag_id (str | FlagId) –
entity (FlagMixIn) –
message (str) –

class dispel.data.flags.FlagSeverity[source]#

Bases: AVEnum

An enumeration for flag severity.

DEVIATION = 1#

INVALIDATION = 2#

class dispel.data.flags.FlagType[source]#

Bases: AVEnum

An enumeration for flag types.

BEHAVIORAL = 2#

TECHNICAL = 1#

class dispel.data.flags.WrappedResult[source]#

Bases: FlagMixIn, Generic[WrappedResultType]

A wrapped result to carry potential flags.

This class provides a convenient way to add flags to values from extract steps that are known to be invalid. This avoids having to write a separate flag step and is useful in cases where the information to flag a result is only accessible in the extract function.

Parameters:: measure_value – The value of the measure returned by the extraction function.

measure_value#: The value of the measure returned by the extraction function.

Examples

Assuming we wanted to flag measures directly inside a custom extraction function based on some metrics calculated, one can do

>>> from dispel.processing.extract import WrappedResult
>>> from dispel.data.flags import Flag
>>> from typing import Union
>>> def custom_aggregation_func(data) -> Union[WrappedResult, float]:
...     result = data.agg('mean')
...     if len(data) < 3:
...         inv = Flag(
...             reason='Not enough data point',
...             flag_severity=FlagSeverity.INVALIDATION
...         )
...         result = WrappedResult(result, inv)
...     return result

During processing, the class ExtractStep allows the transformation function to output WrappedResult objects. The extract step will automatically add any flags present in the WrappedResult object to the measure value. The WrappedResult class supports basic operations with other scalars or WrappedResult object:

>>> from dispel.processing.extract import WrappedResult
>>> res1 = WrappedResult(measure_value=1)
>>> res2 = WrappedResult(measure_value=2)
>>> melted_res = res1 + res2
>>> melted_res2 = res1 + 1

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

Parameters:: measure_value (WrappedResultType) –

dispel.data.flags.verify_flag_uniqueness(flags)[source]#

Check whether all provided flags are unique.

Parameters:: flags (List[Flag]) – The flags whose uniqueness is to be verified.
Raises:: FlagAlreadyExists – Per non-unique flag.