From af202d4fe57fff9e362caf647b7796ae3eab2331 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Tue, 4 Jun 2024 12:12:19 -0700
Subject: [PATCH] pipeline(task): introduce has_finished()

---
 examples/foundational/05a-local-sync-speech-and-image.py | 2 +-
 examples/foundational/09a-local-mirror.py                | 6 +++---
 src/pipecat/pipeline/runner.py                           | 6 ------
 src/pipecat/pipeline/task.py                             | 5 +++++
 4 files changed, 9 insertions(+), 10 deletions(-)

diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py
index d476754fb..a377ebe98 100644
--- a/examples/foundational/05a-local-sync-speech-and-image.py
+++ b/examples/foundational/05a-local-sync-speech-and-image.py
@@ -156,7 +156,7 @@ async def main():
             await runner.stop_when_done()
 
         async def run_tk():
-            while True:
+            while not task.has_finished():
                 tk_root.update()
                 tk_root.update_idletasks()
                 await asyncio.sleep(0.1)
diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py
index 78a2fd97d..92e244d3f 100644
--- a/examples/foundational/09a-local-mirror.py
+++ b/examples/foundational/09a-local-mirror.py
@@ -48,15 +48,15 @@ async def main(room_url, token):
 
     pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
 
-    runner = PipelineRunner()
+    task = PipelineTask(pipeline)
 
     async def run_tk():
-        while runner.is_active():
+        while not task.has_finished():
             tk_root.update()
             tk_root.update_idletasks()
             await asyncio.sleep(0.1)
 
-    task = PipelineTask(pipeline)
+    runner = PipelineRunner()
 
     await asyncio.gather(runner.run(task), run_tk())
 
diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py
index 2d1b8100a..47754d82f 100644
--- a/src/pipecat/pipeline/runner.py
+++ b/src/pipecat/pipeline/runner.py
@@ -20,18 +20,15 @@ class PipelineRunner:
         self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}"
 
         self._tasks = {}
-        self._running = True
 
         if handle_sigint:
             self._setup_sigint()
 
     async def run(self, task: PipelineTask):
         logger.debug(f"Runner {self} started running {task}")
-        self._running = True
         self._tasks[task.name] = task
         await task.run()
         del self._tasks[task.name]
-        self._running = False
         logger.debug(f"Runner {self} finished running {task}")
 
     async def stop_when_done(self):
@@ -42,9 +39,6 @@ class PipelineRunner:
         logger.debug(f"Canceling runner {self}")
         await asyncio.gather(*[t.cancel() for t in self._tasks.values()])
 
-    def is_active(self):
-        return self._running
-
     def _setup_sigint(self):
         loop = asyncio.get_running_loop()
         loop.add_signal_handler(
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index f81281820..e3e7f7e36 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -43,6 +43,7 @@ class PipelineTask:
 
         self._pipeline = pipeline
         self._params = params
+        self._finished = False
 
         self._down_queue = asyncio.Queue()
         self._up_queue = asyncio.Queue()
@@ -50,6 +51,9 @@ class PipelineTask:
         self._source = Source(self._up_queue)
         self._source.link(pipeline)
 
+    def has_finished(self):
+        return self._finished
+
     async def stop_when_done(self):
         logger.debug(f"Task {self} scheduled to stop when done")
         await self.queue_frame(EndFrame())
@@ -67,6 +71,7 @@ class PipelineTask:
         self._process_up_task = asyncio.create_task(self._process_up_queue())
         self._process_down_task = asyncio.create_task(self._process_down_queue())
         await asyncio.gather(self._process_up_task, self._process_down_task)
+        self._finished = True
 
     async def queue_frame(self, frame: Frame):
         await self._down_queue.put(frame)