Skip to content

Backend Documentation

Preprocessing

preprocessing

Preprocess 

Preprocess(file_paths=None, block_size=1024)

The Preprocess class handles everything after importing the user file. It is an interface between Acoular and our internal Signalprocessing class. It contains information about the block size, data, channel count and size and sample frequency. It is able to return the complete data as a Numpy array or iterate over the defined block size.

Attributes:

Name Type Description
file_paths str

A Path to import.
block_size int

Block size for processing.
table_key str

Table key names.
source TimeSamples

Acoular TimeSamples class.
source_result generator

Generator of the current block result.
selected_data_block numpy_array

Selected data block.
current_block_idx int

Index of the current data block.
selected_channel_data numpy_array

All data of selected channel.

Methods:

Name Description
reinitialize_source

Reinitializes the generator
set_channel_data

Gives the complete data of a channel
set_channel_on_data_block

Gives the data block of a channel
set_next_data_block

Gives the next data block
set_current_channel

Changes the current selected channel
set_data_block_to_idx

Returns the data block of an idx
get_channel_size

Gets the channel size
get_channel_count

Gets the channel count
get_abtastrate

Gets the sample frequency
get_table_names

Gets the table names

Parameters:

Name Type Description Default
file_paths object

Get the callback to the data object

 None
block_size int

Length of data block.

 1024
Source code in src/fft_analysator/analysis/preprocessing.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(self, file_paths=None, block_size=1024):
    """
    Constructs all the necessary attributes for the Preprocess object.


    Args:
        file_paths (object): Get the callback to the data object
        block_size (int): Length of data block.
    """

    self.file_paths = file_paths
    self.block_size = block_size

    if file_paths:
        self.table_key = self.get_table_names()[0]
        self.source = ac.TimeSamples(name=self.file_paths)
        self.source_result = self.source.result(num=self.block_size)
        self.selected_data_block = next(self.source_result)

    self.current_block_idx = 0

    self.selected_channel_data = np.array([])

get_abtastrate 

get_abtastrate()

Get_abtastrate returns the sample_freq from the current .h5 file via Acoular

Returns:

Name Type Description
abtasterate int

Sample Frequency
Source code in src/fft_analysator/analysis/preprocessing.py 
149
150
151
152
153
154
155
156
157
158
def get_abtastrate(self):
    """
    Get_abtastrate returns the sample_freq from the current .h5 file via Acoular

    Returns:
        abtasterate (int): Sample Frequency
    """
    abtastrate = self.source.sample_freq

    return abtastrate

get_channel_count 

get_channel_count()

Get_channel_count returns the number of channels of the current .h5 file

Returns:

Name Type Description
count int

Number of channels
Source code in src/fft_analysator/analysis/preprocessing.py 
138
139
140
141
142
143
144
145
146
147
def get_channel_count(self):
    """
    Get_channel_count returns the number of channels of the current .h5 file

    Returns:
        count (int): Number of channels
    """
    count = self.source.numchannels

    return count

get_channel_size 

get_channel_size()

Get_channel_size returns the size of the current channel

Returns:

Name Type Description
size int

Channel size
Source code in src/fft_analysator/analysis/preprocessing.py 
127
128
129
130
131
132
133
134
135
136
def get_channel_size(self):
    """
    Get_channel_size returns the size of the current channel

    Returns:
        size (int): Channel size
    """
    size = np.array(self.converted_file)[:, self.current_channel].shape[0]

    return size

get_table_names 

get_table_names()

Get_table_names returns a list of all data set keyword contained in the current .h5 File

Returns:

Name Type Description
keys list

A list of table names
Source code in src/fft_analysator/analysis/preprocessing.py 
160
161
162
163
164
165
166
167
168
169
170
171
def get_table_names(self):
    """
    Get_table_names returns a list of all data set keyword contained in the current .h5 File

    Returns:
        keys (list): A list of table names
    """
    with h5py.File(self.file_paths, 'r') as file:
        # Zugriff auf den gewünschten Datensatz
        keys = list(file.keys())

    return keys

reinitialize_source 

reinitialize_source()

The reinitialize_source function reinitializes the generator from Acoular. It is used for returning to already viewed data blocks.

Source code in src/fft_analysator/analysis/preprocessing.py 
58
59
60
61
62
63
64
65
66
def reinitialize_source(self):
    """
    The reinitialize_source function reinitializes the generator from Acoular. It is used for returning to already
    viewed data blocks.
    """
    self.current_block_idx = 0
    self.source = ac.TimeSamples(name=self.file_paths)
    self.source_result = self.source.result(num=self.block_size)
    self.selected_data_block = next(self.source_result)

set_channel_data 

set_channel_data(channel)

Set_channel_data sets returns the complete channel by iterating over the generator of Acoular and saving the data blocks in a Numpy Array.

Parameters:

Name Type Description Default
channel int

Channel number.

 required
Source code in src/fft_analysator/analysis/preprocessing.py 
69
70
71
72
73
74
75
76
77
78
79
80
def set_channel_data(self, channel):
    """
    Set_channel_data sets returns the complete channel by iterating over the generator of Acoular and
    saving the data blocks in a Numpy Array.

    Args:
        channel (int): Channel number.
    """
    loop_list = []
    for data in self.source_result:
        loop_list.extend(data[:, channel])
    self.selected_channel_data = np.array(loop_list)

