Using pescador to sample random chunks from spectrograms (saved as npy matrices)

pescador_sampling.py

import numpy as np
import data
import pescador
import random


def excerpt_generator(
    dataset, idx,
    ex_length=100, ex_hop=100,
    shuffle=True, rnd_ex=False, seed=42
):
    X, Y = dataset[idx]
    nb_frames, nb_bins, nb_channels = X.shape

    # get all t indices and shuffle, make sure that excerpt is shorter than
    # number of frames
    if ex_length < nb_frames:
        steps = np.arange(0, nb_frames - ex_length, ex_hop)
    else:
        steps = [0]
        ex_length = nb_frames

    if shuffle:
        np.random.seed(seed)
        np.random.shuffle(steps)

    for s in steps:
        if rnd_ex:
            s = np.random.randint(0, X.shape[0] - ex_length)
        # sample spectrograms
        cur_X = X[s:s+ex_length, ...]
        cur_Y = Y[s:s+ex_length, ...]
        yield dict(X=cur_X, Y=cur_Y)


def mixmux(
    dataset,
    batch_size,
    ex_length=100,
    ex_hop=100,
    rnd_ex=False,
    shuffle=True,
    rnd_track=False,
    seed=42,
    active_streamers=100,
):
    streams = []
    for idx in range(len(dataset)):
        s = pescador.Streamer(
            excerpt_generator,
            dataset, idx,
            ex_length=ex_length,
            ex_hop=ex_hop,
            shuffle=shuffle,
            rnd_ex=rnd_ex,
            seed=seed,
        )
        streams.append(s)

    if not shuffle:
        mux = pescador.ChainMux(streams, mode='exhaustive')
    else:
        random.seed(seed)
        # shuffle the streams, just in case
        random.shuffle(streams)

        if rnd_track:
            mux = pescador.RoundRobinMux(
                streams,
                mode='exhaustive',
                random_state=seed
            )
        else:
            mux = pescador.StochasticMux(
                streams,
                active_streamers,
                rate=None,
                mode='exhaustive',
                random_state=seed
            )

    # get `batch_size` samples from the mux
    batches = pescador.Streamer(pescador.buffer_stream, mux, batch_size)

    return batches