| import numpy as np | |
| from .init import pipe | |
| TASK = "transcribe" | |
| BATCH_SIZE = 16 | |
| class A2T: | |
| def __init__(self, mic): | |
| self.mic = mic | |
| def __transcribe(self, inputs, task: str = None): | |
| if inputs is None: | |
| print("Inputs None") | |
| transcribed_text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True) | |
| print("transcribed_text : ", transcribed_text) | |
| return transcribed_text["text"] | |
| def __preprocces(self, raw: np.ndarray): | |
| chunk = raw.astype(np.float32, order='C') / 32768.0 | |
| return chunk | |
| def predict(self): | |
| try: | |
| if self.mic is not None: | |
| chunk = self.mic.get_array_of_samples() | |
| chunk = np.array(chunk) | |
| audio = self.__preprocces(chunk) | |
| sampling_rate = self.mic.frame_rate | |
| print(f"audio : {audio} \n frame_rate : {sampling_rate} shape : {audio.shape}") | |
| else: | |
| return "please provide audio" | |
| if isinstance(audio , np.ndarray): | |
| inputs = {"sampling_rate": 16_000, "raw": audio} | |
| return self.__transcribe(inputs=inputs, task=TASK) | |
| else: | |
| return "Audio is not np array" | |
| except Exception as e: | |
| print("Predict error", e) | |
| return "Oops some kinda error" | |