127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
import wave
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import signal
|
|
import time
|
|
import utils
|
|
import typing
|
|
import subprocess
|
|
import sox
|
|
import re
|
|
|
|
# To mono audio
|
|
CHANNELS = 1
|
|
|
|
# Target rate is 16K
|
|
RATE = 48000
|
|
CHUNK = 1024
|
|
|
|
# Time limitation 300 seconds
|
|
TIME_LIMIT = 300
|
|
|
|
|
|
# Restart PyAudio
|
|
def restart_audio():
|
|
return
|
|
|
|
|
|
class AlsaRecorder:
|
|
def __init__(self, device_name: str, channels: int = 1, rate: int = RATE, fname: str = None):
|
|
self.channels = channels
|
|
self.rate = rate
|
|
self.device_name = device_name
|
|
self.fname = fname
|
|
|
|
|
|
def __exit__(self, exception, value, traceback):
|
|
self.stop_recording()
|
|
|
|
def close(self):
|
|
self.stop_recording()
|
|
|
|
def start_recording(self):
|
|
utils.log(f'Start recording with device name {self.device_name}, channels {self.channels}, samplerate {self.rate} to {self.fname}')
|
|
# /usr/bin/nice -n -5
|
|
cmd = f'/usr/bin/arecord -D {self.device_name} --format S16_LE --rate {self.rate} -c {self.channels} --buffer-size 262144 {self.fname}'
|
|
utils.log_verbose(cmd)
|
|
self.process_handle = subprocess.Popen(cmd.split())
|
|
return self
|
|
|
|
def stop_recording(self):
|
|
if self.process_handle:
|
|
try:
|
|
self.process_handle.send_signal(signal.SIGINT)
|
|
self.process_handle.wait(timeout=5.0)
|
|
except:
|
|
utils.log_error(f'/usr/bin/arecord timeout on exit')
|
|
|
|
self.process_handle = None
|
|
utils.log(f'ALSA recording stopped.')
|
|
return self
|
|
|
|
|
|
@classmethod
|
|
def find_default(cls) -> str:
|
|
return find_alsa_usb_device('arecord')
|
|
|
|
class AlsaPlayer:
|
|
def __init__(self, device_name: str, channels: int = 1, rate: int = RATE, fname: str = None):
|
|
self.channels = channels
|
|
self.rate = rate
|
|
self.device_name = device_name
|
|
self.fname = fname
|
|
|
|
|
|
def __exit__(self, exception, value, traceback):
|
|
self.stop_playing()
|
|
|
|
def close(self):
|
|
self.stop_playing()
|
|
|
|
def start_playing(self):
|
|
utils.log(f'Start playing with device name {self.device_name}, channels {self.channels}, samplerate {self.rate} from {self.fname}')
|
|
# /usr/bin/nice -n -5
|
|
cmd = f'/usr/bin/aplay -D {self.device_name} --format S16_LE --rate {self.rate} -c {self.channels} --buffer-size 128000 {self.fname}'
|
|
utils.log_verbose(cmd)
|
|
self.process_handle = subprocess.Popen(cmd.split())
|
|
return self
|
|
|
|
def stop_playing(self):
|
|
if self.process_handle:
|
|
try:
|
|
self.process_handle.send_signal(signal.SIGINT)
|
|
self.process_handle.wait(timeout=5.0)
|
|
except:
|
|
utils.log_error(f'/usr/bin/aplay timeout on exit')
|
|
self.process_handle = None
|
|
utils.log(f'ALSA playing stopped.')
|
|
return self
|
|
|
|
@classmethod
|
|
def find_default(cls) -> str:
|
|
return find_alsa_usb_device('aplay')
|
|
|
|
# utility should aplay or arecord
|
|
def find_alsa_usb_device(utility: str) -> str:
|
|
retcode, aplay_output = subprocess.getstatusoutput(f'/usr/bin/{utility} -l')
|
|
if retcode != 0:
|
|
return None
|
|
|
|
# Parse data line by line
|
|
pattern = r'card\s(?P<card_id>\d+):(?P<card_name>.+)device\s(?P<device_id>\d+):(?P<device_name>.+)'
|
|
lines = aplay_output.splitlines()
|
|
|
|
for l in lines:
|
|
found = re.match(pattern, l)
|
|
if found:
|
|
if 'card_id' in found.groupdict() and 'card_name' in found.groupdict() and 'device_id' in found.groupdict() and 'device_name' in found.groupdict():
|
|
card_id = found.group('card_id')
|
|
card_name = found.group('card_name')
|
|
device_id = found.group('device_id')
|
|
device_name = found.group('device_name')
|
|
if 'usb' in card_name.lower() and 'usb' in device_name.lower():
|
|
return f'hw:{card_id},{device_id}'
|
|
|
|
return None
|