# Licensed under a 3-clause BSD style license - see LICENSE.rst
from __future__ import absolute_import, division, print_function, unicode_literals
import copy
import logging
__all__ = ["ObservationFilter"]
log = logging.getLogger(__name__)
[docs]class ObservationFilter(object):
"""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, DataStoreObservation
>>> 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 = DataStoreObservation(obs_id=111630, data_store=ds, obs_filter=my_obs_filter)
"""
EVENT_FILTER_TYPES = dict(
circular_region="select_circular_region", custom="select_parameter"
)
def __init__(self, time_filter=None, event_filters=None):
self.time_filter = time_filter
self.event_filters = event_filters or []
[docs] def filter_events(self, events):
"""Applies the 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):
"""Applies the 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
"""
filtered_gti = self._filter_by_time(gti)
return filtered_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)