From 4432e7e4f790defd2c27fe544d90f7d08890a35e Mon Sep 17 00:00:00 2001
From: Mark Backman <mark@daily.co>
Date: Thu, 6 Feb 2025 11:04:54 -0500
Subject: [PATCH] Add docstrings for PipelineTask and related classes/functions

---
 src/pipecat/pipeline/task.py | 61 +++++++++++++++++++++++++++---------
 1 file changed, 46 insertions(+), 15 deletions(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 21b6af78b..66b1e7647 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -38,6 +38,22 @@ HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
 
 
 class PipelineParams(BaseModel):
+    """Configuration parameters for pipeline execution.
+
+    Attributes:
+        allow_interruptions: Whether to allow pipeline interruptions.
+        audio_in_sample_rate: Input audio sample rate in Hz.
+        audio_out_sample_rate: Output audio sample rate in Hz.
+        enable_heartbeats: Whether to enable heartbeat monitoring.
+        enable_metrics: Whether to enable metrics collection.
+        enable_usage_metrics: Whether to enable usage metrics.
+        heartbeats_period_secs: Period between heartbeats in seconds.
+        observers: List of observers for monitoring pipeline execution.
+        report_only_initial_ttfb: Whether to report only initial time to first byte.
+        send_initial_empty_metrics: Whether to send initial empty metrics.
+        start_metadata: Additional metadata for pipeline start.
+    """
+
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     allow_interruptions: bool = False
@@ -54,11 +70,16 @@ class PipelineParams(BaseModel):
 
 
 class PipelineTaskSource(FrameProcessor):
-    """This is the source processor that is linked at the beginning of the
+    """Source processor for pipeline tasks that handles frame routing.
+
+    This is the source processor that is linked at the beginning of the
     pipeline given to the pipeline task. It allows us to easily push frames
     downstream to the pipeline and also receive upstream frames coming from the
     pipeline.
 
+    Args:
+        up_queue: Queue for upstream frame processing.
+
     """
 
     def __init__(self, up_queue: asyncio.Queue, **kwargs):
@@ -76,10 +97,14 @@ class PipelineTaskSource(FrameProcessor):
 
 
 class PipelineTaskSink(FrameProcessor):
-    """This is the sink processor that is linked at the end of the pipeline
+    """Sink processor for pipeline tasks that handles final frame processing.
+
+    This is the sink processor that is linked at the end of the pipeline
     given to the pipeline task. It allows us to receive downstream frames and
     act on them, for example, waiting to receive an EndFrame.
 
+    Args:
+        down_queue: Queue for downstream frame processing.
     """
 
     def __init__(self, down_queue: asyncio.Queue, **kwargs):
@@ -92,6 +117,14 @@ class PipelineTaskSink(FrameProcessor):
 
 
 class PipelineTask(BaseTask):
+    """Manages the execution of a pipeline, handling frame processing and task lifecycle.
+
+    Args:
+        pipeline: The pipeline to execute.
+        params: Configuration parameters for the pipeline.
+        clock: Clock implementation for timing operations.
+    """
+
     def __init__(
         self,
         pipeline: BasePipeline,
@@ -163,9 +196,7 @@ class PipelineTask(BaseTask):
         await self.queue_frame(EndFrame())
 
     async def cancel(self):
-        """
-        Stops the running pipeline immediately.
-        """
+        """Stops the running pipeline immediately."""
         logger.debug(f"Canceling pipeline task {self}")
         # Make sure everything is cleaned up downstream. This is sent
         # out-of-band from the main streaming task which is what we want since
@@ -175,9 +206,7 @@ class PipelineTask(BaseTask):
         await self._task_manager.cancel_task(self._process_push_task)
 
     async def run(self):
-        """
-        Starts running the given pipeline.
-        """
+        """Starts and manages the pipeline execution until completion or cancellation."""
         if self.has_finished():
             return
         try:
@@ -195,14 +224,18 @@ class PipelineTask(BaseTask):
         self._finished = True
 
     async def queue_frame(self, frame: Frame):
-        """
-        Queue a frame to be pushed down the pipeline.
+        """Queue a single frame to be pushed down the pipeline.
+
+        Args:
+            frame: The frame to be processed.
         """
         await self._push_queue.put(frame)
 
     async def queue_frames(self, frames: Iterable[Frame] | AsyncIterable[Frame]):
-        """
-        Queues multiple frames to be pushed down the pipeline.
+        """Queues multiple frames to be pushed down the pipeline.
+
+        Args:
+            frames: An iterable or async iterable of frames to be processed.
         """
         if isinstance(frames, AsyncIterable):
             async for frame in frames:
@@ -348,9 +381,7 @@ class PipelineTask(BaseTask):
             self._down_queue.task_done()
 
     async def _heartbeat_push_handler(self):
-        """
-        This tasks pushes a heartbeat frame every heartbeat period.
-        """
+        """This tasks pushes a heartbeat frame every heartbeat period."""
         while True:
             # Don't use `queue_frame()` because if an EndFrame is queued the
             # task will just stop waiting for the pipeline to finish not