class WorkforceLogger(WorkforceCallback, WorkforceMetrics):
def __init__(self, workforce_id: str):
def _log_event(self, event_type: str, **kwargs: Any):
def log_task_created(self, event: TaskCreatedEvent):
def log_task_decomposed(self, event: TaskDecomposedEvent):
def log_task_assigned(self, event: TaskAssignedEvent):
def log_task_started(self, event: TaskStartedEvent):
def log_task_completed(self, event: TaskCompletedEvent):
def log_task_failed(self, event: TaskFailedEvent):
def log_worker_created(self, event: WorkerCreatedEvent):
def log_worker_deleted(self, event: WorkerDeletedEvent):
def log_queue_status(self, event: QueueStatusEvent):
def log_all_tasks_completed(self, event: AllTasksCompletedEvent):
def reset_task_data(self):
def dump_to_json(self, file_path: str):
def _get_all_tasks_in_hierarchy(self, task_id: str):
def _get_task_tree_string(
self,
task_id: str,
prefix: str = '',
is_last: bool = True
):
def get_ascii_tree_representation(self):
def get_kpis(self):