asgardpy.config.generator.deep_update() - Code Metrics - Inspection of "Update to support Gammapy v1.2" - chaimain/asgardpy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#165)

by Chaitanya

created 2024-03-19 10:37 UTC

asgardpy.config.generator.deep_update() A

↳ Parent: asgardpy.config.generator

Complexity

Conditions

Size

Total Lines	12
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	2
dl	0
loc	12
rs	10
c	0
b	0
f	0

"""
Main AsgardpyConfig Generator Module
"""

import json
import logging
from collections.abc import Mapping
from enum import Enum
from pathlib import Path

import yaml
from gammapy.modeling.models import Models
from gammapy.utils.scripts import make_path, read_yaml

from asgardpy.analysis.step_base import AnalysisStepEnum
from asgardpy.base import BaseConfig, PathType
from asgardpy.data import (
    Dataset1DConfig,
    Dataset3DConfig,
    FitConfig,
    FluxPointsConfig,
    Target,
)

__all__ = [
    "AsgardpyConfig",
    "GeneralConfig",
    "gammapy_to_asgardpy_model_config",
    "get_model_template",
    "recursive_merge_dicts",
]

CONFIG_PATH = Path(__file__).resolve().parent

log = logging.getLogger(__name__)


# Other general config params
class LogConfig(BaseConfig):
    """Config section for main logging information."""

    level: str = "info"
    filename: str = ""
    filemode: str = "w"
    format: str = ""
    datefmt: str = ""


class ParallelBackendEnum(str, Enum):
    """Config section for list of parallel processing backend methods."""

    multi = "multiprocessing"
    ray = "ray"


class GeneralConfig(BaseConfig):
    """Config section for general information for running AsgardpyAnalysis."""

    log: LogConfig = LogConfig()
    outdir: PathType = "None"
    n_jobs: int = 1
    parallel_backend: ParallelBackendEnum = ParallelBackendEnum.multi
    steps: list[AnalysisStepEnum] = []
    overwrite: bool = True
    stacked_dataset: bool = False


def get_model_template(spec_model_tag):
    """
    Read a particular template model yaml file into AsgardpyConfig object.
    """
    template_files = sorted(list(CONFIG_PATH.glob("model_templates/model_template*yaml")))
    new_model_file = None
    for file in template_files:
        if spec_model_tag == file.name.split("_")[-1].split(".")[0]:
            new_model_file = file
    return new_model_file


def recursive_merge_dicts(base_config, extra_config):
    """
    recursively merge two dictionaries.
    Entries in extra_config override entries in base_config. The built-in
    update function cannot be used for hierarchical dicts.

    Also for the case when there is a list of dicts involved, one has to be
    more careful. The extra_config may have longer list of dicts as compared
    with the base_config, in which case, the extra items are simply added to
    the merged final list.

    Combined here are 2 options from SO.

    See:
    http://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth/3233356#3233356
    and also
    https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth/18394648#18394648

    Parameters
    ----------
    base_config : dict
        dictionary to be merged
    extra_config : dict
        dictionary to be merged
    Returns
    -------
    final_config : dict
        merged dict
    """
    final_config = base_config.copy()

    for key, value in extra_config.items():
        if key in final_config and isinstance(final_config[key], list):
            new_config = []

            for key_, value_ in zip(final_config[key], value, strict=False):
                key_ = recursive_merge_dicts(key_ or {}, value_)
                new_config.append(key_)

            # For example moving from a smaller list of model parameters to a
            # longer list.
            if len(final_config[key]) < len(extra_config[key]):
                for value_ in value[len(final_config[key]) :]:
                    new_config.append(value_)
            final_config[key] = new_config

        elif key in final_config and isinstance(final_config[key], dict):
            final_config[key] = recursive_merge_dicts(final_config.get(key) or {}, value)
        else:
            final_config[key] = value

    return final_config


def deep_update(d, u):
    """
    Recursively update a nested dictionary.

    Just like in Gammapy, taken from: https://stackoverflow.com/a/3233356/19802442
    """
    for k, v in u.items():
        if isinstance(v, Mapping):
            d[k] = deep_update(d.get(k, {}), v)
        else:
            d[k] = v
    return d


