asgardpy.gammapy.read_models.create_source_skymodel() - Code Metrics - Inspection of "Adapt some more code quality suggestions" - chaimain/asgardpy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#157)

by Chaitanya

created 2024-01-18 18:15 UTC

asgardpy.gammapy.read_models.create_source_skymodel() C

↳ Parent: Project

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:
        match source_info["@name"]:
            # Nomenclature as per enrico and fermipy
            case "IsoDiffModel" | "isodiff":
                diffuse_models["iso_diffuse"] = create_iso_diffuse_skymodel(
                    diffuse_models["iso_diffuse"], diffuse_models["key_name"]
                )
                source_info = diffuse_models["iso_diffuse"]

            case "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"]

            case _:
                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:
        match source["@name"]:
            case "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

            case "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 get_target_model_from_config(source_name, asgardpy_target_config):
    """ """
    spectral_model = None
    is_source_target = 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

    return source_name, spectral_model, is_source_target


def add_ebl_model_from_config(spectral_model, asgardpy_target_config=None, is_source_target=False):
    """ """
    # Only of Asgardpy config is provided
    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

    return spectral_model


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

        source_name, spectral_model, is_source_target = get_target_model_from_config(
            source_name, asgardpy_target_config
        )

        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_)

            spectral_model = add_ebl_model_from_config(spectral_model, asgardpy_target_config, is_source_target)

        # 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			match source_info["@name"]:
