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):
    # 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 ) : # 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.
Worker for the ASA algorithm s multiprocessing step .