Spaces:
Paused
Paused
# coding=utf-8 | |
# Copyright 2023 The HuggingFace Inc. team. | |
# Copyright (c) 2022, NVIDIA CORPORATION. All rights reserved. | |
# | |
# 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. | |
""" ConfigMixin base class and utilities.""" | |
import dataclasses | |
import functools | |
import importlib | |
import inspect | |
import json | |
import os | |
import re | |
from collections import OrderedDict | |
from pathlib import PosixPath | |
from typing import Any, Dict, Tuple, Union | |
import numpy as np | |
from huggingface_hub import create_repo, hf_hub_download | |
from huggingface_hub.utils import ( | |
EntryNotFoundError, | |
RepositoryNotFoundError, | |
RevisionNotFoundError, | |
validate_hf_hub_args, | |
) | |
from requests import HTTPError | |
from . import __version__ | |
from .utils import ( | |
HUGGINGFACE_CO_RESOLVE_ENDPOINT, | |
DummyObject, | |
deprecate, | |
extract_commit_hash, | |
http_user_agent, | |
logging, | |
) | |
logger = logging.get_logger(__name__) | |
_re_configuration_file = re.compile(r"config\.(.*)\.json") | |
class FrozenDict(OrderedDict): | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) | |
for key, value in self.items(): | |
setattr(self, key, value) | |
self.__frozen = True | |
def __delitem__(self, *args, **kwargs): | |
raise Exception(f"You cannot use ``__delitem__`` on a {self.__class__.__name__} instance.") | |
def setdefault(self, *args, **kwargs): | |
raise Exception(f"You cannot use ``setdefault`` on a {self.__class__.__name__} instance.") | |
def pop(self, *args, **kwargs): | |
raise Exception(f"You cannot use ``pop`` on a {self.__class__.__name__} instance.") | |
def update(self, *args, **kwargs): | |
raise Exception(f"You cannot use ``update`` on a {self.__class__.__name__} instance.") | |
def __setattr__(self, name, value): | |
if hasattr(self, "__frozen") and self.__frozen: | |
raise Exception(f"You cannot use ``__setattr__`` on a {self.__class__.__name__} instance.") | |
super().__setattr__(name, value) | |
def __setitem__(self, name, value): | |
if hasattr(self, "__frozen") and self.__frozen: | |
raise Exception(f"You cannot use ``__setattr__`` on a {self.__class__.__name__} instance.") | |
super().__setitem__(name, value) | |
class ConfigMixin: | |
r""" | |
Base class for all configuration classes. All configuration parameters are stored under `self.config`. Also | |
provides the [`~ConfigMixin.from_config`] and [`~ConfigMixin.save_config`] methods for loading, downloading, and | |
saving classes that inherit from [`ConfigMixin`]. | |
Class attributes: | |
- **config_name** (`str`) -- A filename under which the config should stored when calling | |
[`~ConfigMixin.save_config`] (should be overridden by parent class). | |
- **ignore_for_config** (`List[str]`) -- A list of attributes that should not be saved in the config (should be | |
overridden by subclass). | |
- **has_compatibles** (`bool`) -- Whether the class has compatible classes (should be overridden by subclass). | |
- **_deprecated_kwargs** (`List[str]`) -- Keyword arguments that are deprecated. Note that the `init` function | |
should only have a `kwargs` argument if at least one argument is deprecated (should be overridden by | |
subclass). | |
""" | |
config_name = None | |
ignore_for_config = [] | |
has_compatibles = False | |
_deprecated_kwargs = [] | |
def register_to_config(self, **kwargs): | |
if self.config_name is None: | |
raise NotImplementedError(f"Make sure that {self.__class__} has defined a class name `config_name`") | |
# Special case for `kwargs` used in deprecation warning added to schedulers | |
# TODO: remove this when we remove the deprecation warning, and the `kwargs` argument, | |
# or solve in a more general way. | |
kwargs.pop("kwargs", None) | |
if not hasattr(self, "_internal_dict"): | |
internal_dict = kwargs | |
else: | |
previous_dict = dict(self._internal_dict) | |
internal_dict = {**self._internal_dict, **kwargs} | |
logger.debug(f"Updating config from {previous_dict} to {internal_dict}") | |
self._internal_dict = FrozenDict(internal_dict) | |
def __getattr__(self, name: str) -> Any: | |
"""The only reason we overwrite `getattr` here is to gracefully deprecate accessing | |
config attributes directly. See https://github.com/huggingface/diffusers/pull/3129 | |
Tihs funtion is mostly copied from PyTorch's __getattr__ overwrite: | |
https://pytorch.org/docs/stable/_modules/torch/nn/modules/module.html#Module | |
""" | |
is_in_config = "_internal_dict" in self.__dict__ and hasattr(self.__dict__["_internal_dict"], name) | |
is_attribute = name in self.__dict__ | |
if is_in_config and not is_attribute: | |
deprecation_message = f"Accessing config attribute `{name}` directly via '{type(self).__name__}' object attribute is deprecated. Please access '{name}' over '{type(self).__name__}'s config object instead, e.g. 'scheduler.config.{name}'." | |
deprecate("direct config name access", "1.0.0", deprecation_message, standard_warn=False) | |
return self._internal_dict[name] | |
raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'") | |
def save_config(self, save_directory: Union[str, os.PathLike], push_to_hub: bool = False, **kwargs): | |
""" | |
Save a configuration object to the directory specified in `save_directory` so that it can be reloaded using the | |
[`~ConfigMixin.from_config`] class method. | |
Args: | |
save_directory (`str` or `os.PathLike`): | |
Directory where the configuration JSON file is saved (will be created if it does not exist). | |
push_to_hub (`bool`, *optional*, defaults to `False`): | |
Whether or not to push your model to the Hugging Face Hub after saving it. You can specify the | |
repository you want to push to with `repo_id` (will default to the name of `save_directory` in your | |
namespace). | |
kwargs (`Dict[str, Any]`, *optional*): | |
Additional keyword arguments passed along to the [`~utils.PushToHubMixin.push_to_hub`] method. | |
""" | |
if os.path.isfile(save_directory): | |
raise AssertionError(f"Provided path ({save_directory}) should be a directory, not a file") | |
os.makedirs(save_directory, exist_ok=True) | |
# If we save using the predefined names, we can load using `from_config` | |
output_config_file = os.path.join(save_directory, self.config_name) | |
self.to_json_file(output_config_file) | |
logger.info(f"Configuration saved in {output_config_file}") | |
if push_to_hub: | |
commit_message = kwargs.pop("commit_message", None) | |
private = kwargs.pop("private", False) | |
create_pr = kwargs.pop("create_pr", False) | |
token = kwargs.pop("token", None) | |
repo_id = kwargs.pop("repo_id", save_directory.split(os.path.sep)[-1]) | |
repo_id = create_repo(repo_id, exist_ok=True, private=private, token=token).repo_id | |
self._upload_folder( | |
save_directory, | |
repo_id, | |
token=token, | |
commit_message=commit_message, | |
create_pr=create_pr, | |
) | |
def from_config(cls, config: Union[FrozenDict, Dict[str, Any]] = None, return_unused_kwargs=False, **kwargs): | |
r""" | |
Instantiate a Python class from a config dictionary. | |
Parameters: | |
config (`Dict[str, Any]`): | |
A config dictionary from which the Python class is instantiated. Make sure to only load configuration | |
files of compatible classes. | |
return_unused_kwargs (`bool`, *optional*, defaults to `False`): | |
Whether kwargs that are not consumed by the Python class should be returned or not. | |
kwargs (remaining dictionary of keyword arguments, *optional*): | |
Can be used to update the configuration object (after it is loaded) and initiate the Python class. | |
`**kwargs` are passed directly to the underlying scheduler/model's `__init__` method and eventually | |
overwrite the same named arguments in `config`. | |
Returns: | |
[`ModelMixin`] or [`SchedulerMixin`]: | |
A model or scheduler object instantiated from a config dictionary. | |
Examples: | |
```python | |
>>> from diffusers import DDPMScheduler, DDIMScheduler, PNDMScheduler | |
>>> # Download scheduler from huggingface.co and cache. | |
>>> scheduler = DDPMScheduler.from_pretrained("google/ddpm-cifar10-32") | |
>>> # Instantiate DDIM scheduler class with same config as DDPM | |
>>> scheduler = DDIMScheduler.from_config(scheduler.config) | |
>>> # Instantiate PNDM scheduler class with same config as DDPM | |
>>> scheduler = PNDMScheduler.from_config(scheduler.config) | |
``` | |
""" | |
# <===== TO BE REMOVED WITH DEPRECATION | |
# TODO(Patrick) - make sure to remove the following lines when config=="model_path" is deprecated | |
if "pretrained_model_name_or_path" in kwargs: | |
config = kwargs.pop("pretrained_model_name_or_path") | |
if config is None: | |
raise ValueError("Please make sure to provide a config as the first positional argument.") | |
# ======> | |
if not isinstance(config, dict): | |
deprecation_message = "It is deprecated to pass a pretrained model name or path to `from_config`." | |
if "Scheduler" in cls.__name__: | |
deprecation_message += ( | |
f"If you were trying to load a scheduler, please use {cls}.from_pretrained(...) instead." | |
" Otherwise, please make sure to pass a configuration dictionary instead. This functionality will" | |
" be removed in v1.0.0." | |
) | |
elif "Model" in cls.__name__: | |
deprecation_message += ( | |
f"If you were trying to load a model, please use {cls}.load_config(...) followed by" | |
f" {cls}.from_config(...) instead. Otherwise, please make sure to pass a configuration dictionary" | |
" instead. This functionality will be removed in v1.0.0." | |
) | |
deprecate("config-passed-as-path", "1.0.0", deprecation_message, standard_warn=False) | |
config, kwargs = cls.load_config(pretrained_model_name_or_path=config, return_unused_kwargs=True, **kwargs) | |
init_dict, unused_kwargs, hidden_dict = cls.extract_init_dict(config, **kwargs) | |
# Allow dtype to be specified on initialization | |
if "dtype" in unused_kwargs: | |
init_dict["dtype"] = unused_kwargs.pop("dtype") | |
# add possible deprecated kwargs | |
for deprecated_kwarg in cls._deprecated_kwargs: | |
if deprecated_kwarg in unused_kwargs: | |
init_dict[deprecated_kwarg] = unused_kwargs.pop(deprecated_kwarg) | |
# Return model and optionally state and/or unused_kwargs | |
model = cls(**init_dict) | |
# make sure to also save config parameters that might be used for compatible classes | |
model.register_to_config(**hidden_dict) | |
# add hidden kwargs of compatible classes to unused_kwargs | |
unused_kwargs = {**unused_kwargs, **hidden_dict} | |
if return_unused_kwargs: | |
return (model, unused_kwargs) | |
else: | |
return model | |
def get_config_dict(cls, *args, **kwargs): | |
deprecation_message = ( | |
f" The function get_config_dict is deprecated. Please use {cls}.load_config instead. This function will be" | |
" removed in version v1.0.0" | |
) | |
deprecate("get_config_dict", "1.0.0", deprecation_message, standard_warn=False) | |
return cls.load_config(*args, **kwargs) | |
def load_config( | |
cls, | |
pretrained_model_name_or_path: Union[str, os.PathLike], | |
return_unused_kwargs=False, | |
return_commit_hash=False, | |
**kwargs, | |
) -> Tuple[Dict[str, Any], Dict[str, Any]]: | |
r""" | |
Load a model or scheduler configuration. | |
Parameters: | |
pretrained_model_name_or_path (`str` or `os.PathLike`, *optional*): | |
Can be either: | |
- A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on | |
the Hub. | |
- A path to a *directory* (for example `./my_model_directory`) containing model weights saved with | |
[`~ConfigMixin.save_config`]. | |
cache_dir (`Union[str, os.PathLike]`, *optional*): | |
Path to a directory where a downloaded pretrained model configuration is cached if the standard cache | |
is not used. | |
force_download (`bool`, *optional*, defaults to `False`): | |
Whether or not to force the (re-)download of the model weights and configuration files, overriding the | |
cached versions if they exist. | |
resume_download (`bool`, *optional*, defaults to `False`): | |
Whether or not to resume downloading the model weights and configuration files. If set to `False`, any | |
incompletely downloaded files are deleted. | |
proxies (`Dict[str, str]`, *optional*): | |
A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128', | |
'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request. | |
output_loading_info(`bool`, *optional*, defaults to `False`): | |
Whether or not to also return a dictionary containing missing keys, unexpected keys and error messages. | |
local_files_only (`bool`, *optional*, defaults to `False`): | |
Whether to only load local model weights and configuration files or not. If set to `True`, the model | |
won't be downloaded from the Hub. | |
token (`str` or *bool*, *optional*): | |
The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from | |
`diffusers-cli login` (stored in `~/.huggingface`) is used. | |
revision (`str`, *optional*, defaults to `"main"`): | |
The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier | |
allowed by Git. | |
subfolder (`str`, *optional*, defaults to `""`): | |
The subfolder location of a model file within a larger model repository on the Hub or locally. | |
return_unused_kwargs (`bool`, *optional*, defaults to `False): | |
Whether unused keyword arguments of the config are returned. | |
return_commit_hash (`bool`, *optional*, defaults to `False): | |
Whether the `commit_hash` of the loaded configuration are returned. | |
Returns: | |
`dict`: | |
A dictionary of all the parameters stored in a JSON configuration file. | |
""" | |
cache_dir = kwargs.pop("cache_dir", None) | |
force_download = kwargs.pop("force_download", False) | |
resume_download = kwargs.pop("resume_download", False) | |
proxies = kwargs.pop("proxies", None) | |
token = kwargs.pop("token", None) | |
local_files_only = kwargs.pop("local_files_only", False) | |
revision = kwargs.pop("revision", None) | |
_ = kwargs.pop("mirror", None) | |
subfolder = kwargs.pop("subfolder", None) | |
user_agent = kwargs.pop("user_agent", {}) | |
user_agent = {**user_agent, "file_type": "config"} | |
user_agent = http_user_agent(user_agent) | |
pretrained_model_name_or_path = str(pretrained_model_name_or_path) | |
if cls.config_name is None: | |
raise ValueError( | |
"`self.config_name` is not defined. Note that one should not load a config from " | |
"`ConfigMixin`. Please make sure to define `config_name` in a class inheriting from `ConfigMixin`" | |
) | |
if os.path.isfile(pretrained_model_name_or_path): | |
config_file = pretrained_model_name_or_path | |
elif os.path.isdir(pretrained_model_name_or_path): | |
if os.path.isfile(os.path.join(pretrained_model_name_or_path, cls.config_name)): | |
# Load from a PyTorch checkpoint | |
config_file = os.path.join(pretrained_model_name_or_path, cls.config_name) | |
elif subfolder is not None and os.path.isfile( | |
os.path.join(pretrained_model_name_or_path, subfolder, cls.config_name) | |
): | |
config_file = os.path.join(pretrained_model_name_or_path, subfolder, cls.config_name) | |
else: | |
raise EnvironmentError( | |
f"Error no file named {cls.config_name} found in directory {pretrained_model_name_or_path}." | |
) | |
else: | |
try: | |
# Load from URL or cache if already cached | |
config_file = hf_hub_download( | |
pretrained_model_name_or_path, | |
filename=cls.config_name, | |
cache_dir=cache_dir, | |
force_download=force_download, | |
proxies=proxies, | |
resume_download=resume_download, | |
local_files_only=local_files_only, | |
token=token, | |
user_agent=user_agent, | |
subfolder=subfolder, | |
revision=revision, | |
) | |
except RepositoryNotFoundError: | |
raise EnvironmentError( | |
f"{pretrained_model_name_or_path} is not a local folder and is not a valid model identifier" | |
" listed on 'https://huggingface.co/models'\nIf this is a private repository, make sure to pass a" | |
" token having permission to this repo with `token` or log in with `huggingface-cli login`." | |
) | |
except RevisionNotFoundError: | |
raise EnvironmentError( | |
f"{revision} is not a valid git identifier (branch name, tag name or commit id) that exists for" | |
" this model name. Check the model page at" | |
f" 'https://huggingface.co/{pretrained_model_name_or_path}' for available revisions." | |
) | |
except EntryNotFoundError: | |
raise EnvironmentError( | |
f"{pretrained_model_name_or_path} does not appear to have a file named {cls.config_name}." | |
) | |
except HTTPError as err: | |
raise EnvironmentError( | |
"There was a specific connection error when trying to load" | |
f" {pretrained_model_name_or_path}:\n{err}" | |
) | |
except ValueError: | |
raise EnvironmentError( | |
f"We couldn't connect to '{HUGGINGFACE_CO_RESOLVE_ENDPOINT}' to load this model, couldn't find it" | |
f" in the cached files and it looks like {pretrained_model_name_or_path} is not the path to a" | |
f" directory containing a {cls.config_name} file.\nCheckout your internet connection or see how to" | |
" run the library in offline mode at" | |
" 'https://huggingface.co/docs/diffusers/installation#offline-mode'." | |
) | |
except EnvironmentError: | |
raise EnvironmentError( | |
f"Can't load config for '{pretrained_model_name_or_path}'. If you were trying to load it from " | |
"'https://huggingface.co/models', make sure you don't have a local directory with the same name. " | |
f"Otherwise, make sure '{pretrained_model_name_or_path}' is the correct path to a directory " | |
f"containing a {cls.config_name} file" | |
) | |
try: | |
# Load config dict | |
config_dict = cls._dict_from_json_file(config_file) | |
commit_hash = extract_commit_hash(config_file) | |
except (json.JSONDecodeError, UnicodeDecodeError): | |
raise EnvironmentError(f"It looks like the config file at '{config_file}' is not a valid JSON file.") | |
if not (return_unused_kwargs or return_commit_hash): | |
return config_dict | |
outputs = (config_dict,) | |
if return_unused_kwargs: | |
outputs += (kwargs,) | |
if return_commit_hash: | |
outputs += (commit_hash,) | |
return outputs | |
def _get_init_keys(cls): | |
return set(dict(inspect.signature(cls.__init__).parameters).keys()) | |
def extract_init_dict(cls, config_dict, **kwargs): | |
# Skip keys that were not present in the original config, so default __init__ values were used | |
used_defaults = config_dict.get("_use_default_values", []) | |
config_dict = {k: v for k, v in config_dict.items() if k not in used_defaults and k != "_use_default_values"} | |
# 0. Copy origin config dict | |
original_dict = dict(config_dict.items()) | |
# 1. Retrieve expected config attributes from __init__ signature | |
expected_keys = cls._get_init_keys(cls) | |
expected_keys.remove("self") | |
# remove general kwargs if present in dict | |
if "kwargs" in expected_keys: | |
expected_keys.remove("kwargs") | |
# remove flax internal keys | |
if hasattr(cls, "_flax_internal_args"): | |
for arg in cls._flax_internal_args: | |
expected_keys.remove(arg) | |
# 2. Remove attributes that cannot be expected from expected config attributes | |
# remove keys to be ignored | |
if len(cls.ignore_for_config) > 0: | |
expected_keys = expected_keys - set(cls.ignore_for_config) | |
# load diffusers library to import compatible and original scheduler | |
diffusers_library = importlib.import_module(__name__.split(".")[0]) | |
if cls.has_compatibles: | |
compatible_classes = [c for c in cls._get_compatibles() if not isinstance(c, DummyObject)] | |
else: | |
compatible_classes = [] | |
expected_keys_comp_cls = set() | |
for c in compatible_classes: | |
expected_keys_c = cls._get_init_keys(c) | |
expected_keys_comp_cls = expected_keys_comp_cls.union(expected_keys_c) | |
expected_keys_comp_cls = expected_keys_comp_cls - cls._get_init_keys(cls) | |
config_dict = {k: v for k, v in config_dict.items() if k not in expected_keys_comp_cls} | |
# remove attributes from orig class that cannot be expected | |
orig_cls_name = config_dict.pop("_class_name", cls.__name__) | |
if ( | |
isinstance(orig_cls_name, str) | |
and orig_cls_name != cls.__name__ | |
and hasattr(diffusers_library, orig_cls_name) | |
): | |
orig_cls = getattr(diffusers_library, orig_cls_name) | |
unexpected_keys_from_orig = cls._get_init_keys(orig_cls) - expected_keys | |
config_dict = {k: v for k, v in config_dict.items() if k not in unexpected_keys_from_orig} | |
elif not isinstance(orig_cls_name, str) and not isinstance(orig_cls_name, (list, tuple)): | |
raise ValueError( | |
"Make sure that the `_class_name` is of type string or list of string (for custom pipelines)." | |
) | |
# remove private attributes | |
config_dict = {k: v for k, v in config_dict.items() if not k.startswith("_")} | |
# 3. Create keyword arguments that will be passed to __init__ from expected keyword arguments | |
init_dict = {} | |
for key in expected_keys: | |
# if config param is passed to kwarg and is present in config dict | |
# it should overwrite existing config dict key | |
if key in kwargs and key in config_dict: | |
config_dict[key] = kwargs.pop(key) | |
if key in kwargs: | |
# overwrite key | |
init_dict[key] = kwargs.pop(key) | |
elif key in config_dict: | |
# use value from config dict | |
init_dict[key] = config_dict.pop(key) | |
# 4. Give nice warning if unexpected values have been passed | |
if len(config_dict) > 0: | |
logger.warning( | |
f"The config attributes {config_dict} were passed to {cls.__name__}, " | |
"but are not expected and will be ignored. Please verify your " | |
f"{cls.config_name} configuration file." | |
) | |
# 5. Give nice info if config attributes are initiliazed to default because they have not been passed | |
passed_keys = set(init_dict.keys()) | |
if len(expected_keys - passed_keys) > 0: | |
logger.info( | |
f"{expected_keys - passed_keys} was not found in config. Values will be initialized to default values." | |
) | |
# 6. Define unused keyword arguments | |
unused_kwargs = {**config_dict, **kwargs} | |
# 7. Define "hidden" config parameters that were saved for compatible classes | |
hidden_config_dict = {k: v for k, v in original_dict.items() if k not in init_dict} | |
return init_dict, unused_kwargs, hidden_config_dict | |
def _dict_from_json_file(cls, json_file: Union[str, os.PathLike]): | |
with open(json_file, "r", encoding="utf-8") as reader: | |
text = reader.read() | |
return json.loads(text) | |
def __repr__(self): | |
return f"{self.__class__.__name__} {self.to_json_string()}" | |
def config(self) -> Dict[str, Any]: | |
""" | |
Returns the config of the class as a frozen dictionary | |
Returns: | |
`Dict[str, Any]`: Config of the class. | |
""" | |
return self._internal_dict | |
def to_json_string(self) -> str: | |
""" | |
Serializes the configuration instance to a JSON string. | |
Returns: | |
`str`: | |
String containing all the attributes that make up the configuration instance in JSON format. | |
""" | |
config_dict = self._internal_dict if hasattr(self, "_internal_dict") else {} | |
config_dict["_class_name"] = self.__class__.__name__ | |
config_dict["_diffusers_version"] = __version__ | |
def to_json_saveable(value): | |
if isinstance(value, np.ndarray): | |
value = value.tolist() | |
elif isinstance(value, PosixPath): | |
value = str(value) | |
return value | |
config_dict = {k: to_json_saveable(v) for k, v in config_dict.items()} | |
# Don't save "_ignore_files" or "_use_default_values" | |
config_dict.pop("_ignore_files", None) | |
config_dict.pop("_use_default_values", None) | |
return json.dumps(config_dict, indent=2, sort_keys=True) + "\n" | |
def to_json_file(self, json_file_path: Union[str, os.PathLike]): | |
""" | |
Save the configuration instance's parameters to a JSON file. | |
Args: | |
json_file_path (`str` or `os.PathLike`): | |
Path to the JSON file to save a configuration instance's parameters. | |
""" | |
with open(json_file_path, "w", encoding="utf-8") as writer: | |
writer.write(self.to_json_string()) | |
def register_to_config(init): | |
r""" | |
Decorator to apply on the init of classes inheriting from [`ConfigMixin`] so that all the arguments are | |
automatically sent to `self.register_for_config`. To ignore a specific argument accepted by the init but that | |
shouldn't be registered in the config, use the `ignore_for_config` class variable | |
Warning: Once decorated, all private arguments (beginning with an underscore) are trashed and not sent to the init! | |
""" | |
def inner_init(self, *args, **kwargs): | |
# Ignore private kwargs in the init. | |
init_kwargs = {k: v for k, v in kwargs.items() if not k.startswith("_")} | |
config_init_kwargs = {k: v for k, v in kwargs.items() if k.startswith("_")} | |
if not isinstance(self, ConfigMixin): | |
raise RuntimeError( | |
f"`@register_for_config` was applied to {self.__class__.__name__} init method, but this class does " | |
"not inherit from `ConfigMixin`." | |
) | |
ignore = getattr(self, "ignore_for_config", []) | |
# Get positional arguments aligned with kwargs | |
new_kwargs = {} | |
signature = inspect.signature(init) | |
parameters = { | |
name: p.default for i, (name, p) in enumerate(signature.parameters.items()) if i > 0 and name not in ignore | |
} | |
for arg, name in zip(args, parameters.keys()): | |
new_kwargs[name] = arg | |
# Then add all kwargs | |
new_kwargs.update( | |
{ | |
k: init_kwargs.get(k, default) | |
for k, default in parameters.items() | |
if k not in ignore and k not in new_kwargs | |
} | |
) | |
# Take note of the parameters that were not present in the loaded config | |
if len(set(new_kwargs.keys()) - set(init_kwargs)) > 0: | |
new_kwargs["_use_default_values"] = list(set(new_kwargs.keys()) - set(init_kwargs)) | |
new_kwargs = {**config_init_kwargs, **new_kwargs} | |
getattr(self, "register_to_config")(**new_kwargs) | |
init(self, *args, **init_kwargs) | |
return inner_init | |
def flax_register_to_config(cls): | |
original_init = cls.__init__ | |
def init(self, *args, **kwargs): | |
if not isinstance(self, ConfigMixin): | |
raise RuntimeError( | |
f"`@register_for_config` was applied to {self.__class__.__name__} init method, but this class does " | |
"not inherit from `ConfigMixin`." | |
) | |
# Ignore private kwargs in the init. Retrieve all passed attributes | |
init_kwargs = dict(kwargs.items()) | |
# Retrieve default values | |
fields = dataclasses.fields(self) | |
default_kwargs = {} | |
for field in fields: | |
# ignore flax specific attributes | |
if field.name in self._flax_internal_args: | |
continue | |
if type(field.default) == dataclasses._MISSING_TYPE: | |
default_kwargs[field.name] = None | |
else: | |
default_kwargs[field.name] = getattr(self, field.name) | |
# Make sure init_kwargs override default kwargs | |
new_kwargs = {**default_kwargs, **init_kwargs} | |
# dtype should be part of `init_kwargs`, but not `new_kwargs` | |
if "dtype" in new_kwargs: | |
new_kwargs.pop("dtype") | |
# Get positional arguments aligned with kwargs | |
for i, arg in enumerate(args): | |
name = fields[i].name | |
new_kwargs[name] = arg | |
# Take note of the parameters that were not present in the loaded config | |
if len(set(new_kwargs.keys()) - set(init_kwargs)) > 0: | |
new_kwargs["_use_default_values"] = list(set(new_kwargs.keys()) - set(init_kwargs)) | |
getattr(self, "register_to_config")(**new_kwargs) | |
original_init(self, *args, **kwargs) | |
cls.__init__ = init | |
return cls | |