set_channel_on_data_block 

set_channel_on_data_block(channel)

Set_channel_data_on_data_block sets returns the block data of a channel as a Numpy Array.

Parameters:

Name Type Description Default
channel int

Channel number.

 required

Returns: numpy_array

Source code in src/fft_analysator/analysis/preprocessing.py 
82
83
84
85
86
87
88
89
90
91
def set_channel_on_data_block(self, channel):
    """
    Set_channel_data_on_data_block sets returns the block data of a channel as a Numpy Array.

    Args:
        channel (int): Channel number.
    Returns:
        numpy_array
    """
    return self.selected_data_block[:, channel]

set_current_channel 

set_current_channel(channel)

Set_current_channel sets the attribute current_channel by taking a channel.

Parameters:

Name Type Description Default
channel int

Channel number switching to.

 required
Source code in src/fft_analysator/analysis/preprocessing.py 
101
102
103
104
105
106
107
108
def set_current_channel(self, channel):
    """
    Set_current_channel sets the attribute current_channel by taking a channel.

    Args:
        channel (int): Channel number switching to.
    """
    self.current_channel = channel

set_data_block_to_idx 

set_data_block_to_idx(idx)

Set_data_block_to_idx sets the attribute selected_data_block by selecting a specific element of the generator from Acoular and saving this data blocks in a Numpy Array.

Parameters:

Name Type Description Default
idx int

Idx element to which the generator iterates to.

 required
Source code in src/fft_analysator/analysis/preprocessing.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def set_data_block_to_idx(self, idx):
    """
    Set_data_block_to_idx sets the attribute selected_data_block by selecting a specific element of
    the generator from Acoular and saving this data blocks in a Numpy Array.

    Args:
        idx (int): Idx element to which the generator iterates to.
    """
    self.reinitialize_source()
    try:
        self.current_block_idx = idx
        if idx > 0:
            for i in range(idx):
                self.selected_data_block = next(self.source_result)
    except StopIteration:
        print('End of file reached')

set_next_data_block 

set_next_data_block()

Set_next_data_block sets the attribute selected_data_block by selecting the next element of the generator from Acoular and saving this data blocks in a Numpy Array.

Source code in src/fft_analysator/analysis/preprocessing.py 
93
94
95
96
97
98
99
def set_next_data_block(self):
    """
    Set_next_data_block sets the attribute selected_data_block by selecting the next element of
    the generator from Acoular and saving this data blocks in a Numpy Array.
    """
    self.current_block_idx += 1
    self.selected_data_block = next(self.source_result)

Signal Processing

signal_processing

Signal_Process 

Signal_Process(channels=[], file_path=None, window='Hanning', block_size=1024, overlap='50%', data_callback=None)

The Signal_Process class handles every signal processing method block-wise and returns its results as a Numpy Array. It is initialized by a required the file_path, a window option for the Fourier transformation, a given block_size and an Overlap percentage. Possible signal processing are CSM calculation (Cross Spectral Matrix), coherence between two signals, frequency and impulse response between input and output data and cross/auto correlation between two given signals.

Attributes:

Name Type Description
file_path str

A Path to import.
window str

Window name for the FFT
block_size int

Block size for processing.
overlap str

Overlap percentage
channels list

List of channels
current_data numpy_array

Current Data for the Signal tab
impulse_response_data numpy_array

Data for the Impulse response
amplitude_response_data numpy_array

Data for the Amplitude response
phase_response_data numpy_array

Data for the Phase
data_callback object
p0 int

Is equal to 2010*-6. Auditory threshold
source MaskedTimeSamples

MaskedTimeSamples class to filter out channles
abtastrate int

Sample frequency
numchannels_total int

Total numbers of channels
invalid_channel_list list

Invalid channel list
powerspectra PowerSpectra

Acoular Powerspectra class for calculation
input_channel int

Input channel
output_channel int

Output channel

Methods:

Name Description
set_parameters

Sets Parameters
invalid_channels

Filters all invalid channels
create_time_axis

Creates time axis
create_frequency_axis

Creates frequncy axis
create_correlation_axis

Create correlation axis
SPL

Calculates Sound Pressure Level
csm

Calculates Cross spectral Matrix
coherence

Calculates Coherence
frequency_response

Calculates frequency response
phase_response

Calculates phase response
impuls_response

Calculates impulse response
correlation

Calculates correlation

Parameters:

Name Type Description Default
file_path str

Get the callback to the data object

 None
window str

window function for the fourier transformation: Allowed options: 'Rectangular', 'Hanning', 'Hamming', 'Bartlett', 'Blackman'

 'Hanning'
block_size int

Length of data block. Allowed values: 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768, 65536

 1024
overlap str

Overlap percentage between two blocks for the Welch-Method. Allowed options: 'None', '50%', '75%', '87.5%'

 '50%'
Source code in src/fft_analysator/analysis/signal_processing.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(self, channels=[], file_path=None, window='Hanning', block_size=1024, overlap='50%', data_callback=None):
    """
    Constructs all the necessary attributes for the Signal_Process object.


    Args:
        file_path (str): Get the callback to the data object
        window (str): window function for the fourier transformation: Allowed options: 'Rectangular', 'Hanning', 'Hamming', 'Bartlett', 'Blackman'
        block_size (int): Length of data block. Allowed values: 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768, 65536
        overlap (str): Overlap percentage between two blocks for the Welch-Method. Allowed options: 'None', '50%', '75%', '87.5%'
    """
    self.file_path = file_path
    self.window = window
    self.block_size = block_size
    self.overlap = overlap
    self.channels = channels
    self.current_data = None
    self.impulse_response_data = None
    self.amplitude_response_data = None
    self.phase_response_data = None
    self.data_callback = data_callback
    self.p0 = 20*10**-6 #auditory threshold

    if file_path:

        self.source = ac.MaskedTimeSamples(name=self.file_path)
        self.abtastrate = self.source.sample_freq
        self.numchannels_total = self.source.numchannels_total
        self.invalid_channel_list = []
        self.powerspectra = None

        if channels:
            if len(channels) == 1:
                self.input_channel = self.channels[0]
                self.output_channel = self.channels[0]
            else:
                self.input_channel = self.channels[0]
                self.output_channel = self.channels[1]

SPL 

SPL(channel)

The SPL function calculates the Sound Pressure Level of the current signal.

Parameters:

Name Type Description Default
channel int

Channel number.

 required

Returns:

Name Type Description
SPL int

Sound Pressure Level of a channel
Source code in src/fft_analysator/analysis/signal_processing.py 
175
176
177
178
179
180
181
182
183
184
185
186
def SPL(self, channel):
    """
    The SPL function calculates the Sound Pressure Level of the current signal.

    Args:
        channel (int): Channel number.

    Returns:
        SPL (int): Sound Pressure Level of a channel
    """

    return 20*np.log10(np.abs(self.data_callback.set_channel_on_data_block(channel))/self.p0)

coherence 

coherence()

The coherence function calculates the coherence between two given signals.

Returns:

Name Type Description
current_data numpy_array

The coherence
Source code in src/fft_analysator/analysis/signal_processing.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def coherence(self):
    """
    The coherence function calculates the coherence between two given signals.

    Args:
        None

    Returns:
        current_data (numpy_array): The coherence
    """
    csm_matrix = self.csm()

    if self.input_channel == self.output_channel:
        coherence = np.abs(csm_matrix[:, 0, 0].real)**2 / (csm_matrix[:, 0, 0].real * csm_matrix[:, 0, 0].real)
    else:
        coherence = np.abs(csm_matrix[:, 0, 1])**2 / (csm_matrix[:, 0, 0].real * csm_matrix[:, 1, 1].real)

    self.current_data = coherence

    return self.current_data

correlation 

correlation(type=None)

The correlation function calculates the correlation response between the two given signals.

Parameters:

Name Type Description Default
type str

Determines if an auto or cross correlation is being calculated. Allowed options: 'xx', 'yy', 'xy'

 None

Returns: current_data (numpy_array): The correlation data

Source code in src/fft_analysator/analysis/signal_processing.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def correlation(self,type=None):
    """
    The correlation function calculates the correlation response between the two given signals.

    Args:
        type (str): Determines if an auto or cross correlation is being calculated. Allowed options: 'xx', 'yy', 'xy'
    Returns:
        current_data (numpy_array): The correlation data
    """
    csm_matrix = self.csm()
    N = len(csm_matrix[:, 0, 0])

    if type == 'xx':
            corr = np.fft.fftshift(np.fft.irfft(csm_matrix[:, 0, 0].real, n=N))
    elif type == 'yy':
        if self.input_channel == self.output_channel:
            corr = np.fft.fftshift(np.fft.irfft(csm_matrix[:, 0, 0].real, n=N))
        else:
            corr = np.fft.fftshift(np.fft.irfft(csm_matrix[:, 1, 1].real, n=N))
    elif type == 'xy':
        if self.input_channel == self.output_channel:
            corr = np.fft.fftshift(np.fft.irfft(csm_matrix[:, 0, 0].real, n=N))
        else:
            corr = np.fft.fftshift(np.fft.irfft(csm_matrix[:, 0, 1], n=N))

    self.current_data = corr / np.max(np.abs(corr)) # normalize the correlation to max_value

    return self.current_data

create_correlation_axis 

create_correlation_axis(N)

The create_correlation_axis function creates the x-Axis for the correlation function

Parameters:

Name Type Description Default
N int

Size of the axis

 required

Returns:

Name Type Description
tau numpy_array

