asgardpy.gammapy.read_models.create_source_skymodel() - Code Metrics - Inspection of "Separate XML model reading functionality into dist..." - chaimain/asgardpy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#155)

by Chaitanya

created 2024-01-08 14:53 UTC

create_source_skymodel() C

↳ Parent: asgardpy.gammapy.read_models

Complexity

Conditions

Size

Total Lines	108
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	50
nop	4
dl	0
loc	108
rs	5.4
c	0
b	0
f	0

How to fix Long Method Complexity

"""
Functions for reading different Models type objects with Gammapy
objects.
"""
import xmltodict
from gammapy.maps import Map
from gammapy.modeling.models import (
    SPATIAL_MODEL_REGISTRY,
    SPECTRAL_MODEL_REGISTRY,
    EBLAbsorptionNormSpectralModel,
    SkyModel,
    create_fermi_isotropic_diffuse_model,
)

from asgardpy.data.target import read_models_from_asgardpy_config
from asgardpy.gammapy.interoperate_models import (
    get_gammapy_spectral_model,
    xml_spatial_model_to_gammapy,
    xml_spectral_model_to_gammapy,
)

__all__ = [
    "create_gal_diffuse_skymodel",
    "create_iso_diffuse_skymodel",
    "create_source_skymodel",
    "read_fermi_xml_models_list",
    "update_aux_info_from_fermi_xml",
]


## Try to make these functions as independent of AConfig as possible
## Make these self-reliant
def read_fermi_xml_models_list(
    list_source_models, dl3_aux_path, xml_file, diffuse_models, asgardpy_target_config=None
):
    """
    Read from the Fermi-XML file to enlist the various objects and get their
    SkyModels.

    Parameters
    ----------
    list_source_models: list
        List of source models to be filled.
    dl3_aux_path: str
        Path location of the DL3 auxiliary files for reading Spatial Models
        from separate files.
    xml_file: str
        Path of the Fermi-XML file to be read.
    diffuse_models: dict
        Dict containing diffuse models' filenames and instrument key.
    asgardpy_target_config: `AsgardpyConfig`
        Config section containing the Target source information.

    Returns
    -------
    list_source_models: list
        List of filled source models.
    diffuse_models: dict
        Dict containing diffuse models objects replacing the filenames.
    """
    with open(xml_file, encoding="utf-8") as file:
        xml_data = xmltodict.parse(file.read())["source_library"]["source"]

    is_target_source = False
    for source_info in xml_data:
        source_name = source_info["@name"]

        # Nomenclature as per enrico and fermipy
        if source_name in ["IsoDiffModel", "isodiff"]:
            diffuse_models["iso_diffuse"] = create_iso_diffuse_skymodel(
                diffuse_models["iso_diffuse"], diffuse_models["key_name"]
            )
            source_info = diffuse_models["iso_diffuse"]
        elif source_name in ["GalDiffModel", "galdiff"]:
            diffuse_models["gal_diffuse"], diffuse_models["gal_diffuse_map"] = create_gal_diffuse_skymodel(
                diffuse_models["gal_diffuse"]
            )
            source_info = diffuse_models["gal_diffuse"]
        else:
            source_info, is_target_source = create_source_skymodel(
                source_info,
                dl3_aux_path,
                base_model_type="Fermi-XML",
                asgardpy_target_config=asgardpy_target_config,
            )
        if is_target_source:
            list_source_models.insert(0, source_info)
            is_target_source = False  # To not let it repeat
        else:
            list_source_models.append(source_info)

    return list_source_models, diffuse_models


