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 c7eb008d6aa8cc5a7ca1436a9071588928b31476 Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sun, 1 Nov 2026 13:57:48 +0100 Subject: [PATCH 3/4] [FIX] window the impulse response before filtering for C80, D50 and support RoomAnalysis took C80, D50, STearly and STlate from the energy decay curve of each band filtered response, so the delay of the band filters moved early energy past the time limits (about 1 dB low C80 at 125 Hz). The broadband response is now windowed before the filtering, as ISO 3382-1:2009, A.3.4, recommends, with the Lundeby truncation and tail correction kept for the late windows. STlate now ends at 1000 ms (ISO 3382-1, C.2), and bypassLundeby=True no longer raises UnboundLocalError. Co-Authored-By: Claude Fable 5.1 --- pytta/classes/analysis.py | 69 ++++++++++++++++++--- tests/test_room_energy_windowing.py | 95 +++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 8 deletions(-) create mode 100644 tests/test_room_energy_windowing.py diff --git a/pytta/classes/analysis.py b/pytta/classes/analysis.py index 7f15924..19d5bd0 100644 --- a/pytta/classes/analysis.py +++ b/pytta/classes/analysis.py @@ -993,15 +993,29 @@ def estimate_energy_parameters(ir: SignalObj, bands: np.ndarray, A dict with parameters by name. """ - listEDC, fhSignal = cumulative_integration(ir, bypassLundeby, plotLundeby, suppressWarnings, **kwargs) + compensation = [] + listEDC, fhSignal = cumulative_integration(ir, bypassLundeby, plotLundeby, suppressWarnings, + compensation=compensation, **kwargs) params = {} params['rms'] = fhSignal.rms() params['SPL'] = fhSignal.spl() params['Ts'] = central_time(fhSignal.timeSignal**2, fhSignal.timeVector) - params['D50'] = definition(listEDC, ir.samplingRate) - params['C80'] = clarity(listEDC, ir.samplingRate) - params['STearly'] = st_early(listEDC, ir.samplingRate) - params['STlate'] = st_late(listEDC, ir.samplingRate) + # The parameters with time windows window the broadband response and + # filter each window afterwards (ISO 3382-1:2009, A.3.4), so that the + # delay of the band filters does not move energy across the limits; + # the late windows keep the Lundeby truncation and tail correction. + early50 = windowed_band_energy(ir, 0, 0.05, **kwargs) + late50 = windowed_band_energy(ir, 0.05, None, compensation, **kwargs) + early80 = windowed_band_energy(ir, 0, 0.08, **kwargs) + late80 = windowed_band_energy(ir, 0.08, None, compensation, **kwargs) + first10 = windowed_band_energy(ir, 0, 0.01, **kwargs) + with np.errstate(divide='ignore', invalid='ignore'): + params['D50'] = np.round(100 * early50 / (early50 + late50), 2) + params['C80'] = np.round(10 * np.log10(early80 / late80), 2) + params['STearly'] = np.round(10 * np.log10( + windowed_band_energy(ir, 0.02, 0.1, **kwargs) / first10), 4) + params['STlate'] = np.round(10 * np.log10( + windowed_band_energy(ir, 0.1, 1.0, **kwargs) / first10), 4) params['EDT'] = reverberation_time('EDT', listEDC) params['T20'] = reverberation_time(20, listEDC) params['T30'] = reverberation_time(30, listEDC) @@ -1417,6 +1431,7 @@ def energy_decay_calculation(band, timeSignal, timeVector, samplingRate, else: interIdx = 0 lateRT = 1 + lundebyParams = None if interIdx == 0: interIdx = -1 @@ -1437,15 +1452,19 @@ def energy_decay_calculation(band, timeSignal, timeVector, samplingRate, print(band, "[Hz] band: could not estimate C factor") C = 0 energyDecay = np.zeros(truncatedTimeVector.size) - return (energyDecay, truncatedTimeVector, lundebyParams) + return (energyDecay, truncatedTimeVector, lundebyParams, (interIdx, C)) def cumulative_integration(inputSignal, bypassLundeby, plotLundebyResults, suppressWarnings=True, + compensation=None, **kwargs): - """Cumulative integration with proper corrections.""" + """Cumulative integration with proper corrections. + + If a list is given as compensation, the Lundeby truncation index and the + tail energy correction C of each band are appended to it.""" def plot_lundeby(): c0, c1, interIdx, BGL = lundebyParams @@ -1476,7 +1495,7 @@ def plot_lundeby(): for ch in range(hSignal.numChannels): signal = hSignal[ch] band = bands[ch] - energyDecay, energyVector, lundebyParams = \ + energyDecay, energyVector, lundebyParams, bandCompensation = \ energy_decay_calculation(band, signal.timeSignal, signal.timeVector, @@ -1487,6 +1506,8 @@ def plot_lundeby(): bypassLundeby, suppressWarnings=suppressWarnings) listEDC.append((energyDecay, energyVector)) + if compensation is not None: + compensation.append(bandCompensation) if plotLundebyResults: # Placed here because Numba can't handle plots. # plot_lundeby(band, timeVector, timeSignal, samplingRate, # lundebyParams) @@ -1530,6 +1551,38 @@ def reverberation_time(decay, listEDC): return np.array(RT, dtype='float32') +def windowed_band_energy(ir, start, stop=None, compensation=None, **kwargs): + """ + Energy in each band of the part of a broadband impulse response between + start and stop seconds (stop None for the end), windowed before the band + filtering, as ISO 3382-1:2009, A.3.4, recommends for the early and late + energy of C80, D50 and the support parameters. The window is filtered + with the filters of _filter (kwargs nthOct, minFreq, maxFreq) after one + second of zeros is appended, so that the energy the filters delay past + the end of the response is kept. + + compensation is the list of (Lundeby truncation index, tail correction + C) of each band that cumulative_integration fills; with it, the filtered + window is cut at the truncation index of each band and C is added, as + the energy decay curve does for the late part of the response. + + Returns an array with one energy per band. + """ + fs = ir.samplingRate + x = np.asarray(ir.timeSignal, dtype=float)[:, 0] + i0 = int(round(start * fs)) + i1 = x.size if stop is None else min(int(round(stop * fs)), x.size) + window = np.zeros(x.size + fs) + window[i0:i1] = x[i0:i1] + y = np.asarray(_filter(SignalObj(window, 'time', fs, signalType='energy'), + **kwargs).timeSignal, dtype=float) + y = y.reshape(y.shape[0], -1) + if compensation is None: + return np.sum(y**2, axis=0) + return np.array([np.sum(y[:cut, band]**2) + C + for band, (cut, C) in enumerate(compensation)]) + + def definition(listEDC: list, fs: int, t: int = 50) -> np.ndarray: """ Room parameter. diff --git a/tests/test_room_energy_windowing.py b/tests/test_room_energy_windowing.py new file mode 100644 index 0000000..33aec3d --- /dev/null +++ b/tests/test_room_energy_windowing.py @@ -0,0 +1,95 @@ +""" +Tests for the energy parameters of pytta.RoomAnalysis computed by windowing +the broadband impulse response before the band filtering (ISO 3382-1:2009, +A.3.4). The responses are synthetic, so the reference values are known. +""" +import os +import unittest + +import numpy as np +import scipy.signal as ss + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import pytta +from pytta.classes.analysis import clarity + +FS = 48000 +BANDS = dict(nthOct=1, minFreq=125, maxFreq=4000) + + +def decay(seed, T=1.0, seconds=1.5): + # Exponentially decaying white noise: the energy falls by 60 dB in T. + rng = np.random.default_rng(seed) + t = np.arange(int(seconds * FS)) / FS + x = np.exp(-6.9078 * t / T) * rng.standard_normal(t.size) + return 0.5 * x / np.abs(x).max() + + +def room(*samples, **kwargs): + # A tuple as first local variable: SignalObj compares it with itself + # while walking the stack, which an array cannot do before #59. + ir = pytta.SignalObj(samples[0], 'time', FS) + return pytta.RoomAnalysis(ir, suppressWarnings=True, **BANDS, **kwargs) + + +class TestRoomEnergyWindowing(unittest.TestCase): + + def test_c80_follows_iso_3382_1_a34(self): + # The early 80 ms and the rest of the cropped broadband response, + # filtered separately with the same band filters (their sos from + # OctFilter, applied with scipy) and summed over the whole output. + filters = pytta.OctFilter(order=4, samplingRate=FS, refFreq=1000, base=10, **BANDS) + k = int(0.08 * FS) + for seed in (1, 2, 3): + r = room(decay(seed), bypassLundeby=True) + x = np.r_[np.asarray(r.ir.timeSignal[:, 0], dtype=float), np.zeros(FS)] + early, late = x.copy(), x.copy() + early[k:], late[:k] = 0, 0 + ref = [10 * np.log10(np.sum(ss.sosfilt(np.ascontiguousarray(filters.sos[:, :, b]), early)**2) + / np.sum(ss.sosfilt(np.ascontiguousarray(filters.sos[:, :, b]), late)**2)) + for b in range(filters.sos.shape[2])] + e = np.abs(r.C80 - ref).max() + print(f'\n C80 against A.3.4, response {seed}: {e:.3f} dB', end='') + self.assertLess(e, 0.011) # C80 is rounded to 0.01 dB + + def test_no_bias_against_the_exponential_decay(self): + # For T = 1 s the expected values are C80 = 10 lg(e^(13.8155 0.08) + # - 1) = 3.05 dB and D50 = 1 - e^(-13.8155 0.05) = 49.88 %. Over 20 + # responses the means stay within three standard errors in every + # band, where the energy decay curve of the filtered response, used + # before, is low by about 1.4 dB at 125 Hz. + new, old, d50 = [], [], [] + for seed in range(20): + r = room(decay(100 + seed)) + new.append(r.C80) + d50.append(r.D50) + old.append(clarity(r.listEDC, FS)) + a = 13.8155 + for name, values, expected in (('C80 windowed', new, 10 * np.log10(np.exp(a * 0.08) - 1)), + ('D50 windowed', d50, 100 * (1 - np.exp(-a * 0.05))), + ('C80 from the EDC', old, 10 * np.log10(np.exp(a * 0.08) - 1))): + v = np.array(values) + z = (v.mean(0) - expected) / (v.std(0, ddof=1) / np.sqrt(len(v))) + print(f'\n {name}: mean - expected {np.round(v.mean(0) - expected, 2)}, ' + f'in standard errors {np.round(z, 1)}', end='') + if name.endswith('windowed'): + self.assertTrue(np.all(np.abs(z) < 3)) + else: + self.assertLess(z[0], -3) + + def test_stlate_ends_at_one_second(self): + # Impulses of energy 1 at the start, 0.25 0.5 s later and 1 1.2 s + # later: STlate = 10 lg(0.25 / 1) = -6.02 dB in every band, since the + # last impulse lies outside its window (ISO 3382-1, C.2). Silence + # before the first impulse lets the start detection find it. + x = np.zeros(int(1.5 * FS)) + n0 = int(0.01 * FS) + x[[n0, n0 + int(0.5 * FS), n0 + int(1.2 * FS)]] = 1, 0.5, 1 + r = room(x, bypassLundeby=True, ircut=1.45) + print(f'\n STlate {np.round(r.STlate, 3)}', end='') + np.testing.assert_allclose(r.STlate, 10 * np.log10(0.25), atol=0.01) + + +if __name__ == '__main__': + unittest.main() From cc9de0e63939d10800c0ceb9d257dadceee3cc67 Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sun, 1 Nov 2026 13:59:05 +0100 Subject: [PATCH 4/4] [FIX] let the late windows subtract a noise level per sample Co-Authored-By: Claude Fable 5.1 --- pytta/classes/analysis.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/pytta/classes/analysis.py b/pytta/classes/analysis.py index 19d5bd0..142f5be 100644 --- a/pytta/classes/analysis.py +++ b/pytta/classes/analysis.py @@ -1564,7 +1564,10 @@ def windowed_band_energy(ir, start, stop=None, compensation=None, **kwargs): compensation is the list of (Lundeby truncation index, tail correction C) of each band that cumulative_integration fills; with it, the filtered window is cut at the truncation index of each band and C is added, as - the energy decay curve does for the late part of the response. + the energy decay curve does for the late part of the response. A third + element, the noise energy per sample, is subtracted from every sample + of the window up to the truncation, for a decay curve computed by noise + subtraction. Returns an array with one energy per band. """ @@ -1579,8 +1582,13 @@ def windowed_band_energy(ir, start, stop=None, compensation=None, **kwargs): y = y.reshape(y.shape[0], -1) if compensation is None: return np.sum(y**2, axis=0) - return np.array([np.sum(y[:cut, band]**2) + C - for band, (cut, C) in enumerate(compensation)]) + energy = [] + for band, comp in enumerate(compensation): + cut, C = comp[:2] + noise = comp[2] if len(comp) > 2 else 0. + end = x.size - 1 if cut < 0 else cut + energy.append(np.sum(y[:cut, band]**2) + C - noise * max(end - i0, 0)) + return np.array(energy) def definition(listEDC: list, fs: int, t: int = 50) -> np.ndarray: