PipelineTask: add new started/stopped/ended/cancelled events

CHANGELOG.md

@@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new `PipelineTask` event handlers `on_pipeline_started`,
+  `on_pipeline_stopped`, `on_pipeline_ended` and `on_pipeline_cancelled`, which
+  correspond to the `StartFrame`, `StopFrame`, `EndFrame` and `CancelFrame`
+  respectively.
+
 - Added additional languages to `LmntTTSService`. Languages include: `hi`, `id`,
-  `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`. 
+  `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`.
 
 - Added a `model` parameter to the `LmntTTSService` constructor, allowing
   switching between LMNT models.

src/pipecat/pipeline/task.py

@@ -144,7 +144,26 @@ class PipelineTask(BaseTask):
     `LLMFullResponseEndFrame` are received within `idle_timeout_secs`.
 
        @task.event_handler("on_idle_timeout")
-       async def on_idle_timeout(task):
+       async def on_pipeline_idle_timeout(task):
+           ...
+
+    There are also events to know if a pipeline has been started, stopped, ended
+    or cancelled.
+
+       @task.event_handler("on_pipeline_started")
+       async def on_pipeline_started(task, frame: StartFrame):
+           ...
+
+       @task.event_handler("on_pipeline_stopped")
+       async def on_pipeline_stopped(task, frame: StopFrame):
+           ...
+
+       @task.event_handler("on_pipeline_ended")
+       async def on_pipeline_ended(task, frame: EndFrame):
+           ...
+
+       @task.event_handler("on_pipeline_cancelled")
+       async def on_pipeline_cancelled(task, frame: CancelFrame):
            ...
 
     Args:
@@ -264,6 +283,10 @@ class PipelineTask(BaseTask):
         self._register_event_handler("on_frame_reached_upstream")
         self._register_event_handler("on_frame_reached_downstream")
         self._register_event_handler("on_idle_timeout")
+        self._register_event_handler("on_pipeline_started")
+        self._register_event_handler("on_pipeline_stopped")
+        self._register_event_handler("on_pipeline_ended")
+        self._register_event_handler("on_pipeline_cancelled")
 
     @property
     def params(self) -> PipelineParams:
@@ -552,8 +575,16 @@ class PipelineTask(BaseTask):
             if isinstance(frame, self._reached_downstream_types):
                 await self._call_event_handler("on_frame_reached_downstream", frame)
 
-            if isinstance(frame, (EndFrame, StopFrame)):
+            if isinstance(frame, StartFrame):
+                await self._call_event_handler("on_pipeline_started", frame)
+            elif isinstance(frame, EndFrame):
+                await self._call_event_handler("on_pipeline_ended", frame)
                 self._pipeline_end_event.set()
+            elif isinstance(frame, StopFrame):
+                await self._call_event_handler("on_pipeline_stopped", frame)
+                self._pipeline_end_event.set()
+            elif isinstance(frame, CancelFrame):
+                await self._call_event_handler("on_pipeline_cancelled", frame)
             elif isinstance(frame, HeartbeatFrame):
                 await self._heartbeat_queue.put(frame)
             self._down_queue.task_done()

tests/test_pipeline.py

@@ -8,7 +8,13 @@ import asyncio
 import time
 import unittest
 
-from pipecat.frames.frames import EndFrame, HeartbeatFrame, StartFrame, TextFrame
+from pipecat.frames.frames import (
+    EndFrame,
+    HeartbeatFrame,
+    StartFrame,
+    StopFrame,
+    TextFrame,
+)
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -95,7 +101,50 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         await task.run()
         assert task.has_finished()
 
-    async def test_task_event_handlers(self):
+    async def test_task_started_ended_event_handler(self):
+        start_received = False
+        end_received = False
+
+        identity = IdentityFilter()
+        pipeline = Pipeline([identity])
+        task = PipelineTask(pipeline)
+        task.set_event_loop(asyncio.get_event_loop())
+
+        @task.event_handler("on_pipeline_started")
+        async def on_pipeline_started(task, frame: StartFrame):
+            nonlocal start_received
+            start_received = True
+
+        @task.event_handler("on_pipeline_ended")
+        async def on_pipeline_ended(task, frame: EndFrame):
+            nonlocal end_received
+            end_received = True
+
+        await task.queue_frame(EndFrame())
+        await task.run()
+
+        assert start_received
+        assert end_received
+
+    async def test_task_stopped_event_handler(self):
+        stop_received = False
+
+        identity = IdentityFilter()
+        pipeline = Pipeline([identity])
+        task = PipelineTask(pipeline)
+        task.set_event_loop(asyncio.get_event_loop())
+
+        @task.event_handler("on_pipeline_stopped")
+        async def on_pipeline_ended(task, frame: StopFrame):
+            nonlocal stop_received
+            stop_received = True
+
+        await task.queue_frame(StopFrame())
+        await task.run()
+
+        assert stop_received
+
+    async def test_task_frame_reached_event_handlers(self):
         upstream_received = False
         downstream_received = False