awendt/pyrelacs
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[buffer] removing logging, adding samplerate, adding check for getting the item at current index

Browse Source
This commit is contained in:
wendtalexander 2024-10-03 08:52:25 +02:00
parent b18f870a6b
commit d127750e7b
1 changed files with 40 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
pyrelacs/dataio/circbuffer.py
View File

@ -1,58 +1,59 @@
from typing import Tuple from typing import Tuple
import numpy as np import numpy as np
from IPython import embed
from pyrelacs.util.logging import config_logging
log = config_logging()
class CircBuffer: class CircBuffer:
def __init__(self, size: int, channels: int = 1): def __init__(self, size: int, channels: int = 1, samplerate: float = 40_000):
self._size = size self.__size = size
self._channels = channels self.__channels = channels
self._buffer = np.zeros( self.__samplereate = samplerate
self.__buffer = np.zeros(
(channels, size), dtype=np.double (channels, size), dtype=np.double
) # or dtype of your choice ) # or dtype of your choice
self._index = [0 for i in range(channels)] self.__index = [0 for i in range(channels)]
self._is_full = [False for i in range(channels)] self.__is_full = [False for i in range(channels)]
self._totalcount = [0 for i in range(channels)] self.__totalcount = [0 for i in range(channels)]
self._overflows = [0 for i in range(channels)] self.__overflows = [0 for i in range(channels)]
@property @property
def size(self): def size(self):
return self._size return self.__size
@property @property
def channel_count(self): def channel_count(self):
return self._channels return self.__channels
def totocount(self, channel: int = 0):
return self.__totalcount[channel]
def is_full(self, channel: int = 0): def is_full(self, channel: int = 0):
return self._is_full[channel] return self.__is_full[channel]
def write_index(self, channel: int = 0): def write_index(self, channel: int = 0):
return self._index[channel] return self.__index[channel]
def append(self, item, channel: int = 0): def append(self, item, channel: int = 0):
self._buffer[channel, self.write_index(channel)] = item self.__buffer[channel, self.write_index(channel)] = item
self._index[channel] = (self.write_index(channel) + 1) % self._size self.__index[channel] = (self.write_index(channel) + 1) % self.__size
self._totalcount[channel] += 1 self.__totalcount[channel] += 1
if self._index[channel] == 0: if self.__index[channel] == 0:
self._is_full[channel] = True self.__is_full[channel] = True
self._overflows[channel] += 1 self.__overflows[channel] += 1
def get_all(self, channel: int = 0): def get_all(self, channel: int = 0):
""" """
Return all valid values from the specified channel Return all valid values from the specified channel
""" """
if self._is_full[channel]: if self.__is_full[channel]:
return np.concatenate( return np.concatenate(
( (
self._buffer[channel, self._index[channel] :], self.__buffer[channel, self.__index[channel] :],
self._buffer[channel, : self._index[channel]], self.__buffer[channel, : self.__index[channel]],
) )
) )
else: else:
return self._buffer[channel, : self._index[channel]] return self.__buffer[channel, : self.__index[channel]]
def has_value(self, index, channel): def has_value(self, index, channel):
if index <= 0 and self.is_full(channel): if index <= 0 and self.is_full(channel):
@ -67,17 +68,17 @@ class CircBuffer:
# and the index is greater than the write index # and the index is greater than the write index
if index > self.write_index(channel) and self.is_full(channel): if index > self.write_index(channel) and self.is_full(channel):
return True return True
elif index > self.write_index(channel) and not self.is_full(channel): elif index >= self.write_index(channel) and not self.is_full(channel):
raise IndexError("Index has no value, not written") raise IndexError("Index has no value, not written")
if index == self.write_index(channel) and self._totalcount[channel] == 0: if index == self.write_index(channel) and self.__totalcount[channel] == 0:
return False return False
return True return True
def valid_range(self, channel: int = 0) -> Tuple[int, int]: def valid_range(self, channel: int = 0) -> Tuple[int, int]:
""" """
Return the start index that are valid within the buffer Return the start index and the extend that are valid within the buffer
Parameters Parameters
---------- ----------
@ -91,32 +92,29 @@ class CircBuffer:
""" """
start = 0 start = 0
extend = 0 extend = 0
if self._totalcount[channel] == 0: if self.__totalcount[channel] == 0:
return start, extend return start, extend
if not self.is_full(channel): if not self.is_full(channel):
extend = self._totalcount[channel] extend = self.__totalcount[channel]
else: else:
extend = self.size extend = self.size
return start, extend return start, extend
def get(self, index: int = -1, channel: int = 0): def get(self, index: int = -1, channel: int = 0):
# easy case first, we can spare the effort of further checking # easy case first, we can spare the effort of further checking
if index >= 0 and index < self.write_index(channel): if index >= 0 and index <= self.write_index(channel):
if self.has_value(index, channel): if self.has_value(index, channel):
return self._buffer[channel, index] return self.__buffer[channel, index]
else: else:
raise IndexError( raise IndexError(
f"Invalid index {index} on ring buffer for channel{channel}" f"Invalid index {index} on ring buffer for channel{channel}"
) )
if index < 0: if index < 0:
log.debug("index is smaller than 0")
log.debug(f"{self.write_index()}")
index = self.write_index() - 1 index = self.write_index() - 1
if self.has_value(index, channel): if self.has_value(index, channel):
log.debug("index has a value") return self.__buffer[channel, index]
return self._buffer[channel, index]
else: else:
raise IndexError( raise IndexError(
f"Invalid index {index} on ring buffer for channel{channel}" f"Invalid index {index} on ring buffer for channel{channel}"
@ -133,9 +131,9 @@ class CircBuffer:
return np.array(self.get(start, channel)) return np.array(self.get(start, channel))
vs, vc = self.valid_range(channel) vs, vc = self.valid_range(channel)
if start > self._totalcount[channel]: if start > self.__totalcount[channel]:
raise IndexError( raise IndexError(
f"Invalid start index {start} is invalid with totalcount {self._totalcount[channel]} for channel{channel}" f"Invalid start index {start} is invalid with totalcount {self.__totalcount[channel]} for channel{channel}"
) )
if start > self.size: if start > self.size:
raise IndexError( raise IndexError(
@ -147,11 +145,11 @@ class CircBuffer:
count = vc count = vc
if (start + count) < self.size: if (start + count) < self.size:
return self._buffer[channel, start : start + count] return self.__buffer[channel, start : start + count]
else: else:
return np.concatenate( return np.concatenate(
( (
self._buffer[channel, start:], self.__buffer[channel, start:],
self._buffer[channel, : count - self.size + start], self.__buffer[channel, : count - self.size + start],
) )
) )