123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100 |
- 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))
|