wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pipeline(task): introduce has_finished()

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-04 12:12:19 -07:00
parent 4057fbbcfd
commit af202d4fe5
4 changed files with 9 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/foundational/05a-local-sync-speech-and-image.py
View File

@@ -156,7 +156,7 @@ async def main():
await runner.stop_when_done() await runner.stop_when_done()
async def run_tk(): async def run_tk():
while True: while not task.has_finished():
tk_root.update() tk_root.update()
tk_root.update_idletasks() tk_root.update_idletasks()
await asyncio.sleep(0.1) await asyncio.sleep(0.1)

6
examples/foundational/09a-local-mirror.py
View File

@@ -48,15 +48,15 @@ async def main(room_url, token):
pipeline = Pipeline([daily_transport.input(), tk_transport.output()]) pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
runner = PipelineRunner() task = PipelineTask(pipeline)
async def run_tk(): async def run_tk():
while runner.is_active(): while not task.has_finished():
tk_root.update() tk_root.update()
tk_root.update_idletasks() tk_root.update_idletasks()
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
task = PipelineTask(pipeline) runner = PipelineRunner()
await asyncio.gather(runner.run(task), run_tk()) await asyncio.gather(runner.run(task), run_tk())

6
src/pipecat/pipeline/runner.py
View File

@@ -20,18 +20,15 @@ class PipelineRunner:
self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}" self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}"
self._tasks = {} self._tasks = {}
self._running = True
if handle_sigint: if handle_sigint:
self._setup_sigint() self._setup_sigint()
async def run(self, task: PipelineTask): async def run(self, task: PipelineTask):
logger.debug(f"Runner {self} started running {task}") logger.debug(f"Runner {self} started running {task}")
self._running = True
self._tasks[task.name] = task self._tasks[task.name] = task
await task.run() await task.run()
del self._tasks[task.name] del self._tasks[task.name]
self._running = False
logger.debug(f"Runner {self} finished running {task}") logger.debug(f"Runner {self} finished running {task}")
async def stop_when_done(self): async def stop_when_done(self):
@@ -42,9 +39,6 @@ class PipelineRunner:
logger.debug(f"Canceling runner {self}") logger.debug(f"Canceling runner {self}")
await asyncio.gather(*[t.cancel() for t in self._tasks.values()]) await asyncio.gather(*[t.cancel() for t in self._tasks.values()])
def is_active(self):
return self._running
def _setup_sigint(self): def _setup_sigint(self):
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
loop.add_signal_handler( loop.add_signal_handler(

5
src/pipecat/pipeline/task.py
View File

@@ -43,6 +43,7 @@ class PipelineTask:
self._pipeline = pipeline self._pipeline = pipeline
self._params = params self._params = params
self._finished = False
self._down_queue = asyncio.Queue() self._down_queue = asyncio.Queue()
self._up_queue = asyncio.Queue() self._up_queue = asyncio.Queue()
@@ -50,6 +51,9 @@ class PipelineTask:
self._source = Source(self._up_queue) self._source = Source(self._up_queue)
self._source.link(pipeline) self._source.link(pipeline)
def has_finished(self):
return self._finished
async def stop_when_done(self): async def stop_when_done(self):
logger.debug(f"Task {self} scheduled to stop when done") logger.debug(f"Task {self} scheduled to stop when done")
await self.queue_frame(EndFrame()) await self.queue_frame(EndFrame())
@@ -67,6 +71,7 @@ class PipelineTask:
self._process_up_task = asyncio.create_task(self._process_up_queue()) self._process_up_task = asyncio.create_task(self._process_up_queue())
self._process_down_task = asyncio.create_task(self._process_down_queue()) self._process_down_task = asyncio.create_task(self._process_down_queue())
await asyncio.gather(self._process_up_task, self._process_down_task) await asyncio.gather(self._process_up_task, self._process_down_task)
self._finished = True
async def queue_frame(self, frame: Frame): async def queue_frame(self, frame: Frame):
await self._down_queue.put(frame) await self._down_queue.put(frame)