edges.filters.filters

Functions that identify and flag bad data in various ways.

edges.filters.filters.apply_flags(*, data: GSData, flags: str | Path | GSFlag)

Apply flags from a file.

edges.filters.filters.aux_filter(*, data: GSData, minima: dict[str, float] | None = None, maxima: dict[str, float] | None = None) → GSFlag

Perform an auxiliary filter on the object.

Parameters:

• minima – Dictionary mapping auxiliary data keys to minimum allowed values.

• maxima – Dictionary mapping auxiliary data keys to maximum allowed values.

Returns:

flags – Boolean array giving which entries are bad.

edges.filters.filters.explicit_day_filter(data: GSData, flag_days: list[tuple[int, int] | tuple[int, int, int] | int | Time]) → GSFlag

Filter out any data coming from specific days.

Parameters:

flag_days – A list of days to flag. Each entry can be a 2-tuple, 3-tuple, astropy.Time or an int. If a 2-tuple, it is interpreted as (year, day_of_year). If a 3-tuple, it is interpreted as (year, month, day). If an int, it is interpreted as a Julian day.

edges.filters.filters.filter_150mhz(*, data: GSData, threshold: float)

Filter data based on power around 150 MHz.

This takes the RMS of the power around 153.5 MHz (in a 1.5 MHz bin), after subtracting the mean, then compares this to the mean power of a 1.5 MHz bin around 157 MHz (which is expected to be cleaner). If this ratio (RMS to mean) is greater than 200 times the threshold given, the integration will be flagged.

edges.filters.filters.flag_frequency_ranges(*, data: GSData, freq_ranges: list[tuple[float, float]], invert: bool = False)

Flag explicit frequency ranges.

Parameters:

• data – The data to flag.

• freq_ranges – A list of tuples, each containing the start and end of a frequency range to flag in MHz.

• invert – If True, invert the flagging (i.e. only keep the data inside the ranges given).

edges.filters.filters.galaxy_filter(*, data: GSData, elevation_range: tuple[~astropy.coordinates.angles.core.Angle, ~astropy.coordinates.angles.core.Angle]=(<Quantity -90. deg>, <Quantity 0. deg>)) → GSFlag

Perform a filter based on the Galactic center position.

Parameters:

elevation_range – The minimum and maximum allowed elevation (as an astropy Angle).

edges.filters.filters.gsdata_filter(multi_data: bool = False)

A decorator to register a filtering function as a potential filter.

Any function that is wrapped by gsdata_filter() must implement the following signature:

def fnc(
    data: GSData | Sequence[GSData],
    use_existing_flags: bool,
    **kwargs
) -> GSFlag

Where the data is either a single GSData object, or sequence of such objects.

The return value should be a GSFlag object, which contains the flags.

Parameters:

multi_data – Whether the filter accepts multiple objects at the same time to filter. This is usually so as to enable more accurate filtering when comparing different days for instance, rather than just performing a loop over the days and flagging each independently.

edges.filters.filters.maxfm_filter(*, data: GSData, threshold: float = 200)

Filter data based on large single-channel spikes in FM band.

This function is only provided as a convenience when comparing to the legacy code that had the same filter with this name. It is really just a very thin wrapper around single_channel_spike_filter, focusing on the FM band.

edges.filters.filters.moon_filter(*, data: GSData, elevation_range: tuple[float, float]) → ndarray

Perform a filter based on moon position.

Parameters:

elevation_range – The minimum and maximum allowed sun elevation.

edges.filters.filters.negative_power_filter(*, data: GSData)

Filter out integrations that have any negative/zero power.

These integrations obviously have some weird stuff going on.

edges.filters.filters.peak_orbcomm_filter(*, data: GSData, threshold: float = 40.0, mean_freq_range: tuple[float, float] | None = (80, 200))

Filter out whole integrations that have high power between (137, 138) MHz.

Parameters:

• threshold – This is the threshold beyond which the peak power causes the integration to be flagged. The units of the threhsold are 10*log10(peak_power / mean), where the mean is the mean power of spectrum in the mean_freq_range (omitting power spikes > peak_power/10)

• mean_freq_range – The range of frequencies over which to take a mean to compare to the peak. By default, the same as the peak_freq_range.

edges.filters.filters.peak_power_filter(*, data: GSData, threshold: float = 40.0, peak_freq_range: tuple[float, float] = (80, 200), mean_freq_range: tuple[float, float] | None = None)

Filter out whole integrations that have high power > 80 MHz.

Parameters:

• threshold – This is the threshold beyond which the peak power causes the integration to be flagged. The units of the threhsold are 10*log10(peak_power / mean), where the mean is the mean power of spectrum in the same frequency range (omitting power spikes > peak_power/10)

• peak_freq_range – The range of frequencies over which to search for the peak.

• mean_freq_range – The range of frequencies over which to take a mean to compare to the peak. By default, the same as the peak_freq_range.

edges.filters.filters.power_percent_filter(*, data: GSData, freq_range: tuple[float, float] = (100, 200), min_threshold: float = 0, max_threshold: float = 3)

Filter data based on the ratio of power in a band compared to entire dataset.

This filter computes the sum of power from the input connected to the antenna within a given band, and finds the ratio within that band compared to the entired dataset. If that ratio is outside the thresholds given for a given timestamp, then the entired integration is flagged.

Note: this is a very bespoke filter. Thresholds that make sense will depend on both the freq_range given, and the frequency range of the data itself. In this regard, it is very flexible, but care must be taken to set the parameters appropriately.

Parameters:

• data (GSData) – The data to be flagged.

• freq_range (tuple[float, float]) – The frequency range of the power to be summed in the numerator, in MHz.

• min_threshold (float) – Threshold of the ratio below which the integration will be flagged.

• max_threshold (float) – Threshold of the ratio above which the integration will be flagged.

edges.filters.filters.prune_flagged_integrations(data: GSData, **kwargs) → GSData

Remove integrations that are flagged for all freq-pol-loads.

edges.filters.filters.rms_filter(data: ~pygsdata.gsdata.GSData, threshold: float, freq_range: tuple[~typing.Annotated[~astropy.units.quantity.Quantity, PhysicalType('frequency')], ~typing.Annotated[~astropy.units.quantity.Quantity, PhysicalType('frequency')]] = (<Quantity 0. MHz>, <Quantity inf MHz>), nsamples_strategy: ~edges.averaging.utils.NsamplesStrategy = NsamplesStrategy.FLAGGED_NSAMPLES, model: ~edges.modeling.core.Model | None = None) → bool

Filter integrations based on the rms of the residuals.

Parameters:

• data – The data to be filtered.

• threshold – The threshold at which to flag integrations.

• freq_range – The frequency range to use in calculating the RMS.

• nsamples_strategy – The strategy to use to infer weights for computing RMS.

• model – A model to be used to fit each integration. Not required if a model already exists on the data.

edges.filters.filters.rms_rfi_filter(data: GSData, threshold: float = 3.0, nsamples_strategy: NsamplesStrategy = NsamplesStrategy.FLAGGED_NSAMPLES) → GSFlag

Flag specific channel-integrations via their outlier-ness compared to RMS.

edges.filters.filters.rmsf_filter(*, data: GSData, threshold: float = 200, freq_range: tuple[float, float] = (60, 80)) → GSFlag

Filter data based on rms calculated between 60 and 80 MHz.

Note that this function is deprecated in favour of the more general rms_filter().

Deprecated since version 8.1.0: This will be removed in 9.0.0. Use the rms_filter function instead

edges.filters.filters.single_channel_spike_filter(*, data: ~pygsdata.gsdata.GSData, threshold: float = 200, freq_range: tuple[~typing.Annotated[~astropy.units.quantity.Quantity, PhysicalType('frequency')], ~typing.Annotated[~astropy.units.quantity.Quantity, PhysicalType('frequency')]] = (<Quantity 88. MHz>, <Quantity 120. MHz>))

Filter data based on single channel spikes.

This filter detrends the data using a simple convolution kernel with weights [0.5, 0, 0.5], which makes single channel spikes stand out. The entire spectrum is flagged if the residual of the original spectrum to the de-trended is larger than the threshold.

edges.filters.filters.sky_coord_filter(*, data: GSData, coord: str | SkyCoord, elevation_range: tuple[Angle, Angle]) → GSFlag

Perform a filter based on a sky coordinate position.

Parameters:

• coord – The sky coordinate to filter on.

• elevation_range – The minimum and maximum allowed elevation (as an astropy Angle).

edges.filters.filters.sun_filter(*, data: GSData, elevation_range: tuple[float, float]) → GSFlag

Perform a filter based on sun position.

Parameters:

elevation_range – The minimum and maximum allowed sun elevation in degrees