| 3 | import wave |
| 4 | |
| 5 | class VoiceRecorder: |
| 6 | def __init__(self, root_path): |
| 7 | |
| 8 | self.root_path = root_path # path to save cache recording file |
| 9 | self.audio = pyaudio.PyAudio() |
| 10 | self.stream = None |
| 11 | self.chunks = [] #.pcm data byte chunks |
| 12 | self.audio_format = pyaudio.paInt16 |
| 13 | self.rate = 16000 |
| 14 | self.channels = 1 |
| 15 | |
| 16 | def start(self): |
| 17 | print("start recording!") |
| 18 | self.stream = self.audio.open( |
| 19 | format= self.audio_format, |
| 20 | channels= self.channels, |
| 21 | rate=self.rate, |
| 22 | input=True, |
| 23 | frames_per_buffer=1280 |
| 24 | ) |
| 25 | self.chunks = [] |
| 26 | |
| 27 | def stop(self): |
| 28 | print("stop recording!") |
| 29 | if self.stream: |
| 30 | self.stream.stop_stream() |
| 31 | self.stream.close() |
| 32 | self.stream = None |
| 33 | file_path = os.path.join(self.root_path,"realtime.pcm") |
| 34 | self.save_to_pcm(file_path,self.chunks) |
| 35 | file_path = os.path.join(self.root_path,"realtime.wav") |
| 36 | self.save_to_wav(file_path, self.chunks) |
| 37 | |
| 38 | def read_audio(self): |
| 39 | if not self.stream: |
| 40 | raise Exception("Stream not open. Call start() first.") |
| 41 | while True: |
| 42 | data = self.stream.read(1280) |
| 43 | yield data |
| 44 | |
| 45 | def save_to_pcm(self, filename, frames): |
| 46 | with open(filename, "wb") as f: |
| 47 | f.write(b"".join(frames)) |
| 48 | |
| 49 | def save_to_wav(self,filename,frames): |
| 50 | wf = wave.open(filename, 'wb') |
| 51 | wf.setnchannels(self.channels) |
| 52 | sampwidth = self.audio.get_sample_size(self.audio_format) |
| 53 | print("sampwidth --->", sampwidth) |
| 54 | wf.setsampwidth(sampwidth) |
| 55 | wf.setframerate(self.rate) |
| 56 | wf.writeframes(b''.join(frames)) |
| 57 | wf.close() |
nothing calls this directly
no outgoing calls
no test coverage detected