def update_aux_info_from_fermi_xml(
    diffuse_models_file_names_dict, xml_file, fetch_iso_diff=False, fetch_gal_diff=False
):
    """
    When no glob_search patterns on axuillary files are provided, fetch
    them from the XML file and update the dict containing diffuse models' file
    names.

    Parameters
    ----------
    diffuse_models_file_names_dict: `dict`
        Dict containing the information on the DL3 files input.
    xml_file: str
        Path of the Fermi-XML file to be read.
    fetch_iso_diff: bool
        Boolean to get the information of the Isotropic diffuse model from the
        Fermi-XML file
    fetch_gal_diff: bool
        Boolean to get the information of the Galactic diffuse model from the
        Fermi-XML file

    Returns
    -------
    diffuse_models_file_names_dict: `dict`
        Dict containing the updated information on the DL3 files input.
    """
    with open(xml_file) as file:
        data = xmltodict.parse(file.read())["source_library"]["source"]

    for source in data:
        source_name = source["@name"]
        if source_name in ["IsoDiffModel", "isodiff"]:
            if fetch_iso_diff:
                file_path = source["spectrum"]["@file"]
                file_name = file_path.split("/")[-1]
                diffuse_models_file_names_dict["iso_diffuse"] = file_name
        if source_name in ["GalDiffModel", "galdiff"]:
            if fetch_gal_diff:
                file_path = source["spatialModel"]["@file"]
                file_name = file_path.split("/")[-1]
                diffuse_models_file_names_dict["gal_diffuse"] = file_name
    return diffuse_models_file_names_dict


def create_source_skymodel(source_info, dl3_aux_path, base_model_type="Fermi-XML", asgardpy_target_config=None):
    """
    Build SkyModels from given base model information.

    If AsgardpyConfig section of the target is provided for the target
    source information, it will be used to check if the target `source_name`
    is provided in the base_model file. If it exists, then check if the model
    information is to be read from AsgardpyConfig using `from_3d` boolean value.
    Also, if EBL model information is provided in the AsgardpyConfig, it will
    be added to the SkyModel object.

    Parameters
    ----------
    source_info: dict
        Dictionary containing the source models information from XML file.
    dl3_aux_path: str
        Path location of the DL3 auxiliary files for reading Spatial Models
        from separate files.
    base_model_type: str
        Name indicating the model format used to read the skymodels from.
    asgardpy_target_config: `AsgardpyConfig`
        Config section containing the Target source information.

    Returns
    -------
    source_sky_model: `gammapy.modeling.SkyModel`
        SkyModels object for the given source information.
    is_source_target: bool
        Boolean to check if the Models belong to the target source.
    """
    if base_model_type == "Fermi-XML":
        source_name = source_info["@name"]
        spectrum_type = source_info["spectrum"]["@type"]
        spectrum_params = source_info["spectrum"]["parameter"]

        # initialized to check for the case if target spectral model information
        # is to be taken from the Config
        spectral_model = None

        # Check if target_source file exists
        is_source_target = False
        ebl_atten = False

        # If Target source model's spectral component is to be taken from Config
        # and not from 3D dataset.
        if asgardpy_target_config:
            source_name_check = source_name.replace("_", "").replace(" ", "")
            target_check = asgardpy_target_config.source_name.replace("_", "").replace(" ", "")

            if source_name_check == target_check:
                source_name = asgardpy_target_config.source_name
                is_source_target = True

                # Only taking the spectral model information right now.
                if not asgardpy_target_config.from_3d:
                    models_ = read_models_from_asgardpy_config(asgardpy_target_config)
                    spectral_model = models_[0].spectral_model

        if spectral_model is None:
            # Define the Spectral Model type for Gammapy
            spectral_model, ebl_atten = get_gammapy_spectral_model(
                spectrum_type,
                ebl_atten,
                base_model_type,
            )
            spectrum_type = spectrum_type.split("EblAtten::")[-1]

            # Read the parameter values from XML file to create SpectralModel
            params_list = xml_spectral_model_to_gammapy(
                spectrum_params,
                spectrum_type,
                is_target=is_source_target,
                keep_sign=ebl_atten,
                base_model_type=base_model_type,
            )

            for param_ in params_list:
                setattr(spectral_model, param_.name, param_)

            if asgardpy_target_config:
                config_spectral = asgardpy_target_config.components[0].spectral
                ebl_absorption_included = config_spectral.ebl_abs.reference != ""

                if is_source_target and ebl_absorption_included:
                    ebl_model = config_spectral.ebl_abs

                    if ebl_model.filename.is_file():
                        ebl_spectral_model = EBLAbsorptionNormSpectralModel.read(
                            str(ebl_model.filename), redshift=ebl_model.redshift
                        )
                        ebl_model.reference = ebl_model.filename.name[:-8].replace("-", "_")
                    else:
                        ebl_spectral_model = EBLAbsorptionNormSpectralModel.read_builtin(
                            ebl_model.reference, redshift=ebl_model.redshift
                        )
                    spectral_model = spectral_model * ebl_spectral_model

        # Reading Spatial model from the XML file
        spatial_model = xml_spatial_model_to_gammapy(dl3_aux_path, source_info["spatialModel"], base_model_type)

        spatial_model.freeze()
        source_sky_model = SkyModel(
            spectral_model=spectral_model,
            spatial_model=spatial_model,
            name=source_name,
        )

    return source_sky_model, is_source_target


