import pyaudio from collections import deque from array import array import numpy as np import requests import json import itertools from config.config import Config # fs = 16000 # seconds = 1 SERVER_URL = Config["server_url"] INFERENCE_STR = Config["inference_path"] TEST_STR = Config["test_path"] class DequeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, deque): return list(obj) return json.JSONEncoder.default(self, obj) class Record(object): fs = 16000 seconds = 1 stride = fs * seconds cnt = 0 def __init__(self): self.cumul_audio = deque([], maxlen = self.fs * 5) self.stream = pyaudio.PyAudio().open(format=pyaudio.paInt16, channels=1, rate=self.fs, input=True, output=True, frames_per_buffer=self.stride) def unit_test(self, output_func): self.cnt += 1 output_func(str(self.cnt)) def record_unit(self, output_func): snd_raw = self.stream.read(self.stride, exception_on_overflow=False) snd_data = array('h', snd_raw) audio = np.float32(snd_data)/32000 audio = audio.tolist() self.cumul_audio.extend(audio) payload = { "device":"0005", "time":"seconds", "recording":{ "filename":"output.mp3", "content_type":"audio/mp3", "content": list(self.cumul_audio) } } headers = {'Content-Type': 'application/json; charset=utf-8'} r = requests.post(SERVER_URL + INFERENCE_STR, data=json.dumps(payload), headers=headers) resJson = r.json() print('resJson = ', resJson) output_func(resJson['output']) def recording(output): while True: snd_raw = stream.read(stride, exception_on_overflow= False) snd_data = array('h', snd_raw) audio = np.float32(snd_data)/32000 audio = audio.tolist() cumul_audio.extend(audio) payload = { "device":"0005", "time":"seconds", "recording":{ "filename":"output.mp3", "content_type":"audio/mp3", "content": cumul_audio } } headers = {'Content-Type': 'application/json; charset=utf-8'} r = requests.post(SERVER_URL + TEST_STR, data=json.dumps(payload), headers=headers) # print('success recording !! ', list(itertools.islice(cumul_audio, 0, 10))) def record_chunk(): snd_raw = stream.read(stride, exception_on_overflow= False) snd_data = array('h', snd_raw) audio = np.float32(snd_data)/32000 audio = audio.tolist() cumul_audio.extend(audio) return list(itertools.islice(cumul_audio, 0, 10))