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 15:22 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",
]


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

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