py-twinmic-aec.py

import threading
import logging

from numpy.fft import fft, ifft
import numpy as np
import jack


def denoice(left, right):
    fft_l = fft(left)
    fft_r = fft(right)
    K = np.abs(np.arctan2(fft_l.imag, fft_l.real) - np.arctan2(fft_r.imag, fft_r.real))
    K = np.clip(1 - (K * 2) ** 2, 0, 1)
    return ifft(fft_l * K).real, ifft(fft_l * K).real


def autoconnect(client):
    capture = client.get_ports(is_physical=True, is_output=True)
    if not capture:
        raise RuntimeError('No physical capture ports')
    for src, dest in zip(capture, client.inports):
        client.connect(src, dest)
    playback = client.get_ports(is_physical=True, is_input=True)
    if not playback:
        raise RuntimeError('No physical playback ports')
    for src, dest in zip(client.outports, playback):
        client.connect(src, dest)


def run():

    shutdown_event = threading.Event()

    client = jack.Client("py-twinmic-aec", no_start_server=True)

    client.inports.register('input_1')
    client.inports.register('input_2')
    out1 = client.outports.register('output_1')
    out2 = client.outports.register('output_2')

    @client.set_xrun_callback
    def xrun(delay):
        logging.warning("xrun delay=%s", delay)

    @client.set_shutdown_callback
    def shutdown(status, reason):
        logging.info("shutdown status=%r reason=%r", status, reason)
        shutdown_event.set()

    @client.set_process_callback
    def process(frames):
        out1.get_array()[:], out2.get_array()[:] = denoice(*[i.get_array() for i in client.inports])

    autoconnect(client)

    with client:
        try:
            shutdown_event.wait()
        except KeyboardInterrupt:
            pass


if __name__ == "__main__":
    logging.basicConfig(level="INFO", format="[%(asctime)s] %(levelname)s | %(message)s")
    run()