def create_iso_diffuse_skymodel(iso_file, key_name):
    """
    Create a SkyModel of the Fermi Isotropic Diffuse Model and assigning
    name as per the observation key. If there are no distinct key types of
    files, the value is None.

    Parameters
    ----------
    iso_file: str
        Path to the isotropic diffuse model file
    key: str
        Instrument key-name, if exists

    Returns
    -------
    diff_iso: `gammapy.modeling.SkyModel`
        SkyModel object of the isotropic diffuse model.
    """
    diff_iso = create_fermi_isotropic_diffuse_model(filename=iso_file, interp_kwargs={"fill_value": None})

    if key_name:
        diff_iso._name = f"{diff_iso.name}-{key_name}"

    # Parameters' limits
    diff_iso.spectral_model.model1.parameters[0].min = 0.001
    diff_iso.spectral_model.model1.parameters[0].max = 10
    diff_iso.spectral_model.model2.parameters[0].min = 0
    diff_iso.spectral_model.model2.parameters[0].max = 10

    return diff_iso


def create_gal_diffuse_skymodel(diff_gal_file):
    """
    Create SkyModel of the Diffuse Galactic sources either by reading a file or
    by using a provided Map object.

    Parameters
    ----------
    diff_gal_file: Path, `gammapy.maps.Map`
        Path to the isotropic diffuse model file, or a Map object already read
        from a file.

    Returns
    -------
    model: `gammapy.modeling.SkyModel`
        SkyModel object read from a given galactic diffuse model file.
    diff_gal: `gammapy.maps.Map`
        Map object of the galactic diffuse model
    """
    if not isinstance(diff_gal_file, Map):
        # assuming it is Path or str type
        diff_gal_filename = diff_gal_file
        diff_gal = Map.read(diff_gal_file)
        diff_gal.meta["filename"] = diff_gal_filename
    else:
        diff_gal = diff_gal_file

    template_diffuse = SPATIAL_MODEL_REGISTRY.get_cls("TemplateSpatialModel")(
        diff_gal, normalize=False, filename=diff_gal.meta["filename"]
    )
    model = SkyModel(
        spectral_model=SPECTRAL_MODEL_REGISTRY.get_cls("PowerLawNormSpectralModel")(),
        spatial_model=template_diffuse,
        name="diffuse-iem",
    )
    model.parameters["norm"].min = 0
    model.parameters["norm"].max = 10
    model.parameters["norm"].frozen = False

    return model, diff_gal


