asgardpy.gammapy.read_models.create_source_skymodel() - Code Metrics - Inspection of "b1da1b7eb6121d79c41d0c61a8b72d31ae7e208c" - chaimain/asgardpy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — main (b1da1b)

by Chaitanya

created 2024-01-19 06:05 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,
    SkyModel,
    create_fermi_isotropic_diffuse_model,
)

from asgardpy.data.target import (
    add_ebl_model_from_config,
    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):
    """
    Function to get the model of the target source from AsgardpyConfig.
    """
    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 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_)

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

                if is_source_target and ebl_absorption_included:
                    spectral_model = add_ebl_model_from_config(spectral_model, model_config)

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

chaimain / asgardpy

Branch — main (b1da1b)

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