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

chaimain / asgardpy

Pull Request — main (#165)

asgardpy.config.generator.deep_update() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like