RingBuffer.py 文件源码

def push(self, values: np.ndarray):
        """
        Push values to buffer. If buffer can't store all values a ValueError is raised
        """
        n = len(values)
        if len(self) + n > self.size:
            raise ValueError("Too much data to push to RingBuffer")

        slide_1 = np.s_[self.right_index:min(self.right_index + n, self.size)]
        slide_2 = np.s_[:max(self.right_index + n - self.size, 0)]
        with self.__data.get_lock():
            data = np.frombuffer(self.__data.get_obj(), dtype=np.complex64)
            data[slide_1] = values[:slide_1.stop - slide_1.start]
            data[slide_2] = values[slide_1.stop - slide_1.start:]
            self.right_index += n

        self.__length.value += n