Source code for gammapy.data.filters

# Licensed under a 3-clause BSD style license - see LICENSE.rst
import copy
import logging

__all__ = ["ObservationFilter"]

log = logging.getLogger(__name__)


[docs]class ObservationFilter: """Holds and applies filters to observation data. Parameters ---------- time_filter : `astropy.time.Time` Start and stop time of the selected time interval. For now we only support a single time interval. event_filters : list of dict An event filter dictionary needs two keys: - **type** : str, one of the keys in `~gammapy.data.ObservationFilter.EVENT_FILTER_TYPES` - **opts** : dict, it is passed on to the method of the `~gammapy.data.EventListBase` class that corresponds to the filter type (see `~gammapy.data.ObservationFilter.EVENT_FILTER_TYPES`) The filtered event list will be an intersection of all filters. A union of filters is not supported yet. Examples -------- >>> from gammapy.data import ObservationFilter, DataStore, Observation >>> from astropy.time import Time >>> from astropy.coordinates import Angle >>> >>> time_filter = Time(['2021-03-27T20:10:00', '2021-03-27T20:20:00']) >>> phase_filter = {'type': 'custom', 'opts': dict(parameter='PHASE', band=(0.2, 0.8))} >>> >>> my_obs_filter = ObservationFilter(time_filter=time_filter, event_filters=[phase_filter]) >>> >>> ds = DataStore.from_dir("$GAMMAPY_DATA/cta-1dc/index/gps") >>> my_obs = ds.obs(obs_id=111630) >>> my_obs.obs_filter = my_obs_filter """ EVENT_FILTER_TYPES = dict(sky_region="select_region", custom="select_parameter") def __init__(self, time_filter=None, event_filters=None): self.time_filter = time_filter self.event_filters = event_filters or [] @property def livetime_fraction(self): """Fraction of the livetime kept when applying the event_filters.""" return self._check_filter_phase(self.event_filters)
[docs] def filter_events(self, events): """Apply filters to an event list. Parameters ---------- events : `~gammapy.data.EventListBase` Event list to which the filters will be applied Returns ------- filtered_events : `~gammapy.data.EventListBase` The filtered event list """ filtered_events = self._filter_by_time(events) for f in self.event_filters: method_str = self.EVENT_FILTER_TYPES[f["type"]] filtered_events = getattr(filtered_events, method_str)(**f["opts"]) return filtered_events
[docs] def filter_gti(self, gti): """Apply filters to a GTI table. Parameters ---------- gti : `~gammapy.data.GTI` GTI table to which the filters will be applied Returns ------- filtered_gti : `~gammapy.data.GTI` The filtered GTI table """ return self._filter_by_time(gti)
def _filter_by_time(self, data): """Returns a new time filtered data object. Calls the `select_time` method of the data object. """ if self.time_filter: return data.select_time(self.time_filter) else: return data
[docs] def copy(self): """Copy the `ObservationFilter` object.""" return copy.deepcopy(self)
@staticmethod def _check_filter_phase(event_filter): if not event_filter: return 1 for f in event_filter: if f.get("opts").get("parameter") == "PHASE": band = f.get("opts").get("band") return band[1] - band[0] else: return 1