| 17 | from streamlit_webrtc import WebRtcMode, webrtc_streamer |
| 18 | |
| 19 | class AudioClient: |
| 20 | def __init__(self, server_url="ws://localhost:8000", token_temp=None, categorical_temp=None, gaussian_temp=None): |
| 21 | # Convert ws:// to http:// for the base URL |
| 22 | self.base_url = server_url.replace("ws://", "http://") |
| 23 | self.server_url = f"{server_url}/audio" |
| 24 | self.sound_check = False |
| 25 | |
| 26 | # Set temperatures if provided |
| 27 | if any(t is not None for t in [token_temp, categorical_temp, gaussian_temp]): |
| 28 | response_message = self.set_temperature_and_echo(token_temp, categorical_temp, gaussian_temp) |
| 29 | print(response_message) |
| 30 | |
| 31 | self.downsampler = torchaudio.transforms.Resample(STREAMING_SAMPLE_RATE, SAMPLE_RATE) |
| 32 | self.upsampler = torchaudio.transforms.Resample(SAMPLE_RATE, STREAMING_SAMPLE_RATE) |
| 33 | self.ws = None |
| 34 | self.in_buffer = None |
| 35 | self.out_buffer = None |
| 36 | |
| 37 | def set_temperature_and_echo(self, token_temp=None, categorical_temp=None, gaussian_temp=None, echo_testing = False): |
| 38 | """Send temperature settings to server""" |
| 39 | params = {} |
| 40 | if token_temp is not None: |
| 41 | params['token_temp'] = token_temp |
| 42 | if categorical_temp is not None: |
| 43 | params['categorical_temp'] = categorical_temp |
| 44 | if gaussian_temp is not None: |
| 45 | params['gaussian_temp'] = gaussian_temp |
| 46 | |
| 47 | response = requests.post(f"{self.base_url}/set_temperature", params=params) |
| 48 | response_message = response.json()['message'] |
| 49 | return response_message |
| 50 | |
| 51 | def _resample(self, audio_data: np.ndarray, resampler: torchaudio.transforms.Resample) -> np.ndarray: |
| 52 | audio_data = audio_data.astype(np.float32) / 32767.0 |
| 53 | audio_data = resampler(torch.tensor(audio_data)).numpy() |
| 54 | audio_data = (audio_data * 32767.0).astype(np.int16) |
| 55 | return audio_data |
| 56 | |
| 57 | def upsample(self, audio_data: np.ndarray) -> np.ndarray: |
| 58 | return self._resample(audio_data, self.upsampler) |
| 59 | |
| 60 | def downsample(self, audio_data: np.ndarray) -> np.ndarray: |
| 61 | return self._resample(audio_data, self.downsampler) |
| 62 | |
| 63 | def from_s16_format(self, audio_data: np.ndarray, channels: int) -> np.ndarray: |
| 64 | if channels == 2: |
| 65 | audio_data = audio_data.reshape(-1, 2).T |
| 66 | else: |
| 67 | audio_data = audio_data.reshape(-1) |
| 68 | return audio_data |
| 69 | |
| 70 | def to_s16_format(self, audio_data: np.ndarray): |
| 71 | if len(audio_data.shape) == 2 and audio_data.shape[0] == 2: |
| 72 | audio_data = audio_data.T.reshape(1, -1) |
| 73 | elif len(audio_data.shape) == 1: |
| 74 | audio_data = audio_data.reshape(1, -1) |
| 75 | return audio_data |
| 76 |
no outgoing calls
no test coverage detected