1			"""
2			Functions for reading different Models type objects with Gammapy
3			objects.
4			"""
5			import xmltodict
6			from gammapy.maps import Map
7			from gammapy.modeling.models import (
8			SPATIAL_MODEL_REGISTRY,
9			SPECTRAL_MODEL_REGISTRY,
10			EBLAbsorptionNormSpectralModel,
11			SkyModel,
12			create_fermi_isotropic_diffuse_model,
13			)
14
15			from asgardpy.data.target import read_models_from_asgardpy_config
16			from asgardpy.gammapy.interoperate_models import (
17			get_gammapy_spectral_model,
18			xml_spatial_model_to_gammapy,
19			xml_spectral_model_to_gammapy,
20			)
21
22			__all__ = [
23			"create_gal_diffuse_skymodel",
24			"create_iso_diffuse_skymodel",
25			"create_source_skymodel",
26			"read_fermi_xml_models_list",
27			"update_aux_info_from_fermi_xml",
28			]
29
30
31			## Try to make these functions as independent of AConfig as possible
32			## Make these self-reliant
33			def read_fermi_xml_models_list(
34			list_source_models, dl3_aux_path, xml_file, diffuse_models, asgardpy_target_config=None
35			):
36			"""
37			Read from the Fermi-XML file to enlist the various objects and get their
38			SkyModels.
39
40			Parameters
41			----------
42			list_source_models: list
43			List of source models to be filled.
44			dl3_aux_path: str
45			Path location of the DL3 auxiliary files for reading Spatial Models
46			from separate files.
47			xml_file: str
48			Path of the Fermi-XML file to be read.
49			diffuse_models: dict
50			Dict containing diffuse models' filenames and instrument key.
51			asgardpy_target_config: `AsgardpyConfig`
52			Config section containing the Target source information.
53
54			Returns
55			-------
56			list_source_models: list
57			List of filled source models.
58			diffuse_models: dict
59			Dict containing diffuse models objects replacing the filenames.
60			"""
61			with open(xml_file, encoding="utf-8") as file:
62			xml_data = xmltodict.parse(file.read())["source_library"]["source"]
63
64			is_target_source = False
65			for source_info in xml_data:
66			source_name = source_info["@name"]
67
68			# Nomenclature as per enrico and fermipy
69			if source_name in ["IsoDiffModel", "isodiff"]:
70			diffuse_models["iso_diffuse"] = create_iso_diffuse_skymodel(
71			diffuse_models["iso_diffuse"], diffuse_models["key_name"]
72			)
73			source_info = diffuse_models["iso_diffuse"]
74			elif source_name in ["GalDiffModel", "galdiff"]:
75			diffuse_models["gal_diffuse"], diffuse_models["gal_diffuse_map"] = create_gal_diffuse_skymodel(
76			diffuse_models["gal_diffuse"]
77			)
78			source_info = diffuse_models["gal_diffuse"]
79			else:
80			source_info, is_target_source = create_source_skymodel(
81			source_info,
82			dl3_aux_path,
83			base_model_type="Fermi-XML",
84			asgardpy_target_config=asgardpy_target_config,
85			)
86			if is_target_source:
87			list_source_models.insert(0, source_info)
88			is_target_source = False # To not let it repeat
89			else:
90			list_source_models.append(source_info)
91
92			return list_source_models, diffuse_models
93
94
95			def update_aux_info_from_fermi_xml(
96			diffuse_models_file_names_dict, xml_file, fetch_iso_diff=False, fetch_gal_diff=False
97			):
98			"""
99			When no glob_search patterns on axuillary files are provided, fetch
100			them from the XML file and update the dict containing diffuse models' file
101			names.
102
103			Parameters
104			----------
105			diffuse_models_file_names_dict: `dict`
106			Dict containing the information on the DL3 files input.
107			xml_file: str
108			Path of the Fermi-XML file to be read.
109			fetch_iso_diff: bool
110			Boolean to get the information of the Isotropic diffuse model from the
111			Fermi-XML file
112			fetch_gal_diff: bool
113			Boolean to get the information of the Galactic diffuse model from the
114			Fermi-XML file
115
116			Returns
117			-------
118			diffuse_models_file_names_dict: `dict`
119			Dict containing the updated information on the DL3 files input.
120			"""
121			with open(xml_file) as file:
122			data = xmltodict.parse(file.read())["source_library"]["source"]
123
124			for source in data:
125			source_name = source["@name"]
126			if source_name in ["IsoDiffModel", "isodiff"]:
127			if fetch_iso_diff:
128			file_path = source["spectrum"]["@file"]
129			file_name = file_path.split("/")[-1]
130			diffuse_models_file_names_dict["iso_diffuse"] = file_name
131			if source_name in ["GalDiffModel", "galdiff"]:
132			if fetch_gal_diff:
133			file_path = source["spatialModel"]["@file"]
134			file_name = file_path.split("/")[-1]
135			diffuse_models_file_names_dict["gal_diffuse"] = file_name
136			return diffuse_models_file_names_dict
137
138
139			def create_source_skymodel(source_info, dl3_aux_path, base_model_type="Fermi-XML", asgardpy_target_config=None):
140			"""
141			Build SkyModels from given base model information.
142
143			If AsgardpyConfig section of the target is provided for the target
144			source information, it will be used to check if the target `source_name`
145			is provided in the base_model file. If it exists, then check if the model
146			information is to be read from AsgardpyConfig using `from_3d` boolean value.
147			Also, if EBL model information is provided in the AsgardpyConfig, it will
148			be added to the SkyModel object.
149
150			Parameters
151			----------
152			source_info: dict
153			Dictionary containing the source models information from XML file.
154			dl3_aux_path: str
155			Path location of the DL3 auxiliary files for reading Spatial Models
156			from separate files.
157			base_model_type: str
158			Name indicating the model format used to read the skymodels from.
159			asgardpy_target_config: `AsgardpyConfig`
160			Config section containing the Target source information.
161
162			Returns
163			-------
164			source_sky_model: `gammapy.modeling.SkyModel`
165			SkyModels object for the given source information.
166			is_source_target: bool
167			Boolean to check if the Models belong to the target source.
168			"""
169			if base_model_type == "Fermi-XML":
170			source_name = source_info["@name"]
171			spectrum_type = source_info["spectrum"]["@type"]
172			spectrum_params = source_info["spectrum"]["parameter"]
173
174			# initialized to check for the case if target spectral model information
175			# is to be taken from the Config
176			spectral_model = None
177
178			# Check if target_source file exists
179			is_source_target = False
180			ebl_atten = False
181
182			# If Target source model's spectral component is to be taken from Config
183			# and not from 3D dataset.
184			if asgardpy_target_config:
185			source_name_check = source_name.replace("_", "").replace(" ", "")
186			target_check = asgardpy_target_config.source_name.replace("_", "").replace(" ", "")
187
188			if source_name_check == target_check:
189			source_name = asgardpy_target_config.source_name
190			is_source_target = True
191
192			# Only taking the spectral model information right now.
193			if not asgardpy_target_config.from_3d:
194			models_ = read_models_from_asgardpy_config(asgardpy_target_config)
195			spectral_model = models_[0].spectral_model
196
197			if spectral_model is None:
198			# Define the Spectral Model type for Gammapy
199			spectral_model, ebl_atten = get_gammapy_spectral_model(
200			spectrum_type,
201			ebl_atten,
202			base_model_type,
203			)
204			spectrum_type = spectrum_type.split("EblAtten::")[-1]
205
206			# Read the parameter values from XML file to create SpectralModel
207			params_list = xml_spectral_model_to_gammapy(
208			spectrum_params,
209			spectrum_type,
210			is_target=is_source_target,
211			keep_sign=ebl_atten,
212			base_model_type=base_model_type,
213			)
214
215			for param_ in params_list:
216			setattr(spectral_model, param_.name, param_)
217
218			if asgardpy_target_config:
219			config_spectral = asgardpy_target_config.components[0].spectral
220			ebl_absorption_included = config_spectral.ebl_abs.reference != ""
221
222			if is_source_target and ebl_absorption_included:
223			ebl_model = config_spectral.ebl_abs
224
225			if ebl_model.filename.is_file():
226			ebl_spectral_model = EBLAbsorptionNormSpectralModel.read(
227			str(ebl_model.filename), redshift=ebl_model.redshift
228			)
229			ebl_model.reference = ebl_model.filename.name[:-8].replace("-", "_")
230			else:
231			ebl_spectral_model = EBLAbsorptionNormSpectralModel.read_builtin(
232			ebl_model.reference, redshift=ebl_model.redshift
233			)
234			spectral_model = spectral_model * ebl_spectral_model
235
236			# Reading Spatial model from the XML file
237			spatial_model = xml_spatial_model_to_gammapy(dl3_aux_path, source_info["spatialModel"], base_model_type)
238
239			spatial_model.freeze()
240			source_sky_model = SkyModel(
241			spectral_model=spectral_model,
242			spatial_model=spatial_model,
243			name=source_name,
244			)
245
246			return source_sky_model, is_source_target
247
248
249			def create_iso_diffuse_skymodel(iso_file, key_name):
250			"""
251			Create a SkyModel of the Fermi Isotropic Diffuse Model and assigning
252			name as per the observation key. If there are no distinct key types of
253			files, the value is None.
254
255			Parameters
256			----------
257			iso_file: str
258			Path to the isotropic diffuse model file
259			key: str
260			Instrument key-name, if exists
261
262			Returns
263			-------
264			diff_iso: `gammapy.modeling.SkyModel`
265			SkyModel object of the isotropic diffuse model.
266			"""
267			diff_iso = create_fermi_isotropic_diffuse_model(filename=iso_file, interp_kwargs={"fill_value": None})
268
269			if key_name:
270			diff_iso._name = f"{diff_iso.name}-{key_name}"
271
272			# Parameters' limits
273			diff_iso.spectral_model.model1.parameters[0].min = 0.001
274			diff_iso.spectral_model.model1.parameters[0].max = 10
275			diff_iso.spectral_model.model2.parameters[0].min = 0
276			diff_iso.spectral_model.model2.parameters[0].max = 10
277
278			return diff_iso
279
280
281			def create_gal_diffuse_skymodel(diff_gal_file):
282			"""
283			Create SkyModel of the Diffuse Galactic sources either by reading a file or
284			by using a provided Map object.
285
286			Parameters
287			----------
288			diff_gal_file: Path, `gammapy.maps.Map`
289			Path to the isotropic diffuse model file, or a Map object already read
290			from a file.
291
292			Returns
293			-------
294			model: `gammapy.modeling.SkyModel`
295			SkyModel object read from a given galactic diffuse model file.
296			diff_gal: `gammapy.maps.Map`
297			Map object of the galactic diffuse model
298			"""
299			if not isinstance(diff_gal_file, Map):
300			# assuming it is Path or str type
301			diff_gal_filename = diff_gal_file
302			diff_gal = Map.read(diff_gal_file)
303			diff_gal.meta["filename"] = diff_gal_filename
304			else:
305			diff_gal = diff_gal_file
306
307			template_diffuse = SPATIAL_MODEL_REGISTRY.get_cls("TemplateSpatialModel")(
308			diff_gal, normalize=False, filename=diff_gal.meta["filename"]
309			)
310			model = SkyModel(
311			spectral_model=SPECTRAL_MODEL_REGISTRY.get_cls("PowerLawNormSpectralModel")(),
312			spatial_model=template_diffuse,
313			name="diffuse-iem",
314			)
315			model.parameters["norm"].min = 0
316			model.parameters["norm"].max = 10
317			model.parameters["norm"].frozen = False
318
319			return model, diff_gal
320

chaimain / asgardpy

Pull Request — main (#155)

create_source_skymodel() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like