VA-Control-for-linux/reference/decompiled/record.py

# uncompyle6 version 3.9.3
# Python bytecode version base 2.6 (62161)
# Decompiled from: Python 3.9.25 (main, Oct 31 2025, 23:16:49) 
# [GCC 14.2.0]
# Embedded file name: record.pyc
# Compiled at: 2022-04-20 07:08:45
if __name__ == '__main__':
    import main
    main.run()
import multiprocessing
from array import array
import pyaudio, numpy as np
from numpy.lib import stride_tricks
from scipy.ndimage.filters import gaussian_filter
import mytime, traceback, sys, math, threading
if __name__ == '__main__':

    class _app(object):

        def __init__(self):
            self.application_stopping = False
            return


    class _one_unit(object):

        def __init__(self):
            self.app = _app()
            return


    one_unit = _one_unit()
else:
    import one_unit
CHUNK_SIZE = 512
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 48000
DEFAULT_SMOOTHING = 0.33
factor = 0.01
testfreqs = [_[1] for i in range(330)]

class _rtadata(object):

    def __init__(self):
        self.avg_data = []
        self.active = -100
        self.recording = False
        self.input = None
        return


def reset_rta_avg():
    rtadata.avg_data = []
    return


rtadata = _rtadata()

def open_stream():
    try:
        p = pyaudio.PyAudio()
        info = p.get_host_api_info_by_index(0)
        numdevices = info.get('deviceCount')
        print mytime.displayTime() + ' Found audio input devices:'
        for i in range(0, numdevices):
            if p.get_device_info_by_host_api_device_index(0, i).get('maxInputChannels') > 0:
                name = p.get_device_info_by_host_api_device_index(0, i).get('name')
                print mytime.displayTime() + ' Input Device id', i, ':', name
                one_unit.app.available_audio_devices[i] = name

        if len(one_unit.app.available_audio_devices.keys()) == 0:
            print mytime.displayTime() + ' No available audio devices!'
            return (None, None)
        input_device = one_unit.app.audio_input_device
        input_device_index = None
        for i in one_unit.app.available_audio_devices.keys():
            dev = one_unit.app.available_audio_devices[i]
            if dev == input_device:
                input_device_index = i
                break

        if input_device_index == None:
            print mytime.displayTime() + ' Input device', one_unit.app.audio_input_device, 'not available, reverting to default'
            input_device_index = one_unit.app.available_audio_devices.keys()[0]
            one_unit.app.audio_input_device = one_unit.app.available_audio_devices[input_device_index]
        print mytime.displayTime() + ' Selected input device', one_unit.app.available_audio_devices[input_device_index]
        rtadata.input = one_unit.app.available_audio_devices[input_device_index]
        stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, output=True, input_device_index=input_device_index, frames_per_buffer=CHUNK_SIZE)
        print mytime.clock(), 'Opened audio stream for RTA'
        return (stream, p)
    except:
        if one_unit.app.mijnpc == True:
            traceback.print_exc(file=sys.stdout)
            print mytime.clock(), 'Failed to open RTA audio stream'
        return (None, None)

    return


def record(stream, size):
    rtadata.recording = True
    data = array('h')
    for i in range(0, size / CHUNK_SIZE):
        chunk = array('h', stream.read(CHUNK_SIZE))
        data.extend(chunk)

    rtadata.recording = False
    return data[:size]


def stft(sig, frameSize, overlapFac=0.5, window=np.hamming):
    win = window(frameSize)
    hopSize = int(frameSize - np.floor(overlapFac * frameSize))
    fs = frameSize / 2.0
    fs = int(np.floor(fs))
    fsl = np.zeros(fs)
    samples = np.append(fsl, sig)
    cols = int(np.ceil((len(samples) - frameSize) / float(hopSize)) + 1)
    samples = np.append(samples, np.zeros(frameSize))
    mstrides = (
     samples.strides[0] * hopSize, samples.strides[0])
    fs = int(frameSize)
    frames = stride_tricks.as_strided(samples, shape=(cols, fs), strides=mstrides).copy()
    frames *= win
    return np.fft.rfft(frames)


