Filter¶

class sofia_redux.scan.filters.filter.Filter(integration=None, data=None)[source]¶

Bases: ABC

Initialize an integration filter.

The filter is designed to filter integration data using an FFT. This is an abstract class designed to hold a few main parameters necessary for filtering. Each filter requires an integration on which to operate, and defines a response at any given frequency.

Parameters:

integrationIntegration, optional
datanumpy.ndarray (float), optional: An array of shape (nt, n_channels) where nt is the nearest power of 2 integer above the number of integration frames. i.e., if n_frames=5, nt=8, or if n_frames=13, nt=16.

Attributes Summary

`channel_dependent_attributes`	Return attributes that are dependent on the parent channels.
`channel_flagspace`	Return the channel flagspace.
`channel_size`	Return the number of channels in the filter.
`configuration`	Return the configuration (taken from the integration).
`flagspace`	Return the frame flagspace.
`frames`	Return the integration frames.
`info`	Return the integration information.
`referenced_attributes`	Return attributes that should be referenced during the copy operation.
`size`	Return the number of integration frames.
`valid_filtering_frames`	Return frames that are valid and not flagged as modeling.

Methods Summary

`apply`([report])	Apply the filter to the integration.
`apply_to_channels`([channels])	Apply the filter the integration for the channel group.
`calc_point_response`()	Returns the point response of a Gaussian source for the filter.
`copy`()	Return a copy of the filter.
`count_parms`()	Return the total dependent count of the filter.
`dft_filter`([channels])	Return the filter rejection using a discrete FFT.
`discard_temp_data`()	Destroy the temporary data.
`fft_filter`([channels])	Apply the FFT filter to the temporary data.
`get_channels`()	Return the associated channel group for the integration filter.
`get_config_name`()	Return the configuration name.
`get_high_pass_index`()	Return the high pass filter frequency index.
`get_id`()	Return the filter ID.
`get_mean_point_response`()	Return the mean point response of the filter.
`get_temp_data`()	Return the temporary data.
`has_option`(key)	Return whether an option is set in the configuration for this filter.
`is_enabled`()	Return whether the filter is enabled.
`level`(signal)	Level the given signal data.
`level_data`()	Level (remove average from) the filter data.
`level_data_for_channels`([channels])	Level (remove average) from the filter data.
`level_for_channels`(signal[, channels])	Level (remove average) from a given signal.
`load_time_streams`([channels])	Load time streams from integration frame data.
`make_temp_data`([channels])	Create the initial temporary data.
`option`(key)	Return the value for a given filter option.
`post_filter`()	Perform the post-filtering steps.
`post_filter_channels`([channels])	Performs the post-filtering channels steps.
`pre_filter`()	Perform the pre-filtering steps.
`pre_filter_channels`([channels])	Performs the pre-filtering channels steps.
`reindex`()	Reindex the channel groups to be consistent with parent channels.
`rejection_at`(fch)	Return the filter rejection at given frequency channel(s).
`remove`([channels])	Subtract the filtered signal from the integration frame data.
`remove_from_frames`(rejected_signal, frames, ...)	Remove the rejected signal from frame data.
`report`()	Add messages to the integration comments regarding the filtering.
`response_at`(fch)	Return the response at a given frequency channel(s).
`set_channels`(channels)	Set the filter channels.
`set_dft`(value)	Set the DFT flag that determines whether to use FFT or DFT filtering.
`set_integration`(integration)	Set the filter integration.
`set_temp_data`(data)	Set the temporary data.
`update_config`()	Update the filter based on configuration settings.
`update_profile`([channels])	Update the profile for given channel(s).

Attributes Documentation

channel_dependent_attributes¶

Return attributes that are dependent on the parent channels.

This is required during a slim operation (reducing number of channels) which may happen to the filter integration.

Returns:

set (str)

channel_flagspace¶

Return the channel flagspace.

Returns:

Flags

channel_size¶

Return the number of channels in the filter.

Returns:

n_channelsint

configuration¶

Return the configuration (taken from the integration).

Returns:

Configuration

flagspace¶

Return the frame flagspace.

Returns:

Flags

frames¶

Return the integration frames.

Returns:

Frames

info¶

Return the integration information.

Returns:

Info

referenced_attributes¶

Return attributes that should be referenced during the copy operation.

Returns:

set (str)

size¶

Return the number of integration frames.

Returns:

n_framesint

valid_filtering_frames¶

Return frames that are valid and not flagged as modeling.

Returns:

valid_framesnumpy.ndarray (bool): An array of shape (n_frames,) where True marks a valid frame to be included in filtering operations.

Methods Documentation

apply(report=True)[source]¶

Apply the filter to the integration.

There are a few steps that are performed during filter application:

Read the configuration to update any settings. - Stop if the filter is not enabled in the configuration.

Pre-filter the channels. This generally involved removing the filter frame and channel dependents form the integration, before zeroing them in the filter in order to be re-calculated.

Apply the filter to the channels.

Post-filter the channels. This generally involves adding the filter frame and channel dependents back to the integration.

Parameters:

reportbool, optional: If True, add messages to the integration comments on the filtering.

Returns:

appliedbool: True if filtering occurred, and False otherwise.

apply_to_channels(channels=None)[source]¶

Apply the filter the integration for the channel group.

Parameters:

channelsChannelGroup, optional: If not supplied, defaults to the filter channels.

Returns:

None

calc_point_response()[source]¶

Returns the point response of a Gaussian source for the filter.

It is assumed a Gaussian source profile is crossed during the integration crossing time such that

sigma(t) = 1 t/2.35 * 2pi * sigma(f) = 1 sigma(f) = 2.35 / (2pi * t) df = 1 / (n dt) sigma(F) = sigma(f) / df = 2.35 * n * dt / (2pi * t)

The filter response is not accounted for below the high-pass timescale.

Returns:

point_responsefloat

copy()[source]¶

Return a copy of the filter.

All attributes are copied aside from the integration and channels which are referenced only.

Returns:

Filter

count_parms()[source]¶

Return the total dependent count of the filter.

The number of filter dependents is calculated as the sum of the filter rejection (varies from 0 to 1) between the high-pass and Nyquist frequency (inclusive).

Returns:

rejectedfloat or numpy.ndarray (float): The sum of the rejected signal above the high pass frequency. If this is a filter that varies by channel, an array will be returned of shape (n_channels,). Otherwise, a singular float value will be returned.

dft_filter(channels=None)[source]¶

Return the filter rejection using a discrete FFT.

The speed of the FFT depends significantly on the number of non-zero rejection values. This will be faster that the fft_filter method under certain circumstances.

Parameters:

channelsChannelGroup, optional: The channel group for which the filtering applied. By default, set to the filtering channels.

Returns:

None

discard_temp_data()[source]¶

Destroy the temporary data.

Returns:

None

fft_filter(channels=None)[source]¶

Apply the FFT filter to the temporary data.

Converts data into a rejected (un-levelled) signal

Parameters:

channelsChannelGroup, optional: The channels for which to apply the filter. If not supplied, defaults to the stored filtering channels.

Returns:

None

get_channels()[source]¶

Return the associated channel group for the integration filter.

Returns:

ChannelGroup

abstract get_config_name()[source]¶

Return the configuration name.

Returns:

config_namestr

get_high_pass_index()[source]¶

Return the high pass filter frequency index.

Returns:

indexint

abstract get_id()[source]¶

Return the filter ID.

Returns:

filter_idstr

abstract get_mean_point_response()[source]¶

Return the mean point response of the filter.

Returns:

responsefloat

get_temp_data()[source]¶

Return the temporary data.

Returns:

numpy.ndarray (float)

has_option(key)[source]¶

Return whether an option is set in the configuration for this filter. The actual key checked for in the configuration will be:

<filter_config_name>.<key>

I.e., the “level” key for a whitening filter will be checked for in the configuration as “filter.whiten.level”.

Parameters:

keystr: The name of the option for the filter.

Returns:

is_configuredbool

is_enabled()[source]¶

Return whether the filter is enabled.

Returns:

bool

level(signal)[source]¶

Level the given signal data.

Parameters:

signalnumpy.ndarray (float): The signal data to level. Leveling occurs for each channel over all frames. The signal shape should be (n_channels, n_frames) or (n_frames,).

Returns:

None

level_data()[source]¶

Level (remove average from) the filter data.

Returns:

None

level_data_for_channels(channels=None)[source]¶

Level (remove average) from the filter data.

Parameters:

channelsChannelGroup, optional: The channels for which to average the data, by default

Returns:

None

level_for_channels(signal, channels=None)[source]¶

Level (remove average) from a given signal.

Parameters:

signalnumpy.ndarray (float): The data to level of shape (channels.size, n_frames).
channelsChannelGroup, optional: The channel group for which the data applied. By default set to the filtering channels.

Returns:

None

load_time_streams(channels=None)[source]¶

Load time streams from integration frame data.

Timestream data is defined as the frame data multiplied by their relative weights with the mean subtracted. Output data are stored in the data temporary array. Invalid data are set to zero. The points array is also populated and contains the sum of frame weights for each channel. Invalid points are set to zero.

Parameters:

channelsChannelGroup, optional: The channels for which to load time streams. If not supplied, defaults to the filtering channels.

Returns:

None

make_temp_data(channels=None)[source]¶

Create the initial temporary data.

Creates a temporary data array of shape (n_channels, n_frames).

Parameters:

channelsChannelGroup, optional: The channels for which to create the temporary data. If not supplied, defaults to the filtering channels.

Returns:

None

option(key)[source]¶

Return the value for a given filter option.

The given key relates specifically to this filter in the configuration. As such, the actual configuration will search for the configuration value associated with <filter_config_name>.<key>. I.e., the “level” key for a whitening filter will be checked for in the configuration as “filter.whiten.level”.

Parameters:

keystr: The name of the option for the filter.

Returns:

str

post_filter()[source]¶

Perform the post-filtering steps.

During the post-filtering steps, dependents are added to the integration and integration channels.

Returns:

None

post_filter_channels(channels=None)[source]¶

Performs the post-filtering channels steps.

Parameters:

channelsChannelGroup, optional: The channel group for which to perform the post-filtering step. If not supplied, defaults to the filtering channels.

Returns:

None

pre_filter()[source]¶

Perform the pre-filtering steps.

During the pre-filtering steps, dependents are retrieved from the filter integration and cleared (subtracted from integration dependents, integration channels, and zeroed).

Returns:

None

pre_filter_channels(channels=None)[source]¶

Performs the pre-filtering channels steps.

Parameters:

channelsChannelGroup, optional: The channel group for which to perform the pre-filtering step. If not supplied, defaults to the filtering channels.

Returns:

None

reindex()[source]¶

Reindex the channel groups to be consistent with parent channels.

When the channels in the filter integration change for some reason (such as a slim operation in which bad channels are removed), filter attributes that map to integration channels will also need to be updated to account for such a change.

At a base level, the filter attributes that need to be updated include channel dependents, and the points (sum of frame weights) for each channel. The data attribute is also set to None, since it is designed to be a temporary workspace that may or may not depend on the integration channels depending on the type of filter in question.

rejection_at(fch)[source]¶

Return the filter rejection at given frequency channel(s).

The filter rejection is equal to 1 minus the filter response:

rejection = 1 - response

Parameters:

fchint or numpy.ndarray (int or bool) or slice.

Returns:

responsefloat or numpy.ndarray (float)

remove(channels=None)[source]¶

Subtract the filtered signal from the integration frame data.

Parameters:

channelsChannelGroup, optional: The channel group for which to remove the signal. If not supplied, defaults to the filtering channels.

Returns:

None

static remove_from_frames(rejected_signal, frames, channels)[source]¶

Remove the rejected signal from frame data.

Parameters:

rejected_signalnumpy.ndarray (float): The rejected signal of shape (filtering_channels.size, n_frames).
framesFrames: The frames for which to remove the rejected signal.
channelsChannelGroup: The channel for which to subtract the signal.

Returns:

None

report()[source]¶

Add messages to the integration comments regarding the filtering.

Returns:

None

abstract response_at(fch)[source]¶

Return the response at a given frequency channel(s).

Parameters:

fchint or numpy.ndarray (int or bool) or slice: The frequency channel or channels in question.

Returns:

responsefloat or numpy.ndarray (float)

set_channels(channels)[source]¶

Set the filter channels.

The channels attribute will be set to a ChannelGroup type.

Parameters:

channelsChannels or ChannelData or ChannelGroup

Returns:

None

set_dft(value)[source]¶

Set the DFT flag that determines whether to use FFT or DFT filtering.

Parameters:

valuebool: If True, sets the filtering method to DFT.

Returns:

None

set_integration(integration)[source]¶

Set the filter integration.

Sets the padding of the FFT filter, the number of frequencies, the frequency spacing, and retrieves the channels from the integration if necessary.

Parameters:

integrationIntegration

Returns:

None

set_temp_data(data)[source]¶

Set the temporary data.

Parameters:

datanumpy.ndarray (float)

Returns:

None

update_config()[source]¶

Update the filter based on configuration settings.

Returns:

None

update_profile(channels=None)[source]¶

Update the profile for given channel(s).

Parameters:

channelsint or numpy.ndarray (int), optional: The channel indices to update.

Returns:

None

Navigation

Filter¶