IPKM/python3-damaris
Public Access
3
0
Fork 0
Code Issues 7 Pull Requests Actions Releases Activity

added preliminary test cases

Browse Source
This commit is contained in:
Markus Rosenstihl
2026-03-20 10:08:07 +01:00
parent 65611b98a0
commit ec8fa09201
3 changed files with 65 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data/ADC_Result.py
+62 -32
View File
@@ -19,7 +19,25 @@ import tables
#############################################################################
class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
def __init__(self, x = None, y = None, index = None, sampl_freq = None, desc = None, job_id = None, job_date = None):
"""
Represents the result of an ADC, encapsulating data and metadata
for processing, visualization, and export.
This class combines data storage and manipulation functionality with interfaces for
drawing and result processing. It manages time-series data across multiple channels,
supports dynamic resizing of datasets, and provides mechanisms to export data in
various formats. Its key roles include ADC result storage, metadata management, and
integration with external systems.
Attributes:
xlabel: Label for the x-axis, used in visualization.
ylabel: Label for the y-axis, used in visualization.
lock: A threading lock for synchronizing access to the data.
nChannels: Number of data channels in the result set.
sampling_rate: Sampling frequency of the ADC data.
job_id: identifier for the job that generated this result.
"""
def __init__(self, x = None, y:list = None, index = None, sampl_freq = None, desc = None, job_id = None, job_date = None):
Resultable.__init__(self)
Drawable.__init__(self)
@@ -33,14 +51,19 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
self.lock=threading.RLock()
self.nChannels = 0
# using no argument for initialization
if (x is None) and (y is None) and (index is None) and (sampl_freq is None) and (desc is None) and (job_id is None) and (job_date is None):
self.cont_data = False
self.sampling_rate = 0
self.index = []
self.x = []
self.y = []
# using all arguments for initialization
elif (x is not None) and (y is not None) and (index is not None) and (sampl_freq is not None) and (desc is not None) and (job_id is not None) and (job_date is not None):
# TODO: insure integer calculations for ADC_Result operations.
#for ch in y:
# if not numpy.issubdtype(ch.dtype, numpy.integer):
# raise TypeError("TypeError: ADC_Result y data must be a list with integer type channels")
self.x = x
self.y = y
self.index = index
@@ -68,9 +91,6 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
print("Warning ADC-Result: Tried to run \"create_data_space()\" more than once.")
return
if channels <= 0:
raise ValueError("ValueError: You cant create an ADC-Result with less than 1 channel!")
if samples <= 0:
raise ValueError("ValueError: You cant create an ADC-Result with less than 1 sample!")
for i in range(channels):
@@ -83,7 +103,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
def contains_data(self):
"Returns true if ADC_Result contains data. (-> create_data_space() was called)"
"""Returns true if ADC_Result contains data. (-> create_data_space() was called)"""
return self.cont_data
@@ -347,7 +367,8 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot add \"%s\" to ADC-Result!" % str(other.__class__))
raise ValueError(f"ValueError: Cannot add \"{other.__class__}\" to ADC-Result!")
def __radd__(self, other):
@@ -368,7 +389,8 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot subtract \"%s\" to ADC-Result!") % str(other.__class__)
raise ValueError(f"ValueError: Cannot subtract \"{other.__class__}\" to ADC-Result!")
def __rsub__(self, other):
@@ -383,9 +405,9 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
r = ADC_Result(x = self.x[:], y = tmp_y, index = self.index[:], sampl_freq = self.sampling_rate, desc = self.description, job_id = self.job_id, job_date = self.job_date)
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot subtract \"%s\" to ADC-Result!") % str(other.__class__)
raise ValueError(f"ValueError: Cannot subtract \"{other.__class__}\" to ADC-Result!")
def __mul__(self, other):
@@ -400,9 +422,8 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
r = ADC_Result(x = self.x[:], y = tmp_y, index = self.index[:], sampl_freq = self.sampling_rate, desc = self.description, job_id = self.job_id, job_date = self.job_date)
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot multiply \"%s\" to ADC-Result!") % str(other.__class__)
raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to ADC-Result!")
def __rmul__(self, other):
@@ -423,26 +444,9 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot multiply \"%s\" to ADC-Result!") % str(other.__class__)
raise ValueError(f"ValueError: Cannot power raise \"{other.__class__}\" to ADC-Result!")
def __div__(self, other):
"Redefining self / other (scalar)"
if isinstance(other, int) or isinstance(other, float):
self.lock.acquire()
tmp_y = []
for i in range(self.get_number_of_channels()):
tmp_y.append(numpy.array(self.y[i], dtype="float32") / other)
r = ADC_Result(x = self.x[:], y = tmp_y, index = self.index[:], sampl_freq = self.sampling_rate, desc = self.description, job_id = self.job_id, job_date = self.job_date)
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot multiply \"%s\" to ADC-Result!") % str(other.__class__)
def __rdiv__(self, other):
def __truediv__(self, other):
"Redefining other (scalar) / self"
if isinstance(other, int) or isinstance(other, float):
self.lock.acquire()
@@ -455,7 +459,33 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath):
self.lock.release()
return r
else:
raise ValueError("ValueError: Cannot multiply \"%s\" to ADC-Result!") % str(other.__class__)
raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!")
def __rtruediv__(self, other):
"Redefining other (scalar) / self"
return self.__truediv__(other)
def __floordiv__(self, other):
"Redefining other (scalar) / self"
if isinstance(other, float):
raise ValueError("ValueError: Cannot use floor division (//) on floats! Use \"//\" instead of \"/\"! ")
if isinstance(other, int):
self.lock.acquire()
tmp_y = []
for i in range(self.get_number_of_channels()):
tmp_y.append(other / numpy.array(self.y[i], dtype="float32"))
r = ADC_Result(x = self.x[:], y = tmp_y, index = self.index[:], sampl_freq = self.sampling_rate, desc = self.description, job_id = self.job_id, job_date = self.job_date)
self.lock.release()
return r
else:
raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!")
def __rfloordiv__(self, other):
"Redefining other (scalar) / self"
return self.__floordiv__(other)
def __neg__(self):
src/data/Accumulation.py
+2
View File
@@ -1,5 +1,7 @@
# -*- coding: iso-8859-1 -*-
from __future__ import annotations
#############################################################################
# #
# Name: Class Accumulation #
src/tests/test_ADC_Result.py
+1 -1
View File
@@ -179,7 +179,7 @@ class TestADCResult(unittest.TestCase):
Test the functionality of __sub__ and __rsub__
"""
adc = self.create_adc_result()
y = adc.y
y = adc.y[0][1]=5
# test integer subtraction
for i in range(adc.get_nChannels()):
self.assertIsNone(np.testing.assert_array_equal(adc.y[i] - 10, y[i] - 10))