Source code for fermipy.diffuse.source_factory

# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
Classes and utilities that create fermipy source objects
"""
from __future__ import absolute_import, division, print_function

from collections import OrderedDict

from astropy.table import Table
from astropy.coordinates import SkyCoord
from fermipy import roi_model
from fermipy import catalog


[docs]def make_point_source(name, src_dict): """Construct and return a `fermipy.roi_model.Source` object """ return roi_model.Source(name, src_dict)
[docs]def make_spatialmap_source(name, Spatial_Filename, spectrum): """Construct and return a `fermipy.roi_model.Source` object """ data = dict(Spatial_Filename=Spatial_Filename, ra=0.0, dec=0.0, SpatialType='SpatialMap', Source_Name=name) if spectrum is not None: data.update(spectrum) return roi_model.Source(name, data)
[docs]def make_mapcube_source(name, Spatial_Filename, spectrum): """Construct and return a `fermipy.roi_model.MapCubeSource` object """ data = dict(Spatial_Filename=Spatial_Filename) if spectrum is not None: data.update(spectrum) return roi_model.MapCubeSource(name, data)
[docs]def make_isotropic_source(name, Spectrum_Filename, spectrum): """Construct and return a `fermipy.roi_model.IsoSource` object """ data = dict(Spectrum_Filename=Spectrum_Filename) if spectrum is not None: data.update(spectrum) return roi_model.IsoSource(name, data)
[docs]def make_composite_source(name, spectrum): """Construct and return a `fermipy.roi_model.CompositeSource` object """ data = dict(SpatialType='CompositeSource', SpatialModel='CompositeSource', SourceType='CompositeSource') if spectrum is not None: data.update(spectrum) return roi_model.CompositeSource(name, data)
[docs]def make_catalog_sources(catalog_roi_model, source_names): """Construct and return dictionary of sources that are a subset of sources in catalog_roi_model. Parameters ---------- catalog_roi_model : dict or `fermipy.roi_model.ROIModel` Input set of sources source_names : list Names of sourcs to extract Returns dict mapping source_name to `fermipy.roi_model.Source` object """ sources = {} for source_name in source_names: sources[source_name] = catalog_roi_model[source_name] return sources
[docs]def make_sources(comp_key, comp_dict): """Make dictionary mapping component keys to a source or set of sources Parameters ---------- comp_key : str Key used to access sources comp_dict : dict Information used to build sources return `OrderedDict` maping comp_key to `fermipy.roi_model.Source` """ srcdict = OrderedDict() try: comp_info = comp_dict.info except AttributeError: comp_info = comp_dict try: spectrum = comp_dict.spectrum except AttributeError: spectrum = None model_type = comp_info.model_type if model_type == 'PointSource': srcdict[comp_key] = make_point_source(comp_info.source_name, comp_info.src_dict) elif model_type == 'SpatialMap': srcdict[comp_key] = make_spatialmap_source(comp_info.source_name, comp_info.Spatial_Filename, spectrum) elif model_type == 'MapCubeSource': srcdict[comp_key] = make_mapcube_source(comp_info.source_name, comp_info.Spatial_Filename, spectrum) elif model_type == 'IsoSource': srcdict[comp_key] = make_isotropic_source(comp_info.source_name, comp_info.Spectral_Filename, spectrum) elif model_type == 'CompositeSource': srcdict[comp_key] = make_composite_source(comp_info.source_name, spectrum) elif model_type == 'CatalogSources': srcdict.update(make_catalog_sources(comp_info.roi_model, comp_info.source_names)) else: raise ValueError("Unrecognized model_type %s" % model_type) return srcdict
[docs]class SourceFactory(object): """Small helper class to build and keep track of sources """ def __init__(self): """C'tor """ self._source_info_dict = OrderedDict() self._sources = OrderedDict() @property def sources(self): """Return the dictionary of sources""" return self._sources @property def source_info_dict(self): """Return the dictionary of source_info objects used to build sources""" return self._source_info_dict
[docs] def add_sources(self, source_info_dict): """Add all of the sources in source_info_dict to this factory """ self._source_info_dict.update(source_info_dict) for key, value in source_info_dict.items(): self._sources.update(make_sources(key, value))
[docs] @staticmethod def build_catalog(**kwargs): """Build a `fermipy.catalog.Catalog` object Parameters ---------- catalog_type : str Specifies catalog type, options include 2FHL | 3FGL | 4FGLP | 4FGL | 4FGL-DR2 | 4FGL-DR3 catalog_file : str FITS file with catalog tables catalog_extdir : str Path to directory with extended source templates """ catalog_type = kwargs.get('catalog_type') catalog_file = kwargs.get('catalog_file') catalog_extdir = kwargs.get('catalog_extdir') if catalog_type == '2FHL': return catalog.Catalog2FHL(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == '3FGL': return catalog.Catalog3FGL(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == '4FGLP': return catalog.Catalog4FGLP(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == '4FGL': return catalog.Catalog4FGL(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == '4FGL-DR2': return catalog.Catalog4FGLDR2(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == '4FGL-DR3': return catalog.Catalog4FGLDR3(fitsfile=catalog_file, extdir=catalog_extdir) elif catalog_type == 'FL8Y': return catalog.CatalogFL8Y(fitsfile=catalog_file, extdir=catalog_extdir) else: table = Table.read(catalog_file) return catalog.Catalog(table, extdir=catalog_extdir)
[docs] @staticmethod def make_fermipy_roi_model_from_catalogs(cataloglist): """Build and return a `fermipy.roi_model.ROIModel object from a list of fermipy.catalog.Catalog` objects """ data = dict(catalogs=cataloglist, src_roiwidth=360.) return roi_model.ROIModel(data, skydir=SkyCoord(0.0, 0.0, unit='deg'))
[docs] @classmethod def make_roi(cls, sources=None): """Build and return a `fermipy.roi_model.ROIModel` object from a dict with information about the sources """ if sources is None: sources = {} src_fact = cls() src_fact.add_sources(sources) ret_model = roi_model.ROIModel( {}, skydir=SkyCoord(0.0, 0.0, unit='deg')) for source in src_fact.sources.values(): ret_model.load_source(source, build_index=False, merge_sources=False) return ret_model
[docs] @classmethod def copy_selected_sources(cls, roi, source_names): """Build and return a `fermipy.roi_model.ROIModel` object by copying selected sources from another such object """ roi_new = cls.make_roi() for source_name in source_names: try: src_cp = roi.copy_source(source_name) except Exception: continue roi_new.load_source(src_cp, build_index=False) return roi_new