def gammapy_to_asgardpy_model_config(gammapy_model, asgardpy_config_file=None, recursive_merge=True):
    """
    Read the Gammapy Models YAML file and save it as AsgardpyConfig object.

    Return
    ------
    asgardpy_config: `asgardpy.config.generator.AsgardpyConfig`
        Updated AsgardpyConfig object
    """
    try:
        models_gpy = Models.read(gammapy_model)
    except KeyError:
        log.error("%s File cannot be read by Gammapy Models", gammapy_model)
        return None

    if not asgardpy_config_file:
        asgardpy_config = AsgardpyConfig()  # Default object
    elif isinstance(asgardpy_config_file, str):  # File path
        asgardpy_config = AsgardpyConfig.read(asgardpy_config_file)
    elif isinstance(asgardpy_config_file, AsgardpyConfig):
        asgardpy_config = asgardpy_config_file

    models_gpy_dict = models_gpy.to_dict()
    asgardpy_config_target_dict = asgardpy_config.model_dump()["target"]


    if recursive_merge:
        temp_target_dict = recursive_merge_dicts(asgardpy_config_target_dict, models_gpy_dict)
    else:
        # Use when there are nans present in the other config file, which are
        # the defaults in Gammapy, but NOT in Asgardpy.
        # E.g. test data Fermi-3fhl-crab model file
        temp_target_dict = deep_update(asgardpy_config_target_dict, models_gpy_dict)
    asgardpy_config.target = temp_target_dict

    return asgardpy_config


# Combine everything!
class AsgardpyConfig(BaseConfig):
    """
    Asgardpy analysis configuration, based on Gammapy Analysis Config.
    """

    general: GeneralConfig = GeneralConfig()

    target: Target = Target()

    dataset3d: Dataset3DConfig = Dataset3DConfig()
    dataset1d: Dataset1DConfig = Dataset1DConfig()

    fit_params: FitConfig = FitConfig()
    flux_points_params: FluxPointsConfig = FluxPointsConfig()

    def __str__(self):
        """
        Display settings in pretty YAML format.
        """
        info = self.__class__.__name__ + "\n\n\t"
        data = self.to_yaml()
        data = data.replace("\n", "\n\t")
        info += data
        return info.expandtabs(tabsize=4)

    @classmethod
    def read(cls, path):
        """
        Reads from YAML file.
        """
        config = read_yaml(path)
        return AsgardpyConfig(**config)

    @classmethod
    def from_yaml(cls, config_str):
        """
        Create from YAML string.
        """
        settings = yaml.safe_load(config_str)
        return AsgardpyConfig(**settings)

    def write(self, path, overwrite=False):
        """
        Write to YAML file.
        """
        path = make_path(path)
        if path.exists() and not overwrite:
            raise OSError(f"File exists already: {path}")
        path.write_text(self.to_yaml())

    def to_yaml(self):
        """
        Convert to YAML string.
        """
        # Here using `dict()` instead of `json()` would be more natural.
        # We should change this once pydantic adds support for custom encoders
        # to `dict()`. See https://github.com/samuelcolvin/pydantic/issues/1043
        data = json.loads(self.model_dump_json())
        return yaml.dump(data, sort_keys=False, indent=4, width=80, default_flow_style=None)

    def set_logging(self):
        """
        Set logging config.
        Calls ``logging.basicConfig``, i.e. adjusts global logging state.
        """
        self.general.log.level = self.general.log.level.upper()
        logging.basicConfig(**self.general.log.model_dump())
        log.info("Setting logging config: %s", self.general.log.model_dump())

    def update(self, config=None, merge_recursive=False):
        """
        Update config with provided settings.
        Parameters
        ----------
        config : string dict or `AsgardpyConfig` object
            The other configuration settings provided in dict() syntax.
        merge_recursive : bool
            Perform a recursive merge from the other config onto the parent config.

        Returns
        -------
        config : `AsgardpyConfig` object
            Updated config object.
        """
        if isinstance(config, str):
            other = AsgardpyConfig.from_yaml(config)
        elif isinstance(config, AsgardpyConfig):
            other = config
        else:
            raise TypeError(f"Invalid type: {config}")

        # Special case of when only updating target model parameters from a
        # separate file, where the name of the source is not provided.
        if other.target.components[0].name == "":
            merge_recursive = True

        if merge_recursive:
            config_new = recursive_merge_dicts(
                self.model_dump(exclude_defaults=True), other.model_dump(exclude_defaults=True)
            )
        else:
            config_new = deep_update(
                self.model_dump(exclude_defaults=True), other.model_dump(exclude_defaults=True)
            )
        return AsgardpyConfig(**config_new)


