(websocket: WebSocket)
| 142 | |
| 143 | @app.websocket("/audio") |
| 144 | async def websocket_endpoint(websocket: WebSocket): |
| 145 | await websocket.accept() |
| 146 | try: |
| 147 | while True: |
| 148 | data = await websocket.receive_text() |
| 149 | audio_data = np.frombuffer( |
| 150 | base64.b64decode(data.split(",")[1]), |
| 151 | dtype=np.int16 |
| 152 | ) |
| 153 | audio_data = audio_data.astype(np.float32) / 32767.0 |
| 154 | processed_audio = await audio_processor.process_audio(audio_data) |
| 155 | processed_audio = (processed_audio * 32767).astype(np.int16) |
| 156 | |
| 157 | processed_data = base64.b64encode(processed_audio.tobytes()).decode('utf-8') |
| 158 | await websocket.send_text(f"data:audio/raw;base64,{processed_data}") |
| 159 | |
| 160 | except Exception as e: |
| 161 | print_colored(f"WebSocket error: {e}", "red") |
| 162 | print_colored(f"Full traceback:\n{traceback.format_exc()}", "red") |
| 163 | finally: |
| 164 | audio_processor.cleanup() |
| 165 | await websocket.close() |
| 166 | |
| 167 | |
| 168 | audio_processor = AudioProcessor(model=model, prompt_path=args.prompt_path) |
nothing calls this directly
no test coverage detected