The correlation axis
Source code in src/fft_analysator/analysis/signal_processing.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def create_correlation_axis(self, N):
    """
    The create_correlation_axis function creates the x-Axis for the correlation function

    Args:
        N (int): Size of the axis

    Returns:
        tau (numpy_array): The correlation axis
    """

    block_size_factor = self.data_callback.source.numsamples / self.block_size
    if N % 2 == 0:
        tau = np.arange(-N//2,N//2-1) * 4 * block_size_factor / self.abtastrate
    else:
        tau = np.arange(-N//2,N//2) * 4 * block_size_factor / self.abtastrate

    return tau

create_frequency_axis 

create_frequency_axis()

The create_frequency_axis function creates the x-Axis for the frequency function

Returns:

Type Description

powerspectra.fftfreq (numpy_array): The frequency axis
Source code in src/fft_analysator/analysis/signal_processing.py 
145
146
147
148
149
150
151
152
153
def create_frequency_axis(self):
    """
    The create_frequency_axis function creates the x-Axis for the frequency function

    Returns:
        powerspectra.fftfreq (numpy_array): The frequency axis
    """

    return self.powerspectra.fftfreq()

create_time_axis 

create_time_axis(N)

The create_time_axis function creates a time axis for the x-Axis

Parameters:

Name Type Description Default
N int

Size of the axis

 required

Returns:

Name Type Description
time_axis numpy_array

The time axis.
Source code in src/fft_analysator/analysis/signal_processing.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def create_time_axis(self, N):
    """
    The create_time_axis function creates a time axis for the x-Axis

    Args:
        N (int): Size of the axis

    Returns:
        time_axis (numpy_array): The time axis.
    """

    time_axis = np.arange(N) / self.abtastrate

    return time_axis

csm 

csm(csm_dB=False)

The csm function calculates the csm (Cross Spectral Matrix) of Acoular for the valid signals. It returns a three-dimensional array with size (number of frequencies, 2, 2) where [:, signal1, signal2] is the Cross spectral density between signal1 and signal2.

Parameters:

Name Type Description Default
csm_dB bool

Return the array in dB values.

 False

Returns:

Name Type Description
current_data numpy_array

The Cross Spectral Matrix
Source code in src/fft_analysator/analysis/signal_processing.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def csm(self,csm_dB=False):
    """
    The csm function calculates the csm (Cross Spectral Matrix) of Acoular for the valid signals. It returns a
    three-dimensional array with size (number of frequencies, 2, 2) where [:, signal1, signal2] is the
    Cross spectral density between signal1 and signal2.

    Args:
        csm_dB (bool): Return the array in dB values.

    Returns:
        current_data (numpy_array): The Cross Spectral Matrix
    """

    if csm_dB:
        self.current_data = 10*np.log10(np.divide(self.powerspectra.csm, 10**-12))
        return self.current_data
    else:
        self.current_data = self.powerspectra.csm
        return self.current_data

frequency_response 

frequency_response(frq_rsp_dB=True)

The frequency_response function calculates the frequency response between the input and output signal. It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Cross Spectral density between signal x and signal y and Gxx is the Power Spectral Density of signal x.

Parameters:

Name Type Description Default
frq_rsp_dB bool

Return the array in dB values.

 True

Returns:

Name Type Description
amplitude_response_data numpy_array

The frequency response
Source code in src/fft_analysator/analysis/signal_processing.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def frequency_response(self, frq_rsp_dB=True):
    """
    The frequency_response function calculates the frequency response between the input and output signal.
    It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Cross Spectral density between signal x and signal y
    and Gxx is the Power Spectral Density of signal x.

    Args:
        frq_rsp_dB (bool): Return the array in dB values.

    Returns:
        amplitude_response_data (numpy_array): The frequency response
    """
    csm_matrix = self.csm()

    if self.input_channel == self.output_channel:
        H = np.divide(csm_matrix[:, 0, 0].real, csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 0].real), where=(np.abs(csm_matrix[:, 0, 0].real) > 1e-10))
    else:
        H = np.divide(csm_matrix[:, 0, 1], csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 1]), where=(np.abs(csm_matrix[:, 0, 1]) > 1e-10))

    if frq_rsp_dB:
        # return SPL(f) based on H1 estimator
        self.amplitude_response_data = 20*np.log10(abs(np.squeeze(H)/self.p0))

    else:
        # absoulte value of H1 estimator
        self.amplitude_response_data = np.abs(np.squeeze(H))

    return self.amplitude_response_data

impuls_response 

impuls_response(imp_dB=False)

The impulse_response function calculates the impulse response between the input and output signal. It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Spectral density between signal x and signal y and Gxx is the Power Spectral Density of signal x.

Parameters:

Name Type Description Default
imp_dB bool

Return the array in dB values.

 False

Returns: impulse_response_data (numpy_array): The impulse response

Source code in src/fft_analysator/analysis/signal_processing.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def impuls_response(self,imp_dB=False):
    """
    The impulse_response function calculates the impulse response between the input and output signal.
    It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Spectral density between signal x and signal y
    and Gxx is the Power Spectral Density of signal x.

    Args:
        imp_dB (bool): Return the array in dB values.
    Returns:
        impulse_response_data (numpy_array): The impulse response
    """
    csm_matrix = self.csm()
    if self.input_channel == self.output_channel:
        H = np.divide(csm_matrix[:, 0, 0].real, csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 0].real), where=(np.abs(csm_matrix[:, 0, 0].real) > 1e-10))
    else:
        H = np.divide(csm_matrix[:, 0, 1], csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 1]), where=(np.abs(csm_matrix[:, 0, 1]) > 1e-10))

    N = len(csm_matrix[:, 0, 0])
    self.impulse_response_data = np.fft.irfft(H, n=N)
    shifted_signal = np.fft.fftshift(self.impulse_response_data)
    peak_index = np.argmax(shifted_signal)
    signal_up_to_peak = shifted_signal[:peak_index + 1]
    self.impulse_response_data = np.flip(signal_up_to_peak)


    if imp_dB:
        if self.input_channel != self.output_channel:
            self.impulse_response_data = 20*np.log10(abs(self.impulse_response_data)/self.p0)
        else:
            self.impulse_response_data = np.ones(N)
            self.impulse_response_data = 20*np.log10(abs(self.impulse_response_data)/self.p0)
    else:
        self.impulse_response_data = self.impulse_response_data

    return self.impulse_response_data

