pyOMA.core.PreProcessingTools.PreProcessSignals#

class pyOMA.core.PreProcessingTools.PreProcessSignals(signals, sampling_rate, ref_channels=None, accel_channels=None, velo_channels=None, disp_channels=None, **kwargs)[source]#

Bases: object

Pre-processor for multi-channel ambient-vibration signals.

Provides signal conditioning (filtering, decimation, offset removal, scaling), spectral estimation (Welch, Blackman-Tukey), and book-keeping of channel metadata (reference channels, measurement quantities, channel-DOF assignments). All downstream pyOMA modules (system identification, stabilisation diagram, mode-shape visualisation) consume an instance of this class.

Todo

  • time-step integration of signals

  • Multi-block Blackman-Tukey PSD

__init__(signals, sampling_rate, ref_channels=None, accel_channels=None, velo_channels=None, disp_channels=None, **kwargs)[source]#
Parameters:
  • signals (np.ndarray, shape (n_samples, n_channels), or str/os.PathLike) – Raw measurement time series; must have more rows than columns. A path to a .npy file is also accepted and loaded via np.load. A .npz path is rejected with a pointer to load_state(), since that format holds a full saved session (multiple fields), not a bare array.

  • sampling_rate (float) – Sampling frequency in Hz.

  • ref_channels (list of int, optional) – Column indices of reference (fixed) sensors. Defaults to all channels.

  • accel_channels (list of int, optional) – Column indices of acceleration channels. Defaults to all channels not in velo_channels or disp_channels.

  • velo_channels (list of int, optional) – Column indices of velocity channels.

  • disp_channels (list of int, optional) – Column indices of displacement channels.

  • setup_name (str, optional) – Label for this measurement setup.

  • channel_headers (list of str, optional) – Human-readable channel names; defaults to [0, 1, 2, ...].

  • start_time (datetime.datetime, optional) – Measurement start timestamp; defaults to datetime.datetime.now().

  • F (np.ndarray, optional) – Optional forcing signal array (used only for FRF-based ERA).

Methods

__init__(signals, sampling_rate[, ...])

add_chan_dofs(chan_dofs)

