-
Notifications
You must be signed in to change notification settings - Fork 0
/
audiomanager.py
96 lines (77 loc) · 3.86 KB
/
audiomanager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import audioanalyzer as aa
import audiostreamer as ast
class AudioManager:
def __init__(self, analyzer_blocksize):
self.audiobuffer = None
self.streamer = None
self.analyzer = None
self.analyzer_blocksize = analyzer_blocksize
def set_play_wav_file(self, filename):
self.audiobuffer = ast.WavBuffer(filename, daemon=True)
self.samplerate = self.audiobuffer.samplerate
self.channel = self.audiobuffer.channel
self.dtype = ast.dic_datawidth_to_numpy_dtype[self.audiobuffer.width]
self.streamer = ast.AudioStreamer(samplerate=self.samplerate,
channel=self.channel,
datatype=self.dtype,
input_flag=False,
daemon=True)
self.analyzer = aa.AudioAnalyzer(blocksize=self.analyzer_blocksize,
overlap=1024,
channel=self.channel,
daemon=True)
self.audiobuffer.register_queue(self.streamer.in_queue)
self.streamer.register_queue(self.analyzer.in_queue)
def set_play_audiodata(self, datablock, samplerate, channel, dtype):
self.dtype = dtype
self.samplerate = samplerate
self.channel = channel
self.audiobuffer = ast.AudioBuffer(datablock, daemon=True)
self.streamer = ast.AudioStreamer(samplerate=self.samplerate,
channel=self.channel,
datatype=self.dtype,
input_flag=False,
daemon=True)
self.analyzer = aa.AudioAnalyzer(blocksize=self.analyzer_blocksize,
channel=self.channel,
daemon=True)
self.audiobuffer.register_queue(self.streamer.in_queue)
self.streamer.register_queue(self.analyzer.in_queue)
def set_record(self, samplerate, channel, dtype, filename):
self.dtype = dtype
self.audiobuffer = None
self.samplerate = samplerate
self.channel = channel
self.streamer = ast.AudioStreamer(samplerate=self.samplerate,
channel=self.channel,
datatype=self.dtype,
input_flag=True,
filename=filename,
daemon=True)
self.analyzer = aa.AudioAnalyzer(blocksize=self.analyzer_blocksize,
channel=self.channel,
daemon=True)
self.streamer.register_queue(self.analyzer.in_queue)
def start(self):
if(self.streamer is not None):
self.streamer.start()
if(self.audiobuffer is not None):
self.audiobuffer.start()
if(self.analyzer is not None):
self.analyzer.start()
def stop(self):
if((self.audiobuffer is not None) and (self.audiobuffer.is_running)):
self.audiobuffer.immediate_join()
if((self.streamer is not None) and (self.streamer.is_running)):
self.streamer.immediate_join()
if((self.analyzer is not None) and (self.analyzer.is_running)):
self.analyzer.immediate_join()
def is_running(self):
result_flag = False
if(self.audiobuffer is not None):
result_flag = self.audiobuffer.is_running()
if(self.streamer is not None):
result_flag = result_flag or self.streamer.is_running()
if(self.analyzer is not None):
result_flag = result_flag or self.analyzer.is_running()
return result_flag