From be055c6d34d57f0477942da3bd9dd8a26dfbb22d Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 19:44:53 +0100 Subject: [PATCH 1/3] [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/3] [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 718dbd56e163573d8933ee8fbbfde7c9cadeb33f Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sun, 1 Nov 2026 14:19:40 +0100 Subject: [PATCH 3/3] [FIX] peak_time with a negative peak, and regularization of short signals Co-Authored-By: Claude Fable 5.1 --- pytta/classes/signal.py | 4 +- pytta/functions.py | 6 +-- tests/test_peak_and_short_regularization.py | 51 +++++++++++++++++++++ 3 files changed, 57 insertions(+), 4 deletions(-) create mode 100644 tests/test_peak_and_short_regularization.py diff --git a/pytta/classes/signal.py b/pytta/classes/signal.py index 7aff876..7fd781c 100644 --- a/pytta/classes/signal.py +++ b/pytta/classes/signal.py @@ -1749,7 +1749,9 @@ def _crossfade_spectruns(self, a, b, freqLims, freqVector): f0idx = np.where(freqVector >= f0)[0][0] f1idx = np.where(freqVector <= f1)[0][-1] totalSamples = a.shape[0] - xsamples = f1idx - f0idx + # A crossfade narrower than one frequency bin (a short signal) + # becomes a switch at the first bin from f0. + xsamples = max(f1idx - f0idx, 0) win = ss.windows.hann(2*xsamples) rightWin = win[xsamples-1:-1] diff --git a/pytta/functions.py b/pytta/functions.py index 9b39239..b903ef5 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -265,9 +265,9 @@ def peak_time(signal): raise TypeError('Signal must be an SignalObj.') peaks_time = [] for chindex in range(signal.numChannels): - maxamp = max(np.abs(signal.timeSignal[:, chindex])) - maxindex = np.where(signal.timeSignal[:, chindex] == np.abs(maxamp))[0] - maxtime = signal.timeVector[maxindex][0] + # the largest magnitude, positive or negative + maxindex = np.argmax(np.abs(signal.timeSignal[:, chindex])) + maxtime = signal.timeVector[maxindex] peaks_time.append(maxtime) if signal.numChannels > 1: return peaks_time diff --git a/tests/test_peak_and_short_regularization.py b/tests/test_peak_and_short_regularization.py new file mode 100644 index 0000000..51e1bc1 --- /dev/null +++ b/tests/test_peak_and_short_regularization.py @@ -0,0 +1,51 @@ +""" +Tests for peak_time with a negative peak and for the regularized +deconvolution of ImpulsiveResponse with short signals, whose crossfade +below freqMin is narrower than one frequency bin. +""" +import os +import unittest + +import numpy as np + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import pytta + +FS = 48000 + + +def signal(*samples, **kwargs): + # A tuple as first local variable: SignalObj compares it with itself + # while walking the stack, which an array cannot do before #59. + return pytta.SignalObj(samples[0], 'time', kwargs.pop('fs', FS), **kwargs) + + +class TestPeakAndShortRegularization(unittest.TestCase): + + def test_peak_time_of_a_negative_peak(self): + x = np.zeros((1000, 2)) + x[300, 0], x[100, 0] = -1, 0.5 # negative peak at 0.3 s + x[700, 1] = 1 # positive peak at 0.7 s + np.testing.assert_allclose(pytta.peak_time(signal(x, fs=1000)), [0.3, 0.7]) + + def test_short_signals_deconvolve(self): + # 4096 samples at 48 kHz: bins of 11.7 Hz, wider than the crossfade + # from freqMin / sqrt(2) to freqMin. The regularized response equals + # the known one between 100 Hz and 10 kHz. + rng = np.random.default_rng(3) + x = rng.standard_normal(4096) * 0.1 + h = np.exp(-np.arange(200) / 30) * rng.standard_normal(200) + y = np.fft.irfft(np.fft.rfft(x) * np.fft.rfft(h, 4096), 4096) + ir = pytta.ImpulsiveResponse(signal(x), signal(y), 'linear') + f = np.fft.rfftfreq(4096, 1 / FS) + band = (f > 100) & (f < 10000) + H = np.fft.rfft(h, 4096) + M = np.fft.rfft(np.asarray(ir.IR.timeSignal[:, 0], dtype=float)) + e = np.abs(M[band] - H[band]).max() / np.abs(H[band]).max() + print(f'\n short regularized deconvolution: {e:.1e}', end='') + self.assertLess(e, 1e-4) # float32 samples + + +if __name__ == '__main__': + unittest.main()