docs: Add task worker docstrings (#1186)

Closes #1179

Author: dan-fernandes

src/blueapi/worker/task_worker.py

@@ -145,6 +145,13 @@ def __init__(
 
     @start_as_current_span(TRACER, "task_id")
     def clear_task(self, task_id: str) -> str:
+        """
+        Remove a task from the worker
+        Args:
+            task_id: The ID of the task to be removed
+        Returns:
+            task_id of the removed task
+        """
         pending = self._pending_tasks.pop(task_id, None)
         task = pending or self._completed_tasks.pop(task_id)
         return task.task_id
@@ -155,6 +162,14 @@ def cancel_active_task(
         failure: bool = False,
         reason: str | None = None,
     ) -> str:
+        """
+        Remove the currently active task from the worker if there is one
+        Args:
+            failure: Flag cancellation as error
+            reason: Reason for cancellation
+        Returns:
+            The task_id of the active task
+        """
         if self._current is None:
             # Persuades type checker that self._current is not None
             # We only allow this method to be called if a Plan is active
@@ -171,14 +186,36 @@ def cancel_active_task(
 
     @start_as_current_span(TRACER)
     def get_tasks(self) -> list[TrackableTask]:
+        """
+        Return a list of all tasks on the worker,
+        any one of which can be triggered with begin_task.
+        Returns:
+            List[TrackableTask[T]]: List of task objects
+        """
         return list(self._pending_tasks.values()) + list(self._completed_tasks.values())
 
     @start_as_current_span(TRACER, "task_id")
     def get_task_by_id(self, task_id: str) -> TrackableTask | None:
+        """
+        Returns a task matching the task ID supplied,
+        if the worker knows of it.
+        Args:
+            task_id: The ID of the task
+        Returns:
+            Optional[TrackableTask[T]]: The task matching the ID,
+                None if the task ID is unknown to the worker.
+        """
         return self._pending_tasks.get(task_id, None) or self._completed_tasks[task_id]
 
     @start_as_current_span(TRACER, "status")
     def get_tasks_by_status(self, status: TaskStatusEnum) -> list[TrackableTask]:
+        """
+        Retrieve a list of tasks based on their status.
+        Args:
+           TaskStatusEnum: The status to filter tasks by.
+        Returns:
+          list[TrackableTask]: A list of tasks that match the given status.
+        """
         if status == TaskStatusEnum.RUNNING:
             return [
                 task
@@ -193,13 +230,26 @@ def get_tasks_by_status(self, status: TaskStatusEnum) -> list[TrackableTask]:
 
     @start_as_current_span(TRACER)
     def get_active_task(self) -> TrackableTask[Task] | None:
+        """
+        Returns the task the worker is currently running
+        Returns:
+            Optional[TrackableTask[Task]]: The current task,
+                None if the worker is idle.
+        """
         current = self._current
         if current is not None:
             add_span_attributes({"Active Task": current.task_id})
         return current
 
     @start_as_current_span(TRACER, "task_id")
     def begin_task(self, task_id: str) -> None:
+        """
+        Trigger a pending task. Will fail if the worker is busy.
+        Args:
+            task_id: The ID of the task to be triggered
+        Throws:
+            KeyError: If the task ID does not exist
+        """
         task = self._pending_tasks.get(task_id)
         if task is None:
             raise KeyError(f"No pending task with ID {task_id}")
@@ -208,6 +258,13 @@ def begin_task(self, task_id: str) -> None:
 
     @start_as_current_span(TRACER, "task.name", "task.params")
     def submit_task(self, task: Task) -> str:
+        """
+        Submit a task to be run on begin_task
+        Args:
+            task: A description of the task
+        Returns:
+            str: A unique ID to refer to this task
+        """
         task.prepare_params(self._ctx)  # Will raise if parameters are invalid
         task_id: str = str(uuid.uuid4())
         add_span_attributes({"TaskId": task_id})
@@ -264,6 +321,10 @@ def mark_task_as_started(event: WorkerEvent, _: str | None) -> None:
 
     @start_as_current_span(TRACER)
     def start(self) -> None:
+        """
+        Start worker in a new thread. Does not block, configures the bluesky
+        event loop in the new thread.
+        """
         if self._started.is_set():
             raise WorkerAlreadyStartedError("Worker is already running")
         self._wait_until_stopped()
@@ -272,6 +333,9 @@ def start(self) -> None:
 
     @start_as_current_span(TRACER)
     def stop(self) -> None:
+        """
+        Command the worker to gracefully stop. Blocks until it has shut down.
+        """
         LOGGER.info("Attempting to stop worker")
 
         # If the worker has not yet started there is nothing to do.
@@ -299,10 +363,16 @@ def _wait_until_stopped(self) -> None:
 
     @property
     def state(self) -> WorkerState:
+        """
+        :return: state of the worker
+        """
         return self._state
 
     @start_as_current_span(TRACER)
     def run(self) -> None:
+        """
+        Run all tasks that are submitted to the worker. Blocks thread.
+        """
         LOGGER.info("Worker starting")
         self._ctx.run_engine.state_hook = self._on_state_change  # type: ignore
         self._ctx.run_engine.subscribe(self._on_document)
@@ -319,11 +389,20 @@ def run(self) -> None:
 
     @start_as_current_span(TRACER, "defer")
     def pause(self, defer=False):
+        """
+        Command the worker to pause.
+
+        Args:
+            defer: Optional, if true wait till next checkpoint
+        """
         LOGGER.info("Requesting to pause the worker")
         self._ctx.run_engine.request_pause(defer)
 
     @start_as_current_span(TRACER)
     def resume(self):
+        """
+        Command the worker to resume
+        """
         LOGGER.info("Requesting to resume the worker")
         self._ctx.run_engine.resume()
 
@@ -378,14 +457,29 @@ def _cycle(self) -> None:
 
     @property
     def worker_events(self) -> EventStream[WorkerEvent, int]:
+        """
+        Events representing changes/errors in worker state
+        Returns:
+            EventStream[WorkerEvent, int]: Subscribable stream of events
+        """
         return self._worker_events
 
     @property
     def progress_events(self) -> EventStream[ProgressEvent, int]:
+        """
+        Events representing progress in running a task
+        Returns:
+            EventStream[ProgressEvent, int]: Subscribable stream of events
+        """
         return self._progress_events
 
     @property
     def data_events(self) -> EventStream[DataEvent, int]:
+        """
+        Events representing collection of data
+        Returns:
+            EventStream[DataEvent, int]: Subscribable stream of events
+        """
         return self._data_events
 
     @start_as_current_span(TRACER, "raw_new_state", "raw_old_state")