-
Notifications
You must be signed in to change notification settings - Fork 0
/
shell.py
85 lines (76 loc) · 2.09 KB
/
shell.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
import imp
import struct
import subprocess
import sys
import threading
import traceback
sample_rate = 44100
channels = 1
export = None
if len(sys.argv) > 1:
export_samples = float(sys.argv.pop(1)) * sample_rate
export = sys.argv.pop(1)
assert len(sys.argv) == 1
if 0:
import pyaudio
p = pyaudio.PyAudio()
#print(p.get_device_count())
#for i in range(p.get_device_count()):
# print(p.get_device_info_by_index(i))
stream = p.open(format=pyaudio.paInt16,
output_device_index=1,
channels=channels,
rate=sample_rate,
output=True)
else:
args = ['sox', '-q', '-r', str(sample_rate), '-b', '16', '-e',
'signed-integer', '-c', str(channels), '-t', 'raw',
'--buffer', '512', '-']
if export:
ftype = export.partition('.')[2]
args += ['-t', ftype, export]
else:
args += ['-d']
p = subprocess.Popen(args,
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
stream = p.stdin
class InputThread(threading.Thread):
def run(self):
global last_audio
while True:
try:
input('> ')
last_audio = audio
imp.reload(audio)
except EOFError:
sys.exit()
except:
traceback.print_exc()
continue
if not export:
tr = InputThread()
tr.daemon=True
tr.start()
import audio
ctx = audio.Context()
ctx.store('sample_rate', sample_rate)
sample = 0
last_audio = audio
while not export or sample < export_samples:
try:
sample += 1
ctx.store('sample', sample)
for channel in range(channels):
ctx.reset()
ctx.store('channel', channel)
value = int(audio.eq.eval(ctx) * (65535 / 20.))
# Hard clipping? Hard clipping.
value = max(-32768, min(32767, value))
stream.write(struct.pack('h', value))
stream.flush()
except KeyboardInterrupt:
sys.exit()
except:
if audio == last_audio:
raise
audio = last_audio