Skip to content

dp3.task_processing.task_hooks

TaskGenericHooksContainer 

TaskGenericHooksContainer(log: Logger, elog: EventGroupType, hook_elog: EventGroupType | None = None)

Container for generic hooks

Possible hooks:

  • on_task_start: receives Task, no return value requirements
Source code in dp3/task_processing/task_hooks.py 
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    log: logging.Logger,
    elog: EventGroupType,
    hook_elog: EventGroupType | None = None,
):
    self.log = log.getChild("genericHooks")
    self.elog = elog
    self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup())

    self._on_start: list[TaskStartHook] = []

TaskEntityHooksContainer 

TaskEntityHooksContainer(entity: str, model_spec: ModelSpec, log: Logger, elog: EventGroupType, hook_elog: EventGroupType | None = None)

Container for entity hooks

Possible hooks:

  • allow_entity_creation: receives eid and Task, may prevent entity record creation (by returning False)
  • on_entity_creation: receives eid and Task, may return list of DataPointTasks
Source code in dp3/task_processing/task_hooks.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    entity: str,
    model_spec: ModelSpec,
    log: logging.Logger,
    elog: EventGroupType,
    hook_elog: EventGroupType | None = None,
):
    self.entity = entity
    self.log = log.getChild(f"entityHooks.{entity}")
    self.elog = elog
    self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup())
    self.model_spec = model_spec

    self._allow_creation: list[AllowEntityCreationHook] = []
    self._on_creation: list[OnEntityCreationHook] = []

TaskAttrHooksContainer 

TaskAttrHooksContainer(entity: str, attr: str, attr_type: AttrType, model_spec: ModelSpec, log: Logger, elog: EventGroupType, hook_elog: EventGroupType | None = None)

Container for attribute hooks

Possible hooks:

  • on_new_plain, on_new_observation, on_new_ts_chunk: receives eid and DataPointBase, may return a list of DataPointTasks
Source code in dp3/task_processing/task_hooks.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def __init__(
    self,
    entity: str,
    attr: str,
    attr_type: AttrType,
    model_spec: ModelSpec,
    log: logging.Logger,
    elog: EventGroupType,
    hook_elog: EventGroupType | None = None,
):
    self.entity = entity
    self.attr = attr
    self.log = log.getChild(f"attributeHooks.{entity}.{attr}")
    self.elog = elog
    self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup())
    self.model_spec = model_spec

    try:
        self.on_new_hook_type = ATTR_TYPE_TO_ON_NEW_HOOK[attr_type]
    except KeyError as e:
        raise ValueError(f"Invalid attribute type '{attr_type}'") from e

    self._on_new: list[OnNewAttributeHook] = []