1			"""
2			Main AsgardpyConfig Generator Module
3			"""
4
5			import json
6			import logging
7			from collections.abc import Mapping
8			from enum import Enum
9			from pathlib import Path
10
11			import yaml
12			from gammapy.modeling.models import Models
13			from gammapy.utils.scripts import make_path, read_yaml
14
15			from asgardpy.analysis.step_base import AnalysisStepEnum
16			from asgardpy.base import BaseConfig, PathType
17			from asgardpy.data import (
18			Dataset1DConfig,
19			Dataset3DConfig,
20			FitConfig,
21			FluxPointsConfig,
22			Target,
23			)
24
25			__all__ = [
26			"AsgardpyConfig",
27			"GeneralConfig",
28			"gammapy_to_asgardpy_model_config",
29			"get_model_template",
30			"recursive_merge_dicts",
31			]
32
33			CONFIG_PATH = Path(__file__).resolve().parent
34
35			log = logging.getLogger(__name__)
36
37
38			# Other general config params
39			class LogConfig(BaseConfig):
40			"""Config section for main logging information."""
41
42			level: str = "info"
43			filename: str = ""
44			filemode: str = "w"
45			format: str = ""
46			datefmt: str = ""
47
48
49			class ParallelBackendEnum(str, Enum):
50			"""Config section for list of parallel processing backend methods."""
51
52			multi = "multiprocessing"
53			ray = "ray"
54
55
56			class GeneralConfig(BaseConfig):
57			"""Config section for general information for running AsgardpyAnalysis."""
58
59			log: LogConfig = LogConfig()
60			outdir: PathType = "None"
61			n_jobs: int = 1
62			parallel_backend: ParallelBackendEnum = ParallelBackendEnum.multi
63			steps: list[AnalysisStepEnum] = []
64			overwrite: bool = True
65			stacked_dataset: bool = False
66
67
68			def get_model_template(spec_model_tag):
69			"""
70			Read a particular template model yaml file into AsgardpyConfig object.
71			"""
72			template_files = sorted(list(CONFIG_PATH.glob("model_templates/model_template*yaml")))
73			new_model_file = None
74			for file in template_files:
75			if spec_model_tag == file.name.split("_")[-1].split(".")[0]:
76			new_model_file = file
77			return new_model_file
78
79
80			def recursive_merge_dicts(base_config, extra_config):
81			"""
82			recursively merge two dictionaries.
83			Entries in extra_config override entries in base_config. The built-in
84			update function cannot be used for hierarchical dicts.
85
86			Also for the case when there is a list of dicts involved, one has to be
87			more careful. The extra_config may have longer list of dicts as compared
88			with the base_config, in which case, the extra items are simply added to
89			the merged final list.
90
91			Combined here are 2 options from SO.
92
93			See:
94			http://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth/3233356#3233356
95			and also
96			https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth/18394648#18394648
97
98			Parameters
99			----------
100			base_config : dict
101			dictionary to be merged
102			extra_config : dict
103			dictionary to be merged
104			Returns
105			-------
106			final_config : dict
107			merged dict
108			"""
109			final_config = base_config.copy()
110
111			for key, value in extra_config.items():
112			if key in final_config and isinstance(final_config[key], list):
113			new_config = []
114
115			for key_, value_ in zip(final_config[key], value, strict=False):
116			key_ = recursive_merge_dicts(key_ or {}, value_)
117			new_config.append(key_)
118
119			# For example moving from a smaller list of model parameters to a
120			# longer list.
121			if len(final_config[key]) < len(extra_config[key]):
122			for value_ in value[len(final_config[key]) :]:
123			new_config.append(value_)
124			final_config[key] = new_config
125
126			elif key in final_config and isinstance(final_config[key], dict):
127			final_config[key] = recursive_merge_dicts(final_config.get(key) or {}, value)
128			else:
129			final_config[key] = value
130
131			return final_config
132
133
134			def deep_update(d, u):
135			"""
136			Recursively update a nested dictionary.
137
138			Just like in Gammapy, taken from: https://stackoverflow.com/a/3233356/19802442
139			"""
140			for k, v in u.items():
141			if isinstance(v, Mapping):
142			d[k] = deep_update(d.get(k, {}), v)
143			else:
144			d[k] = v
145			return d
146
147
148			def gammapy_to_asgardpy_model_config(gammapy_model, asgardpy_config_file=None, recursive_merge=True):
149			"""
150			Read the Gammapy Models YAML file and save it as AsgardpyConfig object.
151
152			Return
153			------
154			asgardpy_config: `asgardpy.config.generator.AsgardpyConfig`
155			Updated AsgardpyConfig object
156			"""
157			try:
158			models_gpy = Models.read(gammapy_model)
159			except KeyError:
160			log.error("%s File cannot be read by Gammapy Models", gammapy_model)
161			return None
162
163			if not asgardpy_config_file:
164			asgardpy_config = AsgardpyConfig() # Default object
165			elif isinstance(asgardpy_config_file, str): # File path
166			asgardpy_config = AsgardpyConfig.read(asgardpy_config_file)
167			elif isinstance(asgardpy_config_file, AsgardpyConfig):
168			asgardpy_config = asgardpy_config_file
169
170			models_gpy_dict = models_gpy.to_dict()
171			asgardpy_config_target_dict = asgardpy_config.model_dump()["target"]
			1 ignored issue – show introduced 2024-01-06 22:55 UTC by Report Bug Copy Issue Report The variable `asgardpy_config` does not seem to be defined for all execution paths. Loading history...
