PyTrickle data channels

from pytrickle import TrickleSubscriber
import json

# Subscribe to the data output channel
data_sub = TrickleSubscriber(url=job.events_url)

async for segment in data_sub:
    payload = json.loads(segment.data.decode('utf-8'))

    if payload.get('type') == 'transcription':
        print(f"[{payload['timestamp']:.1f}s] {payload['text']}")
    elif payload.get('type') == 'detection':
        for box in payload['boxes']:
            print(f"  {box['label']}: ({box['x']}, {box['y']}, {box['w']}, {box['h']})")

from pytrickle import FrameProcessor, VideoFrame

class TranscriptionPipeline(FrameProcessor):
    async def process_frame(self, frame: VideoFrame) -> VideoFrame:
        # Process video as normal
        result = self.model(frame.data)

        # Write transcription to data channel
        await self.publish_data({
            'type': 'transcription',
            'text': result.text,
            'timestamp': frame.pts_time,
            'confidence': result.confidence,
        })

        return VideoFrame(data=result.video, pts=frame.pts)