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.
This commit is contained in:
Mark Backman
2026-03-02 11:01:26 -05:00
parent 68e8732e72
commit 75669b12a2
2 changed files with 32 additions and 8 deletions

View File

@@ -34,9 +34,11 @@ Example::
task = PipelineTask(pipeline, observers=[observer]) task = PipelineTask(pipeline, observers=[observer])
""" """
from dataclasses import dataclass, field import time
from typing import Dict, List, Optional, Tuple, Type from typing import Dict, List, Optional, Tuple, Type
from pydantic import BaseModel, Field
from pipecat.frames.frames import BotConnectedFrame, ClientConnectedFrame, StartFrame from pipecat.frames.frames import BotConnectedFrame, ClientConnectedFrame, StartFrame
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.base_pipeline import BasePipeline
@@ -47,42 +49,45 @@ from pipecat.processors.frame_processor import FrameProcessor
_INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline) _INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline)
@dataclass class ProcessorStartupTiming(BaseModel):
class ProcessorStartupTiming:
"""Startup timing for a single processor. """Startup timing for a single processor.
Parameters: Parameters:
processor_name: The name of the processor. 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. duration_secs: How long the processor's start() took, in seconds.
""" """
processor_name: str processor_name: str
start_time: float
duration_secs: float duration_secs: float
@dataclass class StartupTimingReport(BaseModel):
class StartupTimingReport:
"""Report of startup timings for all measured processors. """Report of startup timings for all measured processors.
Parameters: Parameters:
start_time: Unix timestamp when the first processor began starting.
total_duration_secs: Total wall-clock time from first to last processor start. total_duration_secs: Total wall-clock time from first to last processor start.
processor_timings: Per-processor timing data, in pipeline order. processor_timings: Per-processor timing data, in pipeline order.
""" """
start_time: float
total_duration_secs: float total_duration_secs: float
processor_timings: List[ProcessorStartupTiming] = field(default_factory=list) processor_timings: List[ProcessorStartupTiming] = Field(default_factory=list)
@dataclass class TransportTimingReport(BaseModel):
class TransportTimingReport:
"""Time from pipeline start to transport connection milestones. """Time from pipeline start to transport connection milestones.
Parameters: Parameters:
start_time: Unix timestamp of the StartFrame (pipeline start).
bot_connected_secs: Seconds from StartFrame to first BotConnectedFrame bot_connected_secs: Seconds from StartFrame to first BotConnectedFrame
(only set for SFU transports). (only set for SFU transports).
client_connected_secs: Seconds from StartFrame to first ClientConnectedFrame. client_connected_secs: Seconds from StartFrame to first ClientConnectedFrame.
""" """
start_time: float
bot_connected_secs: Optional[float] = None bot_connected_secs: Optional[float] = None
client_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). # Bot connected timing (stored for inclusion in the transport report).
self._bot_connected_secs: Optional[float] = None 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_startup_timing_report")
self._register_event_handler("on_transport_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: def _should_track(self, processor: FrameProcessor) -> bool:
"""Check if a processor should be tracked for timing. """Check if a processor should be tracked for timing.
@@ -212,6 +227,8 @@ class StartupTimingObserver(BaseObserver):
if self._start_frame_id is None: if self._start_frame_id is None:
self._start_frame_id = data.frame.id self._start_frame_id = data.frame.id
self._start_frame_arrival_ns = data.timestamp 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: elif data.frame.id != self._start_frame_id:
return return
@@ -263,6 +280,7 @@ class StartupTimingObserver(BaseObserver):
self._timings.append( self._timings.append(
ProcessorStartupTiming( ProcessorStartupTiming(
processor_name=processor.name, processor_name=processor.name,
start_time=self._mono_to_wall(arrival_ts),
duration_secs=duration_secs, duration_secs=duration_secs,
) )
) )
@@ -284,6 +302,7 @@ class StartupTimingObserver(BaseObserver):
delta_ns = data.timestamp - self._start_frame_arrival_ns delta_ns = data.timestamp - self._start_frame_arrival_ns
client_connected_secs = delta_ns / 1e9 client_connected_secs = delta_ns / 1e9
report = TransportTimingReport( report = TransportTimingReport(
start_time=self._mono_to_wall(self._start_frame_arrival_ns),
bot_connected_secs=self._bot_connected_secs, bot_connected_secs=self._bot_connected_secs,
client_connected_secs=client_connected_secs, client_connected_secs=client_connected_secs,
) )
@@ -296,8 +315,10 @@ class StartupTimingObserver(BaseObserver):
self._startup_timing_reported = True self._startup_timing_reported = True
total = sum(t.duration_secs for t in self._timings) 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( report = StartupTimingReport(
start_time=start_time,
total_duration_secs=total, total_duration_secs=total,
processor_timings=self._timings, processor_timings=self._timings,
) )

View File

@@ -151,9 +151,11 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
report = reports[0] report = reports[0]
self.assertIsInstance(report, StartupTimingReport) self.assertIsInstance(report, StartupTimingReport)
self.assertIsInstance(report.total_duration_secs, float) self.assertIsInstance(report.total_duration_secs, float)
self.assertGreater(report.start_time, 0)
for timing in report.processor_timings: for timing in report.processor_timings:
self.assertIsInstance(timing.processor_name, str) self.assertIsInstance(timing.processor_name, str)
self.assertIsInstance(timing.duration_secs, float) self.assertIsInstance(timing.duration_secs, float)
self.assertGreater(timing.start_time, 0)
async def test_excludes_internal_processors(self): async def test_excludes_internal_processors(self):
"""Test that internal pipeline processors are excluded by default.""" """Test that internal pipeline processors are excluded by default."""
@@ -211,6 +213,7 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
self.assertEqual(len(transport_reports), 1) self.assertEqual(len(transport_reports), 1)
report = transport_reports[0] report = transport_reports[0]
self.assertIsInstance(report, TransportTimingReport) self.assertIsInstance(report, TransportTimingReport)
self.assertGreater(report.start_time, 0)
self.assertGreater(report.client_connected_secs, 0) self.assertGreater(report.client_connected_secs, 0)
self.assertIsNone(report.bot_connected_secs) self.assertIsNone(report.bot_connected_secs)