invalid_channels 

invalid_channels(valid_channels)

The invalid_channels function fills the invalid_channel_list with two valid channels chosen by the user.

Parameters:

Name Type Description Default
valid_channels list

Contains the two selected channels chosen by the user.

 required
Source code in src/fft_analysator/analysis/signal_processing.py 
117
118
119
120
121
122
123
124
125
126
def invalid_channels(self, valid_channels):
    """
    The invalid_channels function fills the invalid_channel_list with two valid channels chosen by the user.

    Args:
        valid_channels (list): Contains the two selected channels chosen by the user.
    """

    self.invalid_channel_list = [k for k in range(self.numchannels_total) if k not in valid_channels]
    self.source.invalid_channels = self.invalid_channel_list

phase_response 

phase_response(deg=True)

The phase_response function calculates the phase response between the input and output signal. It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Cross Spectral Density between signal x and signal y and Gxx is the Power Spectral Density of signal x.

Parameters:

Name Type Description Default
deg bool

Return the array in degrees

 True

Returns:

Name Type Description
phase_response_data numpy_array

The phase response
Source code in src/fft_analysator/analysis/signal_processing.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def phase_response(self, deg=True):
    """
    The phase_response function calculates the phase response between the input and output signal.
    It uses the H1 estimator H1 = Gxy / Gxx, where Gxy is the Cross Spectral Density between signal x and signal y
    and Gxx is the Power Spectral Density of signal x.

    Args:
        deg (bool): Return the array in degrees

    Returns:
        phase_response_data (numpy_array): The phase response
    """
    csm_matrix = self.csm()

    if self.input_channel == self.output_channel:
        H = np.divide(csm_matrix[:, 0, 0].real, csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 0].real), where=(np.abs(csm_matrix[:, 0, 0].real) > 1e0))
    else:
        H = np.divide(csm_matrix[:, 0, 1], csm_matrix[:, 0, 0].real, out=np.zeros_like(csm_matrix[:, 0, 1]), where=(np.abs(csm_matrix[:, 0, 1]) > 1e-10))

    phase = np.angle(H,deg=deg)
    self.phase_response_data = phase

    return self.phase_response_data

set_parameters 

set_parameters(channels, window, overlap)

The set_parameters function sets parameters

Parameters:

Name Type Description Default
channels list

New selected channels

 required
window str

New window function

 required
overlap str

New Overlap percentage

 required
Source code in src/fft_analysator/analysis/signal_processing.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def set_parameters(self, channels, window, overlap):
    """
    The set_parameters function sets parameters

    Args:
        channels (list): New selected channels
        window (str): New window function
        overlap (str): New Overlap percentage
    """

    if channels:
        self.channels = channels

        if len(channels) == 1:
            self.input_channel = self.channels[0]
            self.output_channel = self.channels[0]
        else:
            self.input_channel = self.channels[0]
            self.output_channel = self.channels[1]

    self.window = window
    self.overlap = overlap

    self.invalid_channels([self.input_channel, self.output_channel])
    self.powerspectra = ac.PowerSpectra(time_data=self.source, block_size=self.block_size,
                                                window=self.window, overlap=self.overlap)

Plotter

plotting

Plotter 

Plotter(signal_process_callback, channels, tabs_callback, data_callback, window, overlap, color_picker_value, stretch_value=None, show_grid=None, x_log=None, y_log=None, db=None)

A class used to process and plot signals.

This class is responsible for processing and plotting signals based on the provided data and parameters.

Attributes:

Name Type Description
signal_process Signal_Process

An instance of the Signal_Process class.
channels list

A list of channels.
color_picker_value list

A list of color values for each channel.
stretch_value bool

A flag indicating whether to stretch the plot width.
input_channel int

The input channel number.
output_channel int

The output channel number.
data_callback function

A callback function to retrieve the data.
tabs function

A callback function to update the tabs.
fs int

Sampling frequency of the data.
block int

Current block index.
block_size int

Size of the data block.
numsamples int

Total number of samples in the data source.
show_grid bool

A flag indicating whether to show the grid.
x_log bool

A flag indicating whether to use a logarithmic scale for the x-axis.
y_log bool

A flag indicating whether to use a logarithmic scale for the y-axis.
db bool

A flag indicating whether to use decibels for the y-axis.

Methods:

Name Description
create_time_plot

Plots the time-domain signal based on the provided data and parameters.
create_frequency_response_plot

Plots the frequency response based on the provided data and parameters.
create_impulse_response_plot

Plots the impulse response based on the provided data and parameters.
create_coherence_plot

Plots the coherence based on the provided data and parameters.
create_auto_and_cross_power_spectrum_plot

Plots the auto and cross power spectra based on the provided data and parameters.
create_correlation_plot

Plots the correlation function based on the provided data and parameters.

Parameters:

Name Type Description Default
signal_process_callback Signal_Process

An instance of the Signal_Process class.

 required
channels list

A list of channels.

 required
tabs_callback function

A callback function to update the tabs.

 required
data_callback function

A callback function to retrieve the data.

 required
