17045496486

Committed 18 Aug 2025 03:46PM UTC coverage: 16.91% (+1.2%) from 15.675%

Build # 17045496486

Build Type

Pull #18

github

Committed by

web-flow

Commit Message

Merge 4cda0da23 into 92b11b3be

Pull Request Pull Request #18: update to v0.18.2

Run Details

277 of 2402 new or added lines in 25 files covered. (11.53%)

13 existing lines in 6 files now uncovered.

3179 of 18800 relevant lines covered (16.91%)

0.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

17.14

/src/sdynpy/signal_processing/sdynpy_buffer.py

# -*- coding: utf-8 -*-
"""
Created on Wed Jul 23 12:34:27 2025

@author: dprohe
"""
import numpy as np

class CircularBufferWithOverlap:
    def __init__(self, buffer_size, block_size, overlap_size, dtype = 'float', data_shape = ()):
        """
        Initialize the circular buffer.
        
        Parameters:
        - buffer_size: Total size of the circular buffer.
        - block_size: Number of samples written in each block.
        - overlap_size: Number of samples from the previous block to include in the read.
        """
        self.buffer_size = buffer_size
        self.block_size = block_size
        self.overlap_size = overlap_size
        self.buffer = np.zeros(tuple(data_shape) + (buffer_size,), dtype=dtype)  # Initialize buffer with zeros
        self.buffer_read = np.ones((buffer_size,),dtype=bool)
        self.write_index = 0  # Index where the next block will be written
        self.read_index = 0 # Index where the next block will be read from
    
    def write_get_data(self,data):
        """
        Writes a block of data and then returns a block if available

        Parameters:
        - data: Array to write to the buffer.
        """
        self.write(data)
        try:
            return self.read()
        except ValueError:
            return None
    
    def write(self, data):
        """
        Write a block of data to the circular buffer.
        
        Parameters:
        - data: Array to write to the buffer.
        """
        # Compute the end index for the write operation
        indices = np.arange(self.write_index,self.write_index+data.shape[-1]+self.overlap_size) % self.buffer_size

        if np.any(~self.buffer_read[indices]):
            raise ValueError('Overwriting data on buffer that has not been read.  Read data before writing again.')

        self.buffer[...,indices[:None if self.overlap_size == 0 else -self.overlap_size]] = data
        self.buffer_read[indices[:None if self.overlap_size == 0 else -self.overlap_size]] = False
        
        # Update the write index
        self.write_index = (self.write_index + data.shape[-1]) % self.buffer_size
        
        print(self.buffer)
        print(self.buffer_read)
        
    def read(self):
        indices = np.arange(self.read_index - self.overlap_size, self.read_index + self.block_size) % self.buffer_size
        if np.any(self.buffer_read[indices[self.overlap_size:]]):
            raise ValueError('Data would be read multiple times.  Write data before reading again.')
        return_data = self.buffer[...,indices]
        self.buffer_read[indices[self.overlap_size:]] = True
        self.read_index = (self.read_index + self.block_size) % self.buffer_size
        print(self.buffer)
        print(self.buffer_read)
        return return_data

1	# -- coding: utf-8 --
2	"""
3	Created on Wed Jul 23 12:34:27 2025
4
5	@author: dprohe
6	"""
7	import numpy as np	1✔
8
9	class CircularBufferWithOverlap:	1✔
10	def __init__(self, buffer_size, block_size, overlap_size, dtype = 'float', data_shape = ()):	1✔
11	"""
12	Initialize the circular buffer.
13
14	Parameters:
15	- buffer_size: Total size of the circular buffer.
16	- block_size: Number of samples written in each block.
17	- overlap_size: Number of samples from the previous block to include in the read.
18	"""
NEW 19	self.buffer_size = buffer_size	×
NEW 20	self.block_size = block_size	×
NEW 21	self.overlap_size = overlap_size	×
NEW 22	self.buffer = np.zeros(tuple(data_shape) + (buffer_size,), dtype=dtype) # Initialize buffer with zeros	×
NEW 23	self.buffer_read = np.ones((buffer_size,),dtype=bool)	×
NEW 24	self.write_index = 0 # Index where the next block will be written	×
NEW 25	self.read_index = 0 # Index where the next block will be read from	×
26
27	def write_get_data(self,data):	1✔
28	"""
29	Writes a block of data and then returns a block if available
30
31	Parameters:
32	- data: Array to write to the buffer.
33	"""
NEW 34	self.write(data)	×
NEW 35	try:	×
NEW 36	return self.read()	×
NEW 37	except ValueError:	×
NEW 38	return None	×
39
40	def write(self, data):	1✔
41	"""
42	Write a block of data to the circular buffer.
43
44	Parameters:
45	- data: Array to write to the buffer.
46	"""
47	# Compute the end index for the write operation
NEW 48	indices = np.arange(self.write_index,self.write_index+data.shape[-1]+self.overlap_size) % self.buffer_size	×
49
NEW 50	if np.any(~self.buffer_read[indices]):	×
NEW 51	raise ValueError('Overwriting data on buffer that has not been read. Read data before writing again.')	×
52
NEW 53	self.buffer[...,indices[:None if self.overlap_size == 0 else -self.overlap_size]] = data	×
NEW 54	self.buffer_read[indices[:None if self.overlap_size == 0 else -self.overlap_size]] = False	×
55
56	# Update the write index
NEW 57	self.write_index = (self.write_index + data.shape[-1]) % self.buffer_size	×
58
NEW 59	print(self.buffer)	×
NEW 60	print(self.buffer_read)	×
61
62	def read(self):	1✔
NEW 63	indices = np.arange(self.read_index - self.overlap_size, self.read_index + self.block_size) % self.buffer_size	×
NEW 64	if np.any(self.buffer_read[indices[self.overlap_size:]]):	×
NEW 65	raise ValueError('Data would be read multiple times. Write data before reading again.')	×
NEW 66	return_data = self.buffer[...,indices]	×
NEW 67	self.buffer_read[indices[self.overlap_size:]] = True	×
NEW 68	self.read_index = (self.read_index + self.block_size) % self.buffer_size	×
NEW 69	print(self.buffer)	×
NEW 70	print(self.buffer_read)	×
NEW 71	return return_data	×

sandialabs / sdynpy / 17045496486

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous