Source code for

# Licensed under a 3-clause BSD style license - see LICENSE.rst
import copy
import logging
import numpy as np
from astropy.coordinates import SkyCoord
from astropy.time import Time
from astropy.units import Quantity
from gammapy.utils.fits import LazyFitsData, earth_location_from_dict
from gammapy.utils.testing import Checker
from .event_list import EventList, EventListChecker
from .filters import ObservationFilter
from .gti import GTI
from .pointing import FixedPointingInfo

__all__ = ["Observation", "Observations"]

log = logging.getLogger(__name__)

[docs]class Observation: """In-memory observation. Parameters ---------- obs_id : int Observation id obs_info : dict Observation info dict aeff : `~gammapy.irf.EffectiveAreaTable2D` Effective area edisp : `~gammapy.irf.EnergyDispersion2D` Energy dispersion psf : `~gammapy.irf.PSF3D` Point spread function bkg : `~gammapy.irf.Background3D` Background rate model rad_max: `~gammapy.irf.RadMax2D` or `~astropy.units.Quantity` Only for point-like IRFs: RAD_MAX table (energy dependent RAD_MAX) or a single angle (global RAD_MAX) gti : `` Table with GTI start and stop time events : `` Event list obs_filter : `ObservationFilter` Observation filter. """ aeff = LazyFitsData(cache=False) edisp = LazyFitsData(cache=False) psf = LazyFitsData(cache=False) bkg = LazyFitsData(cache=False) rad_max = LazyFitsData(cache=False) _events = LazyFitsData(cache=False) _gti = LazyFitsData(cache=False) def __init__( self, obs_id=None, obs_info=None, gti=None, aeff=None, edisp=None, psf=None, bkg=None, rad_max=None, events=None, obs_filter=None, ): self.obs_id = obs_id self.obs_info = obs_info self.aeff = aeff self.edisp = edisp self.psf = psf self.bkg = bkg self.rad_max = rad_max self._gti = gti self._events = events self.obs_filter = obs_filter or ObservationFilter() @property def available_irfs(self): """Which irfs are available""" available_irf = [] for irf in ["aeff", "edisp", "psf", "bkg"]: available = self.__dict__.get(irf, False) available_hdu = self.__dict__.get(f"_{irf}_hdu", False) if available or available_hdu: available_irf.append(irf) return available_irf @property def events(self): events = self.obs_filter.filter_events(self._events) return events @property def gti(self): gti = self.obs_filter.filter_gti(self._gti) return gti @staticmethod def _get_obs_info(pointing, deadtime_fraction): """Create obs info dict from in memory data""" return { "RA_PNT": pointing.icrs.ra.deg, "DEC_PNT": pointing.icrs.dec.deg, "DEADC": 1 - deadtime_fraction, }
[docs] @classmethod def create( cls, pointing, obs_id=0, livetime=None, tstart=None, tstop=None, irfs=None, deadtime_fraction=0.0, reference_time="2000-01-01", ): """Create an observation. User must either provide the livetime, or the start and stop times. Parameters ---------- pointing : `~astropy.coordinates.SkyCoord` Pointing position obs_id : int Observation ID as identifier livetime : ~astropy.units.Quantity` Livetime exposure of the simulated observation tstart : `~astropy.units.Quantity` Start time of observation w.r.t reference_time tstop : `~astropy.units.Quantity` w.r.t reference_time Stop time of observation irfs : dict IRFs used for simulating the observation: `bkg`, `aeff`, `psf`, `edisp` deadtime_fraction : float, optional Deadtime fraction, defaults to 0 reference_time : `~astropy.time.Time` the reference time to use in GTI definition Returns ------- obs : `` """ if tstart is None: tstart = Quantity(0.0, "hr") if tstop is None: tstop = tstart + Quantity(livetime) gti = GTI.create([tstart], [tstop], reference_time=reference_time) obs_info = cls._get_obs_info( pointing=pointing, deadtime_fraction=deadtime_fraction ) return cls( obs_id=obs_id, obs_info=obs_info, gti=gti, aeff=irfs.get("aeff"), bkg=irfs.get("bkg"), edisp=irfs.get("edisp"), psf=irfs.get("psf"), )
@property def tstart(self): """Observation start time (`~astropy.time.Time`).""" return self.gti.time_start[0] @property def tstop(self): """Observation stop time (`~astropy.time.Time`).""" return self.gti.time_stop[0] @property def observation_time_duration(self): """Observation time duration in seconds (`~astropy.units.Quantity`). The wall time, including dead-time. """ return self.gti.time_sum @property def observation_live_time_duration(self): """Live-time duration in seconds (`~astropy.units.Quantity`). The dead-time-corrected observation time. Computed as ``t_live = t_observation * (1 - f_dead)`` where ``f_dead`` is the dead-time fraction. """ return self.observation_time_duration * ( 1 - self.observation_dead_time_fraction ) @property def observation_dead_time_fraction(self): """Dead-time fraction (float). Defined as dead-time over observation time. Dead-time is defined as the time during the observation where the detector didn't record events: The dead-time fraction is used in the live-time computation, which in turn is used in the exposure and flux computation. """ return 1 - self.obs_info["DEADC"] @property def pointing_radec(self): """Pointing RA / DEC sky coordinates (`~astropy.coordinates.SkyCoord`).""" lon, lat = ( self.obs_info.get("RA_PNT", np.nan), self.obs_info.get("DEC_PNT", np.nan), ) return SkyCoord(lon, lat, unit="deg", frame="icrs") @property def pointing_altaz(self): """Pointing ALT / AZ sky coordinates (`~astropy.coordinates.SkyCoord`).""" alt, az = ( self.obs_info.get("ALT_PNT", np.nan), self.obs_info.get("AZ_PNT", np.nan), ) return SkyCoord(az, alt, unit="deg", frame="altaz") @property def pointing_zen(self): """Pointing zenith angle sky (`~astropy.units.Quantity`).""" return Quantity(self.obs_info.get("ZEN_PNT", np.nan), unit="deg") @property def fixed_pointing_info(self): """Fixed pointing info for this observation (`FixedPointingInfo`).""" return FixedPointingInfo( @property def target_radec(self): """Target RA / DEC sky coordinates (`~astropy.coordinates.SkyCoord`).""" lon, lat = ( self.obs_info.get("RA_OBJ", np.nan), self.obs_info.get("DEC_OBJ", np.nan), ) return SkyCoord(lon, lat, unit="deg", frame="icrs") @property def observatory_earth_location(self): """Observatory location (`~astropy.coordinates.EarthLocation`).""" return earth_location_from_dict(self.obs_info) @property def muoneff(self): """Observation muon efficiency.""" return self.obs_info.get("MUONEFF", 1) def __str__(self): ra = self.pointing_radec.ra.deg dec = self.pointing_radec.dec.deg pointing = f"{ra:.1f} deg, {dec:.1f} deg\n" # TODO: Which target was observed? # TODO: print info about available HDUs for this observation ... return ( f"{self.__class__.__name__}\n\n" f"\tobs id : {self.obs_id} \n " f"\ttstart : {self.tstart.mjd:.2f}\n" f"\ttstop : {self.tstop.mjd:.2f}\n" f"\tduration : {self.observation_time_duration:.2f}\n" f"\tpointing (icrs) : {pointing}\n" f"\tdeadtime fraction : {self.observation_dead_time_fraction:.1%}\n" )
[docs] def check(self, checks="all"): """Run checks. This is a generator that yields a list of dicts. """ checker = ObservationChecker(self) return
[docs] def peek(self, figsize=(12, 10)): """Quick-look plots in a few panels. Parameters ---------- figsize : tuple Figure size """ import matplotlib.pyplot as plt n_irfs = len(self.available_irfs) fig, axes = plt.subplots( nrows=n_irfs // 2, ncols=2 + n_irfs % 2, figsize=figsize, gridspec_kw={"wspace": 0.25, "hspace": 0.25}, ) axes_dict = dict(zip(self.available_irfs, axes.flatten())) if "aeff" in self.available_irfs: self.aeff.plot(ax=axes_dict["aeff"]) axes_dict["aeff"].set_title("Effective area") if "bkg" in self.available_irfs: bkg = self.bkg if not bkg.is_offset_dependent: bkg = bkg.to_2d() bkg.plot(ax=axes_dict["bkg"]) axes_dict["bkg"].set_title("Background rate") else: logging.warning(f"No background model found for obs {self.obs_id}.") if "psf" in self.available_irfs: self.psf.plot_containment_radius_vs_energy(ax=axes_dict["psf"]) axes_dict["psf"].set_title("Point spread function") else: logging.warning(f"No PSF found for obs {self.obs_id}.") if "edisp" in self.available_irfs: self.edisp.plot_bias(ax=axes_dict["edisp"], add_cbar=True) axes_dict["edisp"].set_title("Energy dispersion") else: logging.warning(f"No energy dispersion found for obs {self.obs_id}.")
[docs] def select_time(self, time_interval): """Select a time interval of the observation. Parameters ---------- time_interval : `astropy.time.Time` Start and stop time of the selected time interval. For now we only support a single time interval. Returns ------- new_obs : `` A new observation instance of the specified time interval """ new_obs_filter = self.obs_filter.copy() new_obs_filter.time_filter = time_interval obs = copy.deepcopy(self) obs.obs_filter = new_obs_filter return obs
[docs] @classmethod def read(cls, event_file, irf_file=None): """Create an Observation from a Event List and an (optional) IRF file. Parameters ---------- event_file : str, Path path to the .fits file containing the event list and the GTI irf_file : str, Path (optional) path to the .fits file containing the IRF components, if not provided the IRF will be read from the event file Returns ------- observation : `` observation with the events and the irf read from the file """ from import load_irf_dict_from_file events = gti = irf_file = irf_file if irf_file is not None else event_file irf_dict = load_irf_dict_from_file(irf_file) obs_info = events.table.meta return cls( events=events, gti=gti, obs_info=obs_info, obs_id=obs_info.get("OBS_ID"), **irf_dict, )
[docs]class Observations( """Container class that holds a list of observations. Parameters ---------- observations : list A list of `` """ def __init__(self, observations=None): self._observations = observations or [] def __getitem__(self, key): return self._observations[self.index(key)] def __delitem__(self, key): del self._observations[self.index(key)] def __setitem__(self, key, obs): if isinstance(obs, Observation): self._observations[self.index(key)] = obs else: raise TypeError(f"Invalid type: {type(obs)!r}")
[docs] def insert(self, idx, obs): if isinstance(obs, Observation): self._observations.insert(idx, obs) else: raise TypeError(f"Invalid type: {type(obs)!r}")
def __len__(self): return len(self._observations) def __str__(self): s = self.__class__.__name__ + "\n" s += "Number of observations: {}\n".format(len(self)) for obs in self: s += str(obs) return s
[docs] def index(self, key): if isinstance(key, (int, slice)): return key elif isinstance(key, str): return self.ids.index(key) elif isinstance(key, Observation): return self._observations.index(key) else: raise TypeError(f"Invalid type: {type(key)!r}")
@property def ids(self): """List of obs IDs (`list`)""" return [str(obs.obs_id) for obs in self]
[docs] def select_time(self, time_intervals): """Select a time interval of the observations. Parameters ---------- time_intervals : `astropy.time.Time` or list of `astropy.time.Time` list of Start and stop time of the time intervals or one Time interval Returns ------- new_observations : `` A new Observations instance of the specified time intervals """ new_obs_list = [] if isinstance(time_intervals, Time): time_intervals = [time_intervals] for time_interval in time_intervals: for obs in self: if (obs.tstart < time_interval[1]) & (obs.tstop > time_interval[0]): new_obs = obs.select_time(time_interval) new_obs_list.append(new_obs) return self.__class__(new_obs_list)
def _ipython_key_completions_(self): return self.ids
class ObservationChecker(Checker): """Check an observation. Checks data format and a bit about the content. """ CHECKS = { "events": "check_events", "gti": "check_gti", "aeff": "check_aeff", "edisp": "check_edisp", "psf": "check_psf", } def __init__(self, observation): self.observation = observation def _record(self, level="info", msg=None): return {"level": level, "obs_id": self.observation.obs_id, "msg": msg} def check_events(self): yield self._record(level="debug", msg="Starting events check") try: events = except Exception: yield self._record(level="warning", msg="Loading events failed") return yield from EventListChecker(events).run() # TODO: split this out into a GTIChecker def check_gti(self): yield self._record(level="debug", msg="Starting gti check") try: gti = self.observation.gti except Exception: yield self._record(level="warning", msg="Loading GTI failed") return if len(gti.table) == 0: yield self._record(level="error", msg="GTI table has zero rows") columns_required = ["START", "STOP"] for name in columns_required: if name not in gti.table.colnames: yield self._record(level="error", msg=f"Missing table column: {name!r}") # TODO: Check that header keywords agree with table entries # TSTART, TSTOP, MJDREFI, MJDREFF # Check that START and STOP times are consecutive # times = np.ravel(self.table['START'], self.table['STOP']) # # TODO: not sure this is correct ... add test with a multi-gti table from Fermi. # if not np.all(np.diff(times) >= 0): # yield 'GTIs are not consecutive or sorted.' # TODO: add reference times for all instruments and check for this # Use TELESCOP header key to check which instrument it is. def _check_times(self): """Check if various times are consistent. The headers and tables of the FITS EVENTS and GTI extension contain various observation and event time information. """ # # telescope_met_refs = { "FERMI": Time("2001-01-01T00:00:00"), "HESS": Time("2001-01-01T00:00:00"), } meta = self.dset.event_list.table.meta telescope = meta["TELESCOP"] if telescope in telescope_met_refs.keys(): dt = self.time_ref - telescope_met_refs[telescope] if dt > self.accuracy["time"]: yield self._record( level="error", msg="Reference time incorrect for telescope" ) def check_aeff(self): yield self._record(level="debug", msg="Starting aeff check") try: aeff = self.observation.aeff except Exception: yield self._record(level="warning", msg="Loading aeff failed") return # Check that thresholds are meaningful for aeff if ( "LO_THRES" in aeff.meta and "HI_THRES" in aeff.meta and aeff.meta["LO_THRES"] >= aeff.meta["HI_THRES"] ): yield self._record( level="error", msg="LO_THRES >= HI_THRES in effective area meta data" ) # Check that data isn't all null if np.max( <= 0: yield self._record( level="error", msg="maximum entry of effective area is <= 0" ) def check_edisp(self): yield self._record(level="debug", msg="Starting edisp check") try: edisp = self.observation.edisp except Exception: yield self._record(level="warning", msg="Loading edisp failed") return # Check that data isn't all null if np.max( <= 0: yield self._record(level="error", msg="maximum entry of edisp is <= 0") def check_psf(self): yield self._record(level="debug", msg="Starting psf check") try: self.observation.psf except Exception: yield self._record(level="warning", msg="Loading psf failed") return