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 991edb618d6c5c18fa7750396d6086b746f079be Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 19:56:39 +0100 Subject: [PATCH 3/3] [FIX] zero-mean colored noise with random phase in every channel _powerlaw_noise gave the DC bin the largest amplitude of the spectrum (frequency 1/nsamples in the power law), so the noise came out as mostly offset: 7 times its rms. With the silent margins of colored_noise the offset is a rectangular pulse whose 1/f**2 leakage tilts the spectrum: pink measured -3.96 dB per octave instead of -3.01. The DC bin is now 0. imag[-1] = 0 was meant for the Nyquist bin but zeroed the last channel, whose spectrum became real and its waveform its own mirror image in time; with one channel every noise did. It is now imag[:, -1]. Tests pin the power law of the five colors, the absence of offset and a random phase in every channel. Co-Authored-By: Claude Fable 5.1 --- pytta/generate.py | 3 +- tests/test_colored_noise.py | 71 +++++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) create mode 100644 tests/test_colored_noise.py diff --git a/pytta/generate.py b/pytta/generate.py index f9ecd2e..d98986c 100644 --- a/pytta/generate.py +++ b/pytta/generate.py @@ -393,6 +393,7 @@ def _powerlaw_noise(nsamples, nchannels, power, fs): freqs = np.fft.rfftfreq(nsamples, 1/fs) freqs[0] = 1/nsamples scaling = (1/(2 * np.pi * freqs))**(power/2) + scaling[0] = 0 # no DC: a zero-mean noise # For each Fourier freq w_i draw 2 gaussian distributed numbers # multiply them by sqrt(0.5 * S(w_i)) approx (1/w)^(B/2) @@ -403,7 +404,7 @@ def _powerlaw_noise(nsamples, nchannels, power, fs): # If nsamples is even, at nyquist the FFT is real-valued only if not nsamples & 1: - imag[-1] = 0. + imag[:, -1] = 0. # IFFT the spectrum to obtain the time signal out = np.array(np.fft.irfft(real + 1j*imag), diff --git a/tests/test_colored_noise.py b/tests/test_colored_noise.py new file mode 100644 index 0000000..736968a --- /dev/null +++ b/tests/test_colored_noise.py @@ -0,0 +1,71 @@ +""" +Tests for pytta.generate.colored_noise. + +A noise of power law B has a power spectral density proportional to +1 / f**B: -3.01 B dB per octave (white 0, pink -3.01, red -6.02, blue ++3.01, purple +6.02). Each test pins one rule on noise generated with a +fixed seed; the slope is fitted to the mean PSD of octave bands from 63 Hz +to 8 kHz, averaged over eight channels. +""" +import os +import unittest + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import numpy as np + +import pytta + +FS = 48000 +COLORS = {'purple': -2, 'blue': -1, 'white': 0, 'pink': 1, 'red': 2} + + +def octave_slope(signalObj): + psd = np.mean(np.abs(signalObj.freqSignal) ** 2, axis=1) + f = signalObj.freqVector + centres = 63 * 2.0 ** np.arange(8) + levels = [10 * np.log10(psd[(f >= c / 2 ** .5) & (f < c * 2 ** .5)].mean()) + for c in centres] + return np.polyfit(np.log2(centres), levels, 1)[0] + + +class TestColoredNoise(unittest.TestCase): + + def test_spectrum_follows_the_power_law_of_the_color(self): + # With the default kind of use: silent margins around the noise. + for color, B in COLORS.items(): + np.random.seed(0) + noise = pytta.generate.colored_noise( + color, samplingRate=FS, fftDegree=16, numChannels=8, + startMargin=0.1, stopMargin=0.1) + slope = octave_slope(noise) + print(f'\n {color:6s} {slope:+.2f} dB/oct (expected ' + f'{-3.0103 * B:+.2f})') + self.assertAlmostEqual(slope, -3.0103 * B, delta=0.2) + + def test_noise_has_no_dc_offset(self): + np.random.seed(0) + noise = pytta.generate.colored_noise( + 'pink', samplingRate=FS, fftDegree=16, numChannels=4, + startMargin=0, stopMargin=0).timeSignal + offset = np.max(np.abs(noise.mean(axis=0)) / noise.std(axis=0)) + print(f'\n |mean| / rms = {offset:.3f}') + self.assertLess(offset, 0.05) + + def test_every_channel_has_random_phase(self): + # A spectrum with no imaginary part gives a noise that is its own + # mirror image in time, x[n] = x[N - n]. + np.random.seed(0) + noise = pytta.generate.colored_noise( + 'pink', samplingRate=FS, fftDegree=14, numChannels=2, + startMargin=0, stopMargin=0).timeSignal + N = noise.shape[0] + k = np.arange(N // 10, N - N // 10) + for ch in range(2): + r = np.corrcoef(noise[k, ch], noise[N - k, ch])[0, 1] + print(f'\n channel {ch}: corr(x[n], x[N-n]) = {r:+.3f}') + self.assertLess(abs(r), 0.2) + + +if __name__ == '__main__': + unittest.main()