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

add type hints to data classes and fix get_ydata override conflict

Browse Source 
- signature-only type hints for ADC_Result, MeasurementResult,
  Accumulation, and the Drawable base class (x: ndarray | list, y: list)
- get_ydata now takes channel: int = 0 across the hierarchy
  (Drawable, Accumulation, MeasurementResult, Error_Result), fixing
  the LSP violation flagged by pyrefly
- widen _XArray alias to ndarray | list: internal copy constructors
  pass self.x[:] which is that union
- add hinted files to pyrefly project-includes

pyrefly errors in the covered files drop from 102 to 74; the rest is
PyTables stub noise and pre-existing latent bugs left for separate fixes.

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
markusro committed 2026-10-18 15:09:57 +02:00
1 parent e5472484cb
commit c797aeb387
6 files changed
+174 -135
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/damaris/data/Accumulation.py
+42 -36
View File
@@ -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.
"""