Use _ArrivalInfo dataclass instead of tuple for arrival tracking
This commit is contained in:
@@ -35,6 +35,7 @@ Example::
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import time
|
import time
|
||||||
|
from dataclasses import dataclass
|
||||||
from typing import Dict, List, Optional, Tuple, Type
|
from typing import Dict, List, Optional, Tuple, Type
|
||||||
|
|
||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
@@ -49,6 +50,14 @@ from pipecat.processors.frame_processor import FrameProcessor
|
|||||||
_INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline)
|
_INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class _ArrivalInfo:
|
||||||
|
"""Internal record of when a StartFrame arrived at a processor."""
|
||||||
|
|
||||||
|
processor: FrameProcessor
|
||||||
|
arrival_ts_ns: int
|
||||||
|
|
||||||
|
|
||||||
class ProcessorStartupTiming(BaseModel):
|
class ProcessorStartupTiming(BaseModel):
|
||||||
"""Startup timing for a single processor.
|
"""Startup timing for a single processor.
|
||||||
|
|
||||||
@@ -161,8 +170,8 @@ class StartupTimingObserver(BaseObserver):
|
|||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
self._processor_types = processor_types
|
self._processor_types = processor_types
|
||||||
|
|
||||||
# Map processor ID -> (processor, arrival_timestamp_ns)
|
# Map processor ID -> arrival info.
|
||||||
self._arrivals: Dict[int, Tuple[FrameProcessor, int]] = {}
|
self._arrivals: Dict[int, _ArrivalInfo] = {}
|
||||||
|
|
||||||
# Collected timings in pipeline order.
|
# Collected timings in pipeline order.
|
||||||
self._timings: List[ProcessorStartupTiming] = []
|
self._timings: List[ProcessorStartupTiming] = []
|
||||||
@@ -234,7 +243,9 @@ class StartupTimingObserver(BaseObserver):
|
|||||||
return
|
return
|
||||||
|
|
||||||
if self._should_track(data.processor):
|
if self._should_track(data.processor):
|
||||||
self._arrivals[data.processor.id] = (data.processor, data.timestamp)
|
self._arrivals[data.processor.id] = _ArrivalInfo(
|
||||||
|
processor=data.processor, arrival_ts_ns=data.timestamp
|
||||||
|
)
|
||||||
|
|
||||||
async def on_push_frame(self, data: FramePushed):
|
async def on_push_frame(self, data: FramePushed):
|
||||||
"""Record when a StartFrame leaves a processor and compute the delta.
|
"""Record when a StartFrame leaves a processor and compute the delta.
|
||||||
@@ -266,15 +277,13 @@ class StartupTimingObserver(BaseObserver):
|
|||||||
if arrival is None:
|
if arrival is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
processor, arrival_ts = arrival
|
duration_ns = data.timestamp - arrival.arrival_ts_ns
|
||||||
duration_ns = data.timestamp - arrival_ts
|
|
||||||
duration_secs = duration_ns / 1e9
|
duration_secs = duration_ns / 1e9
|
||||||
|
start_offset_secs = (arrival.arrival_ts_ns - self._start_frame_arrival_ns) / 1e9
|
||||||
start_offset_secs = (arrival_ts - self._start_frame_arrival_ns) / 1e9
|
|
||||||
|
|
||||||
self._timings.append(
|
self._timings.append(
|
||||||
ProcessorStartupTiming(
|
ProcessorStartupTiming(
|
||||||
processor_name=processor.name,
|
processor_name=arrival.processor.name,
|
||||||
start_offset_secs=start_offset_secs,
|
start_offset_secs=start_offset_secs,
|
||||||
duration_secs=duration_secs,
|
duration_secs=duration_secs,
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user