-
Notifications
You must be signed in to change notification settings - Fork 0
/
wireFFT.py
102 lines (86 loc) · 3.59 KB
/
wireFFT.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
97
98
99
100
101
102
#!/usr/bin/env python3
"""Pass input directly to output.
https://app.assembla.com/spaces/portaudio/git/source/master/test/patest_wire.c
"""
import argparse
import sounddevice as sd
import numpy # Make sure NumPy is loaded before it is used in the callback
assert numpy # avoid "imported but unused" message (W0611)
def int_or_str(text):
"""Helper function for argument parsing."""
try:
return int(text)
except ValueError:
return text
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument(
'-l', '--list-devices', action='store_true',
help='show list of audio devices and exit')
args, remaining = parser.parse_known_args()
if args.list_devices:
print(sd.query_devices())
parser.exit(0)
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
parents=[parser])
parser.add_argument(
'-i', '--input-device', type=int_or_str,
help='input device (numeric ID or substring)')
parser.add_argument(
'-o', '--output-device', type=int_or_str,
help='output device (numeric ID or substring)')
parser.add_argument(
'-c', '--channels', type=int, default=2,
help='number of channels')
parser.add_argument('--dtype', help='audio data type')
parser.add_argument('--samplerate', type=float, help='sampling rate')
parser.add_argument('--blocksize', type=int, help='block size', default=1024)
parser.add_argument('--latency', type=float, help='latency in seconds')
args = parser.parse_args(remaining)
first = True
parser.parse_args() # needed to prevent blocksize from being 'none' in overlap computation
blocksize = args.blocksize
overlap = blocksize // 2
# create global areas for buffering and processing of channel data
inbuffer = numpy.zeros([blocksize * 3, 2])
outbuffer = numpy.zeros([blocksize + overlap * 2, 2])
timedata = numpy.zeros([blocksize + overlap * 2, 2])
channel = numpy.zeros([blocksize + overlap * 2]) # needs to be power of 2
mask = numpy.hanning(blocksize + overlap * 2) # blackman is another possibility
def callback(indata, outdata, frames, time, status):
global first, blocksize, overlap, inbuffer, outbuffer, mask, timedata, channel
if status:
print(status)
# because of buffering, we introduce a delay of 3 reads before output
# is clean.
inbuffer[blocksize * 2:] = indata # append to inbuffer
timedata = inbuffer[blocksize - overlap:blocksize * 2 + overlap, :]
# expand with loop to do all channels
# replace timedata with inbuffer[blocksize - overlap:blocksize * 2 + overlap, :]?
channel[:] = timedata[:, 0] * mask
# do fft
freqdata = numpy.fft.rfft(channel)
# do filtering here
channel = numpy.fft.irfft(freqdata)
outbuffer[:, 0] += channel
# end of loop for channels
inbuffer[:-blocksize] = inbuffer[blocksize:] # left shift inbuffer
inbuffer[-blocksize:] = 0 # not really needed for input buffer as replacement is done
outbuffer[:-overlap] = outbuffer[overlap:] # left shift outbuffer
outbuffer[-overlap:] = 0 # here, we do a add to the overlap and this zeroed area
outdata[:] = outbuffer[:blocksize]
first = False
try:
with sd.Stream(device=(args.input_device, args.output_device),
samplerate=args.samplerate, blocksize=args.blocksize,
dtype=args.dtype, latency=args.latency,
channels=args.channels, callback=callback):
print('#' * 80)
print('press Return to quit')
print('#' * 80)
input()
except KeyboardInterrupt:
parser.exit('')
except Exception as e:
parser.exit(type(e).__name__ + ': ' + str(e))