[project] adding pytest

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
 
 [[package]]
 name = "asttokens"
@@ -153,6 +153,20 @@ files = [
     {file = "decorator-5.1.1.tar.gz", hash = "sha256:637996211036b6385ef91435e4fae22989472f9d571faba8927ba8253acbc330"},
 ]
 
+[[package]]
+name = "exceptiongroup"
+version = "1.2.2"
+description = "Backport of PEP 654 (exception groups)"
+optional = false
+python-versions = ">=3.7"
+files = [
+    {file = "exceptiongroup-1.2.2-py3-none-any.whl", hash = "sha256:3111b9d131c238bec2f8f516e123e14ba243563fb135d3fe885990585aa7795b"},
+    {file = "exceptiongroup-1.2.2.tar.gz", hash = "sha256:47c2edf7c6738fafb49fd34290706d1a1a2f4d1c6df275526b62cbb4aa5393cc"},
+]
+
+[package.extras]
+test = ["pytest (>=6)"]
+
 [[package]]
 name = "executing"
 version = "2.1.0"
@@ -270,6 +284,17 @@ files = [
 [package.dependencies]
 numpy = ">=1.19.3"
 
+[[package]]
+name = "iniconfig"
+version = "2.0.0"
+description = "brain-dead simple config-ini parsing"
+optional = false
+python-versions = ">=3.7"
+files = [
+    {file = "iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"},
+    {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"},
+]
+
 [[package]]
 name = "ipython"
 version = "8.27.0"
@@ -284,6 +309,7 @@ files = [
 [package.dependencies]
 colorama = {version = "*", markers = "sys_platform == \"win32\""}
 decorator = "*"
+exceptiongroup = {version = "*", markers = "python_version < \"3.11\""}
 jedi = ">=0.16"
 matplotlib-inline = "*"
 pexpect = {version = ">4.3", markers = "sys_platform != \"win32\" and sys_platform != \"emscripten\""}
@@ -291,6 +317,7 @@ prompt-toolkit = ">=3.0.41,<3.1.0"
 pygments = ">=2.4.0"
 stack-data = "*"
 traitlets = ">=5.13.0"
+typing-extensions = {version = ">=4.6", markers = "python_version < \"3.12\""}
 
 [package.extras]
 all = ["ipython[black,doc,kernel,matplotlib,nbconvert,nbformat,notebook,parallel,qtconsole]", "ipython[test,test-extra]"]
@@ -758,6 +785,21 @@ tests = ["check-manifest", "coverage", "defusedxml", "markdown2", "olefile", "pa
 typing = ["typing-extensions"]
 xmp = ["defusedxml"]
 
+[[package]]
+name = "pluggy"
+version = "1.5.0"
+description = "plugin and hook calling mechanisms for python"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669"},
+    {file = "pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1"},
+]
+
+[package.extras]
+dev = ["pre-commit", "tox"]
+testing = ["pytest", "pytest-benchmark"]
+
 [[package]]
 name = "prompt-toolkit"
 version = "3.0.48"
@@ -903,6 +945,28 @@ files = [
 [package.dependencies]
 numpy = ">=1.22.0"
 
+[[package]]
+name = "pytest"
+version = "8.3.3"
+description = "pytest: simple powerful testing with Python"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "pytest-8.3.3-py3-none-any.whl", hash = "sha256:a6853c7375b2663155079443d2e45de913a911a11d669df02a50814944db57b2"},
+    {file = "pytest-8.3.3.tar.gz", hash = "sha256:70b98107bd648308a7952b06e6ca9a50bc660be218d53c257cc1fc94fda10181"},
+]
+
+[package.dependencies]
+colorama = {version = "*", markers = "sys_platform == \"win32\""}
+exceptiongroup = {version = ">=1.0.0rc8", markers = "python_version < \"3.11\""}
+iniconfig = "*"
+packaging = "*"
+pluggy = ">=1.5,<2"
+tomli = {version = ">=1", markers = "python_version < \"3.11\""}
+
+[package.extras]
+dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -1026,6 +1090,17 @@ pure-eval = "*"
 [package.extras]
 tests = ["cython", "littleutils", "pygments", "pytest", "typeguard"]
 
+[[package]]
+name = "tomli"
+version = "2.0.1"
+description = "A lil' TOML parser"
+optional = false
+python-versions = ">=3.7"
+files = [
+    {file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
+    {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
+]
+
 [[package]]
 name = "tomlkit"
 version = "0.13.2"
@@ -1104,5 +1179,5 @@ files = [
 
 [metadata]
 lock-version = "2.0"
-python-versions = "^3.12"
-content-hash = "b1076b7f750e8f7e66542918ec746e74544fbfdb376875158f083fb573d18107"
+python-versions = "^3.10"
+content-hash = "c661a83b9d93748d39d3e4577f20e3b534a35f270231b6f48a9f13498a3658bd"

pyproject.toml

@@ -31,6 +31,7 @@ tomlkit = "^0.13.2"
 scipy = "^1.14.1"
 nixio = "^1.5.3"
 pyqtgraph = "^0.13.7"
+pytest = "^8.3.3"
 
 [tool.poetry.scripts]
 pyrelacs = "pyrelacs.app:main"

pyrelacs/dataio/circbuffer.py

@@ -1,11 +1,18 @@
+from typing import Tuple
 import numpy as np
 
+from pyrelacs.util.logging import config_logging
+
+log = config_logging()
+
 
 class CircBuffer:
     def __init__(self, size: int, channels: int = 1):
         self._size = size
         self._channels = channels
-        self._buffer = np.zeros((channels, size), dtype=np.double)  # or dtype of your choice
+        self._buffer = np.zeros(
+            (channels, size), dtype=np.double
+        )  # or dtype of your choice
         self._index = [0 for i in range(channels)]
         self._is_full = [False for i in range(channels)]
         self._totalcount = [0 for i in range(channels)]
@@ -38,10 +45,14 @@ class CircBuffer:
         Return all valid values from the specified channel
         """
         if self._is_full[channel]:
-            return np.concatenate((self._buffer[channel, self._index[channel]:],
-                                   self._buffer[channel, :self._index[channel]]))
+            return np.concatenate(
+                (
+                    self._buffer[channel, self._index[channel] :],
+                    self._buffer[channel, : self._index[channel]],
+                )
+            )
         else:
-            return self._buffer[channel, :self._index[channel]]
+            return self._buffer[channel, : self._index[channel]]
 
     def has_value(self, index, channel):
         if index < 0 and self.is_full(channel):
@@ -52,59 +63,93 @@ class CircBuffer:
         if index >= self.size:
             return False
 
+        # test if the ring buffer is at the start but
+        # and the index is greater than the write index
         if index > self.write_index(channel) and self.is_full(channel):
             return True
+        elif index > self.write_index(channel) and not self.is_full(channel):
+            raise IndexError("Index has no value, not written")
+
         if index == self.write_index(channel) and self._totalcount[channel] == 0:
             return False
 
-        return True
+    def valid_range(self, channel: int = 0) -> Tuple[int, int]:
+        """
+        Return the start index that are valid within the buffer
 
-    def valid_range(self, channel: int = 0):
-        """ Return the start index that are valid within the buffer
+        Parameters
+        ----------
+        channel : int
+            channel of the buffer
+
+        Returns
+        -------
+        Tuple[int, int]
+            start, extend of the valid range
         """
         start = 0
-        count = 0
+        extend = 0
         if self._totalcount[channel] == 0:
-            return start, count
+            return start, extend
 
         if not self.is_full(channel):
-            count = self._totalcount[channel]
+            extend = self._totalcount[channel]
         else:
-            count = self.size
-        return start, count
+            extend = self.size
+        return start, extend
 
     def get(self, index: int = -1, channel: int = 0):
         # easy case first, we can spare the effort of further checking
         if index >= 0 and index < self.write_index(channel):
-            return self._buffer[channel, index]
+            if self.has_value(index, channel):
+                return self._buffer[channel, index]
+            else:
+                raise IndexError(
+                    f"Invalid index {index} on ring buffer for channel{channel}"
+                )
 
         if index < 0:
+            log.debug("index is smaller than 0")
+            log.debug(f"{self.write_index()}")
             index = self.write_index() - 1
         if self.has_value(index, channel):
+            log.debug("index has a value")
             return self._buffer[channel, index]
         else:
-            raise IndexError(f"Invalid index {index} on ring buffer for channel{channel}")
+            raise IndexError(
+                f"Invalid index {index} on ring buffer for channel{channel}"
+            )
 
     def read(self, start, count=1, channel=0):
-        """ Reads a numpy array from buffer"""
+        """Reads a numpy array from buffer"""
         if start < 0 or count < 0:
-            raise IndexError(f"Invalid start ({start}) or count ({count}) for channel{channel}")
+            raise IndexError(
+                f"Invalid start ({start}) or count ({count}) for channel{channel}"
+            )
 
         if count == 1:
             return np.array(self.get(start, channel))
 
         vs, vc = self.valid_range(channel)
         if start > self._totalcount[channel]:
-            raise IndexError(f"Invalid start index {start} is invalid with totalcount {self._totalcount[channel]} for channel{channel}")
+            raise IndexError(
+                f"Invalid start index {start} is invalid with totalcount {self._totalcount[channel]} for channel{channel}"
+            )
         if start > self.size:
-            raise IndexError(f"Invalid start index {start} for buffer with size {self.size}")
+            raise IndexError(
+                f"Invalid start index {start} for buffer with size {self.size}"
+            )
         if count > self.size:
             count = self.size
         if count > vc:
             count = vc
 
         if (start + count) < self.size:
-            return self._buffer[channel, start:start + count]
+            return self._buffer[channel, start : start + count]
         else:
-            return np.concatenate((self._buffer[channel, start:],
-                                   self._buffer[channel, :count - self.size + start]))
+            return np.concatenate(
+                (
+                    self._buffer[channel, start:],
+                    self._buffer[channel, : count - self.size + start],
+                )
+            )

test/test_buffer.py

@@ -1,18 +1,19 @@
 import pytest
 import numpy as np
+from IPython import embed
 
-from usbdaq import RingBuffer
+from pyrelacs.dataio.circbuffer import CircBuffer
 
 
 def test_init():
-    buff = RingBuffer(1000, 2)
+    buff = CircBuffer(1000, 2)
 
     assert buff.size == 1000
     assert buff.channel_count == 2
 
 
 def test_hasvalue():
-    buff = RingBuffer(1000, 2)
+    buff = CircBuffer(1000, 2)
 
     assert buff.has_value(0, 0) == False
     assert buff.has_value(-1, 0) == False
@@ -39,18 +40,59 @@ def test_hasvalue():
 
 
 def test_validrange():
-    pass
+    buff = CircBuffer(1000, 2)
 
+    # without any values the range is (0, 0)
+    assert buff.valid_range() == (0, 0)
 
-def test_write():
-    buff = RingBuffer(1000, 2)
+    buff.append(0, 0)
+    assert buff.valid_range() == (0, 1)
 
-    samplecount = 1000
+    for i in range(100):
+        buff.append(i, 0)
+
+    assert buff.valid_range() == (0, 101)
+
+    for i in range(1000):
+        buff.append(i, 0)
+
+    assert buff.valid_range() == (0, 1000)
 
 
 def test_get():
-    pass
+    buff = CircBuffer(1000, 2)
+
+    # with no items written to the buffer
+    with pytest.raises(IndexError):
+        item = buff.get(index=-1)
+
+    buff.append(10, 0)
+    item = buff.get(index=-1)
+    assert item == 10
+
+    # Check if index is not written jet
+    with pytest.raises(IndexError):
+        item = buff.get(index=10)
+
+    for i in range(1000):
+        buff.append(i, 0)
+    item = buff.get(index=-1)
+    # the first item should be 999.0 because of we append a value in the earlier test
+    assert item == 999.0
+
+    with pytest.raises(IndexError):
+        item = buff.get(10001)
 
 
 def test_read():
     pass
+
+
+def test_write():
+    buff = CircBuffer(1000, 2)
+
+    samplecount = 1000
+
+
+if __name__ == "__main__":
+    test_get()