"""this module contains the abstract implementation for the execution tracker"""
import datetime
from abc import ABC, abstractmethod
from mosaic_orchestrator.protocols import LifeCycleListener
from mosaic_orchestrator.task import Task
[docs]class ExecutionTracker(ABC, LifeCycleListener):
"""
This is the abstract base class for execution trackers.
You can use it to capture usage statistics or to synchronize some GUI while a generator is running.
You can receive events about important status updates.
"""
def _on_start(self):
"""
Executed before the root task runs
"""
return None
def _on_end(self, exception: Exception = None):
"""
Override this method to optional stop the tracker and clean up resources.
This method is called after each run of the task hierarchy, even if an exception is thrown
during the run, similar to try-catch-finally in python.
Args:
exception: if the run aborted with an exception it is passed as an argument.
"""
return None
[docs] @abstractmethod
def set_entry(self, key: str, value: any) -> None:
"""
Set the key, value pair in the value collection.
Args:
key: Key to use
value: The value to add
"""
[docs] @abstractmethod
def append_to_entry(self, key:str, value:any) -> None:
"""Append to an entry another value (e.g. add a new tool)
Args:
key (str): Key to use
value (any): Value to append
"""
[docs] @abstractmethod
def start_task(
self,
task: Task,
is_root_task: bool = False,
) -> None:
"""
Called when a task execution is started
Args:
task: The task
is_root_task: Is the current task the root task?
"""
[docs] @abstractmethod
def end_task(
self,
task: Task,
cache_used: bool = False,
is_root_task: bool = False,
) -> None:
"""
Called when a task execution ends
Args:
task: The task
cache_used: Cache used?
is_root_task: Is the current task the root task?
"""