diff --git a/docs/pyttadoc/functions.rst b/docs/pyttadoc/functions.rst index add57c0..d56fa09 100644 --- a/docs/pyttadoc/functions.rst +++ b/docs/pyttadoc/functions.rst @@ -26,4 +26,6 @@ Functions .. autofunction:: pytta.plot_bars .. autofunction:: pytta.plot_spectrogram .. autofunction:: pytta.plot_waterfall - +.. autofunction:: pytta.multiple_sweep_excitation +.. autofunction:: pytta.multiple_sweep_irs +.. autofunction:: pytta.triggered_average diff --git a/pytta/__init__.py b/pytta/__init__.py index d8740aa..4ce4069 100644 --- a/pytta/__init__.py +++ b/pytta/__init__.py @@ -113,7 +113,8 @@ list_devices, print_devices, get_device_from_user,\ fft_convolve, find_delay, corr_coef, resample,\ peak_time, save, load, plot_time, plot_time_dB, plot_freq,\ - plot_bars, plot_spectrogram, plot_waterfall, SPL + plot_bars, plot_spectrogram, plot_waterfall, SPL,\ + multiple_sweep_excitation, multiple_sweep_irs, triggered_average # Must go on v0.1.0 (DEPRECATED) from .functions import fft_degree @@ -146,6 +147,9 @@ 'resample', 'corr_coef', 'peak_time', + 'multiple_sweep_excitation', + 'multiple_sweep_irs', + 'triggered_average', 'save', 'load', 'weighting', 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/__init__.py b/pytta/classes/__init__.py index aabc899..5c9e99b 100644 --- a/pytta/classes/__init__.py +++ b/pytta/classes/__init__.py @@ -16,6 +16,7 @@ * RecMeasure * PlayRecMeasure * FRFMeasure + * SimulatedPlayRecMeasure and SimulatedFRFMeasure, without a sound card * Streaming * Monitor * OctFilter @@ -38,7 +39,8 @@ from .signal import SignalObj, ImpulsiveResponse #from .signal import plot_SignalObjs -from .measurement import Measurement, RecMeasure, PlayRecMeasure, FRFMeasure +from .measurement import Measurement, RecMeasure, PlayRecMeasure, FRFMeasure, \ + SimulatedPlayRecMeasure, SimulatedFRFMeasure from .streaming import Streaming, Monitor from .filter import OctFilter, weighting from .analysis import Analysis, RoomAnalysis @@ -51,6 +53,8 @@ 'RecMeasure', 'PlayRecMeasure', 'FRFMeasure', + 'SimulatedPlayRecMeasure', + 'SimulatedFRFMeasure', 'Streaming', 'Monitor', 'OctFilter', 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..de95264 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, @@ -551,37 +537,28 @@ def __repr__(self): f'comment={self.comment!r})') # PlayRec Methods + def _playrec(self, output): + """Play output (samples x output channels) and record the input + channels at the same time; SimulatedPlayRecMeasure replaces it.""" + return sd.playrec(output, + samplerate=self.samplingRate, + input_mapping=self.inChannels.mapping, + output_mapping=self.outChannels.mapping, + device=self.device, + blocking=self.blocking, + latency='low', + dtype='float32') + 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* - self.outputLinearGain, - samplerate=self.samplingRate, - input_mapping=self.inChannels.mapping, - output_mapping=self.outChannels.mapping, - device=self.device, - blocking=self.blocking, - latency='low', - dtype='float32') + recording = self._playrec(self.excitation.timeSignal * + self.outputLinearGain) recording = np.squeeze(recording) recording = SignalObj(signalArray=recording*self.inChannels.CFlist(), domain='time', @@ -807,21 +784,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, @@ -836,6 +799,73 @@ def run(self): return transferfunction +class _Simulated: + """ + Replaces the sound card of a play and record measurement by a simulated + measurement chain, as itaMSTFdummy of the ITA-Toolbox: each output + channel goes through the memoryless nonlinearity y = sum_k a_k x^k + (coefficients a_1, a_2, ...) and then through the impulse responses of + system to every input channel; background noise of rms level noise (dB + re full scale) is added and the result is clipped to +-1 and quantized + to bits bits. Only the samples of the excitation length are recorded, + as with a real sound card. + + system is one array or SignalObj (samples x input channels) per output + channel, in a list when there are several outputs. + """ + + def __init__(self, system=None, nonlinear=(1.,), noise=-200., bits=24, + seed=None, **kwargs): + super().__init__(**kwargs) + self.system = system + self.nonlinear = nonlinear + self.noise = noise + self.bits = bits + self._rng = np.random.default_rng(seed) + + def _paths(self, numOut): + system = self.system if isinstance(self.system, (list, tuple)) \ + else [self.system] + if len(system) != numOut: + raise ValueError('system needs one set of impulse responses per ' + 'output channel.') + paths = [] + for s in system: + h = np.asarray(s.timeSignal if isinstance(s, SignalObj) else s, + dtype=float) + paths.append(h.reshape(h.shape[0], -1)) + return paths + + def _playrec(self, output): + output = np.asarray(output, dtype=float).reshape(len(output), -1) + n = output.shape[0] + recording = np.zeros((n, len(self.inChannels.mapping))) + for x, h in zip(output.T, self._paths(output.shape[1])): + y = sum(a * x ** (k + 1) for k, a in enumerate(self.nonlinear)) + for c in range(recording.shape[1]): + recording[:, c] += np.convolve(y, h[:, c])[:n] + recording += 10 ** (self.noise / 20) * \ + self._rng.standard_normal(recording.shape) + step = 2.0 ** (1 - self.bits) + return np.round(np.clip(recording, -1, 1) / step) * step + + +class SimulatedPlayRecMeasure(_Simulated, PlayRecMeasure): + """PlayRecMeasure on a simulated measurement chain, for tests and + teaching without a sound card; see _Simulated for system, nonlinear, + noise, bits and seed. + + >>> m = pytta.classes.SimulatedPlayRecMeasure(system=ir, + ... excitation=sweep, inChannels=[1], outChannels=[1]) + >>> recording = m.run() + """ + + +class SimulatedFRFMeasure(_Simulated, FRFMeasure): + """FRFMeasure on a simulated measurement chain, as itaMSTFdummy; run() + returns the measured ImpulsiveResponse. See _Simulated.""" + + # Sub functions def _print_max_level(sigObj, kind, gain=1, mapping=None): for chIndex in range(sigObj.numChannels): 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/functions.py b/pytta/functions.py index 9b39239..9a10681 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -274,6 +274,75 @@ def peak_time(signal): else: return peaks_time[0] +def multiple_sweep_excitation(sweep, numSources, delay, tail=0.): + """ + Excitation of the multiple exponential sweep method (Majdak, Balazs and + Laback, Multiple exponential sweep method for fast measurement of + head-related transfer functions, J. Audio Eng. Soc. 55(7/8), 2007), the + interleaved and MIMO measurements of the ITA-Toolbox: channel k plays + sweep delayed by k delay seconds, with tail seconds of silence at the + end for the last responses. delay must exceed the length of the + impulse responses plus the advance of the harmonics that matter. + """ + fs = sweep.samplingRate + s = np.asarray(sweep.timeSignal[:, 0], dtype=float) + d = int(round(delay * fs)) + x = np.zeros((s.size + (numSources - 1) * d + int(round(tail * fs)), numSources)) + for k in range(numSources): + x[k * d:k * d + s.size, k] = s + return SignalObj(x, 'time', fs, freqMin=sweep.freqMin, freqMax=sweep.freqMax) + + +def multiple_sweep_irs(recording, sweep, numSources, delay, irLength, + pretime=0.005, regularization=True): + """ + Impulse responses of each source to every input channel from a + recording of multiple_sweep_excitation: the recording is deconvolved + with the single sweep by ImpulsiveResponse, regularized between the + freqMin and freqMax of sweep unless regularization is False, and the + response of source k is cut from pretime seconds before k delay, + irLength seconds long, so that the onset of the band-limited impulse is + kept. The regularization limits the band and so spreads each impulse + in time, and part of it falls in the windows of the other sources + (about 1 % with the default band); without it the division is exact + but raises the noise outside the band of the sweep. Returns one + SignalObj per source, with one channel per input channel. + """ + fs = recording.samplingRate + n = recording.timeSignal.shape[0] + s = np.zeros(n) + s[:sweep.timeSignal.shape[0]] = sweep.timeSignal[:, 0] + excitation = SignalObj(s, 'time', fs, freqMin=sweep.freqMin, freqMax=sweep.freqMax) + ir = ImpulsiveResponse(excitation, recording, 'linear', regularization=regularization) + h = np.asarray(ir.IR.timeSignal, dtype=float) + d, m = int(round(delay * fs)), int(round(irLength * fs)) + h = np.roll(h, int(round(pretime * fs)), axis=0) + return [SignalObj(h[k * d:k * d + m], 'time', fs) for k in range(numSources)] + + +def triggered_average(recording, threshold, duration, pretime=0., channel=0): + """ + Average of the events of a triggered measurement (e.g. impact hammer), + as ita_measurement_trigger: an event starts where the trigger channel + first exceeds threshold dB re full scale after the previous event, and + is cut from pretime seconds before that sample, duration seconds long, + in every channel. Returns (mean SignalObj, list of the events). + """ + fs = recording.samplingRate + x = np.asarray(recording.timeSignal, dtype=float) + pre, m = int(round(pretime * fs)), int(round(duration * fs)) + above = np.flatnonzero(np.abs(x[:, channel]) > 10 ** (threshold / 20)) + runs, end = [], -1 + for i in above: + if i >= end and i - pre >= 0 and i - pre + m <= x.shape[0]: + runs.append(SignalObj(x[i - pre:i - pre + m], 'time', fs)) + end = i - pre + m + if not runs: + raise ValueError('No event above the threshold.') + mean = np.mean([r.timeSignal for r in runs], axis=0) + return SignalObj(mean, 'time', fs), runs + + def fft_degree(*args,**kwargs): """ DEPRECATED 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_simulated_measurement.py b/tests/test_simulated_measurement.py new file mode 100644 index 0000000..5f1d0d7 --- /dev/null +++ b/tests/test_simulated_measurement.py @@ -0,0 +1,114 @@ +""" +Tests for the simulated measurement chain (SimulatedPlayRecMeasure and +SimulatedFRFMeasure, as itaMSTFdummy of the ITA-Toolbox) and for +multiple_sweep_excitation, multiple_sweep_irs and triggered_average of +pytta.functions. Each test measures a known system without a sound card. +""" +import unittest + +import numpy as np + +import pytta +from pytta.classes import SimulatedFRFMeasure, SimulatedPlayRecMeasure + +FS = 48000 + + +def signal(*samples, **kwargs): + # A tuple as first local variable: SignalObj compares it with itself + # while walking the stack, which an array cannot do before #59. + return pytta.SignalObj(samples[0], 'time', FS, **kwargs) + + +def exponential_sweep(n, tail): + t = np.arange(n) / FS + L = n / FS / np.log(1000) + x = 0.5 * np.sin(2 * np.pi * 20 * L * (np.exp(t / L) - 1)) + return signal(np.r_[x, np.zeros(tail)], freqMin=20, freqMax=20000) + + +def room(seed, n=2400, channels=2): + rng = np.random.default_rng(seed) + return 0.01 * np.exp(-np.arange(n) / 300)[:, None] * rng.standard_normal((n, channels)) + + +def band_error(measured, true, n): + # Largest relative difference of the spectra between 100 Hz and 10 kHz. + f = np.fft.rfftfreq(n, 1 / FS) + band = (f > 100) & (f < 10000) + M, T = np.fft.rfft(measured, n, axis=0), np.fft.rfft(true, n, axis=0) + return np.abs(M[band] - T[band]).max() / np.abs(T[band]).max() + + +class TestSimulatedMeasurement(unittest.TestCase): + + def test_frf_of_a_known_system(self): + label = 'x' + h = room(1) + m = SimulatedFRFMeasure(system=h, excitation=exponential_sweep(2 ** 16, 2 ** 14), + inChannels=[1, 2], outChannels=[1], device=0) + # The regularization limits the band, which spreads the response + # in time, so the whole response is compared in the band. + ir = np.asarray(m.run().IR.timeSignal, dtype=float) + e = band_error(ir, h, ir.shape[0]) + print(f'\n simulated FRF: {e:.1e}', end='') + self.assertLess(e, 1e-3) + + def test_noise_and_quantization(self): + # Silence gives the background noise level; a 16-bit chain adds the + # quantization error of power step^2 / 12. + label = 'x' + quiet = signal(np.zeros(FS)) + rec = SimulatedPlayRecMeasure(system=room(2, channels=1), noise=-60., seed=3, excitation=quiet, + inChannels=[1], outChannels=[1], device=0).run() + level = 10 * np.log10(np.mean(np.asarray(rec.timeSignal, dtype=float) ** 2)) + tone = signal(0.5 * np.sin(2 * np.pi * 997 * np.arange(FS) / FS)) + ideal = SimulatedPlayRecMeasure(system=[[1.]], bits=53, excitation=tone, + inChannels=[1], outChannels=[1], device=0).run() + coarse = SimulatedPlayRecMeasure(system=[[1.]], bits=16, excitation=tone, + inChannels=[1], outChannels=[1], device=0).run() + e = np.asarray(coarse.timeSignal, dtype=float) - np.asarray(ideal.timeSignal, dtype=float) + power = np.mean(e ** 2) / (2.0 ** -15) ** 2 * 12 + print(f'\n noise {level:.2f} dB, quantization power {power:.3f} of step^2/12', end='') + self.assertLess(abs(level + 60), 0.1) + self.assertLess(abs(power - 1), 0.05) + + def test_multiple_sweeps_separate_the_sources(self): + # Three sources, two microphones, sweeps 0.3 s apart, each response + # cut 0.3 s long from 5 ms before its delay; without regularization + # the deconvolution is exact, so the separation is. + label = 'x' + sweep = exponential_sweep(2 ** 15, 0) + paths = [room(10 + k) for k in range(3)] + x = pytta.multiple_sweep_excitation(sweep, 3, 0.3, tail=0.4) + rec = SimulatedPlayRecMeasure(system=paths, excitation=x, inChannels=[1, 2], + outChannels=[1, 2, 3], device=0).run() + irs = pytta.multiple_sweep_irs(rec, sweep, 3, 0.3, 0.3, pretime=0.005, + regularization=False) + pre = int(0.005 * FS) + e = max(band_error(np.asarray(ir.timeSignal, dtype=float), + np.pad(p, ((pre, 0), (0, 0))), int(0.3 * FS)) + for ir, p in zip(irs, paths)) + print(f'\n multiple sweeps: {e:.1e}', end='') + self.assertLess(e, 1e-3) + + def test_triggered_average(self): + # Five hits of the same pulse in noise: five events, each starting + # 1 ms before the first sample of the pulse above -20 dB, whose mean + # is the pulse within the averaged noise. + label = 'x' + rng = np.random.default_rng(4) + pulse = np.exp(-np.arange(960) / 100) * np.sin(2 * np.pi * 300 * np.arange(960) / FS) + x = 1e-3 * rng.standard_normal(5 * FS) + for k in range(5): + x[k * FS + 1000:k * FS + 1960] += pulse + mean, runs = pytta.triggered_average(signal(x), -20., 0.05, 0.001) + i0 = np.argmax(np.abs(pulse) > 0.1) + e = np.abs(np.asarray(mean.timeSignal[48:48 + 900, 0], dtype=float) - pulse[i0:i0 + 900]).max() + print(f'\n triggered average: {len(runs)} events, {e:.1e}', end='') + self.assertEqual(len(runs), 5) + self.assertLess(e, 3e-3) + + +if __name__ == '__main__': + unittest.main()