1+ #!/usr/bin/env python3
2+ """
3+ Video/Audio Passthrough with Text Publishing using StreamProcessor
4+
5+ This example demonstrates:
6+ - Video passthrough (no processing)
7+ - Audio passthrough (no processing)
8+ - Text publishing every 400 audio frames using simple text queue
9+ """
10+
11+ import logging
12+ import json
13+ import time
14+ from pytrickle import StreamProcessor
15+ from pytrickle .frames import VideoFrame , AudioFrame
16+ from typing import List
17+
18+ logging .basicConfig (level = logging .INFO )
19+ logger = logging .getLogger (__name__ )
20+
21+ # Global state
22+ audio_frame_count = 0
23+ text_publish_interval = 400
24+ ready = False
25+ start_time = None
26+ _stream_processor = None # Reference to StreamProcessor for text publishing
27+
28+ def load_model (** kwargs ):
29+ """Initialize processor state - called during model loading phase."""
30+ global text_publish_interval , ready , start_time
31+
32+ logger .info (f"load_model called with kwargs: { kwargs } )
33+
34+ # Set processor variables from kwargs or use defaults
35+ text_publish_interval = kwargs .get ('text_publish_interval' , 400 )
36+ text_publish_interval = max (1 , int (text_publish_interval ))
37+
38+ ready = True
39+ start_time = time .time ()
40+ logger .info (f"✅ Video/Audio passthrough with text publishing ready (interval: { text_publish_interval } )
41+
42+ async def process_video (frame : VideoFrame ) -> VideoFrame :
43+ """Pass through video frames unchanged."""
44+ global ready
45+
46+ if not ready :
47+ return frame
48+
49+ # Simply pass through the video frame without any processing
50+ return frame
51+
52+ async def process_audio (frame : AudioFrame ) -> List [AudioFrame ]:
53+ """Pass through audio frames and publish text data periodically."""
54+ global audio_frame_count , text_publish_interval , ready , start_time , _stream_processor
55+
56+ if not ready :
57+ return [frame ]
58+
59+ # Increment frame counter
60+ audio_frame_count += 1
61+
62+ # Check if we should publish text data
63+ if audio_frame_count % text_publish_interval == 0 :
64+ # Calculate elapsed time
65+ elapsed_time = time .time () - start_time
66+
67+ # Create JSONL data with audio processing statistics
68+ text_data = {
69+ "type" : "audio_stats" ,
70+ "timestamp" : time .time (),
71+ "elapsed_time_seconds" : round (elapsed_time , 2 ),
72+ "total_audio_frames" : audio_frame_count ,
73+ "frames_per_second" : round (audio_frame_count / elapsed_time , 2 ) if elapsed_time > 0 else 0 ,
74+ "frame_shape" : list (frame .samples .shape ) if hasattr (frame , 'samples' ) else None ,
75+ "sample_rate" : getattr (frame , 'sample_rate' , None ),
76+ "channels" : getattr (frame , 'channels' , None ),
77+ "message" : f"Processed { audio_frame_count } { elapsed_time :.2f}
78+ }
79+
80+ # Publish as JSONL - just add text to the queue!
81+ jsonl_line = json .dumps (text_data )
82+ if _stream_processor :
83+ await _stream_processor .publish_data_output (jsonl_line )
84+
85+ logger .info (f"📊 Published stats: { audio_frame_count } { elapsed_time :.2f} )
86+
87+ # Pass through the audio frame unchanged
88+ return [frame ]
89+
90+ def update_params (params : dict ):
91+ """Update text publishing interval."""
92+ global text_publish_interval
93+
94+ if "text_publish_interval" in params :
95+ old = text_publish_interval
96+ text_publish_interval = max (1 , int (params ["text_publish_interval" ]))
97+ if old != text_publish_interval :
98+ logger .info (f"Text publish interval: { old } { text_publish_interval } )
99+
100+ if "reset_counter" in params and params ["reset_counter" ]:
101+ global audio_frame_count , start_time
102+ audio_frame_count = 0
103+ start_time = time .time ()
104+ logger .info ("🔄 Reset audio frame counter and timer" )
105+
106+
107+ # Create and run StreamProcessor
108+ if __name__ == "__main__" :
109+ processor = StreamProcessor (
110+ video_processor = process_video ,
111+ audio_processor = process_audio ,
112+ model_loader = load_model ,
113+ param_updater = update_params ,
114+ name = "passthrough-with-text" ,
115+ port = 8000
116+ )
117+
118+ # Set reference for text publishing
119+ _stream_processor = processor
120+
121+ logger .info ("🚀 Starting passthrough processor with text publishing..." )
122+ logger .info (f"📝 Will publish JSONL stats every { text_publish_interval } )
123+ logger .info ("🔧 Update parameters via /api/update_params:" )
124+ logger .info (" - text_publish_interval: number of frames between text publications" )
125+ logger .info (" - reset_counter: true to reset frame counter and timer" )
126+
127+ processor .run ()
0 commit comments