ปรับปรุงการสุ่มตัวอย่างต่ำสุด/สูงสุด

ฉันจัดการเพื่อให้ได้ประสิทธิภาพที่ดีขึ้นโดยใช้เอาต์พุตของ arg(min|max) โดยตรงเพื่อสร้างดัชนีอาร์เรย์ข้อมูล ซึ่งต้องเสียค่าใช้จ่ายในการเรียก np.sort เพิ่มเติม แต่แกนที่จะจัดเรียงมีเพียงสององค์ประกอบเท่านั้น (ดัชนีต่ำสุด / สูงสุด) และอาร์เรย์โดยรวมค่อนข้างเล็ก (จำนวนถังขยะ):

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]

ฉันตรวจสอบกำหนดเวลาสำหรับตัวอย่างของคุณและฉันได้รับ:

• 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

ฉันยังตรวจสอบการกลับมาโดยตรงหลังจากการโทรไปที่ arg(min|max) และผลลัพธ์ก็เท่ากับ 164 ms นั่นคือไม่มีค่าใช้จ่ายจริงหลังจากนั้นอีกต่อไป

ดังนั้น นี่ไม่ได้กล่าวถึงการเร่งความเร็วฟังก์ชันเฉพาะที่เป็นปัญหา แต่แสดงวิธีการสองสามวิธีในการวางแผนเส้นที่มีจุดจำนวนมากค่อนข้างมีประสิทธิภาพ วิธีนี้จะถือว่าจุด x ได้รับการเรียงลำดับและสุ่มตัวอย่างสม่ำเสมอ (หรือใกล้เคียงกัน)

ติดตั้ง

from pylab import *

นี่คือฟังก์ชันที่ฉันชอบซึ่งจะลดจำนวนคะแนนโดยการสุ่มเลือกหนึ่งจุดในแต่ละช่วงเวลา ไม่รับประกันว่าจะแสดงทุกจุดสูงสุดของข้อมูล แต่ไม่มีปัญหามากเท่ากับการทำลายข้อมูลโดยตรงและรวดเร็ว

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]

และนี่คือค่าต่ำสุดและสูงสุดเพื่อดูซองสัญญาณ

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()

ฟังก์ชันต่อไปนี้สามารถใช้อย่างใดอย่างหนึ่งเหล่านี้ และใช้เพื่อลดข้อมูลที่ถูกดึงออกมา จำนวนคะแนนที่ได้รับจริงคือ 5,000 คะแนนโดยค่าเริ่มต้น ซึ่งน่าจะเกินความละเอียดของจอภาพอย่างมาก ข้อมูลจะถูกแคชหลังจากที่ลดลง หน่วยความจำอาจเป็นปัญหา โดยเฉพาะอย่างยิ่งกับข้อมูลจำนวนมาก แต่ไม่ควรเกินปริมาณที่สัญญาณดั้งเดิมต้องการ

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]

นี่คือรหัสทดสอบบางส่วน

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)

บนเครื่องของฉันสิ่งนี้แสดงเร็วมาก การซูมมีความล่าช้าเล็กน้อย โดยเฉพาะอย่างยิ่งเมื่อมีจำนวนมาก การแพนไม่มีปัญหา การใช้การเลือกแบบสุ่มแทนค่าต่ำสุดและสูงสุดนั้นค่อนข้างเร็วกว่าเล็กน้อย และมีปัญหาเฉพาะในระดับการซูมที่สูงมากเท่านั้น

แก้ไข: เพิ่ม Parallel=True ถึง numba ... เร็วขึ้นอีก

ฉันลงเอยด้วยการสร้างไฮบริดของรูทีน argmin+max ผ่านครั้งเดียวและการจัดทำดัชนีที่ปรับปรุงจากคำตอบของ @a_guest และลิงก์ไปยัง คำถามสูงสุดขั้นต่ำที่เกี่ยวข้องพร้อมกันนี้

เวอร์ชันนี้ส่งคืนค่า x ที่ถูกต้องสำหรับแต่ละคู่ min/max y แต่ละคู่ และต้องขอบคุณ numba ที่ทำให้เร็วกว่าเวอร์ชัน "เร็วแต่ไม่ถูกต้อง" เล็กน้อย

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]

การเปรียบเทียบความเร็วโดยใช้ timeit แสดงว่าโค้ด numba เร็วกว่าประมาณ 2.6 เท่า และให้ผลลัพธ์ที่ดีกว่า v1 มันเร็วกว่าการทำ argmin & argmax ของ numpy มากกว่า 10 เท่าเล็กน้อยในซีรีย์

%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)

คุณลองใช้ pyqtgraph เพื่อการวางแผนเชิงโต้ตอบหรือไม่? มันตอบสนองได้ดีกว่า matplotlib

เคล็ดลับหนึ่งที่ฉันใช้สำหรับการลดขนาดตัวอย่างคือการใช้ array_split และคำนวณขั้นต่ำและสูงสุดสำหรับการแยก การแยกจะกระทำตามจำนวนตัวอย่างต่อพิกเซลของพื้นที่ลงจุด