asgardpy.analysis.analysis.AsgardpyAnalysis.add_to_instrument_info() - 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 22:25 UTC

AsgardpyAnalysis.add_to_instrument_info() A

↳ Parent: asgardpy.analysis.analysis

Complexity

Conditions

Size

Total Lines	13
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	2
dl	0
loc	13
rs	10
c	0
b	0
f	0

"""
Config-driven high level analysis interface.
"""
import logging

from gammapy.datasets import Datasets
from gammapy.modeling.models import Models
from pydantic import ValidationError

from asgardpy.analysis.step import AnalysisStep
from asgardpy.config.generator import AsgardpyConfig, gammapy_to_asgardpy_model_config
from asgardpy.data.target import set_models
from asgardpy.stats.stats import get_goodness_of_fit_stats

log = logging.getLogger(__name__)

__all__ = ["AsgardpyAnalysis"]


class AsgardpyAnalysis:
    """
    Config-driven high level analysis interface.

    It is initialized by default with a set of configuration parameters and
    values declared in an internal high level interface model, though the user
    can also provide configuration parameters passed as a nested dictionary at
    the moment of instantiation. In that case these parameters will overwrite
    the default values of those present in the configuration file.

    A specific example of an upgrade of the configuration parameters is when
    the Target Models information is provided as a path to a separate yaml file,
    which is readable with AsgardpyConfig. In this case, the configuration used
    in AsgardpyAnalysis is updated in the initialization step itself.

    Parameters
    ----------
    config : dict or `AsgardpyConfig`
        Configuration options following `AsgardpyConfig` schema
    """

    def __init__(self, config):
        self.log = log
        self.config = config

        if self.config.target.models_file.is_file():
            try:
                other_config = AsgardpyConfig.read(self.config.target.models_file)
                self.config = self.config.update(other_config)
            except ValidationError:
                self.config = gammapy_to_asgardpy_model_config(
                    gammapy_model=self.config.target.models_file,
                    asgardpy_config_file=self.config,
                )

        self.config.set_logging()
        self.datasets = Datasets()
        self.instrument_spectral_info = {
            "name": [],
            "spectral_energy_ranges": [],
            "en_bins": 0,
            "free_params": 0,
            "DoF": 0,
        }
        self.dataset_name_list = []

        self.final_model = Models()
        self.final_data_products = ["fit", "fit_result", "flux_points"]

        for data_product in self.final_data_products:
            setattr(self, data_product, None)

    @property
    def models(self):
        """
        Display the assigned Models.
        """
        if not self.datasets:
            raise RuntimeError("No datasets defined. Impossible to set models.")
        return self.datasets.models

    @property
    def config(self):
        """
        Analysis configuration (`AsgardpyConfig`)
        """
        return self._config

    @config.setter
    def config(self, value):
        if isinstance(value, dict):
            self._config = AsgardpyConfig(**value)
        elif isinstance(value, AsgardpyConfig):
            self._config = value
        else:
            raise TypeError("config must be dict or AsgardpyConfig.")

    def update_models_list(self, models_list):
        """
        This function updates the final_model object with a given list of
        models, only if the target source has a Spatial model included.

        This step is only valid for 3D Datasets, and might need reconsideration
        in the future.
        """
        if models_list:
            target_source_model = models_list[self.config.target.source_name]

            if target_source_model.spatial_model:
                for model_ in models_list:
                    self.final_model.append(model_)
            else:  # pragma: no cover
                self.log.info(
                    "The target source %s only has spectral model",
                    self.config.target.source_name,
                )

    def add_to_instrument_info(self, info_dict):
        """
        Update the name, Degrees of Freedom and spectral energy ranges for each
        instrument Datasets, to be used for the DL4 to DL5 processes.
        """
        for name in info_dict["name"]:
            self.instrument_spectral_info["name"].append(name)

        for edges in info_dict["spectral_energy_ranges"]:
            self.instrument_spectral_info["spectral_energy_ranges"].append(edges)

        self.instrument_spectral_info["en_bins"] += info_dict["en_bins"]
        self.instrument_spectral_info["free_params"] += info_dict["free_params"]

    def update_dof_value(self):
        """
        Simple function to update total Degrees of Freedom value in the
        instrument_spectral_info dict from a filled final_model object.
        """
        if len(self.final_model) > 0:
            # Add to the total number of free model parameters
            n_free_params = len(list(self.final_model.parameters.free_parameters))
            self.instrument_spectral_info["free_params"] += n_free_params

            # Get the final degrees of freedom as en_bins - free_params
            self.instrument_spectral_info["DoF"] = (
                self.instrument_spectral_info["en_bins"] - self.instrument_spectral_info["free_params"]
            )

    def run(self, steps=None, **kwargs):
        """
        Main function to run the AnalaysisSteps provided.

        Currently overwrite option is used from the value in AsgardpyConfig,
        and is True by default.
        """
        if steps is None:
            steps = self.config.general.steps
            self.overwrite = self.config.general.overwrite

        dl3_dl4_steps = [step for step in steps if "datasets" in step]
        dl4_dl5_steps = [step for step in steps if "datasets" not in step]

        if len(dl3_dl4_steps) > 0:
            self.log.info("Perform DL3 to DL4 process!")

            for step in dl3_dl4_steps:
                analysis_step = AnalysisStep.create(step, self.config, **kwargs)

                datasets_list, models_list, instrument_spectral_info = analysis_step.run()

                self.update_models_list(models_list)

                # To get all datasets_names from the datasets and update the final datasets list
                for data in datasets_list:
                    if data.name not in self.dataset_name_list:
                        self.dataset_name_list.append(data.name)
                    self.datasets.append(data)

                self.add_to_instrument_info(instrument_spectral_info)

            self.datasets, self.final_model = set_models(
                self.config.target,
                self.datasets,
                self.dataset_name_list,
                models=self.final_model,
            )
            self.log.info("Models have been associated with the Datasets")

            self.update_dof_value()

        if len(dl4_dl5_steps) > 0:
            self.log.info("Perform DL4 to DL5 processes!")

            for step in dl4_dl5_steps:
                analysis_step = AnalysisStep.create(step, self.config, **kwargs)

                analysis_step.run(datasets=self.datasets, instrument_spectral_info=self.instrument_spectral_info)

                # Update the final data product objects
                for data_product in self.final_data_products:
                    if hasattr(analysis_step, data_product):
                        setattr(self, data_product, getattr(analysis_step, data_product))

        if self.fit_result:
            self.instrument_spectral_info, message = get_goodness_of_fit_stats(
                self.datasets, self.instrument_spectral_info
            )
            self.log.info(message)

    # keep these methods to be backward compatible
    def get_1d_datasets(self):
        """Produce stacked 1D datasets."""
        self.run(steps=["datasets-1d"])

    def get_3d_datasets(self):
        """Produce stacked 3D datasets."""
        self.run(steps=["datasets-3d"])

    def run_fit(self):
        """Fitting reduced datasets to model."""
        self.run(steps=["fit"])

    def get_flux_points(self):
        """Calculate flux points for a specific model component."""
        self.run(steps=["flux-points"])

    def update_config(self, config):
        """Update the primary config with another config."""
        self.config = self.config.update(config=config)


