From 7807cbeb397804ad0c500432d39b24ee10f21743 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:18:24 -0700 Subject: [PATCH] pipeline(task): add a clock to the pipeline task --- src/pipecat/pipeline/task.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 102b4528b..03fd5c734 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -10,6 +10,8 @@ from typing import AsyncIterable, Iterable from pydantic import BaseModel +from pipecat.clocks.base_clock import BaseClock +from pipecat.clocks.system_clock import SystemClock from pipecat.frames.frames import ( CancelFrame, EndFrame, @@ -60,11 +62,16 @@ class Source(FrameProcessor): class PipelineTask: - def __init__(self, pipeline: BasePipeline, params: PipelineParams = PipelineParams()): + def __init__( + self, + pipeline: BasePipeline, + params: PipelineParams = PipelineParams(), + clock: BaseClock = SystemClock()): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self._pipeline = pipeline + self._clock = clock self._params = params self._finished = False @@ -116,11 +123,14 @@ class PipelineTask: return MetricsFrame(ttfb=ttfb, processing=processing) async def _process_down_queue(self): + self._clock.start() + start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics, enable_usage_metrics=self._params.enable_metrics, - report_only_initial_ttfb=self._params.report_only_initial_ttfb + report_only_initial_ttfb=self._params.report_only_initial_ttfb, + clock=self._clock ) await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM)