1

#!/usr/bin/env python

     2

# -*- coding: utf-8 -*-

     3

     4

# This file is part of Beremiz

     5

# Copyright (C) 2021: Edouard TISSERANT

     6

#

     7

# See COPYING file for copyrights details.

     8

     9

# Based on Eelco Hoogendoorn stackoverflow answer about RingBuffer with numpy

    10

    11

import numpy as np

    12

    13

    14

class RingBuffer(object):

    15

    def __init__(self, width=None, size=131072, padding=None):

    16

        self.size = size

    17

        self.padding = size if padding is None else padding

    18

        shape = (self.size+self.padding,)

    19

        if width :

    20

            shape += (width,)

    21

        self.buffer = np.zeros(shape)

    22

        self.cursor = 0

    23

    24

    def append(self, data):

    25

        """this is an O(n) operation"""

    26

        data = data[-self.size:]

    27

        n = len(data)

    28

        if self.size + self.padding - self.cursor < n:

    29

            self.compact()

    30

        self.buffer[self.cursor:][:n] = data

    31

        self.cursor += n

    32

    33

    @property

    34

    def count(self):

    35

        return min(self.size, self.cursor)

    36

    37

    @property

    38

    def view(self):

    39

        """this is always an O(1) operation"""

    40

        return self.buffer[max(0, self.cursor - self.size):][:self.count]

    41

    42

    def compact(self):

    43

        """

    44

        note: only when this function is called, is an O(size) performance hit incurred,

    45

        and this cost is amortized over the whole padding space

    46

        """

    47

        self.buffer[:self.count] = self.view

    48

        self.cursor -= self.size

    49