1			"""
2			Config-driven high level analysis interface.
3			"""
4			import logging
5
6			from gammapy.datasets import Datasets
7			from gammapy.modeling.models import Models
8			from pydantic import ValidationError
9
10			from asgardpy.analysis.step import AnalysisStep
11			from asgardpy.config.generator import AsgardpyConfig, gammapy_to_asgardpy_model_config
12			from asgardpy.data.target import set_models
13			from asgardpy.stats.stats import get_goodness_of_fit_stats
14
15			log = logging.getLogger(__name__)
16
17			__all__ = ["AsgardpyAnalysis"]
18
19
20			class AsgardpyAnalysis:
21			"""
22			Config-driven high level analysis interface.
23
24			It is initialized by default with a set of configuration parameters and
25			values declared in an internal high level interface model, though the user
26			can also provide configuration parameters passed as a nested dictionary at
27			the moment of instantiation. In that case these parameters will overwrite
28			the default values of those present in the configuration file.
29
30			A specific example of an upgrade of the configuration parameters is when
31			the Target Models information is provided as a path to a separate yaml file,
32			which is readable with AsgardpyConfig. In this case, the configuration used
33			in AsgardpyAnalysis is updated in the initialization step itself.
34
35			Parameters
36			----------
37			config : dict or `AsgardpyConfig`
38			Configuration options following `AsgardpyConfig` schema
39			"""
40
41			def __init__(self, config):
42			self.log = log
43			self.config = config
44
45			if self.config.target.models_file.is_file():
46			try:
47			other_config = AsgardpyConfig.read(self.config.target.models_file)
48			self.config = self.config.update(other_config)
49			except ValidationError:
50			self.config = gammapy_to_asgardpy_model_config(
51			gammapy_model=self.config.target.models_file,
52			asgardpy_config_file=self.config,
53			)
54
55			self.config.set_logging()
56			self.datasets = Datasets()
57			self.instrument_spectral_info = {
58			"name": [],
59			"spectral_energy_ranges": [],
60			"en_bins": 0,
61			"free_params": 0,
62			"DoF": 0,
63			}
64			self.dataset_name_list = []
65
66			self.final_model = Models()
67			self.final_data_products = ["fit", "fit_result", "flux_points"]
68
69			for data_product in self.final_data_products:
70			setattr(self, data_product, None)
71
72			@property
73			def models(self):
74			"""
75			Display the assigned Models.
76			"""
77			if not self.datasets:
78			raise RuntimeError("No datasets defined. Impossible to set models.")
79			return self.datasets.models
80
81			@property
82			def config(self):
83			"""
84			Analysis configuration (`AsgardpyConfig`)
85			"""
86			return self._config
87
88			@config.setter
89			def config(self, value):
90			if isinstance(value, dict):
91			self._config = AsgardpyConfig(**value)
92			elif isinstance(value, AsgardpyConfig):
93			self._config = value
94			else:
95			raise TypeError("config must be dict or AsgardpyConfig.")
96
97			def update_models_list(self, models_list):
98			"""
99			This function updates the final_model object with a given list of
100			models, only if the target source has a Spatial model included.
101
102			This step is only valid for 3D Datasets, and might need reconsideration
103			in the future.
104			"""
105			if models_list:
106			target_source_model = models_list[self.config.target.source_name]
107
108			if target_source_model.spatial_model:
109			for model_ in models_list:
110			self.final_model.append(model_)
111			else: # pragma: no cover
112			self.log.info(
113			"The target source %s only has spectral model",
114			self.config.target.source_name,
115			)
116
117			def add_to_instrument_info(self, info_dict):
118			"""
119			Update the name, Degrees of Freedom and spectral energy ranges for each
120			instrument Datasets, to be used for the DL4 to DL5 processes.
121			"""
122			for name in info_dict["name"]:
123			self.instrument_spectral_info["name"].append(name)
124
125			for edges in info_dict["spectral_energy_ranges"]:
126			self.instrument_spectral_info["spectral_energy_ranges"].append(edges)
127
128			self.instrument_spectral_info["en_bins"] += info_dict["en_bins"]
129			self.instrument_spectral_info["free_params"] += info_dict["free_params"]
130
131			def update_dof_value(self):
132			"""
133			Simple function to update total Degrees of Freedom value in the
134			instrument_spectral_info dict from a filled final_model object.
135			"""
136			if len(self.final_model) > 0:
137			# Add to the total number of free model parameters
138			n_free_params = len(list(self.final_model.parameters.free_parameters))
139			self.instrument_spectral_info["free_params"] += n_free_params
140
141			# Get the final degrees of freedom as en_bins - free_params
142			self.instrument_spectral_info["DoF"] = (
143			self.instrument_spectral_info["en_bins"] - self.instrument_spectral_info["free_params"]
144			)
145
146			def run(self, steps=None, **kwargs):
147			"""
148			Main function to run the AnalaysisSteps provided.
149
150			Currently overwrite option is used from the value in AsgardpyConfig,
151			and is True by default.
152			"""
153			if steps is None:
154			steps = self.config.general.steps
155			self.overwrite = self.config.general.overwrite
156
157			dl3_dl4_steps = [step for step in steps if "datasets" in step]
158			dl4_dl5_steps = [step for step in steps if "datasets" not in step]
159
160			if len(dl3_dl4_steps) > 0:
161			self.log.info("Perform DL3 to DL4 process!")
162
163			for step in dl3_dl4_steps:
164			analysis_step = AnalysisStep.create(step, self.config, **kwargs)
165
166			datasets_list, models_list, instrument_spectral_info = analysis_step.run()
167
168			self.update_models_list(models_list)
169
170			# To get all datasets_names from the datasets and update the final datasets list
171			for data in datasets_list:
172			if data.name not in self.dataset_name_list:
173			self.dataset_name_list.append(data.name)
174			self.datasets.append(data)
175
176			self.add_to_instrument_info(instrument_spectral_info)
177
178			self.datasets, self.final_model = set_models(
179			self.config.target,
180			self.datasets,
181			self.dataset_name_list,
182			models=self.final_model,
183			)
184			self.log.info("Models have been associated with the Datasets")
185
186			self.update_dof_value()
187
188			if len(dl4_dl5_steps) > 0:
189			self.log.info("Perform DL4 to DL5 processes!")
190
191			for step in dl4_dl5_steps:
192			analysis_step = AnalysisStep.create(step, self.config, **kwargs)
193
194			analysis_step.run(datasets=self.datasets, instrument_spectral_info=self.instrument_spectral_info)
195
196			# Update the final data product objects
197			for data_product in self.final_data_products:
198			if hasattr(analysis_step, data_product):
199			setattr(self, data_product, getattr(analysis_step, data_product))
200
201			if self.fit_result:
202			self.instrument_spectral_info, message = get_goodness_of_fit_stats(
203			self.datasets, self.instrument_spectral_info
204			)
205			self.log.info(message)
206
207			# keep these methods to be backward compatible
208			def get_1d_datasets(self):
209			"""Produce stacked 1D datasets."""
210			self.run(steps=["datasets-1d"])
211
212			def get_3d_datasets(self):
213			"""Produce stacked 3D datasets."""
214			self.run(steps=["datasets-3d"])
215
216			def run_fit(self):
217			"""Fitting reduced datasets to model."""
218			self.run(steps=["fit"])
219
220			def get_flux_points(self):
221			"""Calculate flux points for a specific model component."""
222			self.run(steps=["flux-points"])
223
224			def update_config(self, config):
225			"""Update the primary config with another config."""
226			self.config = self.config.update(config=config)
227

chaimain / asgardpy

Pull Request — main (#157)

AsgardpyAnalysis.add_to_instrument_info() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like