algorithms/asa.py
def _asa_task(q, masks, stft, sample_width, frame_rate, nsamples_for_each_fft):
    """
    Worker for the ASA algorithm's multiprocessing step.
    """
    # Convert each mask to (1 or 0) rather than (ID or 0)
    for mask in masks:
        mask = np.where(mask > 0, 1, 0)

    # Multiply the masks against STFTs
    masks = [mask * stft for mask in masks]

    nparrs = []
    dtype_dict = {1: np.int8, 2: np.int16, 4: np.int32}
    dtype = dtype_dict[sample_width]
    for m in masks:
        _times, nparr = signal.istft(m, frame_rate, nperseg=nsamples_for_each_fft)
        nparr = nparr.astype(dtype)
        nparrs.append(nparr)

    for m in nparrs:
        q.put(m)
    q.put("DONE")
def _asa_task(q, masks, stft, sample_width, frame_rate, nsamples_for_each_fft):
    """
    Worker for the ASA algorithm's multiprocessing step.
    """
    # Convert each mask to (1 or 0) rather than (ID or 0)
    for mask in masks:
        mask = np.where(mask > 0, 1, 0)

    # Multiply the masks against STFTs
    masks = [mask * stft for mask in masks]

    nparrs = []
    dtype_dict = {1: np.int8, 2: np.int16, 4: np.int32}
    dtype = dtype_dict[sample_width]
    for m in masks:
        _times, nparr = signal.istft(m, frame_rate, nperseg=nsamples_for_each_fft)
        nparr = nparr.astype(dtype)
        nparrs.append(nparr)

    for m in nparrs:
        q.put(m)
    q.put("DONE")
Worker for the ASA algorithm's multiprocessing step.