Améliorer le sous-échantillonnage min / max

J'ai réussi à améliorer les performances en utilisant la sortie de arg (min | max) directement pour indexer les tableaux de données. Cela se fait au prix d'un appel supplémentaire à np.sort mais l'axe à trier ne comporte que deux éléments (les indices min. / Max.) Et le tableau global est plutôt petit (nombre de bins ):

def min_max_downsample_v3(x, y, num_bins):
    pts_per_bin = x.size // num_bins

    x_view = x[:pts_per_bin*num_bins].reshape(num_bins, pts_per_bin)
    y_view = y[:pts_per_bin*num_bins].reshape(num_bins, pts_per_bin)
    i_min = np.argmin(y_view, axis=1)
    i_max = np.argmax(y_view, axis=1)

    r_index = np.repeat(np.arange(num_bins), 2)
    c_index = np.sort(np.stack((i_min, i_max), axis=1)).ravel()

    return x_view[r_index, c_index], y_view[r_index, c_index]

J'ai vérifié les horaires de votre exemple et j'ai obtenu:

• min_max_downsample_v1 : 110 ms ± 5 ms
• min_max_downsample_v2 : 240 ms ± 8,01 ms
• min_max_downsample_v3 : 164 ms ± 1,23 ms

J'ai également vérifié le retour directement après les appels à arg (min | max) et le résultat était également de 164 ms, c'est-à-dire qu'il n'y a plus de véritable surcharge après cela.

Donc, cela ne concerne pas l'accélération de la fonction spécifique en question, mais cela montre quelques façons de tracer une ligne avec un grand nombre de points de manière assez efficace. Cela suppose que les x points sont ordonnés et uniformément (ou presque uniformément) échantillonnés.

Configuration

L=int(100e6)
x=linspace(0,1,L)
y=0.1*randn(L)+sin(2*pi*18*x)
plot_bigly(x,y, red=calc_env)

Voici une fonction que j'aime qui réduit le nombre de points de manière aléatoire en choisir un dans chaque intervalle. Il n'est pas garanti d'afficher tous les pics des données, mais il n'a pas autant de problèmes que de décimer directement les données, et est rapide.