window int

The window size for the spectrogram.

 required
overlap int

The overlap size for the spectrogram.

 required
color_picker_value list

A list of color values for each channel.

 required
stretch_value bool

A flag indicating whether to stretch the plot width. Defaults to None.

 None
show_grid bool

A flag indicating whether to show the grid. Defaults to None.

 None
x_log bool

A flag indicating whether to use a logarithmic scale for the x-axis. Defaults to None.

 None
y_log bool

A flag indicating whether to use a logarithmic scale for the y-axis. Defaults to None.

 None
db bool

A flag indicating whether to use decibels for the y-axis. Defaults to None.

 None
Source code in src/fft_analysator/analysis/plotting.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(self, signal_process_callback, channels, tabs_callback, data_callback,
            window, overlap, color_picker_value,stretch_value=None,show_grid=None,x_log=None,y_log=None,db=None):
    """
    Constructs all the necessary attributes for the Plotter object.

    Args:
        signal_process_callback (Signal_Process): An instance of the Signal_Process class.
        channels (list): A list of channels.
        tabs_callback (function): A callback function to update the tabs.
        data_callback (function): A callback function to retrieve the data.
        window (int): The window size for the spectrogram.
        overlap (int): The overlap size for the spectrogram.
        color_picker_value (list): A list of color values for each channel.
        stretch_value (bool, optional): A flag indicating whether to stretch the plot width. Defaults to None.
        show_grid (bool, optional): A flag indicating whether to show the grid. Defaults to None.
        x_log (bool, optional): A flag indicating whether to use a logarithmic scale for the x-axis. Defaults to None.
        y_log (bool, optional): A flag indicating whether to use a logarithmic scale for the y-axis. Defaults to None.
        db (bool, optional): A flag indicating whether to use decibels for the y-axis. Defaults to None.
    """
    self.data_callback = data_callback
    self.tabs = tabs_callback
    self.fs = self.data_callback.get_abtastrate()
    self.block = data_callback.current_block_idx
    self.block_size = data_callback.block_size
    self.numsamples = data_callback.source.numsamples
    self.signal_process = signal_process_callback
    self.channels = channels
    self.color_picker_value = color_picker_value
    self.stretch_value = stretch_value
    self.show_grid = show_grid
    self.x_log = x_log
    self.y_log = y_log
    self.db = db

    if channels:
        if len(channels) == 1:
            self.input_channel = self.channels[0]
            self.output_channel = self.channels[0]
        else:
            self.input_channel = self.channels[0]
            self.output_channel = self.channels[1]

create_auto_and_cross_power_spectrum_plot 

create_auto_and_cross_power_spectrum_plot(type=None)

Plots the auto and cross power spectra based on the provided data and parameters.

Parameters:

Name Type Description Default
type str

Specifies the type of spectrum to plot ('xx', 'yy', 'xy'). Defaults to None.

 None
Source code in src/fft_analysator/analysis/plotting.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def create_auto_and_cross_power_spectrum_plot(self,type=None):
    """
    Plots the auto and cross power spectra based on the provided data and parameters.

    Args:
        type (str, optional): Specifies the type of spectrum to plot ('xx', 'yy', 'xy'). Defaults to None.
    """
    # set a own color picker for results
    color_value = self.color_picker_value[2]

    csm =  self.signal_process.csm()
    csm_dB = self.signal_process.csm(csm_dB=True)

    f = self.signal_process.create_frequency_axis()

    if type == 'xx':
        title = "Auto Power Spectrum - Input"
        if self.db:
            csm_value = csm_dB[:,0,0]
            y_label = r'PSD in $$\mathrm{dB}/\mathrm{Hz}$$'
        else:
            csm_value = csm[:,0,0]
            y_label = r'PSD in $$\mathrm{Pa}^{2}/\mathrm{Hz}$$'

    elif type == 'yy':
        title = "Auto Power Spectrum - Output"

        if self.db:
            y_label = r'PSD in $$\mathrm{dB}/\mathrm{Hz}$$'
            if self.input_channel == self.output_channel:
                csm_value = csm_dB[:,0,0]
            else:
                csm_value = csm_dB[:,1,1]
        else:
            y_label = r'PSD in $$\mathrm{Pa}^{2}/\mathrm{Hz}$$'
            if self.input_channel == self.output_channel:
                csm_value = csm[:,0,0]
            else:
                csm_value = csm[:,1,1]

    elif type == 'xy':
        title = "Cross Power Spectrum"

        if self.db:
            y_label = r'PSD in $$\mathrm{dB}/\mathrm{Hz}$$'
            if self.input_channel == self.output_channel:
                csm_value = csm_dB[:,0,0]
            else:
                csm_value = csm_dB[:,0,1]
        else:
            y_label = r'PSD in $$\mathrm{Pa}^{2}/\mathrm{Hz}$$'
            if self.input_channel == self.output_channel:
                csm_value = csm[:,0,0]
            else:
                csm_value = csm[:,0,1]



    fig = hv.Curve((f,np.abs(csm_value)),kdims="Frequency in Hz", vdims= y_label, label=title)
    fig.opts(color=color_value, shared_axes=False, width=750, height=350, show_grid=self.show_grid,
            logx=self.x_log, logy=self.y_log, xlim=(f[1], None) if self.x_log else (0, None),
            ylim=(np.min(np.abs(csm_value))-0.2*np.min(np.abs(csm_value)), np.max(np.abs(csm_value)) +
                    0.2*np.max(np.abs(csm_value)))) #if not self.y_log else (, None)))

    # Create a HoloViews pane for the figure
    plot_pane = HoloViews(fig,  sizing_mode='stretch_width' if self.stretch_value else None)

    # Update the corresponding tab with new signals
    self.tabs.component[3] = (self.tabs.str_analysis_function_tab, plot_pane)

