18979016529

Committed 31 Oct 2025 04:35PM UTC coverage: 17.134% (+0.03%) from 17.109%

Build # 18979016529

Build Type

push

github

Committed by

dprohe

Commit Message

Update to 0.21.0

Run Details

85 of 758 new or added lines in 10 files covered. (11.21%)

6 existing lines in 5 files now uncovered.

3364 of 19633 relevant lines covered (17.13%)

0.68 hits per line

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

20.51

/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 report_buffer_state(self):
        read_samples = np.sum(self.buffer_read)
        write_samples = self.buffer_size - read_samples
        print(f'{read_samples} of {self.buffer_size} have been read')
        print(f'{write_samples} of {self.buffer_size} have been written but not read')

    def write_get_data(self,data, read_remaining = False):
        """
        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(read_remaining)
        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, read_remaining = False):
        indices = np.arange(self.read_index - self.overlap_size, self.read_index + self.block_size) % self.buffer_size
        if read_remaining:
            # Pick out just the indices that are ok to read
            # print('Reading Remaining:')
            # print(f"{indices.copy()=}")
            indices = np.concatenate((indices[:self.overlap_size],indices[self.overlap_size:][~self.buffer_read[indices[self.overlap_size:]]]))
            # print(f"{indices.copy()=}")
        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 + (return_data.shape[-1]-self.overlap_size)) % self.buffer_size
        # print(self.buffer)
        # print(self.buffer_read)
        return return_data

1	# -- coding: utf-8 --
2	"""	1✔
3	Created on Wed Jul 23 12:34:27 2025
4
5	@author: dprohe
6	"""
7	import numpy as np	4✔
8
9	class CircularBufferWithOverlap:	4✔
10	def __init__(self, buffer_size, block_size, overlap_size, dtype = 'float', data_shape = ()):	4✔
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	"""
19	self.buffer_size = buffer_size	×
20	self.block_size = block_size	×
21	self.overlap_size = overlap_size	×
22	self.buffer = np.zeros(tuple(data_shape) + (buffer_size,), dtype=dtype) # Initialize buffer with zeros	×
23	self.buffer_read = np.ones((buffer_size,),dtype=bool)	×
24	self.write_index = 0 # Index where the next block will be written	×
25	self.read_index = 0 # Index where the next block will be read from	×
26
27	def report_buffer_state(self):	4✔
NEW 28	read_samples = np.sum(self.buffer_read)	×
NEW 29	write_samples = self.buffer_size - read_samples	×
NEW 30	print(f'{read_samples} of {self.buffer_size} have been read')	×
NEW 31	print(f'{write_samples} of {self.buffer_size} have been written but not read')	×
32
33	def write_get_data(self,data, read_remaining = False):	4✔
34	"""
35	Writes a block of data and then returns a block if available
36
37	Parameters:
38	- data: Array to write to the buffer.
39	"""
40	self.write(data)	×
41	try:	×
NEW 42	return self.read(read_remaining)	×
43	except ValueError:	×
44	return None	×
45
46	def write(self, data):	4✔
47	"""
48	Write a block of data to the circular buffer.
49
50	Parameters:
51	- data: Array to write to the buffer.
52	"""
53	# Compute the end index for the write operation
54	indices = np.arange(self.write_index,self.write_index+data.shape[-1]+self.overlap_size) % self.buffer_size	×
55
56	if np.any(~self.buffer_read[indices]):	×
57	raise ValueError('Overwriting data on buffer that has not been read. Read data before writing again.')	×
58
59	self.buffer[...,indices[:None if self.overlap_size == 0 else -self.overlap_size]] = data	×
60	self.buffer_read[indices[:None if self.overlap_size == 0 else -self.overlap_size]] = False	×
61
62	# Update the write index
63	self.write_index = (self.write_index + data.shape[-1]) % self.buffer_size	×
64
65	# print(self.buffer)
66	# print(self.buffer_read)
67
68	def read(self, read_remaining = False):	4✔
69	indices = np.arange(self.read_index - self.overlap_size, self.read_index + self.block_size) % self.buffer_size	×
NEW 70	if read_remaining:	×
71	# Pick out just the indices that are ok to read
72	# print('Reading Remaining:')
73	# print(f"{indices.copy()=}")
NEW 74	indices = np.concatenate((indices[:self.overlap_size],indices[self.overlap_size:][~self.buffer_read[indices[self.overlap_size:]]]))	×
75	# print(f"{indices.copy()=}")
76	if np.any(self.buffer_read[indices[self.overlap_size:]]):	×
77	raise ValueError('Data would be read multiple times. Write data before reading again.')	×
78	return_data = self.buffer[...,indices]	×
79	self.buffer_read[indices[self.overlap_size:]] = True	×
NEW 80	self.read_index = (self.read_index + (return_data.shape[-1]-self.overlap_size)) % self.buffer_size	×
81	# print(self.buffer)
82	# print(self.buffer_read)
UNCOV 83	return return_data	×

sandialabs / sdynpy / 18979016529

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