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/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/analysis.py b/pytta/classes/analysis.py index 7f15924..142f5be 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,46 @@ 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. 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. + """ + 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) + 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: """ Room parameter. 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/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 1979f54..f9ecd2e 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 @@ -266,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 @@ -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 @@ -459,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 @@ -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() 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()