create_coherence_plot 

create_coherence_plot()

Plots the coherence based on the provided data and parameters.

Source code in src/fft_analysator/analysis/plotting.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def create_coherence_plot(self):
    """
    Plots the coherence based on the provided data and parameters.
    """

    # set a own color picker for results
    color_value = self.color_picker_value[2]

    coherence =  self.signal_process.coherence()
    f = self.signal_process.create_frequency_axis()

    fig = hv.Curve((f,coherence), kdims="Frequency in Hz", vdims=r'$$\gamma_{XY}^2(f)$$', label="Coherence")
    fig.opts(color=color_value, shared_axes=False, width=750, height=350, show_grid=self.show_grid,
               logx=self.x_log, logy=False,ylim=(-0.1,1.1), xlim=(f[1], None) if self.x_log else (0, None))

    # Create a HoloViews pane for the figure
    plot_pane = HoloViews(fig,  sizing_mode='stretch_width' if self.stretch_value else None)

    # Update the corresponding tab with new signals
    self.tabs.component[3] = (self.tabs.str_analysis_function_tab, plot_pane)

create_correlation_plot 

create_correlation_plot(type=None)

Plots the correlation function based on the provided data and parameters.

Parameters:

Name Type Description Default
type str

Specifies the type of correlation to plot ('xx', 'yy', 'xy'). Defaults to None.

 None
Source code in src/fft_analysator/analysis/plotting.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def create_correlation_plot(self,type=None):
    """
    Plots the correlation function based on the provided data and parameters.

    Args:
        type (str, optional): Specifies the type of correlation to plot ('xx', 'yy', 'xy'). Defaults to None.
    """
    # set a own color picker for results
    color_value = self.color_picker_value[2]

    if type == 'xx':
        title = "Auto Correlation - Input"
        y_label = r'$$\mathrm{\psi}_{xx}(\mathrm{\tau})$$'
        corr = self.signal_process.correlation(type='xx')

    elif type == 'yy':
        title = "Auto Correlation - Output"
        y_label = r'$$\mathrm{\psi}_{yy}(\mathrm{\tau})$$'
        corr = self.signal_process.correlation(type='yy')

    elif type == 'xy':
        title = "Cross Correlation"
        y_label = r'$$\mathrm{\psi}_{xy}(\mathrm{\tau})$$'
        corr = self.signal_process.correlation(type='xy')

    if len(corr) % 2 != 0:
        corr = np.roll(corr,1) # shift correlation to 1 sample

    # create time delay axis
    tau = self.signal_process.create_correlation_axis(len(corr))

    fig = hv.Curve((tau,corr),kdims=r'$$\mathrm{\tau}$$ in s', vdims=y_label, label=title )
    fig.opts(color=color_value, shared_axes=False, width=750, height=350, show_grid=self.show_grid,
             xlim=(np.min(tau)+0.1*np.min(tau),np.max(tau)+0.1*np.max(tau)),
             ylim=(np.min(corr) - 0.1, np.max(corr) + 0.1) ,logx=False, logy=False)

    # Create a HoloViews pane for the figure
    plot_pane = HoloViews(fig,  sizing_mode='stretch_width' if self.stretch_value else None)

    # Update the corresponding tab with new signals
    self.tabs.component[3] = (self.tabs.str_analysis_function_tab, plot_pane)

create_frequency_response_plot 

create_frequency_response_plot()

Plots the frequency response based on the provided data and parameters.

Source code in src/fft_analysator/analysis/plotting.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def create_frequency_response_plot(self):
    """
    Plots the frequency response based on the provided data and parameters.
    """

    # set the signals column
    signals = pn.Column(sizing_mode='stretch_width')

    # set a own color picker for results
    color_value = self.color_picker_value[2]

    # values in dB
    if not self.db:
        H =  self.signal_process.frequency_response(frq_rsp_dB=False)
        y_label = r'Sound Pressure in $$\mathrm{Pa}$$'
    else:
        H =  self.signal_process.frequency_response(frq_rsp_dB=True)
        y_label = r'$$\mathrm{L_{p}}$$ in $$\mathrm{dB}$$'# SPL

    # Create the phase response
    phi = self.signal_process.phase_response(deg=True)

    # Create frequency axis for the given blocks
    f = self.signal_process.create_frequency_axis()

    # Create frequency response fig
    fig1 = hv.Curve((f,H), kdims="Frequency in Hz", vdims=y_label, label=f'Amplitude Response')
    fig1.opts(color=color_value, shared_axes=False, width=750, height=350,show_grid=self.show_grid,
              logx=self.x_log, logy=self.y_log, xlim=(f[1], None) if self.x_log else (0, None))

    fig2 = hv.Curve((f,phi), kdims="Frequency in Hz", vdims="Phase in Degree °", label=f'Phase Response')
    fig2.opts(color=color_value, shared_axes=False, width=750, height=350,show_grid=self.show_grid,
              logx=self.x_log, logy=False, xlim=(f[1], None) if self.x_log else (0, None))

    for fig in [fig1, fig2]:

        # Create a HoloViews pane for the figure
        plot_pane = HoloViews(fig, backend='bokeh', sizing_mode='stretch_width' if self.stretch_value else None)

        # Append the plot pane to the signals column
        signals.append(plot_pane)

    # Update the corresponding tab with new signals
    self.tabs.component[1] = (self.tabs.str_frequency_response_tab, signals)

