From be055c6d34d57f0477942da3bd9dd8a26dfbb22d Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 19:44:53 +0100 Subject: [PATCH 1/4] [FIX] read creation names from the stack on Python 3.12 The name given to generated objects was read by a block copied in 20 places that walked the stack with traceback.walk_stack(None). Since Python 3.12 it starts four frames up instead of two, so in a plain script it walks past the end of the stack and pytta.generate.sweep() raises AttributeError at module level and UnboundLocalError inside a function. One helper, _instanceinfo.caller_name, replaces the copies: it walks from the caller explicitly and returns an empty name when the source line is not available (python -c). Tests run real scripts in a subprocess, since under pytest the stack is deep enough to hide the failure. Co-Authored-By: Claude Fable 5.1 --- pytta/apps/roomir.py | 98 ++-------------------------- pytta/classes/_instanceinfo.py | 14 ++++ pytta/classes/measurement.py | 50 ++------------- pytta/generate.py | 114 +++------------------------------ pytta/iso3741.py | 18 +----- pytta/rooms.py | 50 ++------------- tests/test_creation_name.py | 56 ++++++++++++++++ 7 files changed, 95 insertions(+), 305 deletions(-) create mode 100644 tests/test_creation_name.py diff --git a/pytta/apps/roomir.py b/pytta/apps/roomir.py index 697bdd6..8eb4a0c 100644 --- a/pytta/apps/roomir.py +++ b/pytta/apps/roomir.py @@ -67,7 +67,7 @@ from os import getcwd, listdir, mkdir from os.path import isfile, join, exists import copy as cp -import traceback +from pytta.classes._instanceinfo import caller_name # Dict with the measurementKinds # TO DO FUTURE: add 'inchcalibration', 'outchcalibration' @@ -2236,21 +2236,7 @@ def RT(self, roomirsGetDict, decay=20, IREndManualCut=None): each source-receiver configuration, which is a key; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # TR in all positions and averagßes # TR_avgs = {'S1R1': [Analysis_avg1, An_avg2, ..., An_avgn]} @@ -2320,21 +2306,7 @@ def G_Lps(self, recalibirsGetDict): level; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() recalibirs = recalibirsGetDict # Calculating Lps for each recalibir MeasuredThing @@ -2401,21 +2373,7 @@ def G_Lpe_inSitu(self, roomirsGetDict, IREndManualCut=None): key; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # Lpe in all positions and averagßes # Lpe_avgs = {'S1R1': [Analysis_avg1, An_avg2, ..., An_avgn]} @@ -2461,21 +2419,7 @@ def G_Lpe_revCh(self, roomirsGetDict, IREndManualCut=None): all the reverberation chamber's impulsive responses; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() SigObjs = [] if isinstance(roomirsGetDict, dict): @@ -2548,21 +2492,7 @@ def G(self, Lpe_avgs, Lpe_revCh, V_revCh, T_revCh, Lps_revCh, Lps_inSitu): source-receiver configuration, which is a key; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # G (omfg) calculation # G averages in all positions @@ -2627,21 +2557,7 @@ def G_T_revCh(self, roomirsGetDict, IREndManualCut=None, T=20): all the reverberation chamber's impulsive responses; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() roomirs = roomirsGetDict RTs = [] diff --git a/pytta/classes/_instanceinfo.py b/pytta/classes/_instanceinfo.py index ab2a11c..cd98adf 100644 --- a/pytta/classes/_instanceinfo.py +++ b/pytta/classes/_instanceinfo.py @@ -13,11 +13,25 @@ """ +import sys import time import traceback import threading +def caller_name(): + """ + Name assigned on the module-level line that led to the current call: + 'x' for ``x = pytta.generate.sweep()``. The text before '=' (the whole + line if there is none), or '' when the source line is not available. + """ + frame = sys._getframe(2) # the caller of the function asking + while frame.f_code.co_varnames and frame.f_back is not None: + frame = frame.f_back + line = traceback.extract_stack(frame, 1)[0].line or '' + return line.split('=')[0].strip() + + class InstanceCreationError(Exception): pass diff --git a/pytta/classes/measurement.py b/pytta/classes/measurement.py index 7440531..0910647 100644 --- a/pytta/classes/measurement.py +++ b/pytta/classes/measurement.py @@ -10,7 +10,7 @@ from pytta.classes.streaming import Monitor, Streaming from pytta.classes.signal import SignalObj, ImpulsiveResponse from pytta import _h5utils as _h5 -import traceback +from pytta.classes._instanceinfo import caller_name # Measurement class @@ -418,21 +418,7 @@ def run(self): Run method: starts recording during Tmax seconds Outputs a signalObj with the recording content """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # Record recording = sd.rec(frames=self.numSamples, @@ -556,21 +542,7 @@ def run(self): Starts reproducing the excitation signal and recording at the same time Outputs a signalObj with the recording content """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(framenline, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() timeStamp = time.ctime(time.time()) recording = sd.playrec(self.excitation.timeSignal* @@ -807,21 +779,7 @@ def run(self): Starts reproducing the excitation signal and recording at the same time Outputs the transferfunction ImpulsiveResponse """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() recording = super().run() transferfunction = ImpulsiveResponse(self.excitation, diff --git a/pytta/generate.py b/pytta/generate.py index 1979f54..da441f7 100644 --- a/pytta/generate.py +++ b/pytta/generate.py @@ -37,7 +37,7 @@ from pytta.classes import OctFilter as _OctFilter from scipy import signal as ss import numpy as np -import traceback +from pytta.classes._instanceinfo import caller_name from warnings import warn @@ -75,21 +75,7 @@ def sin(Arms=0.5, Nothing to say; """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() if fftDegree is not None: timeLength = 2**(fftDegree)/samplingRate @@ -150,21 +136,7 @@ def sweep(freqMin=None, """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # It was done like this because a function default argument is a value @@ -342,21 +314,7 @@ def colored_noise(color: str or int = 'white', @author: Chum4k3r """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # It was done like this because a function default argument is a value # assigned at import time, and PyTTa have a default object that handles @@ -472,21 +430,7 @@ def impulse(samplingRate=None, Generates a normalized impulse signal at time zero, with zeros to fill the time length """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() if samplingRate is None: samplingRate = default.samplingRate @@ -581,21 +525,7 @@ def measurement(kind='playrec', - same as for (kind='playrec'); - regularization: [boolean] option for Kirkeby regularization """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # Default Parameters if freqMin is None: @@ -707,21 +637,7 @@ def stream(IO='IO', *args, **kwargs): """ """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() if device is None: device = default.device @@ -800,21 +716,7 @@ def octfilter(order: int = 4, maxFreq: float = 20000, refFreq: float = 1000, base: int = 10) -> _OctFilter: - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() of = _OctFilter(order, nthOct, samplingRate, minFreq, maxFreq, refFreq, base) diff --git a/pytta/iso3741.py b/pytta/iso3741.py index ff321e0..e6cbcc4 100644 --- a/pytta/iso3741.py +++ b/pytta/iso3741.py @@ -21,7 +21,7 @@ from pytta import SignalObj, OctFilter, Analysis import numpy as np from pytta.classes.filter import fractional_octave_frequencies as FOF -import traceback +from pytta.classes._instanceinfo import caller_name def _filter(signal, order: int = 4, @@ -75,21 +75,7 @@ def Lp_ST(sigObjList, nthOct, minFreq, maxFreq, IRManualCut=None): pytta.Analysis object. """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() Leqs = [] diff --git a/pytta/rooms.py b/pytta/rooms.py index 8ab8949..5e36fdb 100644 --- a/pytta/rooms.py +++ b/pytta/rooms.py @@ -37,7 +37,7 @@ from pytta.utils import fractional_octave_frequencies as FOF, freq_to_band from pytta.classes.analysis import _circular_time_shift, _filter, crop_IR, \ cumulative_integration, reverberation_time -import traceback +from pytta.classes._instanceinfo import caller_name import copy as cp from warnings import warn @@ -65,21 +65,7 @@ def G_Lpe(IR, nthOct, minFreq, maxFreq, IREndManualCut=None): :return: Analysis object with the calculated parameter :rtype: Analysis """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # firstChNum = IR.systemSignal.channels.mapping[0] # if not IR.systemSignal.channels[firstChNum].calibCheck: @@ -145,21 +131,7 @@ def G_Lps(IR, nthOct, minFreq, maxFreq): :return: Analysis object with the calculated parameter :rtype: Analysis """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() # firstChNum = IR.systemSignal.channels.mapping[0] # if not IR.systemSignal.channels[firstChNum].calibCheck: @@ -336,21 +308,7 @@ def analyse(obj, *params, For more tips check the examples folder. """ - # Code snippet to guarantee that generated object name is - # the declared at global scope - # for frame, line in traceback.walk_stack(None): - for framenline in traceback.walk_stack(None): - # varnames = frame.f_code.co_varnames - varnames = framenline[0].f_code.co_varnames - if varnames == (): - break - # creation_file, creation_line, creation_function, \ - # creation_text = \ - extracted_text = \ - traceback.extract_stack(framenline[0], 1)[0] - # traceback.extract_stack(frame, 1)[0] - # creation_name = creation_text.split("=")[0].strip() - creation_name = extracted_text[3].split("=")[0].strip() + creation_name = caller_name() warn(DeprecationWarning("Function 'pytta.Analyse' is DEPRECATED and " + "being replaced by the class pytta.RoomAnalysis" + diff --git a/tests/test_creation_name.py b/tests/test_creation_name.py new file mode 100644 index 0000000..2bcc5ea --- /dev/null +++ b/tests/test_creation_name.py @@ -0,0 +1,56 @@ +""" +Tests for the creation name that pytta.generate and the measurement run +methods give to the objects they return: the name assigned on the +module-level line of the calling script. + +The name is read from the call stack, so the tests run real scripts in a +subprocess: under pytest the stack is deep and hides failures that a plain +script hits (Python 3.12 changed where traceback.walk_stack(None) starts). +""" +import os +import subprocess +import sys +import tempfile +import unittest + +ENV = dict(os.environ, MPLBACKEND='Agg') + + +def run(*args): + out = subprocess.run([sys.executable, *args], capture_output=True, + text=True, env=ENV) + return out.returncode, out.stdout.strip(), out.stderr + + +class TestCreationName(unittest.TestCase): + + def test_script_gets_the_module_level_name(self): + # At module level the name is the assigned variable. Inside a + # function the generator still works; the name comes from the + # module-level line that led to the call. + script = ('import pytta\n' + 'x = pytta.generate.sweep()\n' + 'print(x.creation_name)\n' + 'def main():\n' + ' y = pytta.generate.sin()\n' + ' print(y.creation_name)\n' + 'main()\n') + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'script.py') + with open(path, 'w') as f: + f.write(script) + code, out, err = run(path) + self.assertEqual(code, 0, err) + self.assertEqual(out.splitlines(), ['x', 'main()']) + + def test_works_without_source_lines(self): + # Under `python -c` the source of the calling line is not + # available: the object is created with an empty name. + code, out, err = run('-c', 'import pytta; x = pytta.generate.sweep(); ' + 'print(repr(x.creation_name))') + self.assertEqual(code, 0, err) + self.assertEqual(out, "''") + + +if __name__ == '__main__': + unittest.main() From d653f20720ca8106564a4bbc32f396eb6a26f994 Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 19:44:53 +0100 Subject: [PATCH 2/4] [FIX] use scipy.signal.windows for hann and tukey scipy.signal.hann and scipy.signal.tukey were removed in SciPy 1.13. The five calls now use scipy.signal.windows, the same functions with the same defaults. Co-Authored-By: Claude Fable 5.1 --- pytta/_plot.py | 2 +- pytta/classes/signal.py | 2 +- pytta/generate.py | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pytta/_plot.py b/pytta/_plot.py index c0a0970..57fe689 100644 --- a/pytta/_plot.py +++ b/pytta/_plot.py @@ -1136,7 +1136,7 @@ def waterfall(sigObjs, step=2 ** 9, n=2 ** 13, fmin=None, fmax=None, pmin=None, # Apply rolling window sliced = np.asarray(list(more_itertools.windowed(ht, n=n, step=step, fillvalue=0))) window = np.asarray( - [ss.tukey(n, alpha=winAlpha) for s in sliced]) # Tukey window (alpha=0 is a rectangular window) + [ss.windows.tukey(n, alpha=winAlpha) for s in sliced]) # Tukey window (alpha=0 is a rectangular window) windowed = sliced * window # Apply window windowedFFT = abs(2 / n * fft(windowed)) # Apply FFT windowedFFT = windowedFFT[:, idx_min_FFT - fpad: idx_max_FFT + 1 + fpad] # Crop the FFT diff --git a/pytta/classes/signal.py b/pytta/classes/signal.py index 79151c3..7aff876 100644 --- a/pytta/classes/signal.py +++ b/pytta/classes/signal.py @@ -1750,7 +1750,7 @@ def _crossfade_spectruns(self, a, b, freqLims, freqVector): f1idx = np.where(freqVector <= f1)[0][-1] totalSamples = a.shape[0] xsamples = f1idx - f0idx - win = ss.hann(2*xsamples) + win = ss.windows.hann(2*xsamples) rightWin = win[xsamples-1:-1] fullRightWin = np.concatenate((np.ones(f0idx), diff --git a/pytta/generate.py b/pytta/generate.py index da441f7..f9ecd2e 100644 --- a/pytta/generate.py +++ b/pytta/generate.py @@ -238,8 +238,8 @@ def __do_sweep_windowing(inputSweep, # exact sample where the chirp reaches freqMax [Hz] freqMaxSample = np.where(freqSweep <= freqMax) freqMaxSample = len(freqSweep) - freqMaxSample[-1][-1] - windowStart = ss.hann(2*freqMinSample) - windowEnd = ss.hann(2*freqMaxSample) + windowStart = ss.windows.hann(2*freqMinSample) + windowEnd = ss.windows.hann(2*freqMaxSample) # Uses first half of windowStart, last half of windowEnd, and a vector of # ones with the remaining length, in between the half windows @@ -417,7 +417,7 @@ def __do_noise_windowing(inputNoise, window): # sample equivalent to the first five percent of noise duration fivePercentSample = int((5/100) * (noiseSamples)) - windowStart = ss.hann(2*fivePercentSample) + windowStart = ss.windows.hann(2*fivePercentSample) fullWindow = np.concatenate((windowStart[0:fivePercentSample], np.ones(int(noiseSamples-fivePercentSample)))) newNoise = (fullWindow * inputNoise.T).T From 038ebcc3183580f19e8ef91e851c4d37d12efa8e Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 20:23:26 +0100 Subject: [PATCH 3/4] [FIX] round trip of measurements through .pytta files Saving and loading a RecMeasure, PlayRecMeasure or FRFMeasure with pytta.save and pytta.load failed or lost data: - json.dump failed on NumPy scalars (fftDegree is np.int64); every json.dump now converts them with .item(). - ChannelsList refused NumPy integers, which load passes; it now accepts any NumPy number. - load renumbered a channel and then looked it up again by its old number; channels are now restored by position. - RecMeasure was loaded with channel 0, and the output channels of PlayRecMeasure and FRFMeasure were never restored. - the base fields of a measurement (sampling rate, frequency limits, length domain, comment) were not saved, since _to_dict of the measurements did not call the base one, and the channel code was neither saved nor restored. With this, all tests in tests/test_pyttasave.py and tests/test_h5io.py pass. The coordinates and orientation of channels are saved but still not restored. Co-Authored-By: Claude Fable 5.1 --- pytta/classes/_base.py | 5 +++-- pytta/classes/measurement.py | 14 ++++++------ pytta/classes/signal.py | 4 ++-- pytta/functions.py | 42 +++++++++++++++++++++++------------- 4 files changed, 40 insertions(+), 25 deletions(-) diff --git a/pytta/classes/_base.py b/pytta/classes/_base.py index 13cb139..8b46874 100644 --- a/pytta/classes/_base.py +++ b/pytta/classes/_base.py @@ -549,6 +549,7 @@ def _to_dict(self): out = {'calib': [self.CF, self.calibCheck], 'unit': self.unit, 'name': self.name, + 'code': self.code, 'coordinates': self.coordinates._to_dict(), 'orientation': self.orientation._to_dict()} return out @@ -578,11 +579,11 @@ def __init__(self, chList=None): for memb in chList: if type(memb) is ChannelObj: self._channels.append(memb) - elif (type(memb) is int) or (type(memb) is float): + elif isinstance(memb, (int, float, np.number)): self._channels.append(ChannelObj(int(memb))) else: raise TypeError("Could not resolve ChannelsList initialization parameters.") - elif type(chList) is int: + elif isinstance(chList, (int, np.integer)): self._channels.append(ChannelObj(chList)) elif type(chList) is ChannelObj: self._channels.append(chList) diff --git a/pytta/classes/measurement.py b/pytta/classes/measurement.py index 0910647..64afdc3 100644 --- a/pytta/classes/measurement.py +++ b/pytta/classes/measurement.py @@ -92,9 +92,10 @@ def __repr__(self): f'timeLength={self.timeLength!r}') def _to_dict(self): - out = {'device': self.device, - 'inChannels': self.inChannels._to_dict(), - 'outChannels': self.outChannels._to_dict()} + out = super()._to_dict() + out.update({'device': self.device, + 'inChannels': self.inChannels._to_dict(), + 'outChannels': self.outChannels._to_dict()}) return out def _h5_save(self, h5group): @@ -361,6 +362,7 @@ def __repr__(self): def _to_dict(self): sup = super()._to_dict() sup['fftDegree'] = self.fftDegree + sup['timeLength'] = self.timeLength return sup def pytta_save(self, dirname=time.ctime(time.time())): @@ -368,7 +370,7 @@ def pytta_save(self, dirname=time.ctime(time.time())): name = dirname + '.pytta' with zipfile.ZipFile(name, 'w') as zdir: with open('RecMeasure.json', 'w') as f: - json.dump(dic, f, indent=4) + json.dump(dic, f, indent=4, default=lambda o: o.item()) zdir.write('RecMeasure.json') return name @@ -584,7 +586,7 @@ def pytta_save(self, dirname=time.ctime(time.time())): zdir.write(excit) os.remove(excit) with open('PlayRecMeasure.json', 'w') as f: - json.dump(dic, f, indent=4) + json.dump(dic, f, indent=4, default=lambda o: o.item()) zdir.write('PlayRecMeasure.json') os.remove('PlayRecMeasure.json') return name @@ -751,7 +753,7 @@ def pytta_save(self, dirname=time.ctime(time.time())): zdir.write(excit) os.remove(excit) with open('FRFMeasure.json', 'w') as f: - json.dump(dic, f, indent=4) + json.dump(dic, f, indent=4, default=lambda o: o.item()) zdir.write('FRFMeasure.json') os.remove('FRFMeasure.json') return name diff --git a/pytta/classes/signal.py b/pytta/classes/signal.py index 7aff876..7c54b49 100644 --- a/pytta/classes/signal.py +++ b/pytta/classes/signal.py @@ -900,7 +900,7 @@ def pytta_save(self, dirname=time.ctime(time.time())): mySigObj['timeSignalAddress'] = filename filename = 'SignalObj.json' with open(filename, 'w') as f: - json.dump(mySigObj, f, indent=4) + json.dump(mySigObj, f, indent=4, default=lambda o: o.item()) zdir.write(filename, compress_type=zipfile.ZIP_DEFLATED) os.remove(filename) return dirname + '.pytta' @@ -1354,7 +1354,7 @@ def pytta_save(self, dirname=time.ctime(time.time())): out = self._to_dict() out['SignalAddress'] = {'ir': ir} with open('ImpulsiveResponse.json', 'w') as f: - json.dump(out, f, indent=4) + json.dump(out, f, indent=4, default=lambda o: o.item()) zdir.write('ImpulsiveResponse.json') os.remove('ImpulsiveResponse.json') return dirname + '.pytta' diff --git a/pytta/functions.py b/pytta/functions.py index 9b39239..656d61e 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -722,7 +722,7 @@ def pytta_save(fileName: str = time.ctime(time.time()), *PyTTaObjs): zdir.write(sobj) os.remove(sobj) with open('Meta.json', 'w') as f: - json.dump(meta, f, indent=4) + json.dump(meta, f, indent=4, default=lambda o: o.item()) zdir.write('Meta.json') os.remove('Meta.json') return fileName + '.pytta' @@ -766,29 +766,42 @@ def __parse_load(className): os.remove(openJson['SignalAddress']['ir']) elif name == 'RecMeasure': - inch = list(np.arange(len(openJson['inChannels']))) + inch = list(1 + np.arange(len(openJson['inChannels']))) out = RecMeasure(device=openJson['device'], inChannels=inch, - lengthDomain='samples', - fftDegree=openJson['fftDegree']) + lengthDomain=openJson['lengthDomain'], + fftDegree=openJson['fftDegree'], + timeLength=openJson['timeLength'], + samplingRate=openJson['samplingRate'], + freqMin=openJson['freqLims'][0], + freqMax=openJson['freqLims'][1], + comment=openJson['comment']) out.inChannels = __parse_channels(openJson['inChannels'], out.inChannels) elif name == 'PlayRecMeasure': inch = list(1 + np.arange(len(openJson['inChannels']))) + outch = list(1 + np.arange(len(openJson['outChannels']))) excit = pytta_load(openJson['excitationAddress']) out = PlayRecMeasure(excitation=excit, - device=openJson['device'], inChannels=inch) + device=openJson['device'], inChannels=inch, + outChannels=outch, comment=openJson['comment']) out.inChannels = __parse_channels(openJson['inChannels'], out.inChannels) + out.outChannels = __parse_channels(openJson['outChannels'], + out.outChannels) os.remove(openJson['excitationAddress']) elif name == 'FRFMeasure': inch = list(1 + np.arange(len(openJson['inChannels']))) + outch = list(1 + np.arange(len(openJson['outChannels']))) excit = pytta_load(openJson['excitationAddress']) out = FRFMeasure(excitation=excit, device=openJson['device'], - inChannels=inch) + inChannels=inch, + outChannels=outch, comment=openJson['comment']) out.inChannels = __parse_channels(openJson['inChannels'], out.inChannels) + out.outChannels = __parse_channels(openJson['outChannels'], + out.outChannels) os.remove(openJson['excitationAddress']) elif name == 'Meta': @@ -802,15 +815,14 @@ def __parse_load(className): def __parse_channels(chDict, chList): - ch = 1 - for key in chDict.keys(): - chList[ch].num = key - chList[ch].unit = chDict[key]['unit'] - chList[ch].name = chDict[key]['name'] - chList[ch].CF = chDict[key]['calib'][0] - chList[ch].calibCheck\ - = chDict[key]['calib'][1] - ch += 1 + # by position: renumbering a channel must not change which one is next + for channel, key in zip(chList._channels, chDict.keys()): + channel.num = key + channel.unit = chDict[key]['unit'] + channel.name = chDict[key]['name'] + channel.code = chDict[key].get('code', 'Ch' + str(key)) + channel.CF = chDict[key]['calib'][0] + channel.calibCheck = chDict[key]['calib'][1] return chList From cb1f5f99a9e32e2e79d23bac85a5437370853d28 Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 20:51:28 +0100 Subject: [PATCH 4/4] [FIX] restore channel coordinates and orientation from .pytta files The coordinates and orientation of each channel were saved but not restored on load; they now are, and files without them keep the defaults. Setting a point with x = 0, such as an orientation along the y axis, raised ZeroDivisionError because the azimuth was arctan(y / x); it is now arctan2(y, x), which also gives the right quadrant for x < 0. Co-Authored-By: Claude Fable 5.1 --- pytta/classes/_base.py | 3 +-- pytta/functions.py | 5 ++++- tests/test_pytta_channels.py | 39 ++++++++++++++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 3 deletions(-) create mode 100644 tests/test_pytta_channels.py diff --git a/pytta/classes/_base.py b/pytta/classes/_base.py index 8b46874..3ad1542 100644 --- a/pytta/classes/_base.py +++ b/pytta/classes/_base.py @@ -231,8 +231,7 @@ def point(self, newpoint): r = (self._point[0]**2+self._point[1]**2+self._point[2]**2)**(1/2) if r != 0: elev = np.arccos(self._point[2]/r) - azi = np.arctan(self._point[1] / - self._point[0]) + azi = np.arctan2(self._point[1], self._point[0]) else: elev = 0 azi = 0 diff --git a/pytta/functions.py b/pytta/functions.py index 656d61e..7b6a152 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -43,7 +43,7 @@ from pytta.classes import SignalObj, ImpulsiveResponse, \ RecMeasure, PlayRecMeasure, FRFMeasure, \ Analysis, OctFilter -from pytta.classes._base import ChannelsList, ChannelObj +from pytta.classes._base import ChannelsList, ChannelObj, CoordinateObj from pytta.generate import measurement # TODO: Change to class instantiation. from pytta import _h5utils as _h5 import copy as cp @@ -821,6 +821,9 @@ def __parse_channels(chDict, chList): channel.unit = chDict[key]['unit'] channel.name = chDict[key]['name'] channel.code = chDict[key].get('code', 'Ch' + str(key)) + for name in ('coordinates', 'orientation'): + if name in chDict[key]: + setattr(channel, name, CoordinateObj(**chDict[key][name])) channel.CF = chDict[key]['calib'][0] channel.calibCheck = chDict[key]['calib'][1] return chList diff --git a/tests/test_pytta_channels.py b/tests/test_pytta_channels.py new file mode 100644 index 0000000..e549aa7 --- /dev/null +++ b/tests/test_pytta_channels.py @@ -0,0 +1,39 @@ +""" +Test for the channel metadata of SignalObj through .pytta files: the +coordinates and orientation of a channel are saved and restored. +""" +import os +import tempfile +import unittest + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import numpy as np + +import pytta + + +class TestPyttaChannels(unittest.TestCase): + + def test_coordinates_and_orientation_round_trip(self): + sig = pytta.SignalObj(np.ones(480), 'time', 48000) + sig.channels[1].coordinates = pytta.classes._base.CoordinateObj( + point=[1.0, 2.0, 3.0], ref='stage centre', unit='m') + sig.channels[1].orientation = [0.0, 1.0, 0.0] + with tempfile.TemporaryDirectory() as folder: + cwd = os.getcwd() + os.chdir(folder) # pytta.save writes next to the file + try: + pytta.save('channels.pytta', sig) + loaded = pytta.load('channels.pytta') + finally: + os.chdir(cwd) + channel = (loaded[0] if isinstance(loaded, list) else loaded).channels[1] + self.assertEqual(repr(channel.coordinates), + repr(sig.channels[1].coordinates)) + self.assertEqual(repr(channel.orientation), + repr(sig.channels[1].orientation)) + + +if __name__ == '__main__': + unittest.main()