def makefft(stream):
    size = one_unit.app.FFT_len
    start_time = mytime.clock()
    samples = None
    for i in range(3):
        try:
            samples = record(stream, size)
            break
        except:
            pass

    if samples == None:
        if one_unit.app.mijnpc == True:
            print mytime.displayTime() + ' Failed to read RTA stream'
        res = [
         0.0] * size
        rtadata.avg_data = res
        return (
         res, False)
    else:
        res = abs(stft(samples, size * 2))[0]
        if len(rtadata.avg_data) == len(res):
            res = (rtadata.avg_data * (one_unit.app.rta_time_avg - 1) + res) / one_unit.app.rta_time_avg
        rtadata.avg_data = res
        return (res, True)


def resample_fft(data, outbins, avg=None, smoothing=DEFAULT_SMOOTHING):
    in_size = len(data)
    try:
        fft_scale = math.sqrt((len(data) - 1) / 4096.0)
    except:
        fft_scale = 1.0

    out_size = len(outbins)
    out = [1e-30] * out_size
    if in_size == 0:
        return (out, avg)
    else:
        freqstep = float(RATE) / float(in_size) / 2
        bin = 0
        num = 0.0
        for i in range(in_size):
            freq = freqstep * i
            num += 1.0
            df = 0.0115 * freq
            if freq > outbins[bin] + df:
                out[bin] /= num
                out[bin] /= fft_scale
                num = 0.0
                bin += 1
                if bin < out_size:
                    while freq > outbins[bin] + df:
                        out[bin] = out[bin - 1]
                        bin += 1
                        if bin >= out_size:
                            break

            if bin < out_size:
                out[bin] += data[i]
            else:
                break

        if smoothing > 0.0:
            f_smoothing = smoothing * (out_size / 20)
            out = gaussian_filter(out, f_smoothing)
        f_low = 0
        f_high = 0
        for i in range(out_size):
            if outbins[i] < 250:
                f_low = i
            if outbins[i] > 10000:
                f_high = i
                break

        out = 20.0 * np.log10(out)
        if avg == None:
            avg = np.mean(out[f_low:f_high])
            pk = np.amax(out)
            if pk > 15:
                diff = pk - 15
                if diff > avg:
                    avg = diff
        out -= avg
        return (
         out, avg)


def process():
    stream = None
    p = None
    while one_unit.app.application_stopping == False:
        if one_unit.app.rta_on == True:
            if stream == None:
                (stream, p) = open_stream()
                mytime.sleep(0.01)
                continue
            (rtadata.res, allIsOK) = makefft(stream)
            if allIsOK == False:
                (stream, p) = open_stream()
            one_unit.app.rta_new_data = True
            if rtadata.input != one_unit.app.audio_input_device:
                stream.stop_stream()
                stream.close()
                p.terminate()
                stream = None
                print mytime.clock(), 'Closed RTA stream'
                mytime.sleep(0.1)
        elif stream != None and rtadata.recording == False:
            stream.stop_stream()
            stream.close()
            p.terminate()
            stream = None
            print mytime.clock(), 'Closed RTA stream'
        mytime.sleep(0.01)

    try:
        stream.stop_stream()
        stream.close()
        print mytime.clock(), 'Closed RTA stream'
    except:
        pass

    try:
        p.terminate()
        print mytime.clock(), 'Terminated audio handle'
    except:
        pass

    print mytime.clock(), 'Closed RTA thread'
    return


def getfft():
    rtadata.active = mytime.clock()
    return rtadata.avg_data


recording_thread = threading.Thread(name='ALLDSP RTA', target=process)
recording_thread.start()
if __name__ == '__main__':
    i = 0
    while i < 30:
        data = getfft(testfreqs)
        mytime.sleep(0.05)
        i += 1

    one_unit.app.application_stopping = True

# okay decompiling pycode/record.pyc