Handle key press events.
(self, event: events.Key)
| 257 | stream.close() |
| 258 | |
| 259 | async def on_key(self, event: events.Key) -> None: |
| 260 | """Handle key press events.""" |
| 261 | if event.key == "enter": |
| 262 | self.query_one(Button).press() |
| 263 | return |
| 264 | |
| 265 | if event.key == "q": |
| 266 | self.exit() |
| 267 | return |
| 268 | |
| 269 | if event.key == "k": |
| 270 | status_indicator = self.query_one(AudioStatusIndicator) |
| 271 | if status_indicator.is_recording: |
| 272 | self.should_send_audio.clear() |
| 273 | status_indicator.is_recording = False |
| 274 | |
| 275 | if self.session and self.session.turn_detection is None: |
| 276 | # The default in the API is that the model will automatically detect when the user has |
| 277 | # stopped talking and then start responding itself. |
| 278 | # |
| 279 | # However if we're in manual `turn_detection` mode then we need to |
| 280 | # manually tell the model to commit the audio buffer and start responding. |
| 281 | conn = await self._get_connection() |
| 282 | await conn.input_audio_buffer.commit() |
| 283 | await conn.response.create() |
| 284 | else: |
| 285 | self.should_send_audio.set() |
| 286 | status_indicator.is_recording = True |
| 287 | |
| 288 | |
| 289 | if __name__ == "__main__": |
nothing calls this directly
no test coverage detected