chan_dofs = [ (chan_num, node_name, az, elev, chan_name) , .

add_noise([amplitude, snr])

Add Gaussian white noise to the signals (useful for simulation studies).

corr_blackman_tukey(m_lags[, n_segments, ...])

Estimate the (cross- and auto-) correlation functions (C/ACF), by direct computation of the standard un-biased estimator:

corr_welch([m_lags, n_segments, refs_only])

Estimate the (cross- and auto-) correlation functions (C/ACF), by the inverse Fourier Transform of Power Spectral Densities, estimated according to Welch's method.

correct_offset()

corrects a constant offset from measured signals

correlation([m_lags, method])

A convenience method for obtaining the correlation sequence by the default or any specified estimation method.

decimate_signals(decimate_factor[, nyq_rat, ...])

Decimate the signals by an integer factor.

delete_channels(channels)

Remove one or more channels from the signals and all associated channel-index bookkeeping.

filter_signals([lowpass, highpass, ...])

Apply a zero-phase IIR or FIR filter to the measurement signals.

get_chan_dof(channel)

Return the (node, az, elev) assignment for channel, or None.

init_from_config(conf_file, meas_file[, ...])

initializes the PreProcessor object with a configuration file

load_chan_dofs(fname)

chan_dofs[i] = (chan_num, node_name, az, elev, chan_name)

load_measurement_file(fname, **kwargs)

A method for loading a signals file

load_state(fname)

precondition_signals([method])

Remove the DC offset and scale each channel by its spread.

psd([n_lines, method])

A convenience method for obtaining the PSD by the default or any specified estimation method.

psd_blackman_tukey([n_lines, refs_only, window])

Estimate the (cross- and auto-) power spectral densities (PSD), by Fourier Transform of correlation functions estimated according to Blackman-Tukey's method.

psd_welch([n_lines, n_segments, refs_only, ...])

Estimate the (cross- and auto-) power spectral densities (PSD), according to Welch's method.

remove_chan_dof(channel)

Remove any existing DOF assignment for channel, if present.

rename_channel(channel, new_name)

Rename a channel, keeping any chan_dofs annotation in sync.

save_chan_dofs(fname)

Write self.chan_dofs to fname in the format read by load_chan_dofs().

save_config(fname[, delete_channels])

Write a config file readable by init_from_config().

save_state(fname)

save_undo_snapshot()

Capture the current state for a single-step undo.

set_chan_dof(channel, node, az, elev)

Assign channel to a node and direction (azimuth/elevation).

signal_clarity_score()

sv_psd([n_lines])

Compute the singular values of the power spectral density matrices, for which the complete (all cross spectral densities) matrices are used.

undo()

Restore the state captured by the last save_undo_snapshot() call.

validate_channels(channels[, quant_check])

welch(n_lines, **kwargs)

Attributes

accel_channels

corr_matrices

corr_matrix

disp_channels

dt

duration

freqs

returns: freqs -- Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

freqs_bt

returns: freqs -- Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

freqs_wl

returns: freqs -- Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

lags

lags_bt

lags_wl

m_lags

n_lines

n_segments

num_analised_channels

num_ref_channels

psd_matrix

ref_channels

signal_power

signal_rms

t

total_time_steps

undo_available

Whether a single-step undo snapshot is available.

velo_channels

add_chan_dofs(chan_dofs)[source]#

chan_dofs = [ (chan_num, node_name, az, elev, chan_name) , … ] This function is not checking if channels or nodes actually exist the former should be added the latter might only be possible, if the geometry object is known to the class

add_noise(amplitude=0, snr=0)[source]#

Add Gaussian white noise to the signals (useful for simulation studies).

Parameters:
  • amplitude (float, optional) – Absolute noise amplitude (standard deviation). Ignored when snr is non-zero.

  • snr (float, optional) – Noise amplitude as a fraction of the per-channel RMS. At least one of amplitude or snr must be non-zero.

corr_blackman_tukey(m_lags, n_segments=None, refs_only=True, **kwargs)[source]#

Estimate the (cross- and auto-) correlation functions (C/ACF), by direct computation of the standard un-biased estimator:

\[\hat{R}_{fg}[m] = \frac{1}{N - m}\sum_{n=0}^{N - m - 1} f[n] g[n + m]\]

Computes correlation functions of all channels with selected reference channels up to, but excluding, a time lag of m_lags. Normalization is done according to the unbiased estimator, i.e. 0-lag correlation value must be multiplied by n_lines to get the signals cross-power.

Variance estimation for each time lag is performed by dividing the signals into n_segments non-overlapping blocks for individual estimation of correlation functions. With increasing numbers of non-overlapping blocks the confidence intervals of the correlation functions increase (“worsen”), especially at higher lags and short block lengths, due to a larger number of time steps being discarded.

Note that:

m_lags = n_lines // 2 + 1

n_lines = (m_lags - 1) * 2

Parameters:
  • m_lags (integer, optional) – Number of lags (positive). Note: this includes the 0-lag, therefore excludes the “m_lags”-lag.

  • n_segments (integer, optional) – Number of blocks to perform averaging over. If blocks are shorter than m_lags it raises a ValueError.

  • refs_only (bool, optional) – Compute cross-ACFss only with reference channels

  • kwargs – Additional kwargs are currently not used

Returns:

corr_matrix – Array of shape (num_channels, num_ref_channels, m_lags) containing the correlation values of the respective channels and lags

Return type:

np.ndarray

See also

corr_welch

Correlation function estimation by Welch’s method, possibly faster, but distorted for short segments and biased through windowing.

corr_welch(m_lags=None, n_segments=None, refs_only=True, **kwargs)[source]#

Estimate the (cross- and auto-) correlation functions (C/ACF), by the inverse Fourier Transform of Power Spectral Densities, estimated according to Welch’s method. Bias due to windowing of the underlying PSD persists. Normalization is done according to the unbiased estimator, i.e. 0-lag correlation value must be multiplied by n_lines to get the signals cross-power.

Note that:

m_lags = n_lines // 2 + 1

n_lines = (m_lags - 1) * 2

N_segment = N // n_segments

Parameters:
  • m_lags (integer, optional) – Total number of lags (positive). Note: this includes the 0-lag, therefore excludes the m_lags-lag.

  • n_segments (integer, optional) – Number of segments to perform averaging over resulting segment length must be smaller or equal n_lines

  • refs_only (bool, optional) – Compute cross-ACFss only with reference channels

  • kwargs – Additional kwargs are passed to self.psd_welch and further

Returns:

corr_matrix – Array of shape (num_channels, num_ref_channels, m_lags) containing the correlation values of the respective channels and lags

Return type:

np.ndarray

See also

psd_welch

PSD estimation algorithm used by this method.

Todo

  • deconvolve window (if possible)

correct_offset()[source]#

corrects a constant offset from measured signals

..TODO::
  • remove linear, … ofsets as well

correlation(m_lags=None, method=None, **kwargs)[source]#

A convenience method for obtaining the correlation sequence by the default or any specified estimation method.

Parameters:
  • m_lags (integer, optional) – Number of lags (positive). Note: this includes the 0-lag, therefore excludes the m_lags-lag.

  • method (str, optional) – The method to use for spectral estimation

  • kwargs – Additional parameters are passed to the spectral estimation method

Returns:

corr_matrix – Array of shape (num_channels, num_ref_channels, m_lags) containing the correlation values of the respective channels and lags

Return type:

np.ndarray

decimate_signals(decimate_factor, nyq_rat=2.5, highpass=None, order=None, filter_type='cheby1')[source]#

Decimate the signals by an integer factor.

An anti-aliasing lowpass filter is applied before downsampling. To achieve large total reduction factors, call this method multiple times with moderate per-step factors (e.g. two passes of x3 instead of one pass of x9).

Parameters:
  • decimate_factor (int) – Integer downsampling factor (must be >= 1).

  • nyq_rat (float, optional) – The lowpass corner frequency is set to sampling_rate / (decimate_factor * nyq_rat). Must be >= 2. Default is 2.5.

  • highpass (float or None, optional) – Additional highpass corner frequency in Hz applied simultaneously.

  • order (int, optional) – Anti-aliasing filter order. Defaults to 8 (IIR) or 21 * decimate_factor - 1 (FIR).

  • filter_type (str, optional) – Filter type passed to filter_signals(). Default is 'cheby1'.

delete_channels(channels)[source]#

Remove one or more channels from the signals and all associated channel-index bookkeeping.

Parameters:

channels (int, str, or list of int/str) – The channel(s) to remove, by index or name (see _channel_numbers() for accepted formats).

filter_signals(lowpass=None, highpass=None, overwrite=True, order=None, ftype='butter', RpRs=None, plot_ax=None)[source]#

Apply a zero-phase IIR or FIR filter to the measurement signals.

Parameters:
  • lowpass (float, optional) – Lowpass corner frequency in Hz.

  • highpass (float, optional) – Highpass corner frequency in Hz. At least one of lowpass or highpass must be given; providing both creates a bandpass filter.

  • overwrite (bool, optional) – If True (default), store the filtered signals in self.signals. If False, return the filtered array without modifying self.

  • order (int, optional) – Filter order. Defaults to 4 for IIR types and 21 for FIR types.

  • ftype (str, optional) – Filter type: 'butter', 'cheby1', 'cheby2', 'ellip', 'bessel', 'moving_average', or 'brickwall'.

  • RpRs (list of float, optional) – [rp, rs] — maximum passband ripple and minimum stopband attenuation (dB) for Chebyshev/elliptic filters.

  • plot_ax (matplotlib.axes.Axes or list of Axes, optional) – When provided, plot the filter frequency response (and optionally impulse response) into the given axes.

Returns:

Filtered signal array (returned regardless of overwrite).

Return type:

np.ndarray

property freqs#

returns: freqs – Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

property freqs_bt#

returns: freqs – Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

property freqs_wl#

returns: freqs – Array with the frequency lines corresponding to the spectral values :rtype: np.ndarray (n_lines, )

get_chan_dof(channel)[source]#

Return the (node, az, elev) assignment for channel, or None.

Parameters:

channel (int) – Channel index.

classmethod init_from_config(conf_file, meas_file, chan_dofs_file=None, **kwargs)[source]#

initializes the PreProcessor object with a configuration file

to remove channels at loading time use ‘usecols’ keyword argument if delete_channels are specified, these will be checked against all other channel definitions, which will be adjusted accordingly

static load_chan_dofs(fname)[source]#
chan_dofs[i] = (chan_num, node_name, az, elev, chan_name)

= (int, str, float,float, str)

azimuth angle starting from x axis towards y axis elevation defined from x-y plane up x: 0.0, 0.0 y: 90.0, 0.0 z: 0.0, 90.0 channels not present in the file will be removed later nodes do not have to, but should exist, as this information is also used for merging multiple setups, which does not rely on any “real” geometry

static load_measurement_file(fname, **kwargs)[source]#

A method for loading a signals file

Parameters:

fname (str) – The full path of the signals file

Returns:

  • headers (list of str) – The names of all channels

  • units (list of str) – The units of all channels

  • start_time (datetime.datetime) – The starting time of the measured signal

  • sample_rate (float) – The sample rate, at wich the signal was acquired

  • signals (ndarray) – Array of shape (num_timesteps, num_channels) which contains the acquired signal

Notes

This default implementation only handles a bare .npy array (returned as-is; init_from_config() then synthesizes headers/units/start_time/sample_rate). For any other measurement-file format, assign a custom loader to PreProcessSignals.load_measurement_file before calling init_from_config() - see scripts/converters/ for examples.

precondition_signals(method='iqr')[source]#

Remove the DC offset and scale each channel by its spread.

First calls correct_offset(), then divides each channel by either the inter-quartile range (IQR, 5th-95th percentile) or the full signal range. Scaling factors are stored in self.channel_factors.

Parameters:

method ({'iqr', 'range'}, optional) – Spreading measure used for normalisation. Default is 'iqr'.

psd(n_lines=None, method=None, **kwargs)[source]#

A convenience method for obtaining the PSD by the default or any specified estimation method.

Parameters:
  • n_lines (integer, optional) – Number of frequency lines (positive + negative)

  • method – The method to use for spectral estimation

  • **kwargs – Additional parameters are passed to the spectral estimation method

Returns:

psd_matrix – Array of shape (num_channels, num_ref_channels, n_lines // 2 + 1) containing the power density values of the respective channels and frequencies

Return type:

np.ndarray

psd_blackman_tukey(n_lines=None, refs_only=True, window='hamming', **kwargs)[source]#

Estimate the (cross- and auto-) power spectral densities (PSD), by Fourier Transform of correlation functions estimated according to Blackman-Tukey’s method. Non-negativeness of the PSD is ensured by using a lag window, i.e. convolving the temporal window with itself. Normalization is applied w.r.t. conservation of energy, i.e. magnitudes will change with n_lines but power stays constant.

Note that:

m_lags = n_lines // 2 + 1 n_lines = (m_lags - 1) * 2

Parameters:
  • n_lines (integer, optional) – Number of frequency lines (positive + negative)

  • refs_only (bool, optional) – Compute cross-PDSs only with reference channels

  • window (str or tuple or array_like, optional) – Desired temporal window to be applied to the correlation sequence after conversion to a lag window by “self-convolution” See scipy.signal.get_window() for more information

  • kwargs – Additional kwargs are passed to self.corr_blackman_tukey

Returns:

psd_matrix – Array of shape (num_channels, num_ref_channels, n_lines // 2 + 1) containing the power density values of the respective channels and frequencies

Return type:

np.ndarray

psd_welch(n_lines=None, n_segments=None, refs_only=True, window='hamming', **kwargs)[source]#

Estimate the (cross- and auto-) power spectral densities (PSD), according to Welch’s method. No overlapping is allowed (deliberately to ensure statistical independence of blocks for variance estimation). Segments are n_lines // 2 long and zero padded to n_lines to allow estimation of the full correlation sequence, which is twice as long as the input signal. Normalization is applied w.r.t. conservation of energy, i.e. magnitudes will change with n_lines but power stays constant.

Parameters:
  • n_lines (integer, optional) – Number of frequency lines (positive + negative)

  • n_segments (integer, optional) – Number of segments to perform averaging over resulting segment length must be smaller or equal n_lines

  • refs_only (bool, optional) – Compute cross-PDSs only with reference channels

  • window (str or tuple or array_like, optional) – Desired window to use. See scipy.signal.get_window() for more information

  • kwargs – Additional kwargs are passed to scipy.signals.csd

Returns:

psd_matrix – Array of shape (num_channels, num_ref_channels, n_lines // 2 + 1) containing the power density values of the respective channels and frequencies

Return type:

np.ndarray

remove_chan_dof(channel)[source]#

Remove any existing DOF assignment for channel, if present.

Parameters:

channel (int) – Channel index.

rename_channel(channel, new_name)[source]#

Rename a channel, keeping any chan_dofs annotation in sync.

Parameters:
  • channel (int) – Channel index.

  • new_name (str) – New name for the channel. Must be unique among all channel names, since channels can be looked up by name elsewhere (e.g. _str_channel_to_index()).

save_chan_dofs(fname)[source]#

Write self.chan_dofs to fname in the format read by load_chan_dofs().

save_config(fname, delete_channels=None)[source]#

Write a config file readable by init_from_config().

Parameters:
  • fname (str)

  • delete_channels (list of int, optional) – Recorded as-is (informational only — channels are already absent from self.signals by the time this is called).

save_undo_snapshot()[source]#

Capture the current state for a single-step undo.

Stub: not yet implemented. Intended to be called at the start of each mutating action (correct_offset(), add_noise(), filter_signals(), decimate_signals(), delete_channels(), …) so undo() can restore exactly the state before that action. Only one snapshot is kept - a new call overwrites the previous one (single-step, not a full history).

set_chan_dof(channel, node, az, elev)[source]#

Assign channel to a node and direction (azimuth/elevation).

Replaces any existing assignment for this channel - a channel measures at most one DOF.

Parameters:
  • channel (int) – Channel index.

  • node (str) – Name of the node in geometry_data.nodes this channel is attached to.

  • az (float) – Azimuth and elevation (degrees) of the measured direction.

  • elev (float) – Azimuth and elevation (degrees) of the measured direction.

sv_psd(n_lines=None, **kwargs)[source]#

Compute the singular values of the power spectral density matrices, for which the complete (all cross spectral densities) matrices are used.

Parameters:
  • n_lines (integer, optional) – Number of frequency lines (positive + negative)

  • kwargs – Additional parameters are passed to the spectral estimation method

undo()[source]#

Restore the state captured by the last save_undo_snapshot() call.

Stub: not yet implemented.

property undo_available#

Whether a single-step undo snapshot is available.

Stub: snapshot capture is not implemented yet, so this is always False. Once save_undo_snapshot()/undo() are filled in, this should reflect whether a snapshot has been captured and not yet consumed by undo().