asgardpy.analysis.analysis.AsgardpyAnalysis.update_dof_value() - 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

AsgardpyAnalysis.update_dof_value() A

↳ Parent: asgardpy.analysis.analysis

Complexity

Conditions

Size

Total Lines	11
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	1
dl	0
loc	11
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):
        """ """
        if models_list:
            # This step is only valid for 3D Datasets which have a list of models
            target_source_model = models_list[self.config.target.source_name]

            if target_source_model.spatial_model:
                # If the target source has Spatial model included,
                # only then (?) get all the models as final_model.
                # Needs reconsideration.
                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, DoF 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):
        """ """
        # Only when the final model object has been updated
        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, overwrite=None, **kwargs):
        """
        Main function to run the AnalaysisSteps provided.
        """
        if steps is None:
            steps = self.config.general.steps
            overwrite = self.config.general.overwrite
        else:
            if overwrite is None:
                overwrite = True

        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			if models_list:
100			# This step is only valid for 3D Datasets which have a list of models
101			target_source_model = models_list[self.config.target.source_name]
102
103			if target_source_model.spatial_model:
104			# If the target source has Spatial model included,
105			# only then (?) get all the models as final_model.
106			# Needs reconsideration.
107			for model_ in models_list:
108			self.final_model.append(model_)
109			else: # pragma: no cover
110			self.log.info(
111			"The target source %s only has spectral model",
112			self.config.target.source_name,
113			)
114
115			def add_to_instrument_info(self, info_dict):
116			""" """
117			# Update the name, DoF and spectral energy ranges for each
118			# instrument Datasets, to be used for the DL4 to DL5 processes.
119			for name in info_dict["name"]:
120			self.instrument_spectral_info["name"].append(name)
121
122			for edges in info_dict["spectral_energy_ranges"]:
123			self.instrument_spectral_info["spectral_energy_ranges"].append(edges)
124
125			self.instrument_spectral_info["en_bins"] += info_dict["en_bins"]
126			self.instrument_spectral_info["free_params"] += info_dict["free_params"]
127
128			def update_dof_value(self):
129			""" """
130			# Only when the final model object has been updated
131			if len(self.final_model) > 0:
132			# Add to the total number of free model parameters
133			n_free_params = len(list(self.final_model.parameters.free_parameters))
134			self.instrument_spectral_info["free_params"] += n_free_params
135
136			# Get the final degrees of freedom as en_bins - free_params
137			self.instrument_spectral_info["DoF"] = (
138			self.instrument_spectral_info["en_bins"] - self.instrument_spectral_info["free_params"]
139			)
140
141			def run(self, steps=None, overwrite=None, **kwargs):
142			"""
143			Main function to run the AnalaysisSteps provided.
144			"""
145			if steps is None:
146			steps = self.config.general.steps
147			overwrite = self.config.general.overwrite
148			else:
149			if overwrite is None:
150			overwrite = True
151
152			dl3_dl4_steps = [step for step in steps if "datasets" in step]
153			dl4_dl5_steps = [step for step in steps if "datasets" not in step]
154
155			if len(dl3_dl4_steps) > 0:
156			self.log.info("Perform DL3 to DL4 process!")
157
158			for step in dl3_dl4_steps:
159			analysis_step = AnalysisStep.create(step, self.config, **kwargs)
160
161			datasets_list, models_list, instrument_spectral_info = analysis_step.run()
162
163			self.update_models_list(models_list)
164
165			# To get all datasets_names from the datasets and update the final datasets list
166			for data in datasets_list:
167			if data.name not in self.dataset_name_list:
168			self.dataset_name_list.append(data.name)
169			self.datasets.append(data)
170
171			self.add_to_instrument_info(instrument_spectral_info)
172
173			self.datasets, self.final_model = set_models(
174			self.config.target,
175			self.datasets,
176			self.dataset_name_list,
177			models=self.final_model,
178			)
179			self.log.info("Models have been associated with the Datasets")
180
181			self.update_dof_value()
182
183			if len(dl4_dl5_steps) > 0:
184			self.log.info("Perform DL4 to DL5 processes!")
185
186			for step in dl4_dl5_steps:
187			analysis_step = AnalysisStep.create(step, self.config, **kwargs)
188
189			analysis_step.run(datasets=self.datasets, instrument_spectral_info=self.instrument_spectral_info)
190
191			# Update the final data product objects
192			for data_product in self.final_data_products:
193			if hasattr(analysis_step, data_product):
194			setattr(self, data_product, getattr(analysis_step, data_product))
195
196			if self.fit_result:
197			self.instrument_spectral_info, message = get_goodness_of_fit_stats(
198			self.datasets, self.instrument_spectral_info
199			)
200			self.log.info(message)
201
202			# keep these methods to be backward compatible
203			def get_1d_datasets(self):
204			"""Produce stacked 1D datasets."""
205			self.run(steps=["datasets-1d"])
206
207			def get_3d_datasets(self):
208			"""Produce stacked 3D datasets."""
209			self.run(steps=["datasets-3d"])
210
211			def run_fit(self):
212			"""Fitting reduced datasets to model."""
213			self.run(steps=["fit"])
214
215			def get_flux_points(self):
216			"""Calculate flux points for a specific model component."""
217			self.run(steps=["flux-points"])
218
219			def update_config(self, config):
220			"""Update the primary config with another config."""
221			self.config = self.config.update(config=config)
222

chaimain / asgardpy

Pull Request — main (#157)

AsgardpyAnalysis.update_dof_value() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like