create_impulse_response_plot 

create_impulse_response_plot()

Plots the impulse response based on the provided data and parameters.

Source code in src/fft_analysator/analysis/plotting.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def create_impulse_response_plot(self):
    """
    Plots the impulse response based on the provided data and parameters.
    """
    # set a own color picker for results
    color_value = self.color_picker_value[2]

    # values in dB
    if not self.db:
        h = self.signal_process.impuls_response(imp_dB=False).real
        scale_min_factor = 0.2 # set min scale factor
        y_label = r'Sound Pressure in $$\mathrm{Pa}$$'
    else:
        h = self.signal_process.impuls_response(imp_dB=True).real
        scale_min_factor = -0.2 # set min scale factor
        y_label = r'$$\mathrm{L_{p}}$$ in $$\mathrm{dB}$$'# SPL

    # set max scale factor
    scale_max_factor = 0.2

    # Create and scaling time axis
    block_size_factor = self.data_callback.source.numsamples / self.block_size
    t = self.signal_process.create_time_axis(len(h)) * 8 * block_size_factor

    # Create frequency response fig
    fig = hv.Curve((t,h), kdims="Time in s", vdims=y_label, label = f'Impulse Response')
    fig.opts(color=color_value, shared_axes=False, width=750, height=350,show_grid=self.show_grid,
            logx = False, logy = self.y_log, xlim=(-0.1, np.max(t) + 0.1),
            ylim=(np.min(h) + scale_min_factor *
            np.min(h), np.max(h) + scale_max_factor * np.max(h)) if not self.y_log else (0, None))

    # Create a HoloViews pane for the figure
    plot_pane = HoloViews(fig,  sizing_mode='stretch_width' if self.stretch_value else None)

    # Update the corresponding tab with new signals
    self.tabs.component[2] = (self.tabs.str_impulse_response_tab, plot_pane)

create_time_plot 

create_time_plot()

Plots the time-domain signal based on the provided data and parameters.

Source code in src/fft_analysator/analysis/plotting.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def create_time_plot(self):
    """
    Plots the time-domain signal based on the provided data and parameters.
    """

    # set the signals column
    signals = pn.Column(sizing_mode='stretch_width')

    for i, channel in enumerate(list(dict.fromkeys(self.channels))):

        # set color picker value
        color_value = self.color_picker_value[i] if i < len(self.color_picker_value) else "default_color"

        # values in dB
        if not self.db:
            # Get the time_data block wise for the given channel
            time_data = self.data_callback.set_channel_on_data_block(channel)
            scale_min_factor = 0.2 # set min scale factor
            y_label = r'Sound Pressure in $$\mathrm{Pa}$$'
        else:
            time_data = self.signal_process.SPL(channel)
            scale_min_factor = -0.2 # set min scale factor
            y_label = r'$$\mathrm{L_{p}}$$ in $$\mathrm{dB}$$' # SPL

        # set max scale factor
        scale_max_factor = 0.2

        # Create the dynamic time axis for the given blocks
        t = self.signal_process.create_time_axis(N=len(time_data))
        amount_steps = self.numsamples/self.block_size
        fractional_part, integer_part = math.modf(amount_steps)

        # set time axis for current block
        if self.block != 0:

            if fractional_part == 0:
                t = t + self.block * (self.block_size/self.fs)
            else:
                if integer_part > (self.block):
                    t = t + self.block * (self.block_size/self.fs)
                else:
                    amount_samp = self.numsamples - (self.block) * self.block_size
                    t = t + (self.block-1) * (self.block_size/self.fs) + (amount_samp/self.fs) / fractional_part

        # set title
        if self.input_channel == self.output_channel:
            title = f"Input/Output signal - Channel {channel}"
        elif channel == self.input_channel:
            role = "Input"
            title =  f"{role} signal - Channel {channel}"
        elif channel == self.output_channel:
            role = "Output"
            title =  f"{role} signal - Channel {channel}"


        # Create the figure
        fig = hv.Curve((t, time_data), kdims="Time in s", vdims=y_label, label=  title)
        fig.opts(color=color_value, shared_axes=False, width=750, height=350,show_grid=self.show_grid,
                 logx = False, logy = self.y_log, ylim=(np.min(time_data) + scale_min_factor *
                np.min(time_data), np.max(time_data) + scale_max_factor * np.max(time_data)))

        # Create a HoloViews pane for the figure
        plot_pane = HoloViews(fig, sizing_mode='stretch_width' if self.stretch_value else None)

        # Append the plot pane to the signals column
        signals.append(plot_pane)

    # Update the corresponding tab with new signals
    self.tabs.component[0] = (self.tabs.str_signal_tab, signals)