def plot_bigly(x, y, *, ax=None, M=5000, red=calc_env, **kwargs):
    """
    x : the x data
    y : the y data
    ax : axis to plot on
    M : The maximum number of line points to display at any given time
    kwargs : passed to line
    """
    assert x.shape == y.shape, "x and y data must have same shape!"
    if ax is None:
        ax = gca()

    cached = {}

    # Setup line to be drawn beforehand, note this doesn't increment line properties so
    #  style needs to be passed in explicitly
    line = plt.Line2D([],[], **kwargs)
    def update(xmin, xmax):
        """
        Update line data

        precomputes and caches entire line at each level, so initial
        display may be slow but panning and zooming should speed up after that
        """
        # Find nearest power of two as a factor to downsample by
        imin = max(np.searchsorted(x, xmin)-1, 0)
        imax = min(np.searchsorted(x, xmax) + 1, y.shape[0])
        L = imax - imin + 1
        factor = max(2**int(round(np.log(L/M) / np.log(2))), 1)

        # only calculate reduction if it hasn't been cached, do reduction using nearest cached version if possible
        if factor not in cached:
            cached[factor] = red(y, factor=factor)

        ## Make sure lengths match correctly here, by ensuring at least
        #   "factor" points for each x point, then matching y length
        #  this assumes x has uniform sample spacing - but could be modified
        newx = x[imin:imin + ((imax-imin)//factor)* factor:factor]
        start = imin//factor
        newy = cached[factor][start:start + newx.shape[-1]]
        assert newx.shape == newy.shape, "decimation error {}/{}!".format(newx.shape, newy.shape)

        ## Update line data
        line.set_xdata(newx)
        line.set_ydata(newy)

    update(x[0], x[-1])
    ax.add_line(line)
    ## Manually update limits of axis, as adding line doesn't do this
    #   if drawing multiple lines this can quickly slow things down, and some
    #   sort of check should be included to prevent unnecessary changes in limits
    #   when a line is first drawn.
    ax.set_xlim(min(ax.get_xlim()[0], x[0]), max(ax.get_xlim()[1], x[1]))
    ax.set_ylim(min(ax.get_ylim()[0], np.min(y)), max(ax.get_ylim()[1], np.max(y)))

    def callback(*ignore):
        lims = ax.get_xlim()
        update(*lims)

    ax.callbacks.connect('xlim_changed', callback)

    return [line]

Et voici min et max pour voir l'enveloppe du signal

def calc_env(y, factor):
    """
    y : 1D signal
    factor : amount to reduce y by (actually returns twice this for min and max)
    Calculate envelope (interleaved min and max points) for y
    """
    split = y[:len(y)//factor*factor].reshape(-1, factor)
    upper = split.max(axis=-1)
    lower = split.min(axis=-1)
    return c_[upper, lower].flatten()

La fonction suivante peut prendre l'un de ces deux éléments et les utilise pour réduire les données en cours de dessin. Le nombre de points réellement pris est de 5000 par défaut, ce qui devrait largement dépasser la résolution d'un moniteur. Les données sont mises en cache après avoir été réduites. La mémoire peut être un problème, en particulier avec de grandes quantités de données, mais elle ne doit pas dépasser la quantité requise par le signal d'origine.

def calc_rand(y, factor):
    split = y[:len(y)//factor*factor].reshape(-1, factor)
    idx = randint(0, split.shape[-1], split.shape[0])
    return split[arange(split.shape[0]), idx]

Voici un code de test

from pylab import *

Sur ma machine, cela s'affiche très rapidement. Le zoom a un peu de retard, surtout lorsqu'il est important. Le panoramique n'a aucun problème. L'utilisation de la sélection aléatoire au lieu du min et du max est un peu plus rapide et ne pose des problèmes que sur des niveaux de zoom très élevés.

EDIT: Ajout de parallel = True to numba ... encore plus rapide

J'ai fini par faire un hybride d'une seule passe argmin + max routine et de l'indexation améliorée de la réponse et du lien de @ a_guest à cette question min max simultanée associée .

Cette version renvoie les valeurs x correctes pour chaque paire y min / max et grâce à numba est en fait un peu plus rapide que la version "rapide mais pas tout à fait correcte".

%timeit min_max_downsample_v1(x_big ,y_big ,2000)
96 ms ± 2.46 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

%timeit min_max_downsample_v2(x_big ,y_big ,2000)
507 ms ± 4.75 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit min_max_downsample_v3(x_big ,y_big ,2000)
365 ms ± 1.27 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit min_max_downsample_v4(x_big ,y_big ,2000)
36.2 ms ± 487 µs per loop (mean ± std. dev. of 7 runs, 1 loop each)

La comparaison des vitesses en utilisant timeit montre que le code numba est environ 2,6 fois plus rapide et fournit de meilleurs résultats que la v1. C'est un peu plus de 10 fois plus rapide que de faire l'argmin et l'argmax de numpy en série.

from numba import jit, prange
@jit(parallel=True)
def min_max_downsample_v4(x, y, num_bins):
    pts_per_bin = x.size // num_bins
    x_view = x[:pts_per_bin*num_bins].reshape(num_bins, pts_per_bin)
    y_view = y[:pts_per_bin*num_bins].reshape(num_bins, pts_per_bin)    
    i_min = np.zeros(num_bins,dtype='int64')
    i_max = np.zeros(num_bins,dtype='int64')

    for r in prange(num_bins):
        min_val = y_view[r,0]
        max_val = y_view[r,0]
        for c in range(pts_per_bin):
            if y_view[r,c] < min_val:
                min_val = y_view[r,c]
                i_min[r] = c
            elif y_view[r,c] > max_val:
                max_val = y_view[r,c]
                i_max[r] = c                
    r_index = np.repeat(np.arange(num_bins), 2)
    c_index = np.sort(np.stack((i_min, i_max), axis=1)).ravel()        
    return x_view[r_index, c_index], y_view[r_index, c_index]

Avez-vous essayé pyqtgraph pour le traçage interactif? Il est plus réactif que matplotlib.

Une astuce que j'utilise pour le sous-échantillonnage consiste à utiliser array_split < / a> et calculez min et max pour les fractionnements. Le fractionnement est effectué en fonction du nombre d'échantillons par pixel de la zone de tracé.