PaddleX/paddlex/repo_apis/PaddleSeg_api/base_seg_config.py

# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
#
# 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.

from urllib.parse import urlparse

import yaml
from paddleseg.utils import NoAliasDumper
from paddleseg.cvlibs.config import parse_from_yaml, merge_config_dicts

from ..base import BaseConfig
from ...utils.misc import abspath


class BaseSegConfig(BaseConfig):
    """BaseSegConfig"""

    def update(self, dict_like_obj):
        """update"""
        dict_ = merge_config_dicts(dict_like_obj, self.dict)
        self.reset_from_dict(dict_)

    def load(self, config_path):
        """load"""
        dict_ = parse_from_yaml(config_path)
        if not isinstance(dict_, dict):
            raise TypeError
        self.reset_from_dict(dict_)

    def dump(self, config_path):
        """dump"""
        with open(config_path, "w", encoding="utf-8") as f:
            yaml.dump(self.dict, f, Dumper=NoAliasDumper)

    def update_learning_rate(self, learning_rate):
        """update_learning_rate"""
        if "lr_scheduler" not in self:
            raise RuntimeError(
                "Not able to update learning rate, because no LR scheduler config was found."
            )
        self.lr_scheduler["learning_rate"] = learning_rate

    def update_batch_size(self, batch_size, mode="train"):
        """update_batch_size"""
        if mode == "train":
            self.set_val("batch_size", batch_size)
        else:
            raise ValueError(
                f"Setting `batch_size` in {repr(mode)} mode is not supported."
            )

    def update_log_ranks(self, device):
        """update log ranks

        Args:
            device (str): the running device to set
        """
        log_ranks = device.split(":")[1]
        self.set_val("log_ranks", log_ranks)

    def update_print_mem_info(self, print_mem_info: bool):
        """setting print memory info"""
        assert isinstance(print_mem_info, bool), "print_mem_info should be a bool"
        self.set_val("print_mem_info", print_mem_info)

    def update_shuffle(self, shuffle: bool):
        """setting print memory info"""
        assert isinstance(shuffle, bool), "shuffle should be a bool"
        self.set_val("shuffle", shuffle)

    def update_pretrained_weights(self, weight_path, is_backbone=False):
        """update_pretrained_weights"""
        if "model" not in self:
            raise RuntimeError(
                "Not able to update pretrained weight path, because no model config was found."
            )
        if isinstance(weight_path, str):
            if urlparse(weight_path).scheme == "":
                # If `weight_path` is a string but not URL (with scheme present),
                # it will be recognized as a local file path.
                weight_path = abspath(weight_path)
        else:
            if weight_path is not None:
                raise TypeError("`weight_path` must be string or None.")
        if is_backbone:
            if "backbone" not in self.model:
                raise RuntimeError(
                    "Not able to update pretrained weight path of backbone, because no backbone config was found."
                )
            self.model["backbone"]["pretrained"] = weight_path
        else:
            self.model["pretrained"] = weight_path

    def update_dy2st(self, dy2st):
        """update_dy2st"""
        self.set_val("to_static_training", dy2st)

    def update_dataset(self, dataset_dir, dataset_type=None):
        """update_dataset"""
        raise NotImplementedError

    def get_epochs_iters(self):
        """get_epochs_iters"""
        raise NotImplementedError

    def get_learning_rate(self):
        """get_learning_rate"""
        raise NotImplementedError

    def get_batch_size(self, mode="train"):
        """get_batch_size"""
        raise NotImplementedError

    def get_qat_epochs_iters(self):
        """get_qat_epochs_iters"""
        return self.get_epochs_iters() // 2

    def get_qat_learning_rate(self):
        """get_qat_learning_rate"""
        return self.get_learning_rate() / 2