Source code for mindspore_rl.utils.utils

# Copyright 2022-2023 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""
Utils.
"""
import os
from importlib import import_module
import yaml
import numpy as np
from typing import Tuple, Union


def _update_dict(dest, src) -> None:
    """update config dict"""
    if src is not None:
        for key in src:
            if key in dest.keys() and isinstance(dest.get(key), dict):
                if isinstance(src.get(key), dict):
                    for v in src.get(key):
                        if isinstance(src.get(key).get(v), dict) and v in dest.get(key) and \
                                isinstance(dest.get(key).get(v), dict):
                            _update_dict(dest[key], src[key])
                        elif isinstance(dest.get(key).get(v), dict):
                            dest[key][v].update(src.get(key).get(v))
                        else:
                            dest[key][v] = src.get(key).get(v)
                else:
                    dest[key].update(src[key])
            else:
                dest[key] = src[key]


[docs]def update_config(config, env_yaml, algo_yaml) -> None: r''' Update the config by the provided yamls. Eg: see `mindspore_rl/algorithm/dqn/config.py`, `mindspore_rl/example/env_yaml/` and `mindspore_rl/example/algo_yaml/` for usage. Args: config (dict): the config to be update. env_yaml (str): the environment yaml file. algo_yaml (str): the algorithm yaml file. ''' if env_yaml: if os.path.exists(env_yaml): with open(env_yaml) as f: data = yaml.safe_load(f) config.collect_env_params['name'] = data.get('env') config.eval_env_params['name'] = data.get('env') _update_dict(config.collect_env_params, data.get('collect_env_params')) _update_dict(config.eval_env_params, data.get('eval_env_params')) if data.get('env_class') and data.get('env_type'): try: env_class = import_module(data.get('env_class')) env_type = getattr(env_class, data.get('env_type')) config.algorithm_config['collect_environment']['type'] = env_type config.algorithm_config['eval_environment']['type'] = env_type except: raise ValueError(f"Import {data.get('env_class')} failed") else: print(f"File {env_yaml} is not exists.") return if algo_yaml: if os.path.exists(algo_yaml): with open(algo_yaml) as f: data = yaml.safe_load(f) if data.get('algorithm_config'): _update_dict(config.algorithm_config, data.get('algorithm_config')) if data.get('policy_params'): _update_dict(config.policy_params, data.get('policy_params')) if data.get('trainer_params'): _update_dict(config.trainer_params, data.get('trainer_params')) if data.get('learner_params'): _update_dict(config.learner_params, data.get('learner_params')) if data.get('learner_class') and data.get('learner_type'): try: learner_class = import_module(data.get('learner_class')) learner_type = getattr(learner_class, data.get('learner_type')) config.algorithm_config['learner']['type'] = learner_type except: raise ValueError(f"Import {data.get('learner_class')} failed") else: print(f"File {algo_yaml} is not exiddsts.") return
def check_type(input_type, items, debug_str): if isinstance(items, list) or isinstance(items, tuple): [check_type(input_type, item, debug_str) for item in items] return items, len(items) elif isinstance(items, input_type): return items, 1 else: raise TypeError(f"input item {debug_str} expects {input_type}, but got {type(items)}") def check_valid_return_value(return_value: Union[Tuple, np.ndarray], debug_str: str) -> int: num_valid_output = 0 if isinstance(return_value, tuple): list_valid_output = [check_valid_return_value(item, debug_str) for item in return_value] num_valid_output = sum(list_valid_output) elif isinstance(return_value, np.ndarray): num_valid_output = 1 else: raise TypeError(f"For {debug_str}, its output must be tuple or np.ndarray, but got {type(return_value)}") return num_valid_output