From 75669b12a2a4a07c396932a0e67afa951733db2c Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 2 Mar 2026 11:01:26 -0500 Subject: [PATCH] Convert observer data models to Pydantic BaseModel with timestamps Switch ProcessorStartupTiming, StartupTimingReport, and TransportTimingReport from dataclasses to Pydantic BaseModel. Add start_time (Unix timestamp) fields and wall clock conversion for monotonic observer timestamps. --- .../observers/startup_timing_observer.py | 37 +++++++++++++++---- tests/test_startup_timing_observer.py | 3 ++ 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py index 555a10cb0..6dd574cdc 100644 --- a/src/pipecat/observers/startup_timing_observer.py +++ b/src/pipecat/observers/startup_timing_observer.py @@ -34,9 +34,11 @@ Example:: task = PipelineTask(pipeline, observers=[observer]) """ -from dataclasses import dataclass, field +import time from typing import Dict, List, Optional, Tuple, Type +from pydantic import BaseModel, Field + from pipecat.frames.frames import BotConnectedFrame, ClientConnectedFrame, StartFrame from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed from pipecat.pipeline.base_pipeline import BasePipeline @@ -47,42 +49,45 @@ from pipecat.processors.frame_processor import FrameProcessor _INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline) -@dataclass -class ProcessorStartupTiming: +class ProcessorStartupTiming(BaseModel): """Startup timing for a single processor. Parameters: processor_name: The name of the processor. + start_time: Unix timestamp when the processor's start() began. duration_secs: How long the processor's start() took, in seconds. """ processor_name: str + start_time: float duration_secs: float -@dataclass -class StartupTimingReport: +class StartupTimingReport(BaseModel): """Report of startup timings for all measured processors. Parameters: + start_time: Unix timestamp when the first processor began starting. total_duration_secs: Total wall-clock time from first to last processor start. processor_timings: Per-processor timing data, in pipeline order. """ + start_time: float total_duration_secs: float - processor_timings: List[ProcessorStartupTiming] = field(default_factory=list) + processor_timings: List[ProcessorStartupTiming] = Field(default_factory=list) -@dataclass -class TransportTimingReport: +class TransportTimingReport(BaseModel): """Time from pipeline start to transport connection milestones. Parameters: + start_time: Unix timestamp of the StartFrame (pipeline start). bot_connected_secs: Seconds from StartFrame to first BotConnectedFrame (only set for SFU transports). client_connected_secs: Seconds from StartFrame to first ClientConnectedFrame. """ + start_time: float bot_connected_secs: Optional[float] = None client_connected_secs: Optional[float] = None @@ -176,9 +181,19 @@ class StartupTimingObserver(BaseObserver): # Bot connected timing (stored for inclusion in the transport report). self._bot_connected_secs: Optional[float] = None + # Wall clock reference for converting monotonic ns to Unix timestamps. + self._wall_clock_ref: Optional[float] = None + self._mono_clock_ref_ns: Optional[int] = None + self._register_event_handler("on_startup_timing_report") self._register_event_handler("on_transport_timing_report") + def _mono_to_wall(self, mono_ns: int) -> float: + """Convert a monotonic nanosecond timestamp to a Unix wall clock time.""" + if self._wall_clock_ref is None or self._mono_clock_ref_ns is None: + return 0.0 + return self._wall_clock_ref + (mono_ns - self._mono_clock_ref_ns) / 1e9 + def _should_track(self, processor: FrameProcessor) -> bool: """Check if a processor should be tracked for timing. @@ -212,6 +227,8 @@ class StartupTimingObserver(BaseObserver): if self._start_frame_id is None: self._start_frame_id = data.frame.id self._start_frame_arrival_ns = data.timestamp + self._wall_clock_ref = time.time() + self._mono_clock_ref_ns = data.timestamp elif data.frame.id != self._start_frame_id: return @@ -263,6 +280,7 @@ class StartupTimingObserver(BaseObserver): self._timings.append( ProcessorStartupTiming( processor_name=processor.name, + start_time=self._mono_to_wall(arrival_ts), duration_secs=duration_secs, ) ) @@ -284,6 +302,7 @@ class StartupTimingObserver(BaseObserver): delta_ns = data.timestamp - self._start_frame_arrival_ns client_connected_secs = delta_ns / 1e9 report = TransportTimingReport( + start_time=self._mono_to_wall(self._start_frame_arrival_ns), bot_connected_secs=self._bot_connected_secs, client_connected_secs=client_connected_secs, ) @@ -296,8 +315,10 @@ class StartupTimingObserver(BaseObserver): self._startup_timing_reported = True total = sum(t.duration_secs for t in self._timings) + start_time = self._timings[0].start_time if self._timings else 0.0 report = StartupTimingReport( + start_time=start_time, total_duration_secs=total, processor_timings=self._timings, ) diff --git a/tests/test_startup_timing_observer.py b/tests/test_startup_timing_observer.py index 3c89b9ca3..2bc246754 100644 --- a/tests/test_startup_timing_observer.py +++ b/tests/test_startup_timing_observer.py @@ -151,9 +151,11 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): report = reports[0] self.assertIsInstance(report, StartupTimingReport) self.assertIsInstance(report.total_duration_secs, float) + self.assertGreater(report.start_time, 0) for timing in report.processor_timings: self.assertIsInstance(timing.processor_name, str) self.assertIsInstance(timing.duration_secs, float) + self.assertGreater(timing.start_time, 0) async def test_excludes_internal_processors(self): """Test that internal pipeline processors are excluded by default.""" @@ -211,6 +213,7 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(transport_reports), 1) report = transport_reports[0] self.assertIsInstance(report, TransportTimingReport) + self.assertGreater(report.start_time, 0) self.assertGreater(report.client_connected_secs, 0) self.assertIsNone(report.bot_connected_secs)