172
173			if recursive_merge:
174			temp_target_dict = recursive_merge_dicts(asgardpy_config_target_dict, models_gpy_dict)
175			else:
176			# Use when there are nans present in the other config file, which are
177			# the defaults in Gammapy, but NOT in Asgardpy.
178			# E.g. test data Fermi-3fhl-crab model file
179			temp_target_dict = deep_update(asgardpy_config_target_dict, models_gpy_dict)
180			asgardpy_config.target = temp_target_dict
181
182			return asgardpy_config
183
184
185			# Combine everything!
186			class AsgardpyConfig(BaseConfig):
187			"""
188			Asgardpy analysis configuration, based on Gammapy Analysis Config.
189			"""
190
191			general: GeneralConfig = GeneralConfig()
192
193			target: Target = Target()
194
195			dataset3d: Dataset3DConfig = Dataset3DConfig()
196			dataset1d: Dataset1DConfig = Dataset1DConfig()
197
198			fit_params: FitConfig = FitConfig()
199			flux_points_params: FluxPointsConfig = FluxPointsConfig()
200
201			def __str__(self):
202			"""
203			Display settings in pretty YAML format.
204			"""
205			info = self.__class__.__name__ + "\n\n\t"
206			data = self.to_yaml()
207			data = data.replace("\n", "\n\t")
208			info += data
209			return info.expandtabs(tabsize=4)
210
211			@classmethod
212			def read(cls, path):
213			"""
214			Reads from YAML file.
215			"""
216			config = read_yaml(path)
217			return AsgardpyConfig(**config)
218
219			@classmethod
220			def from_yaml(cls, config_str):
221			"""
222			Create from YAML string.
223			"""
224			settings = yaml.safe_load(config_str)
225			return AsgardpyConfig(**settings)
226
227			def write(self, path, overwrite=False):
228			"""
229			Write to YAML file.
230			"""
231			path = make_path(path)
232			if path.exists() and not overwrite:
233			raise OSError(f"File exists already: {path}")
234			path.write_text(self.to_yaml())
235
236			def to_yaml(self):
237			"""
238			Convert to YAML string.
239			"""
240			# Here using `dict()` instead of `json()` would be more natural.
241			# We should change this once pydantic adds support for custom encoders
242			# to `dict()`. See https://github.com/samuelcolvin/pydantic/issues/1043
243			data = json.loads(self.model_dump_json())
244			return yaml.dump(data, sort_keys=False, indent=4, width=80, default_flow_style=None)
245
246			def set_logging(self):
247			"""
248			Set logging config.
249			Calls ``logging.basicConfig``, i.e. adjusts global logging state.
250			"""
251			self.general.log.level = self.general.log.level.upper()
252			logging.basicConfig(**self.general.log.model_dump())
253			log.info("Setting logging config: %s", self.general.log.model_dump())
254
255			def update(self, config=None, merge_recursive=False):
256			"""
257			Update config with provided settings.
258			Parameters
259			----------
260			config : string dict or `AsgardpyConfig` object
261			The other configuration settings provided in dict() syntax.
262			merge_recursive : bool
263			Perform a recursive merge from the other config onto the parent config.
264
265			Returns
266			-------
267			config : `AsgardpyConfig` object
268			Updated config object.
269			"""
270			if isinstance(config, str):
271			other = AsgardpyConfig.from_yaml(config)
272			elif isinstance(config, AsgardpyConfig):
273			other = config
274			else:
275			raise TypeError(f"Invalid type: {config}")
276
277			# Special case of when only updating target model parameters from a
278			# separate file, where the name of the source is not provided.
279			if other.target.components[0].name == "":
280			merge_recursive = True
281
282			if merge_recursive:
283			config_new = recursive_merge_dicts(
284			self.model_dump(exclude_defaults=True), other.model_dump(exclude_defaults=True)
285			)
286			else:
287			config_new = deep_update(
288			self.model_dump(exclude_defaults=True), other.model_dump(exclude_defaults=True)
289			)
290			return AsgardpyConfig(**config_new)
291

chaimain / asgardpy

Pull Request — main (#165)

asgardpy.config.generator.deep_update() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like