diff --git a/pyproject.toml b/pyproject.toml index 25c1e05..01f6231 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,9 @@ version = {attr = "damaris.__version__"} project-includes = [ "src/damaris/experiments/Experiment.py", "src/damaris/experiments/dac.py", + "src/damaris/data/ADC_Result.py", + "src/damaris/data/MeasurementResult.py", + "src/damaris/data/Accumulation.py", ] search-path = ["src"] python-version = "3.11" diff --git a/src/damaris/data/ADC_Result.py b/src/damaris/data/ADC_Result.py index d8b3a6e..067bd59 100644 --- a/src/damaris/data/ADC_Result.py +++ b/src/damaris/data/ADC_Result.py @@ -1,5 +1,7 @@ # -*- coding: iso-8859-1 -*- +from __future__ import annotations + from .Resultable import Resultable from .Drawable import Drawable from .Signalpath import Signalpath @@ -9,7 +11,13 @@ import numpy import sys import datetime import tables +from typing import IO from scipy.signal import filtfilt, remez + +_XArray = numpy.ndarray | list +_YChannels = list[numpy.ndarray] +_IndexList = list[tuple[int, int]] +_Description = dict[str, str] ############################################################################# # # # Name: Class ADC_Result # @@ -39,7 +47,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): 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): + def __init__(self, x: _XArray | None = None, y: _YChannels | None = None, index: _IndexList | None = None, sampl_freq: float | None = None, desc: _Description | None = None, job_id: int | None = None, job_date: datetime.datetime | None = None): Resultable.__init__(self) Drawable.__init__(self) @@ -87,7 +95,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): raise ValueError("Wrong usage of __init__!") - def create_data_space(self, channels, samples): + def create_data_space(self, channels: int, samples: int) -> None: "Initialises the internal data-structures" if self.contains_data(): @@ -104,12 +112,12 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): self.cont_data = True - def contains_data(self): + def contains_data(self) -> bool: """Returns true if ADC_Result contains data. (-> create_data_space() was called)""" return self.cont_data - def check_clipping(self, bit_depth=None): + def check_clipping(self, bit_depth: int | None = None) -> bool: """ Check if any data in y is clipped. For a 14-bit ADC, clipping occurs at 2**13 (8192) or 2**13-1 (8191). @@ -147,7 +155,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): return self.is_clipped - def lowpass(self, cutoff, numtaps=None): + def lowpass(self, cutoff: float, numtaps: int | None = None) -> ADC_Result: """ Apply a zero-phase lowpass FIR filter to all channels. @@ -239,7 +247,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): return r - def add_sample_space(self, samples): + def add_sample_space(self, samples: int) -> None: "Adds space for n samples, where n can also be negative (deletes space). New space is filled up with \"0\"" self.lock.acquire() @@ -259,7 +267,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): self.lock.release() - def get_result_by_index(self, index): + def get_result_by_index(self, index: int) -> ADC_Result: self.lock.acquire() try: @@ -281,33 +289,33 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): return r - def get_sampling_rate(self): + def get_sampling_rate(self) -> float: """Returns the samplingfrequency""" return self.sampling_rate + 0 - def set_sampling_rate(self, hz): + def set_sampling_rate(self, hz: float) -> None: """Sets the samplingfrequency in hz""" self.sampling_rate = float(hz) - def get_nChannels(self): + def get_nChannels(self) -> int: """Gets the number of channels""" return self.nChannels + 0 - def set_nChannels(self, channels): + def set_nChannels(self, channels: int) -> None: """Sets the number of channels""" self.nChannels = int(channels) - def get_index_bounds(self, index): + def get_index_bounds(self, index: int) -> tuple[int, int]: "Returns a tuple with (start, end) of the wanted result" return self.index[index] - def uses_statistics(self): + def uses_statistics(self) -> bool: return False - def write_to_csv(self, destination=sys.stdout, delimiter=" "): + def write_to_csv(self, destination: str | IO[str] = sys.stdout, delimiter: str = " ") -> None: """ writes the data to a file or to sys.stdout destination can be a file or a filename @@ -319,7 +327,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: self._write_to_csv_dest(destination, delimiter) - def _write_to_csv_dest(self, dest, delimiter): + def _write_to_csv_dest(self, dest: IO[str], delimiter: str) -> None: dest.write("# adc_result\n") dest.write("# t y0 y1 ...\n") self.lock.acquire() @@ -335,7 +343,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): finally: self.lock.release() - def write_to_simpson(self, destination=sys.stdout, delimiter=" "): + def write_to_simpson(self, destination: str | IO[str] = sys.stdout, delimiter: str = " ") -> None: """ writes the data to a text file or sys.stdout in Simpson format, for further processing with the NMRnotebook software; @@ -347,7 +355,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: self._write_to_simpson_dest(destination, delimiter) - def _write_to_simpson_dest(self, dest, delimiter): + def _write_to_simpson_dest(self, dest: IO[str], delimiter: str) -> None: self.lock.acquire() try: xdata=self.get_xdata() @@ -366,7 +374,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): finally: self.lock.release() - def write_to_hdf(self, hdffile, where, name, title, complib=None, complevel=None): + def write_to_hdf(self, hdffile: tables.File, where: str, name: str, title: str, complib: str | None = None, complevel: int | None = None) -> None: accu_group=hdffile.create_group(where=where,name=name,title=title) accu_group._v_attrs.damaris_type="ADC_Result" if self.contains_data(): @@ -463,13 +471,13 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): # Ueberladen von Operatoren und Built-Ins ------------------------------------------------------- - def __len__(self): + def __len__(self) -> int: "Redefining len(ADC_Result obj), returns the number of samples in one channel and 0 without data" if len(self.y)>0: return len(self.y[0]) return 0 - def __repr__(self): + def __repr__(self) -> str: """ writes job meta data and data to string returned """ @@ -486,12 +494,12 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): return tmp_string - def _copy_metadata(self, r): + def _copy_metadata(self, r: ADC_Result) -> None: "Copy channel metadata (nChannels, is_clipped) from self to the new result r" r.nChannels = self.nChannels r.is_clipped = self.is_clipped - def __add__(self, other): + def __add__(self, other: int | float) -> ADC_Result: "Redefining self + other (scalar)" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -509,12 +517,12 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): - def __radd__(self, other): + def __radd__(self, other: int | float) -> ADC_Result: "Redefining other (scalar) + self" return self.__add__(other) - def __sub__(self, other): + def __sub__(self, other: int | float) -> ADC_Result: "Redefining self - other (scalar)" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -532,7 +540,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): - def __rsub__(self, other): + def __rsub__(self, other: int | float) -> ADC_Result: "Redefining other (scalar) - self" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -550,7 +558,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): - def __mul__(self, other): + def __mul__(self, other: int | float) -> ADC_Result: "Redefining self * other (scalar)" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -567,12 +575,12 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to ADC-Result!") - def __rmul__(self, other): + def __rmul__(self, other: int | float) -> ADC_Result: "Redefining other (scalar) * self" return self.__mul__(other) - def __pow__(self, other): + def __pow__(self, other: int | float) -> ADC_Result: "Redefining self ** other (scalar)" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -588,7 +596,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot power raise \"{other.__class__}\" to ADC-Result!") - def __truediv__(self, other): + def __truediv__(self, other: int | float) -> ADC_Result: "Redefining self / other (scalar)" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -603,7 +611,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!") - def __rtruediv__(self, other): + def __rtruediv__(self, other: int | float) -> ADC_Result: "Redefining other (scalar) / self" if isinstance(other, int) or isinstance(other, float): self.lock.acquire() @@ -617,14 +625,14 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!") - def __floordiv__(self, other): + def __floordiv__(self, other: int) -> ADC_Result: "Redefining self // other (scalar)" 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 = [] + tmp_y: list[numpy.ndarray] = [] 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) @@ -634,14 +642,14 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!") - def __rfloordiv__(self, other): + def __rfloordiv__(self, other: int) -> ADC_Result: "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 = [] + tmp_y: list[numpy.ndarray] = [] 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) @@ -652,7 +660,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to ADC-Result!") - def __neg__(self): + def __neg__(self) -> ADC_Result: "Redefining -self" self.lock.acquire() tmp_y = [] @@ -666,7 +674,7 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): return r -def read_from_hdf(hdf_node): +def read_from_hdf(hdf_node: object) -> ADC_Result | None: """ read accumulation data from HDF node and return it. """ diff --git a/src/damaris/data/Accumulation.py b/src/damaris/data/Accumulation.py index 4519a09..6b2666b 100644 --- a/src/damaris/data/Accumulation.py +++ b/src/damaris/data/Accumulation.py @@ -25,6 +25,12 @@ import datetime # added by Oleg Petrov import ctypes # added by Oleg Petrov import struct # added by Oleg Petrov import os # added by Oleg Petrov +from typing import IO + +_XArray = numpy.ndarray | list +_YChannels = list[numpy.ndarray] +_IndexList = list[tuple[int, int]] +_Description = dict[str, str] class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): @@ -50,7 +56,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): x (list): The x-axis data points. y (list): The y-axis data points. """ - def __init__(self, x = None, y = None, y_2 = None, n = None, index = None, sampl_freq = None, error = False): + def __init__(self, x: _XArray | None = None, y: _YChannels | None = None, y_2: _YChannels | numpy.ndarray | None = None, n: int | None = None, index: _IndexList | None = None, sampl_freq: float | None = None, error: bool = False) -> None: Errorable.__init__(self) Drawable.__init__(self) @@ -102,7 +108,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError("Wrong usage of __init__!") - def _copy_provenance(self, r): + def _copy_provenance(self, r: "Accumulation") -> None: r.is_clipped = self.is_clipped r.job_id = self.job_id r.job_ids = self.job_ids.copy() @@ -110,7 +116,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): if self.common_descriptions is not None: r.common_descriptions = self.common_descriptions.copy() - def get_accu_by_index(self, index: int): + def get_accu_by_index(self, index: int) -> "Accumulation": self.lock.acquire() try: start = self.index[index][0] @@ -138,13 +144,13 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.lock.release() return r - def get_result_by_index(self, index): + def get_result_by_index(self, index: int) -> "Accumulation": return self.get_accu_by_index(index) - def get_Result_by_index(self, index): + def get_Result_by_index(self, index: int) -> "Accumulation": return self.get_accu_by_index(index) - def get_ysquare(self, channel): + def get_ysquare(self, channel: int) -> numpy.ndarray | None: if self.uses_statistics(): try: return self.y_square[channel] @@ -172,7 +178,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): # external interface -------------------------------------------------------------------- - def get_yerr(self, channel): + def get_yerr(self, channel: int) -> numpy.ndarray | list: """ return error (std.dev/sqrt(n)) of mean """ @@ -196,7 +202,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): # sample standard deviation / sqrt(n) return numpy.nan_to_num(numpy.sqrt(variance_over_n)) - def get_ydata(self, channel): + def get_ydata(self, channel: int = 0) -> numpy.ndarray | list: """ return mean data """ @@ -214,7 +220,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): return tmp_y - def get_ymin(self): + def get_ymin(self) -> float|int: if not self.contains_data(): return 0 @@ -249,10 +255,10 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.lock.release() return max(tmp_max) - def get_job_id(self) -> int: + def get_job_id(self) -> int | None: return self.job_id # modified by Oleg Petrov - def write_to_csv(self, destination=sys.stdout, delimiter=" "): + def write_to_csv(self, destination: str | IO[str] = sys.stdout, delimiter: str = " ") -> None: """ writes the data to a file. destination can be a filehandle or a filename, default sys.stdout @@ -263,7 +269,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: self._write_to_csv_dest(destination, delimiter) - def _write_to_csv_dest(self, dest, delimiter): + def _write_to_csv_dest(self, dest: IO[str], delimiter: str) -> None: dest.write("# accumulation %d\n"%self.n) self.lock.acquire() try: @@ -296,7 +302,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.lock.release() # ------------- added by Oleg Petrov, 14 Feb 2012 ---------------------- - def write_to_simpson(self, destination=sys.stdout, delimiter=" ", frequency=100e6): + def write_to_simpson(self, destination: str | IO[str] = sys.stdout, delimiter: str = " ", frequency: float = 100e6) -> None: """ writes the data to a text file or sys.stdout in Simpson format, for further processing with the NMRnotebook software; @@ -308,7 +314,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: self._write_to_simpson_dest(destination, delimiter, frequency) - def _write_to_simpson_dest(self, dest, delimiter, frequency): + def _write_to_simpson_dest(self, dest: IO[str], delimiter: str, frequency: float) -> None: self.lock.acquire() try: xdata=self.get_xdata() @@ -329,11 +335,11 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.lock.release() # ------------- added by Oleg Petrov, 10 Sep 2013 ----------------------- - def write_to_tecmag(self, destination=sys.stdout, nrecords=1,\ - frequency=100.,\ - last_delay = 1.,\ - receiver_phase=0.,\ - nucleus='1H'): + def write_to_tecmag(self, destination = sys.stdout, nrecords: int = 1,\ + frequency: float = 100.,\ + last_delay: float = 1.,\ + receiver_phase: float = 0.,\ + nucleus: str = '1H') -> None: """ writes the data to a binary file in TecMag format; destination can be a file object or a filename; @@ -456,7 +462,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): # ----------------------------------------------------------------------- - def write_to_hdf(self, hdffile, where, name, title, complib=None, complevel=None): + def write_to_hdf(self, hdffile: tables.File, where: str, name: str, title: str, complib: str | None = None, complevel: int | None = None) -> None: """ Write the accumulation data to an HDF5 file. @@ -617,7 +623,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): return tmp_string - def __add__(self, other): + def __add__(self, other: "int | float | ADC_Result | Accumulation | None") -> "Accumulation": """ Redefining self + other We define this for different types of other: @@ -801,22 +807,22 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError("can not add "+repr(type(other))+" to Accumulation") - def __radd__(self, other): + def __radd__(self, other: "int | float | ADC_Result | Accumulation | None") -> "Accumulation": """Redefining other + self""" return self.__add__(other) - def __sub__(self, other): + def __sub__(self, other: "int | float | ADC_Result | Accumulation") -> "Accumulation": """Redefining self - other""" return self.__add__(-other) - def __rsub__(self, other): + def __rsub__(self, other: "int | float | ADC_Result | Accumulation") -> "Accumulation": """Redefining other - self""" return (-self) + other - def __mul__(self, other): + def __mul__(self, other: int | float) -> "Accumulation": """Redefining self * other (scalar)""" if isinstance(other, (int, float)): if not self.contains_data(): @@ -843,11 +849,11 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to Accumulation!") - def __rmul__(self, other): + def __rmul__(self, other: int | float) -> "Accumulation": """Redefining other (scalar) * self""" return self.__mul__(other) - def __imul__(self, other): + def __imul__(self, other: int | float) -> "Accumulation": """Redefining self *= other (scalar)""" if isinstance(other, (int, float)): if not self.contains_data(): @@ -862,7 +868,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to Accumulation!") - def __truediv__(self, other): + def __truediv__(self, other: int | float) -> "Accumulation": """Redefining self / other (scalar)""" if isinstance(other, (int, float)): if not self.contains_data(): @@ -889,7 +895,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") - def __itruediv__(self, other): + def __itruediv__(self, other: int | float) -> "Accumulation": """Redefining self /= other (scalar)""" if isinstance(other, (int, float)): if not self.contains_data(): @@ -904,7 +910,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") - def __floordiv__(self, other): + def __floordiv__(self, other: int) -> "Accumulation": """Redefining self // other (scalar)""" if isinstance(other, float): raise ValueError("ValueError: Cannot use floor division (//) on floats! Use \"/\" instead of \"//\"! ") @@ -913,7 +919,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): if not self.contains_data(): raise ValueError("Accumulation: You cant divide an empty accumulation") self.lock.acquire() - tmp_y = [] + tmp_y: list[numpy.ndarray] = [] for i in range(self.get_number_of_channels()): tmp_y.append(self.y[i] // other) @@ -929,7 +935,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") - def __ifloordiv__(self, other): + def __ifloordiv__(self, other: int) -> "Accumulation": """Redefining self //= other (scalar)""" if isinstance(other, float): raise ValueError("ValueError: Cannot use floor division (//) on floats! Use \"/\" instead of \"//\"! ") @@ -950,7 +956,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") - def __iadd__(self, other): + def __iadd__(self, other: "int | float | ADC_Result | Accumulation | None") -> "Accumulation": """Redefining self += other""" # Float or int if isinstance(other, int) or isinstance(other, float): @@ -1123,12 +1129,12 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): raise ValueError("can not add "+repr(type(other))+" to Accumulation") - def __isub__(self, other): + def __isub__(self, other: "int | float | ADC_Result | Accumulation") -> "Accumulation": """Redefining self -= other""" return self.__iadd__(-other) - def __neg__(self): + def __neg__(self) -> "Accumulation": """Redefining -self""" if not self.contains_data(): @@ -1149,7 +1155,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): return r -def read_from_hdf(hdf_node): +def read_from_hdf(hdf_node: object) -> "Accumulation | None": """ read accumulation data from HDF node and return it. """ diff --git a/src/damaris/data/Drawable.py b/src/damaris/data/Drawable.py index b1990f5..5689425 100644 --- a/src/damaris/data/Drawable.py +++ b/src/damaris/data/Drawable.py @@ -1,5 +1,7 @@ # -*- coding: iso-8859-1 -*- +from __future__ import annotations + ############################################################################# # # # Name: Class Drawable # @@ -8,23 +10,26 @@ # # ############################################################################# +import numpy + + class Drawable: def __init__(self): # Will be set correctly in one of the subclasses - self.x = [] - self.y = [] + self.x: numpy.ndarray | list = [] + self.y: list = [] - self.styles = { } + self.styles: dict = { } - self.xlabel = None - self.ylabel = None + self.xlabel: str | None = None + self.ylabel: str | None = None - self.title = None + self.title: str | None = None - self.legend = { } + self.legend: dict = { } - self.text = {} + self.text: dict = {} self.xmin = 0 self.xmax = 0 @@ -32,12 +37,12 @@ class Drawable: self.ymax = 0 - def get_xdata(self): + def get_xdata(self) -> numpy.ndarray | list: "Returns a reference to the x-Plotdata (array)" return self.x - def set_xdata(self, pos, value): + def set_xdata(self, pos: int, value: float) -> None: "Sets a point in x" try: self.x[pos] = value @@ -45,7 +50,7 @@ class Drawable: raise - def get_ydata(self, channel): + def get_ydata(self, channel: int = 0) -> numpy.ndarray | list: "Returns the y-Plotdata of channel n (array)" try: return self.y[channel] @@ -53,51 +58,52 @@ class Drawable: raise - def set_ydata(self, channel, pos, value): + def set_ydata(self, channel: int, pos: int, value: float) -> None: "Sets a point in y" try: self.y[channel][pos] = value except: raise - def get_number_of_channels(self): + + def get_number_of_channels(self) -> int: "Returns the number of channels in y" return len(self.y) - def get_style(self): + def get_style(self) -> dict: "Returns a reference to plot-styles (dictionary)" return self.styles - def set_style(self, channel, value): + def set_style(self, channel: int | str, value: object) -> None: "Sets a channel to a certain plot-style" if channel in self.styles: print("Drawable Warning: Style key \"%s\" will be overwritten with \"%s\"" % (str(channel), str(value))) self.styles[channel] = str(value) - def get_xlabel(self): + def get_xlabel(self) -> str | None: "Returns the label for the x-axis" return self.xlabel - def set_xlabel(self, label): + def set_xlabel(self, label: object) -> None: "Sets the label for the x-axis" self.xlabel = str(label) - def get_ylabel(self): + def get_ylabel(self) -> str | None: "Gets the label for the y-axis" return self.ylabel - def set_ylabel(self, label): + def set_ylabel(self, label: object) -> None: "Sets the label for the y-axis" self.ylabel = str(label) - def get_text(self, index): + def get_text(self, index: int | str) -> str | None: "Returns labels to be plotted (List)" if index in self.text: return self.text[index] @@ -105,61 +111,68 @@ class Drawable: return None - def set_text(self, index, text): + def set_text(self, index: int | str, text: object) -> None: "Sets labels to be plotted " self.text[index] = str(text) - def get_title(self): + def get_title(self) -> str | None: "Returns the title of the plot" return self.title - def set_title(self, title): + def set_title(self, title: object) -> None: "Sets the title of the plot" self.title = str(title) - def get_legend(self): + def get_legend(self) -> dict: "Returns the legend of the plot (Dictionary)" return self.legend - def set_legend(self, channel, value): + def set_legend(self, channel: int | str, value: object) -> None: "Sets the legend of the plot" if channel in self.legend: print("Drawable Warning: Legend key \"%s\" will be overwritten with \"%s\"" % (str(channel), str(value))) self.legend[channel] = str(value) - def get_xmin(self): + + def get_xmin(self) -> float: "Returns minimun of x" return self.x.min() - def get_xminpos(self): + + def get_xminpos(self) -> float: "Returns smallest positive value of x" mask = self.x > 0 return self.x[mask].min() - def set_xmin(self, xmin): + + def set_xmin(self, xmin: float) -> None: "Sets minimum of x" self.xmin = xmin - def get_xmax(self): + + def get_xmax(self) -> float: "Returns maximum of x" return self.x.max() - def set_xmax(self, xmax): + + def set_xmax(self, xmax: float) -> None: "Sets maximum of x" self.xmax = xmax - def get_ymin(self): + + def get_ymin(self) -> float | int: "Returns minimum of y" if isinstance(self.y, list): return min([yarr.min() for yarr in self.y]) else: return self.y.min() - def get_yminpos(self): + + def get_yminpos(self) -> float | int: "Returns smallest positive value of y" if isinstance(self.y, list): ymins = [] @@ -172,17 +185,20 @@ class Drawable: ymin = self.y[mask].min() return ymin - def set_ymin(self, ymin): + + def set_ymin(self, ymin: float) -> None: "Sets minimum of y" self.ymin = ymin - def get_ymax(self): + + def get_ymax(self) -> float | int: "Returns maximimum of y" if isinstance(self.y, list): return max([yarr.max() for yarr in self.y]) else: return self.y.max() - def set_ymax(self, ymax): + + def set_ymax(self, ymax: float) -> None: "Sets maximum of y" self.ymax = ymax diff --git a/src/damaris/data/Error_Result.py b/src/damaris/data/Error_Result.py index c576265..d2b557d 100644 --- a/src/damaris/data/Error_Result.py +++ b/src/damaris/data/Error_Result.py @@ -46,7 +46,7 @@ class Error_Result(Resultable, Drawable): return False # Nothing to plot - def get_ydata(self): + def get_ydata(self, channel=0): return [0.0] # Nothing to plot diff --git a/src/damaris/data/MeasurementResult.py b/src/damaris/data/MeasurementResult.py index 97d34eb..97cfdcd 100644 --- a/src/damaris/data/MeasurementResult.py +++ b/src/damaris/data/MeasurementResult.py @@ -4,6 +4,7 @@ import sys import tables import numpy import collections +from typing import IO from . import Drawable ## provide gaussian statistics for a series of measured data points @@ -12,7 +13,7 @@ from . import Drawable # internally it keeps the sum, the sum of squares and the number of data points class AccumulatedValue: - def __init__(self, mean=None, mean_err=None, n=None): + def __init__(self, mean: float | None = None, mean_err: float | None = None, n: int | None = None): """ one value with std. deviation can be initialized by: @@ -42,7 +43,7 @@ class AccumulatedValue: self.y=float(mean)*self.n self.y2=float(mean_err)**2*n*(n-1.0)+float(mean)**2*n - def __add__(self,y): + def __add__(self, y: "AccumulatedValue | float") -> "AccumulatedValue": new_one=AccumulatedValue() if isinstance(y, AccumulatedValue): new_one.y=self.y+y.y @@ -54,7 +55,7 @@ class AccumulatedValue: new_one.n=self.n+1 return new_one - def __iadd__(self,y): + def __iadd__(self, y: "AccumulatedValue | float") -> "AccumulatedValue": if isinstance(y, AccumulatedValue): self.y+=y.y self.y2+=y.y2 @@ -65,14 +66,14 @@ class AccumulatedValue: self.n+=1 return self - def copy(self): + def copy(self) -> "AccumulatedValue": a=AccumulatedValue() a.y=self.y a.y2=self.y2 a.n=self.n return a - def mean(self): + def mean(self) -> float | None: """ returns the mean of all added/accumulated values """ @@ -81,7 +82,7 @@ class AccumulatedValue: else: return self.y/self.n - def sigma(self): + def sigma(self) -> float | None: """ returns the standard deviation added/accumulated values """ @@ -97,7 +98,7 @@ class AccumulatedValue: else: return None - def mean_error(self): + def mean_error(self) -> float | None: """ returns the mean's error (=std.dev/sqrt(n)) of all added/accumulated values """ @@ -113,7 +114,7 @@ class AccumulatedValue: else: return None - def __str__(self): + def __str__(self) -> str: if self.n==0: return "no value" elif self.n==1: @@ -121,12 +122,12 @@ class AccumulatedValue: else: return "%g +/- %g (%d accumulations)"%(self.mean(),self.mean_error(),self.n) - def __repr__(self): + def __repr__(self) -> str: return str(self) class MeasurementResult(Drawable.Drawable, collections.UserDict): - def __init__(self, quantity_name): + def __init__(self, quantity_name: str) -> None: """ convenient accumulation and interface to plot functions @@ -138,7 +139,7 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): self.lock = threading.RLock() # get the selected item, if it does not exist, create an empty one - def __getitem__(self, key): + def __getitem__(self, key: float) -> "AccumulatedValue": if key not in self: a=AccumulatedValue() self.data[float(key)]=a @@ -146,14 +147,14 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): else: return self.data[float(key)] - def __setitem__(self,key,value): - if not isinstance(value, AccumulatedValue): - value=AccumulatedValue(float(value)) + def __setitem__(self,key: float,item: "AccumulatedValue | float")-> None: + if not isinstance(item, AccumulatedValue): + item=AccumulatedValue(float(item)) return collections.UserDict.__setitem__(self, float(key), - value) + item) - def __add__(self, right_value): + def __add__(self, right_value: int) -> "MeasurementResult": if right_value==0: new_one=MeasurementResult(self.quantity_name) for key in self.data: @@ -162,10 +163,10 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): else: raise Exception("not implemented") - def get_title(self): + def get_title(self) -> str: return self.quantity_name - def get_xdata(self): + def get_xdata(self) -> numpy.ndarray: """ sorted array of all dictionary entries without Accumulated Value objects with n==0 """ @@ -174,32 +175,37 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): keys.sort() return keys - def get_ydata(self): + def get_ydata(self, channel: int = 0) -> numpy.ndarray: + """ + y data (mean values); the channel argument is accepted for + compatibility with the Drawable interface, MeasurementResult + is single-channel + """ return self.get_xydata()[1] - def get_xydata(self): + def get_xydata(self) -> list[numpy.ndarray]: k=self.get_xdata() v=numpy.array([self.data[key].mean() for key in k], dtype="float64") return [k,v] - def get_errorplotdata(self): + def get_errorplotdata(self) -> list[numpy.ndarray]: k=self.get_xdata() v=numpy.array([self.data[key].mean() for key in k], dtype="float64") e=numpy.array([self.data[key].mean_error() for key in k], dtype="float64") return [k,v,e] - def get_lineplotdata(self): + def get_lineplotdata(self) -> list[numpy.ndarray]: k=self.get_xdata() v=numpy.array(self.y, dtype="float64") return [k, v] - def uses_statistics(self): + def uses_statistics(self) -> bool: """ drawable interface method returns True """ return True - def write_to_csv(self,destination=sys.stdout, delimiter=" "): + def write_to_csv(self,destination: str | IO[str] = sys.stdout, delimiter: str = " ") -> None: """ writes the data to a file or to sys.stdout destination can be a file or a filename @@ -211,7 +217,7 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): else: self._write_to_csv_dest(destination, delimiter) - def _write_to_csv_dest(self, dest, delimiter): + def _write_to_csv_dest(self, dest: IO[str], delimiter: str) -> None: dest.write("# quantity:"+str(self.quantity_name)+"\n") dest.write("# x y ysigma n\n") for x in self.get_xdata(): @@ -228,7 +234,7 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): y.n)) - def write_to_hdf(self, hdffile, where, name, title, complib=None, complevel=None): + def write_to_hdf(self, hdffile: tables.File, where: str, name: str, title: str, complib: str | None = None, complevel: int | None = None) -> None: h5_table_format= { "x" : tables.Float32Col(), "y" : tables.Float32Col(), @@ -271,7 +277,7 @@ class MeasurementResult(Drawable.Drawable, collections.UserDict): mr_table.flush() self.lock.release() -def read_from_hdf(hdf_node): +def read_from_hdf(hdf_node: object) -> "MeasurementResult | None": """ reads a MeasurementResult object from the hdf_node or None if the node is not suitable