65			# Nomenclature as per enrico and fermipy
66			case "IsoDiffModel" \| "isodiff":
67			diffuse_models["iso_diffuse"] = create_iso_diffuse_skymodel(
68			diffuse_models["iso_diffuse"], diffuse_models["key_name"]
69			)
70			source_info = diffuse_models["iso_diffuse"]
71
72			case "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
78			case _:
79			source_info, is_target_source = create_source_skymodel(
80			source_info,
81			dl3_aux_path,
82			base_model_type="Fermi-XML",
83			asgardpy_target_config=asgardpy_target_config,
84			)
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			match source["@name"]:
126			case "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
132			case "GalDiffModel" \| "galdiff":
133			if fetch_gal_diff:
134			file_path = source["spatialModel"]["@file"]
135			file_name = file_path.split("/")[-1]
136			diffuse_models_file_names_dict["gal_diffuse"] = file_name
137
138			return diffuse_models_file_names_dict
139
140
141			def get_target_model_from_config(source_name, asgardpy_target_config):
142			""" """
143			spectral_model = None
144			is_source_target = False
145
146			# If Target source model's spectral component is to be taken from Config
147			# and not from 3D dataset.
148			if asgardpy_target_config:
149			source_name_check = source_name.replace("_", "").replace(" ", "")
150			target_check = asgardpy_target_config.source_name.replace("_", "").replace(" ", "")
151
152			if source_name_check == target_check:
153			source_name = asgardpy_target_config.source_name
154			is_source_target = True
155
156			# Only taking the spectral model information right now.
157			if not asgardpy_target_config.from_3d:
158			models_ = read_models_from_asgardpy_config(asgardpy_target_config)
159			spectral_model = models_[0].spectral_model
160
161			return source_name, spectral_model, is_source_target
162
163
164			def add_ebl_model_from_config(spectral_model, asgardpy_target_config=None, is_source_target=False):
165			""" """
166			# Only of Asgardpy config is provided
167			if asgardpy_target_config:
168			config_spectral = asgardpy_target_config.components[0].spectral
169			ebl_absorption_included = config_spectral.ebl_abs.reference != ""
170
171			if is_source_target and ebl_absorption_included:
172			ebl_model = config_spectral.ebl_abs
173
174			if ebl_model.filename.is_file():
175			ebl_spectral_model = EBLAbsorptionNormSpectralModel.read(
176			str(ebl_model.filename), redshift=ebl_model.redshift
177			)
178			ebl_model.reference = ebl_model.filename.name[:-8].replace("-", "_")
179			else:
180			ebl_spectral_model = EBLAbsorptionNormSpectralModel.read_builtin(
181			ebl_model.reference, redshift=ebl_model.redshift
182			)
183			spectral_model = spectral_model * ebl_spectral_model
184
185			return spectral_model
186
187
188			def create_source_skymodel(source_info, dl3_aux_path, base_model_type="Fermi-XML", asgardpy_target_config=None):
189			"""
190			Build SkyModels from given base model information.
191
192			If AsgardpyConfig section of the target is provided for the target
193			source information, it will be used to check if the target `source_name`
194			is provided in the base_model file. If it exists, then check if the model
195			information is to be read from AsgardpyConfig using `from_3d` boolean value.
196			Also, if EBL model information is provided in the AsgardpyConfig, it will
197			be added to the SkyModel object.
198
199			Parameters
200			----------
201			source_info: dict
202			Dictionary containing the source models information from XML file.
203			dl3_aux_path: str
204			Path location of the DL3 auxiliary files for reading Spatial Models
205			from separate files.
206			base_model_type: str
207			Name indicating the model format used to read the skymodels from.
208			asgardpy_target_config: `AsgardpyConfig`
209			Config section containing the Target source information.
210
211			Returns
212			-------
213			source_sky_model: `gammapy.modeling.SkyModel`
214			SkyModels object for the given source information.
215			is_source_target: bool
216			Boolean to check if the Models belong to the target source.
217			"""
218			if base_model_type == "Fermi-XML":
219			source_name = source_info["@name"]
220			spectrum_type = source_info["spectrum"]["@type"]
221			spectrum_params = source_info["spectrum"]["parameter"]
222
223			# initialized to check for the case if target spectral model information
224			# is to be taken from the Config
225			spectral_model = None
226
227			# Check if target_source file exists
228			is_source_target = False
229			ebl_atten = False
230
231			source_name, spectral_model, is_source_target = get_target_model_from_config(
232			source_name, asgardpy_target_config
233			)
234
235			if spectral_model is None:
236			# Define the Spectral Model type for Gammapy
237			spectral_model, ebl_atten = get_gammapy_spectral_model(
238			spectrum_type,
239			ebl_atten,
240			base_model_type,
241			)
242			spectrum_type = spectrum_type.split("EblAtten::")[-1]
243
244			# Read the parameter values from XML file to create SpectralModel
245			params_list = xml_spectral_model_to_gammapy(
246			spectrum_params,
247			spectrum_type,
248			is_target=is_source_target,
249			keep_sign=ebl_atten,
250			base_model_type=base_model_type,
251			)
252
253			for param_ in params_list:
254			setattr(spectral_model, param_.name, param_)
255
256			spectral_model = add_ebl_model_from_config(spectral_model, asgardpy_target_config, is_source_target)
257
258			# Reading Spatial model from the XML file
259			spatial_model = xml_spatial_model_to_gammapy(dl3_aux_path, source_info["spatialModel"], base_model_type)
260
261			spatial_model.freeze()
262			source_sky_model = SkyModel(
263			spectral_model=spectral_model,
264			spatial_model=spatial_model,
265			name=source_name,
266			)
267
268			return source_sky_model, is_source_target
269
270
271			def create_iso_diffuse_skymodel(iso_file, key_name):
272			"""
273			Create a SkyModel of the Fermi Isotropic Diffuse Model and assigning
274			name as per the observation key. If there are no distinct key types of
275			files, the value is None.
276
277			Parameters
278			----------
279			iso_file: str
280			Path to the isotropic diffuse model file
281			key: str
282			Instrument key-name, if exists
283
284			Returns
285			-------
286			diff_iso: `gammapy.modeling.SkyModel`
287			SkyModel object of the isotropic diffuse model.
288			"""
289			diff_iso = create_fermi_isotropic_diffuse_model(filename=iso_file, interp_kwargs={"fill_value": None})
290
291			if key_name:
292			diff_iso._name = f"{diff_iso.name}-{key_name}"
293
294			# Parameters' limits
295			diff_iso.spectral_model.model1.parameters[0].min = 0.001
296			diff_iso.spectral_model.model1.parameters[0].max = 10
297			diff_iso.spectral_model.model2.parameters[0].min = 0
298			diff_iso.spectral_model.model2.parameters[0].max = 10
299
300			return diff_iso
301
302
303			def create_gal_diffuse_skymodel(diff_gal_file):
304			"""
305			Create SkyModel of the Diffuse Galactic sources either by reading a file or
306			by using a provided Map object.
307
308			Parameters
309			----------
310			diff_gal_file: Path, `gammapy.maps.Map`
311			Path to the isotropic diffuse model file, or a Map object already read
312			from a file.
313
314			Returns
315			-------
316			model: `gammapy.modeling.SkyModel`
317			SkyModel object read from a given galactic diffuse model file.
318			diff_gal: `gammapy.maps.Map`
319			Map object of the galactic diffuse model
320			"""
321			if not isinstance(diff_gal_file, Map):
322			# assuming it is Path or str type
323			diff_gal_filename = diff_gal_file
324			diff_gal = Map.read(diff_gal_file)
325			diff_gal.meta["filename"] = diff_gal_filename
326			else:
327			diff_gal = diff_gal_file
328
329			template_diffuse = SPATIAL_MODEL_REGISTRY.get_cls("TemplateSpatialModel")(
330			diff_gal, normalize=False, filename=diff_gal.meta["filename"]
331			)
332			model = SkyModel(
333			spectral_model=SPECTRAL_MODEL_REGISTRY.get_cls("PowerLawNormSpectralModel")(),
334			spatial_model=template_diffuse,
335			name="diffuse-iem",
336			)
337			model.parameters["norm"].min = 0
338			model.parameters["norm"].max = 10
339			model.parameters["norm"].frozen = False
340
341			return model, diff_gal
342

chaimain / asgardpy

Pull Request — main (#157)

asgardpy.gammapy.read_models.create_source_skymodel() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like