Skip to content

hyfi.composer

BaseConfig

Bases: BaseModel

Base class for all config classes.

Source code in hyfi/composer/config.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class BaseConfig(BaseModel):
    """
    Base class for all config classes.
    """

    _config_name_: str = "__init__"
    _auto_populate_: bool = True

    verbose: bool = False

    _init_args_: Dict[str, Any] = {}
    _property_set_methods_: Dict[str, str] = {}
    _subconfigs_: Dict[str, Any] = {}

    def __init__(self, **config_kwargs):
        logger.debug(
            "init %s with %s args", self.__class__.__name__, len(config_kwargs)
        )
        super().__init__(**config_kwargs)
        # self.initialize_subconfigs(config_kwargs)
        self._init_args_ = config_kwargs.copy()

    def __setattr__(self, key, val):
        """
        Overrides the default __setattr__ method to allow for custom property set methods.

        Args:
            key (str): The name of the attribute to set.
            val (Any): The value to set the attribute to.
        """
        if method := self._property_set_methods_.get(key):  # type: ignore
            logger.debug(
                "Setting %s to %s",
                key,
                val if isinstance(val, (str, int)) else type(val),
            )
            getattr(self, method)(val)
        super().__setattr__(key, val)

    def export_config(
        self,
        exclude: Optional[Union[str, List[str], Set[str], None]] = None,
        exclude_none: bool = True,
        only_include: Optional[Union[str, List[str], Set[str], None]] = None,
    ) -> Dict[str, Any]:
        """
        Export the configuration to a dictionary.

        Args:
            exclude (Optional[Union[str, List[str], Set[str], None]]): Keys to exclude from the saved configuration.
                Defaults to None.
            exclude_none (bool): Whether to exclude keys with None values. Defaults to True.
            only_include (Optional[Union[str, List[str], Set[str], None]]): Keys to include in the saved configuration.
                Defaults to None.

        Returns:
            Dict[str, Any]: The configuration dictionary.
        """
        if not exclude:
            exclude = self._exclude_  # type: ignore
        if isinstance(exclude, str):
            exclude = [exclude]
        if exclude is None:
            exclude = []
        if isinstance(only_include, str):
            only_include = [only_include]
        if only_include is None:
            only_include = []

        config = self.model_dump(exclude=exclude, exclude_none=exclude_none)  # type: ignore
        if only_include:
            config = {key: config[key] for key in only_include if key in config}

        return config

    def save_config(
        self,
        filepath: Union[str, Path],
        exclude: Optional[Union[str, List[str], Set[str], None]] = None,
        exclude_none: bool = True,
        only_include: Optional[Union[str, List[str], Set[str], None]] = None,
    ) -> str:
        """
        Save the batch configuration to file.

        Args:
            filepath ([Union[str, Path]): The filepath to save the configuration to.
            exclude (Optional[Union[str, List[str], Set[str], None]]): Keys to exclude from the saved configuration.
                Defaults to None.
            exclude_none (bool): Whether to exclude keys with None values. Defaults to True.
            only_include (Optional[Union[str, List[str], Set[str], None]]): Keys to include in the saved configuration.
                Defaults to None.

        Returns:
            str: The filename of the saved configuration.
        """
        logger.info("Saving config to %s", filepath)

        config_to_save = self.export_config(
            exclude=exclude, exclude_none=exclude_none, only_include=only_include
        )

        Composer.save(config_to_save, filepath)
        return str(filepath)

    def save_config_as_json(
        self,
        filepath: Union[str, Path],
        exclude: Optional[Union[str, List[str], Set[str], None]] = None,
        exclude_none: bool = True,
        only_include: Optional[Union[str, List[str], Set[str], None]] = None,
    ) -> str:
        def dumper(obj):
            return Composer.to_dict(obj) if isinstance(obj, DictConfig) else str(obj)

        config_to_save = self.export_config(
            exclude=exclude, exclude_none=exclude_none, only_include=only_include
        )
        logger.info("Saving config to %s", filepath)
        Composer.save_json(config_to_save, filepath, default=dumper)
        return str(filepath)

    def print_config(
        self,
    ):
        Composer.print(self.model_dump())

__setattr__(key, val)

Overrides the default setattr method to allow for custom property set methods.

Parameters:

Name Type Description Default
key str

The name of the attribute to set.

 required
val Any

The value to set the attribute to.

 required
Source code in hyfi/composer/config.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __setattr__(self, key, val):
    """
    Overrides the default __setattr__ method to allow for custom property set methods.

    Args:
        key (str): The name of the attribute to set.
        val (Any): The value to set the attribute to.
    """
    if method := self._property_set_methods_.get(key):  # type: ignore
        logger.debug(
            "Setting %s to %s",
            key,
            val if isinstance(val, (str, int)) else type(val),
        )
        getattr(self, method)(val)
    super().__setattr__(key, val)

export_config(exclude=None, exclude_none=True, only_include=None)

Export the configuration to a dictionary.

Parameters:

Name Type Description Default
exclude Optional[Union[str, List[str], Set[str], None]]

Keys to exclude from the saved configuration. Defaults to None.

 None
exclude_none bool

Whether to exclude keys with None values. Defaults to True.

 True
only_include Optional[Union[str, List[str], Set[str], None]]

Keys to include in the saved configuration. Defaults to None.

 None

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The configuration dictionary.
Source code in hyfi/composer/config.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def export_config(
    self,
    exclude: Optional[Union[str, List[str], Set[str], None]] = None,
    exclude_none: bool = True,
    only_include: Optional[Union[str, List[str], Set[str], None]] = None,
) -> Dict[str, Any]:
    """
    Export the configuration to a dictionary.

    Args:
        exclude (Optional[Union[str, List[str], Set[str], None]]): Keys to exclude from the saved configuration.
            Defaults to None.
        exclude_none (bool): Whether to exclude keys with None values. Defaults to True.
        only_include (Optional[Union[str, List[str], Set[str], None]]): Keys to include in the saved configuration.
            Defaults to None.

    Returns:
        Dict[str, Any]: The configuration dictionary.
    """
    if not exclude:
        exclude = self._exclude_  # type: ignore
    if isinstance(exclude, str):
        exclude = [exclude]
    if exclude is None:
        exclude = []
    if isinstance(only_include, str):
        only_include = [only_include]
    if only_include is None:
        only_include = []

    config = self.model_dump(exclude=exclude, exclude_none=exclude_none)  # type: ignore
    if only_include:
        config = {key: config[key] for key in only_include if key in config}

    return config

save_config(filepath, exclude=None, exclude_none=True, only_include=None)

Save the batch configuration to file.

Parameters:

Name Type Description Default
filepath [Union[str, Path]

The filepath to save the configuration to.

 required
exclude Optional[Union[str, List[str], Set[str], None]]

Keys to exclude from the saved configuration. Defaults to None.

 None
exclude_none bool

Whether to exclude keys with None values. Defaults to True.

 True
only_include Optional[Union[str, List[str], Set[str], None]]

Keys to include in the saved configuration. Defaults to None.

 None

Returns:

Name Type Description
str str

The filename of the saved configuration.
Source code in hyfi/composer/config.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def save_config(
    self,
    filepath: Union[str, Path],
    exclude: Optional[Union[str, List[str], Set[str], None]] = None,
    exclude_none: bool = True,
    only_include: Optional[Union[str, List[str], Set[str], None]] = None,
) -> str:
    """
    Save the batch configuration to file.

    Args:
        filepath ([Union[str, Path]): The filepath to save the configuration to.
        exclude (Optional[Union[str, List[str], Set[str], None]]): Keys to exclude from the saved configuration.
            Defaults to None.
        exclude_none (bool): Whether to exclude keys with None values. Defaults to True.
        only_include (Optional[Union[str, List[str], Set[str], None]]): Keys to include in the saved configuration.
            Defaults to None.

    Returns:
        str: The filename of the saved configuration.
    """
    logger.info("Saving config to %s", filepath)

    config_to_save = self.export_config(
        exclude=exclude, exclude_none=exclude_none, only_include=only_include
    )

    Composer.save(config_to_save, filepath)
    return str(filepath)

BaseModel

Bases: BaseModel

Base class for all Pydantic models.

Attributes:

Name Type Description
_config_name_ str

The name of the model.
_config_group_ str

The group of the model.
_auto_populate_ bool

Whether to auto-populate the model with defaults from the config.
_auto_generate_ bool

Whether to auto-generate the config for the model.
Source code in hyfi/composer/model.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class BaseModel(PydanticBaseModel):
    """
    Base class for all Pydantic models.

    Attributes:
        _config_name_ (str): The name of the model.
        _config_group_ (str): The group of the model.
        _auto_populate_ (bool): Whether to auto-populate the model with defaults from the config.
        _auto_generate_ (bool): Whether to auto-generate the config for the model.

    """

    _config_name_: str = "__init__"
    _config_group_: str = "/composer"
    _auto_populate_: bool = False
    _auto_generate_: bool = False
    _exclude_: Set[str] = set()
    _exclude_keys_ = {
        "_exclude_keys_": True,
        "_target_": True,
        "_config_name_": True,
        "_config_group_": True,
        "_auto_populate_": True,
        "_auto_generate_": True,
        "_exclude_": True,
    }

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        extra="allow",
        validate_assignment=False,
    )  # type: ignore

    @model_validator(mode="before")
    def validate_model_config_before(cls, data):
        # logger.debug("Validating model config before validating each field.")
        _auto_populate_ = data.get("_auto_populate_", getattr(cls._auto_populate_, "default", False))  # type: ignore
        if not _auto_populate_:
            if global_hyfi.verbosity > 1:
                logger.debug("Auto-populate is disabled for class `%s`.", cls.__name__)
            return data
        _config_name_ = data.get("_config_name_", getattr(cls._config_name_, "default", "__init__"))  # type: ignore
        _config_group_ = data.get("_config_group_", getattr(cls._config_group_, "default"))  # type: ignore
        _class_name_ = cls.__name__  # type: ignore
        if not _config_group_:
            if global_hyfi.verbosity > 0:
                logger.debug("There is no config group specified.")
            return data
        # Initialize the config with the given config_name.
        if global_hyfi.verbosity > 1:
            logger.info(
                "Composing `%s` class with `%s` config in `%s` group.",
                _class_name_,
                _config_name_,
                _config_group_,
            )
        config_group = f"{_config_group_}={_config_name_}"
        cfg = Composer.compose_as_dict(
            config_group=config_group,
            config_data=data,
            throw_on_compose_failure=False,
        )
        data = Composer.update(cfg, data)
        # Exclude any attributes specified in the class's `exclude` list.
        exclude = getattr(cls._exclude_, "default", set())  # type: ignore
        for name in exclude:
            if name in data:
                logger.debug("Excluding `%s` from the config.", name)
                del data[name]  # type: ignore
        return data

    @property
    def kwargs(self) -> Dict[str, Any]:
        """
        Returns the model as a dictionary excluding any keys specified in the class's `exclude_keys` list.
        """
        return self.model_dump(exclude=self._exclude_keys_)

    @property
    def config_name(self) -> str:
        """
        Returns the name of the model.
        """
        return self._config_name_

    @property
    def config_group(self) -> str:
        """
        Returns the group of the model.
        """
        return self._config_group_

    @property
    def auto_populate(self) -> bool:
        """
        Returns whether the model should be auto-populated.
        """
        return self._auto_populate_

    @property
    def auto_generate(self) -> bool:
        """
        Returns whether the model should be auto-generated.
        """
        return self._auto_generate_

    @classmethod
    def generate_config(
        cls,
        config_name: Optional[str] = None,
        config_path: str = None,
        config_root: Optional[str] = None,
        save: bool = True,
    ) -> Dict[str, Any]:
        """
        Saves a HyFI config for itself.

        Args:
            cls (BaseModel): The class to generate a config for.
            config_name (Optional[str]): The name of the config. If not provided, the name of the target will be used.
            config_path (Optional[str]): The path to save the config to (relative to the config root). Defaults to "run".
            config_root (Optional[str]): The root of the config path. If not provided, the global hyfi config directory will be used.
            **kwargs_for_target: Keyword arguments to pass to the target.
        """

        target = f"{cls.__module__}.{cls.__name__}"
        cfg = {
            "_target_": target,
            "_config_name_": getattr(cls._config_name_, "default"),
            "_config_group_": getattr(cls._config_group_, "default"),
        }

        model_fields = {
            key: getattr(value, "default") for key, value in cls.model_fields.items()
        }
        cfg.update(model_fields)
        cfg = cls.sanitized_config(cfg)

        if not save:
            return cfg

        config_name = (
            config_name or getattr(cls._config_name_, "default")
        ) or cls._config_name_
        _save_config(
            config=cfg,
            class_name=cls.__name__,
            config_group=getattr(cls._config_group_, "default"),
            config_name=config_name,
            config_path=config_path,
            config_root=config_root,
        )
        return cfg

    @staticmethod
    def sanitized_config(
        config: Dict[str, Any],
    ) -> Dict[str, Any]:
        """
        Converts a config to Hydra-supported type if necessary and possible.

        Args:
            config (Dict[str, Any]): The config to sanitize.

        Returns:
            Dict[str, Any]: The sanitized config.
        """
        defaults = []
        sanitized_config = {}
        _config = {}
        for key, value in config.items():
            if hasattr(value, "_config_group_") and hasattr(value, "_config_name_"):
                config_name = (
                    getattr(value, "model_extra", {}).get("_config_name_")
                    or value._config_name_
                )
                if value._config_group_ == key:
                    defaults.append({f"{value._config_group_}": config_name})
                else:
                    defaults.append({f"{value._config_group_}@{key}": config_name})
            else:
                value = sanitized_default_value(value)
                _config[key] = value
        if defaults:
            sanitized_config["defaults"] = defaults
        sanitized_config.update(_config)
        return sanitized_config

auto_generate: bool property

Returns whether the model should be auto-generated.

auto_populate: bool property

Returns whether the model should be auto-populated.

config_group: str property

Returns the group of the model.

config_name: str property

Returns the name of the model.

kwargs: Dict[str, Any] property

Returns the model as a dictionary excluding any keys specified in the class's exclude_keys list.

generate_config(config_name=None, config_path=None, config_root=None, save=True) classmethod

Saves a HyFI config for itself.

Parameters:

Name Type Description Default
cls BaseModel

The class to generate a config for.

 required
config_name Optional[str]

The name of the config. If not provided, the name of the target will be used.

 None
config_path Optional[str]

The path to save the config to (relative to the config root). Defaults to "run".

 None
config_root Optional[str]

The root of the config path. If not provided, the global hyfi config directory will be used.

 None
**kwargs_for_target

Keyword arguments to pass to the target.

 required
Source code in hyfi/composer/model.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@classmethod
def generate_config(
    cls,
    config_name: Optional[str] = None,
    config_path: str = None,
    config_root: Optional[str] = None,
    save: bool = True,
) -> Dict[str, Any]:
    """
    Saves a HyFI config for itself.

    Args:
        cls (BaseModel): The class to generate a config for.
        config_name (Optional[str]): The name of the config. If not provided, the name of the target will be used.
        config_path (Optional[str]): The path to save the config to (relative to the config root). Defaults to "run".
        config_root (Optional[str]): The root of the config path. If not provided, the global hyfi config directory will be used.
        **kwargs_for_target: Keyword arguments to pass to the target.
    """

    target = f"{cls.__module__}.{cls.__name__}"
    cfg = {
        "_target_": target,
        "_config_name_": getattr(cls._config_name_, "default"),
        "_config_group_": getattr(cls._config_group_, "default"),
    }

    model_fields = {
        key: getattr(value, "default") for key, value in cls.model_fields.items()
    }
    cfg.update(model_fields)
    cfg = cls.sanitized_config(cfg)

    if not save:
        return cfg

    config_name = (
        config_name or getattr(cls._config_name_, "default")
    ) or cls._config_name_
    _save_config(
        config=cfg,
        class_name=cls.__name__,
        config_group=getattr(cls._config_group_, "default"),
        config_name=config_name,
        config_path=config_path,
        config_root=config_root,
    )
    return cfg

sanitized_config(config) staticmethod

Converts a config to Hydra-supported type if necessary and possible.

Parameters:

Name Type Description Default
config Dict[str, Any]

The config to sanitize.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The sanitized config.
Source code in hyfi/composer/model.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@staticmethod
def sanitized_config(
    config: Dict[str, Any],
) -> Dict[str, Any]:
    """
    Converts a config to Hydra-supported type if necessary and possible.

    Args:
        config (Dict[str, Any]): The config to sanitize.

    Returns:
        Dict[str, Any]: The sanitized config.
    """
    defaults = []
    sanitized_config = {}
    _config = {}
    for key, value in config.items():
        if hasattr(value, "_config_group_") and hasattr(value, "_config_name_"):
            config_name = (
                getattr(value, "model_extra", {}).get("_config_name_")
                or value._config_name_
            )
            if value._config_group_ == key:
                defaults.append({f"{value._config_group_}": config_name})
            else:
                defaults.append({f"{value._config_group_}@{key}": config_name})
        else:
            value = sanitized_default_value(value)
            _config[key] = value
    if defaults:
        sanitized_config["defaults"] = defaults
    sanitized_config.update(_config)
    return sanitized_config

BaseSettings

Bases: BaseSettings, ENVs

Configuration class for environment variables in HyFI.

!! The variables are read-only and cannot be changed after initialization.

Source code in hyfi/composer/settings.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class BaseSettings(PydanticBaseSettings, ENVs):
    """
    Configuration class for environment variables in HyFI.

    !! The variables are read-only and cannot be changed after initialization.
    """

    """Environment variables for HyFI"""

    _config_name_: str = "__init__"
    _config_group_: str = "/env"

    DOTENV_FILENAME: Optional[str] = ".env"
    DOTENV_DIR: Optional[str] = None
    DOTENV_FILE: Optional[str] = None
    HYFI_SECRETS_DIR: Optional[str] = None

    model_config = SettingsConfigDict(
        env_prefix="",
        env_nested_delimiter="__",
        case_sentive=False,
        env_file=global_hyfi.dotenv_file,
        env_file_encoding="utf-8",
        validate_assignment=True,
        extra="allow",
        secrets_dir=global_hyfi.secrets_dir,
    )  # type: ignore

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: Type[PydanticBaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> Tuple[PydanticBaseSettingsSource, ...]:
        # BaseSettings.load_dotenv(dotenv_file=global_hyfi.dotenv_file)
        return (
            init_settings,
            file_secret_settings,
            DotEnvSettingsSource(settings_cls),
            env_settings,
        )

    @model_validator(mode="before")  # type: ignore
    def check_and_set_values(cls, data):
        if not isinstance(data, dict):
            return data
        env_file = os.environ.get("DOTENV_FILE", "")
        if os.path.isfile(env_file):
            data["DOTENV_FILE"] = env_file
            dotenv_filename = os.path.basename(env_file)
            dotenv_dir = os.path.dirname(env_file)
            data["DOTENV_DIR"] = dotenv_dir
            data["DOTENV_FILENAME"] = dotenv_filename
        data["HYFI_SECRETS_DIR"] = global_hyfi.secrets_dir
        return BaseSettings.check_and_set_osenv_vars(data)

    @property
    def os(self):
        """Returns the OS environment variables."""
        return os.environ

os property

Returns the OS environment variables.

Composer

Bases: UTILs, GENERATOR

Compose a configuration by applying overrides

Source code in hyfi/composer/composer.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
class Composer(UTILs, GENERATOR):
    """
    Compose a configuration by applying overrides
    """

    @staticmethod
    def hydra_compose(
        root_config_name: Optional[str] = None,
        config_module: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        plugins: Optional[List[str]] = None,
    ):
        is_initialized = GlobalHydra.instance().is_initialized()  # type: ignore
        config_module = config_module or global_hyfi.config_module
        plugins = plugins or global_hyfi.plugins

        overrides = overrides or []
        # by adding the variables=__init__ override,
        # we can access the variables in the config whenever we want
        override = "+variables=__init__"
        if override not in overrides:
            overrides.append(override)
            logger.debug(
                "Overriding `about` config group with `%s`",
                global_hyfi.package_name,
            )
        # logger.debug("config_module: %s", config_module)
        if is_initialized:
            # Hydra is already initialized.
            # logger.debug("Hydra is already initialized")
            cfg = hydra.compose(config_name=root_config_name, overrides=overrides)
        else:
            with hyfi_hydra.initialize_config(
                config_module=config_module,
                config_dir=global_hyfi.user_config_path,
                plugins=plugins,
                version_base=global_hyfi.hydra_version_base,
            ):
                cfg = hydra.compose(config_name=root_config_name, overrides=overrides)
        return cfg

    @staticmethod
    def compose_as_dict(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        throw_on_compose_failure: bool = True,
        throw_on_resolution_failure: bool = True,
        throw_on_missing: bool = False,
        root_config_name: Optional[str] = None,
        config_module: Optional[str] = None,
        global_package: bool = False,
        **kwargs,
    ) -> Dict:
        return Composer.to_dict(
            Composer.compose(
                config_group=config_group,
                overrides=overrides,
                config_data=config_data,
                throw_on_compose_failure=throw_on_compose_failure,
                throw_on_resolution_failure=throw_on_resolution_failure,
                throw_on_missing=throw_on_missing,
                root_config_name=root_config_name,
                config_module=config_module,
                global_package=global_package,
                **kwargs,
            )
        )

    @staticmethod
    def compose(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        throw_on_compose_failure: bool = True,
        throw_on_resolution_failure: bool = True,
        throw_on_missing: bool = False,
        root_config_name: Optional[str] = None,
        config_module: Optional[str] = None,
        global_package: bool = False,
        **kwargs,
    ) -> DictConfig:
        try:
            return Composer._compose_internal(
                config_group=config_group,
                overrides=overrides,
                config_data=config_data,
                throw_on_resolution_failure=throw_on_resolution_failure,
                throw_on_missing=throw_on_missing,
                root_config_name=root_config_name,
                config_module=config_module,
                global_package=global_package,
                **kwargs,
            )
        except Exception as e:
            logger.error("Error composing config: %s", e)
            if throw_on_compose_failure:
                raise e
            return DictConfig(config_data) if config_data else DictConfig({})

    @staticmethod
    def _compose_internal(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        throw_on_resolution_failure: bool = True,
        throw_on_missing: bool = False,
        root_config_name: Optional[str] = None,
        config_module: Optional[str] = None,
        global_package: bool = False,
        **kwargs,
    ) -> DictConfig:
        if isinstance(config_data, DictConfig):
            logger.debug("returning config_data without composing")
            return config_data
        cg = ConfigGroup(config_group=config_group)
        overrides = cg.get_overrides(
            overrides=overrides,
            root_config_name=root_config_name,
            config_module=config_module,
        )

        logger.debug(f"compose config with overrides: {overrides}")
        # Initialize hydra and return the configuration.
        cfg = Composer.hydra_compose(
            root_config_name=root_config_name,
            config_module=config_module,
            overrides=overrides,
        )
        # Add config group overrides to overrides list.
        global_package = global_package or cg.global_package
        if cg.group_key and not global_package:
            group_overrides: List[str] = []
            group_cfg = Composer.select(
                cfg,
                key=cg.group_key,
                default=None,
                throw_on_missing=False,
                throw_on_resolution_failure=False,
            )
            if config_data and group_cfg:
                group_overrides.extend(
                    (
                        f"{cg.group_key}.{k}='{v}'"
                        if isinstance(v, str)
                        else f"{cg.group_key}.{k}={v}"
                    )
                    for k, v in config_data.items()
                    if isinstance(v, (str, int, float, bool)) and k in group_cfg
                )
            if group_overrides:
                overrides.extend(group_overrides)
                cfg = Composer.hydra_compose(
                    root_config_name=root_config_name,
                    config_module=config_module,
                    overrides=overrides,
                )

            # Select the group_key from the configuration.
            cfg = Composer.select(
                cfg,
                key=cg.group_key,
                default=None,
                throw_on_missing=throw_on_missing,
                throw_on_resolution_failure=throw_on_resolution_failure,
            )
        return cfg

    @staticmethod
    def is_composable(
        config_group: str,
        config_module: Optional[str] = None,
    ) -> bool:
        """
        Determines whether the input configuration object is composable.

        Args:
            config_group (str): The name of the configuration group to check.
            config_module (Optional[str], optional): The name of the configuration module to check. Defaults to None.

        Returns:
            bool: True if the configuration object is composable, False otherwise.
        """
        try:
            cfg = Composer.compose(
                config_group=config_group,
                config_module=config_module,
            )
            return cfg is not None
        except Exception as e:
            logger.error("Error composing config: %s", e)
            return False

    @staticmethod
    def is_instantiatable(cfg: Any):
        """
        Determines whether the input configuration object is instantiatable.

        Args:
            cfg (Any): The configuration object to check.

        Returns:
            bool: True if the configuration object is instantiatable, False otherwise.
        """
        return Composer.is_config(cfg) and SpecialKeys.TARGET in cfg

    @staticmethod
    def replace_special_keys(_dict: Mapping[str, Any]) -> Mapping:
        """
        Replace special keys in a dictionary.

        Args:
            _dict (Mapping[str, Any]): The dictionary to update.

        Returns:
            Mapping: The updated dictionary.
        """
        _new_dict = {}
        for k, v in _dict.items():
            key = Composer.generate_alias_for_special_keys(k)
            if isinstance(v, collections.abc.Mapping):
                _new_dict[key] = Composer.replace_special_keys(v)
            else:
                _new_dict[key] = v
        return _new_dict

    @staticmethod
    def generate_alias_for_special_keys(key: str) -> str:
        """
        Generate an alias for special keys.
        _with_ -> run
        _pipe_ -> pipe_target
        _run_ -> run

        Args:
            key (str): The special key to generate an alias for.

        Returns:
            str: The alias for the special key.
        """
        # replace the exact `with`, `pipe` with `run_with`, `run_pipe`
        key_ = re.sub(r"^with$", "run", key)
        # replace the prefix `_` with `run_`
        key_ = re.sub(r"^_with_$", "run", key_)
        key_ = re.sub(r"^_pipe_$", "pipe_target", key_)
        key_ = re.sub(r"^_run_$", "run", key_)
        return key_

    @staticmethod
    def partial(
        config: Union[str, Dict],
        *args: Any,
        **kwargs: Any,
    ) -> Callable:
        """
        Returns a callable object that is a partial version of the function or class specified in the config object.

        Args:
            config: An config object describing what to call and what params to use.
                    In addition to the parameters, the config must contain:
                    _target_ : target class or callable name (str)
                    And may contain:
                    _partial_: If True, return functools.partial wrapped method or object
                                False by default. Configure per target.
            args: Optional positional parameters pass-through
            kwargs: Optional named parameters to override
                    parameters in the config object. Parameters not present
                    in the config objects are being passed as is to the target.
                    IMPORTANT: dataclasses instances in kwargs are interpreted as config
                                and cannot be used as passthrough
        Returns:
            A callable object that is a partial version of the function or class specified in the config object.
        """
        if isinstance(config, str):
            config = {SpecialKeys.TARGET.value: config}
        else:
            config = Composer.to_dict(config)
        if not isinstance(config, dict):
            raise ValueError("config must be a dict or a str")
        config[SpecialKeys.PARTIAL.value] = True
        rc_kwargs_ = config.pop(SpecialKeys.KWARGS, {})
        if rc_kwargs_ and kwargs:
            kwargs.update(rc_kwargs_)
        return Composer.instantiate(config, *args, **kwargs)

    @staticmethod
    def instantiate(config: Any, *args: Any, **kwargs: Any) -> Any:
        """
        Instantiates an object using the provided config object.

        Args:
            config: An config object describing what to call and what params to use.
                    In addition to the parameters, the config must contain:
                    _target_ : target class or callable name (str)
                    And may contain:
                    _args_: List-like of positional arguments to pass to the target
                    _recursive_: Construct nested objects as well (bool).
                                    False by default.
                                    may be overridden via a _recursive_ key in
                                    the kwargs
                    _convert_: Conversion strategy
                            none    : Passed objects are DictConfig and ListConfig, default
                            partial : Passed objects are converted to dict and list, with
                                    the exception of Structured Configs (and their fields).
                            all     : Passed objects are dicts, lists and primitives without
                                    a trace of OmegaConf containers
                    _partial_: If True, return functools.partial wrapped method or object
                                False by default. Configure per target.
                    _args_: List-like of positional arguments
            args: Optional positional parameters pass-through
            kwargs: Optional named parameters to override
                    parameters in the config object. Parameters not present
                    in the config objects are being passed as is to the target.
                    IMPORTANT: dataclasses instances in kwargs are interpreted as config
                                and cannot be used as passthrough

        Returns:
            if _target_ is a class name: the instantiated object
            if _target_ is a callable: the return value of the call
        """
        verbose = config.get("verbose", False)
        if not Composer.is_instantiatable(config):
            if verbose:
                logger.info("Config is not instantiatable, returning config")
            return config
        _recursive_ = config.get(SpecialKeys.RECURSIVE, False)
        if SpecialKeys.RECURSIVE not in kwargs:
            kwargs[SpecialKeys.RECURSIVE.value] = _recursive_
        if verbose:
            logger.info("instantiating %s ...", config.get(SpecialKeys.TARGET))
        return hydra.utils.instantiate(config, *args, **kwargs)

    @staticmethod
    def getsource(obj: Any) -> str:
        """
        Return the source code of the object.

        Args:
            obj: The object to get the source code of.

        Returns:
            The source code of the object as a string.

        """
        try:
            target_string = ""
            if Composer.is_config(obj):
                if SpecialKeys.TARGET in obj:
                    target_string = obj[SpecialKeys.TARGET]
            elif isinstance(obj, str):
                target_string = obj
            return UTILs.getsource(target_string) if target_string else ""
        except Exception as e:
            logger.error(f"Error getting source: {e}")
            return ""

    @staticmethod
    def viewsource(obj: Any):
        """
        Print the source code of the object.

        Args:
            obj: The object to print the source code of.

        """
        print(Composer.getsource(obj))

    @staticmethod
    def instantiate_config(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        global_package: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """
        Instantiates an object using the provided config group and overrides.

        Args:
            config_group: Name of the config group to compose (`config_group=name`)
            overrides: List of config groups to apply overrides to (`overrides=["override_name"]`)
            config_data: Keyword arguments to override config group values (will be converted to overrides of the form `config_group_name.key=value`)
            global_package: If True, the config assumed to be a global package
            args: Optional positional parameters pass-through
            kwargs: Optional named parameters to override
                    parameters in the config object. Parameters not present
                    in the config objects are being passed as is to the target.
                    IMPORTANT: dataclasses instances in kwargs are interpreted as config
                                and cannot be used as passthrough

        Returns:
            if _target_ is a class name: the instantiated object
            if _target_ is a callable: the return value of the call
        """
        cfg = Composer.compose(
            config_group=config_group,
            overrides=overrides,
            config_data=config_data,
            global_package=global_package,
        )
        return Composer.instantiate(cfg, *args, **kwargs)

    @staticmethod
    def print_config(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        global_package: bool = False,
    ):
        """
        Print the configuration

        Args:
            config_group: Name of the config group to compose (`config_group=name`)
            overrides: List of config groups to apply overrides to (`overrides=["override_name"]`)
            config_data: Keyword arguments to override config group values (will be converted to overrides of the form `config_group_name.key=value`)
            global_package: If True, the config assumed to be a global package
        """
        cfg = Composer.compose(
            config_group=config_group,
            overrides=overrides,
            config_data=config_data,
            global_package=global_package,
        )
        Composer.print(cfg)

generate_alias_for_special_keys(key) staticmethod

Generate an alias for special keys. with -> run pipe -> pipe_target run -> run

Parameters:

Name Type Description Default
key str

The special key to generate an alias for.

 required

Returns:

Name Type Description
str str

The alias for the special key.
Source code in hyfi/composer/composer.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
@staticmethod
def generate_alias_for_special_keys(key: str) -> str:
    """
    Generate an alias for special keys.
    _with_ -> run
    _pipe_ -> pipe_target
    _run_ -> run

    Args:
        key (str): The special key to generate an alias for.

    Returns:
        str: The alias for the special key.
    """
    # replace the exact `with`, `pipe` with `run_with`, `run_pipe`
    key_ = re.sub(r"^with$", "run", key)
    # replace the prefix `_` with `run_`
    key_ = re.sub(r"^_with_$", "run", key_)
    key_ = re.sub(r"^_pipe_$", "pipe_target", key_)
    key_ = re.sub(r"^_run_$", "run", key_)
    return key_

getsource(obj) staticmethod

Return the source code of the object.

Parameters:

Name Type Description Default
obj Any

The object to get the source code of.

 required

Returns:

Type Description
str

The source code of the object as a string.
Source code in hyfi/composer/composer.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
@staticmethod
def getsource(obj: Any) -> str:
    """
    Return the source code of the object.

    Args:
        obj: The object to get the source code of.

    Returns:
        The source code of the object as a string.

    """
    try:
        target_string = ""
        if Composer.is_config(obj):
            if SpecialKeys.TARGET in obj:
                target_string = obj[SpecialKeys.TARGET]
        elif isinstance(obj, str):
            target_string = obj
        return UTILs.getsource(target_string) if target_string else ""
    except Exception as e:
        logger.error(f"Error getting source: {e}")
        return ""

instantiate(config, *args, **kwargs) staticmethod

Instantiates an object using the provided config object.

Parameters:

Name Type Description Default
config Any

An config object describing what to call and what params to use. In addition to the parameters, the config must contain: target : target class or callable name (str) And may contain: args: List-like of positional arguments to pass to the target recursive: Construct nested objects as well (bool). False by default. may be overridden via a recursive key in the kwargs convert: Conversion strategy none : Passed objects are DictConfig and ListConfig, default partial : Passed objects are converted to dict and list, with the exception of Structured Configs (and their fields). all : Passed objects are dicts, lists and primitives without a trace of OmegaConf containers partial: If True, return functools.partial wrapped method or object False by default. Configure per target. args: List-like of positional arguments

 required
args Any

Optional positional parameters pass-through

 ()
kwargs Any

Optional named parameters to override parameters in the config object. Parameters not present in the config objects are being passed as is to the target. IMPORTANT: dataclasses instances in kwargs are interpreted as config and cannot be used as passthrough

 {}

Returns:

Type Description
Any

if target is a class name: the instantiated object
Any

if target is a callable: the return value of the call
Source code in hyfi/composer/composer.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
@staticmethod
def instantiate(config: Any, *args: Any, **kwargs: Any) -> Any:
    """
    Instantiates an object using the provided config object.

    Args:
        config: An config object describing what to call and what params to use.
                In addition to the parameters, the config must contain:
                _target_ : target class or callable name (str)
                And may contain:
                _args_: List-like of positional arguments to pass to the target
                _recursive_: Construct nested objects as well (bool).
                                False by default.
                                may be overridden via a _recursive_ key in
                                the kwargs
                _convert_: Conversion strategy
                        none    : Passed objects are DictConfig and ListConfig, default
                        partial : Passed objects are converted to dict and list, with
                                the exception of Structured Configs (and their fields).
                        all     : Passed objects are dicts, lists and primitives without
                                a trace of OmegaConf containers
                _partial_: If True, return functools.partial wrapped method or object
                            False by default. Configure per target.
                _args_: List-like of positional arguments
        args: Optional positional parameters pass-through
        kwargs: Optional named parameters to override
                parameters in the config object. Parameters not present
                in the config objects are being passed as is to the target.
                IMPORTANT: dataclasses instances in kwargs are interpreted as config
                            and cannot be used as passthrough

    Returns:
        if _target_ is a class name: the instantiated object
        if _target_ is a callable: the return value of the call
    """
    verbose = config.get("verbose", False)
    if not Composer.is_instantiatable(config):
        if verbose:
            logger.info("Config is not instantiatable, returning config")
        return config
    _recursive_ = config.get(SpecialKeys.RECURSIVE, False)
    if SpecialKeys.RECURSIVE not in kwargs:
        kwargs[SpecialKeys.RECURSIVE.value] = _recursive_
    if verbose:
        logger.info("instantiating %s ...", config.get(SpecialKeys.TARGET))
    return hydra.utils.instantiate(config, *args, **kwargs)

instantiate_config(config_group=None, overrides=None, config_data=None, global_package=False, *args, **kwargs) staticmethod

Instantiates an object using the provided config group and overrides.

Parameters:

Name Type Description Default
config_group Optional[str]

Name of the config group to compose (config_group=name)

 None
overrides Optional[List[str]]

List of config groups to apply overrides to (overrides=["override_name"])

 None
config_data Optional[Union[Dict[str, Any], DictConfig]]

Keyword arguments to override config group values (will be converted to overrides of the form config_group_name.key=value)

 None
global_package bool

If True, the config assumed to be a global package

 False
args Any

Optional positional parameters pass-through

 ()
kwargs Any

Optional named parameters to override parameters in the config object. Parameters not present in the config objects are being passed as is to the target. IMPORTANT: dataclasses instances in kwargs are interpreted as config and cannot be used as passthrough

 {}

Returns:

Type Description
Any

if target is a class name: the instantiated object
Any

if target is a callable: the return value of the call
Source code in hyfi/composer/composer.py 
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
@staticmethod
def instantiate_config(
    config_group: Optional[str] = None,
    overrides: Optional[List[str]] = None,
    config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
    global_package: bool = False,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """
    Instantiates an object using the provided config group and overrides.

    Args:
        config_group: Name of the config group to compose (`config_group=name`)
        overrides: List of config groups to apply overrides to (`overrides=["override_name"]`)
        config_data: Keyword arguments to override config group values (will be converted to overrides of the form `config_group_name.key=value`)
        global_package: If True, the config assumed to be a global package
        args: Optional positional parameters pass-through
        kwargs: Optional named parameters to override
                parameters in the config object. Parameters not present
                in the config objects are being passed as is to the target.
                IMPORTANT: dataclasses instances in kwargs are interpreted as config
                            and cannot be used as passthrough

    Returns:
        if _target_ is a class name: the instantiated object
        if _target_ is a callable: the return value of the call
    """
    cfg = Composer.compose(
        config_group=config_group,
        overrides=overrides,
        config_data=config_data,
        global_package=global_package,
    )
    return Composer.instantiate(cfg, *args, **kwargs)

is_composable(config_group, config_module=None) staticmethod

Determines whether the input configuration object is composable.

Parameters:

Name Type Description Default
config_group str

The name of the configuration group to check.

 required
config_module Optional[str]

The name of the configuration module to check. Defaults to None.

 None

Returns:

Name Type Description
bool bool

True if the configuration object is composable, False otherwise.
Source code in hyfi/composer/composer.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
@staticmethod
def is_composable(
    config_group: str,
    config_module: Optional[str] = None,
) -> bool:
    """
    Determines whether the input configuration object is composable.

    Args:
        config_group (str): The name of the configuration group to check.
        config_module (Optional[str], optional): The name of the configuration module to check. Defaults to None.

    Returns:
        bool: True if the configuration object is composable, False otherwise.
    """
    try:
        cfg = Composer.compose(
            config_group=config_group,
            config_module=config_module,
        )
        return cfg is not None
    except Exception as e:
        logger.error("Error composing config: %s", e)
        return False

is_instantiatable(cfg) staticmethod

Determines whether the input configuration object is instantiatable.

Parameters:

Name Type Description Default
cfg Any

The configuration object to check.

 required

Returns:

Name Type Description
bool

True if the configuration object is instantiatable, False otherwise.
Source code in hyfi/composer/composer.py 
327
328
329
330
331
332
333
334
335
336
337
338
@staticmethod
def is_instantiatable(cfg: Any):
    """
    Determines whether the input configuration object is instantiatable.

    Args:
        cfg (Any): The configuration object to check.

    Returns:
        bool: True if the configuration object is instantiatable, False otherwise.
    """
    return Composer.is_config(cfg) and SpecialKeys.TARGET in cfg

partial(config, *args, **kwargs) staticmethod

Returns a callable object that is a partial version of the function or class specified in the config object.

Parameters:

Name Type Description Default
config Union[str, Dict]

An config object describing what to call and what params to use. In addition to the parameters, the config must contain: target : target class or callable name (str) And may contain: partial: If True, return functools.partial wrapped method or object False by default. Configure per target.

 required
args Any

Optional positional parameters pass-through

 ()
kwargs Any

Optional named parameters to override parameters in the config object. Parameters not present in the config objects are being passed as is to the target. IMPORTANT: dataclasses instances in kwargs are interpreted as config and cannot be used as passthrough

 {}

Returns: A callable object that is a partial version of the function or class specified in the config object.

Source code in hyfi/composer/composer.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
@staticmethod
def partial(
    config: Union[str, Dict],
    *args: Any,
    **kwargs: Any,
) -> Callable:
    """
    Returns a callable object that is a partial version of the function or class specified in the config object.

    Args:
        config: An config object describing what to call and what params to use.
                In addition to the parameters, the config must contain:
                _target_ : target class or callable name (str)
                And may contain:
                _partial_: If True, return functools.partial wrapped method or object
                            False by default. Configure per target.
        args: Optional positional parameters pass-through
        kwargs: Optional named parameters to override
                parameters in the config object. Parameters not present
                in the config objects are being passed as is to the target.
                IMPORTANT: dataclasses instances in kwargs are interpreted as config
                            and cannot be used as passthrough
    Returns:
        A callable object that is a partial version of the function or class specified in the config object.
    """
    if isinstance(config, str):
        config = {SpecialKeys.TARGET.value: config}
    else:
        config = Composer.to_dict(config)
    if not isinstance(config, dict):
        raise ValueError("config must be a dict or a str")
    config[SpecialKeys.PARTIAL.value] = True
    rc_kwargs_ = config.pop(SpecialKeys.KWARGS, {})
    if rc_kwargs_ and kwargs:
        kwargs.update(rc_kwargs_)
    return Composer.instantiate(config, *args, **kwargs)

print_config(config_group=None, overrides=None, config_data=None, global_package=False) staticmethod

Print the configuration

Parameters:

Name Type Description Default
config_group Optional[str]

Name of the config group to compose (config_group=name)

 None
overrides Optional[List[str]]

List of config groups to apply overrides to (overrides=["override_name"])

 None
config_data Optional[Union[Dict[str, Any], DictConfig]]

Keyword arguments to override config group values (will be converted to overrides of the form config_group_name.key=value)

 None
global_package bool

If True, the config assumed to be a global package

 False
Source code in hyfi/composer/composer.py 
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
@staticmethod
def print_config(
    config_group: Optional[str] = None,
    overrides: Optional[List[str]] = None,
    config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
    global_package: bool = False,
):
    """
    Print the configuration

    Args:
        config_group: Name of the config group to compose (`config_group=name`)
        overrides: List of config groups to apply overrides to (`overrides=["override_name"]`)
        config_data: Keyword arguments to override config group values (will be converted to overrides of the form `config_group_name.key=value`)
        global_package: If True, the config assumed to be a global package
    """
    cfg = Composer.compose(
        config_group=config_group,
        overrides=overrides,
        config_data=config_data,
        global_package=global_package,
    )
    Composer.print(cfg)

replace_special_keys(_dict) staticmethod

Replace special keys in a dictionary.

Parameters:

Name Type Description Default
_dict Mapping[str, Any]

The dictionary to update.

 required

Returns:

Name Type Description
Mapping Mapping

The updated dictionary.
Source code in hyfi/composer/composer.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
@staticmethod
def replace_special_keys(_dict: Mapping[str, Any]) -> Mapping:
    """
    Replace special keys in a dictionary.

    Args:
        _dict (Mapping[str, Any]): The dictionary to update.

    Returns:
        Mapping: The updated dictionary.
    """
    _new_dict = {}
    for k, v in _dict.items():
        key = Composer.generate_alias_for_special_keys(k)
        if isinstance(v, collections.abc.Mapping):
            _new_dict[key] = Composer.replace_special_keys(v)
        else:
            _new_dict[key] = v
    return _new_dict

viewsource(obj) staticmethod

Print the source code of the object.

Parameters:

Name Type Description Default
obj Any

The object to print the source code of.

 required
Source code in hyfi/composer/composer.py 
490
491
492
493
494
495
496
497
498
499
@staticmethod
def viewsource(obj: Any):
    """
    Print the source code of the object.

    Args:
        obj: The object to print the source code of.

    """
    print(Composer.getsource(obj))

DocGenerator

Bases: BaseModel

A class for generating reference documentation and configuration documentation.

This class provides methods for generating reference documentation for modules and configuration documentation for directories.

Attributes:

Name Type Description
_config_name_ str

The name of the configuration.
_config_group_ str

The group of the configuration.
config_docs_dirname str

The name of the directory for configuration documentation.
reference_docs_dirname str

The name of the directory for reference documentation.
exclude_configs List[str]

A list of configurations to exclude.
exclude_references List[str]

A list of references to exclude.
Properties

root_dir: The root directory. package_dir: The package directory. package_name: The package name. config_dir: The configuration directory. config_docs_dir: The directory for configuration documentation. reference_docs_dir: The directory for reference documentation.

Methods:

Name Description
generate_reference_docs

Generates reference documentation for modules.
write_ref_doc

Writes reference documentation for a module.
generate_config_docs

Generates configuration documentation for directories.
write_config_doc

Writes configuration documentation for a directory.
Example 
doc_generator = DocGenerator()
doc_generator.generate_reference_docs()
doc_generator.generate_config_docs()
Source code in hyfi/composer/docs.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class DocGenerator(BaseModel):
    """
    A class for generating reference documentation and configuration documentation.

    This class provides methods for generating reference documentation for modules and configuration documentation for directories.

    Attributes:
        _config_name_: The name of the configuration.
        _config_group_: The group of the configuration.
        config_docs_dirname: The name of the directory for configuration documentation.
        reference_docs_dirname: The name of the directory for reference documentation.
        exclude_configs: A list of configurations to exclude.
        exclude_references: A list of references to exclude.

    Properties:
        root_dir: The root directory.
        package_dir: The package directory.
        package_name: The package name.
        config_dir: The configuration directory.
        config_docs_dir: The directory for configuration documentation.
        reference_docs_dir: The directory for reference documentation.

    Methods:
        generate_reference_docs: Generates reference documentation for modules.
        write_ref_doc: Writes reference documentation for a module.
        generate_config_docs: Generates configuration documentation for directories.
        write_config_doc: Writes configuration documentation for a directory.

    Example:
        ```python
        doc_generator = DocGenerator()
        doc_generator.generate_reference_docs()
        doc_generator.generate_config_docs()
        ```
    """

    _config_name_: str = "__init__"
    _config_group_: str = "/docs"

    config_docs_dirname: str = "docs/configs"
    reference_docs_dirname: str = "docs/reference"

    exclude_configs: List[str] = [
        "docs",
        "hydra",
    ]
    exclude_references: List[str] = [
        "conf",
        "__cli__.py",
        "__click__.py",
        "__init__.py",
        "_version.py",
        "__pycache__",
    ]

    def __call__(self) -> None:
        self.generate_config_docs()
        self.generate_reference_docs()

    @property
    def root_dir(self) -> Path:
        return Path(ENVs.getcwd())

    @property
    def package_dir(self) -> Path:
        return Path(global_hyfi.package_path)

    @property
    def package_name(self) -> str:
        return global_hyfi.package_name

    @property
    def config_dir(self) -> Path:
        return self.package_dir / global_hyfi.config_dirname

    @property
    def config_docs_dir(self) -> Path:
        path_ = self.root_dir / self.config_docs_dirname
        path_.mkdir(parents=True, exist_ok=True)
        return path_

    @property
    def reference_docs_dir(self) -> Path:
        path_ = self.root_dir / self.reference_docs_dirname
        path_.mkdir(parents=True, exist_ok=True)
        return path_

    def generate_reference_docs(self):
        exclude_refs = self.exclude_references or []
        logger.info("Generating reference documentation excluding %s", exclude_refs)
        for _path in self.package_dir.iterdir():
            module_name = _path.name
            if _path.is_file() or module_name in exclude_refs:
                continue
            self.write_ref_doc(_path)
            # for file in _path.iterdir():
            #     if file.name in exclude_refs:
            #         continue
            #     self.write_ref_doc(file)

    def write_ref_doc(self, module_path: Path):
        module_name = module_path.relative_to(self.package_dir)
        if module_path.is_dir():
            ref_file = self.reference_docs_dir / str(module_name) / "index.md"
        else:
            module_name = module_name.with_suffix("")
            ref_file = self.reference_docs_dir / f"{module_name}.md"
        module_name = module_name.as_posix().replace("/", ".")
        module_name = f"{self.package_name}.{module_name}"
        ref_file.parent.mkdir(parents=True, exist_ok=True)
        logger.info(
            "Writing reference documentation for %s to %s", module_name, ref_file
        )
        with open(ref_file, "w") as f:
            f.write(f"# `{module_name}`\n\n")
            f.write(f"::: {module_name}\n")

    def generate_config_docs(self):
        exclude_configs = self.exclude_configs or []
        logger.info(
            "Generating configuration documentation excluding %s", exclude_configs
        )
        for config_path in sorted(self.config_dir.iterdir()):
            if (
                config_path.is_file()
                or config_path.name in exclude_configs
                or config_path.name.startswith("__")
            ):
                continue
            self.write_config_doc(config_path)

    def write_config_doc(self, config_path: Path):
        dirname = config_path.name
        output_file = self.config_docs_dir / f"{dirname}.md"
        logger.info(
            "Writing configuration documentation for %s to %s", dirname, output_file
        )
        with open(output_file, "w") as f:
            f.write(f"# {dirname}\n\n")
            f.write(f"Config location: `conf/{dirname}`\n\n")
            for file in sorted(config_path.iterdir()):
                if file.is_dir():
                    continue
                if file.suffix == ".yaml":
                    f.write(f"## `{file.name}`\n\n")
                    f.write("```yaml\n")
                    lvl = self.config_docs_dir.relative_to(self.root_dir)
                    lvl = "/".join([".."] * len(lvl.as_posix().split("/")))
                    rel_path = f"{lvl}/{file.relative_to(self.root_dir)}"
                    f.write("{% include '" + str(rel_path) + "' %}")
                    f.write("\n```\n\n")

GENERATOR

Generates Hydra configs for functions and classes.

Source code in hyfi/composer/generator.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class GENERATOR:
    """
    Generates Hydra configs for functions and classes.
    """

    @staticmethod
    def generate_pipe_config(
        target: Callable,
        pipe_target_type: PipeTargetTypes = PipeTargetTypes.GENERAL_EXTERNAL_FUNCS,
        use_pipe_obj: bool = True,
        pipe_obj_arg_name: Optional[str] = None,
        return_pipe_obj: bool = False,
        pipe_prefix: Optional[str] = None,
        config_name: Optional[str] = None,
        config_root: Optional[str] = None,
        **kwargs_for_target,
    ) -> str:
        """
        Generates HyFI pipe config for a given target.

        Args:
            target: Target function or class.
            pipe_target_type: Type of target function or class.
            use_pipe_obj: Whether to use pipe object as the first argument.
            pipe_obj_arg_name: Name of the pipe object argument.
            return_pipe_obj: Whether to return pipe object.
            pipe_prefix: Prefix for pipe object argument.
            config_name: Name of the config.
            config_root: Root of the config.
            **kwargs_for_target: Keyword arguments for the target.
        """
        use_first_arg_as_pipe_obj = not pipe_obj_arg_name and use_pipe_obj
        config_name = config_name or target.__name__
        config_name = f"{pipe_prefix}_{config_name}" if pipe_prefix else config_name
        config_root = config_root or global_hyfi.config_root

        run_config_name = GENERATOR.generate_callable_config(
            target,
            config_name=config_name,
            config_path="run",
            config_root=config_root,
            use_first_arg_as_pipe_obj=use_first_arg_as_pipe_obj,
            **kwargs_for_target,
        )

        cfg = {
            "defaults": [
                pipe_target_type.value,
                {"/run": run_config_name},
            ],
            "use_pipe_obj": use_pipe_obj,
            "pipe_obj_arg_name": pipe_obj_arg_name,
            "return_pipe_obj": return_pipe_obj,
        }

        filename = f"{config_name}.yaml"
        config_path = Path(config_root) / "pipe"
        config_path.mkdir(parents=True, exist_ok=True)
        config_path /= filename
        CONFs.save(cfg, config_path)
        logger.info(f"Saved HyFI pipe config for {target.__name__} to {config_path}")

        return config_name

    @staticmethod
    def generate_callable_config(
        target: Callable,
        config_name: Optional[str] = None,
        config_path: str = "run",
        config_root: Optional[str] = None,
        use_first_arg_as_pipe_obj: bool = False,
        **kwargs_for_target,
    ) -> str:
        """
        Saves a HyFI config to a file.

        Args:
            target (Callable): The function or class to generate a config for.
            use_first_arg_as_pipe_obj (bool): Whether to use the first argument as the pipe object.
            config_name (Optional[str]): The name of the config. If not provided, the name of the target will be used.
            config_path (Optional[str]): The path to save the config to (relative to the config root). Defaults to "run".
            config_root (Optional[str]): The root of the config path. If not provided, the global hyfi config directory will be used.
            **kwargs_for_target: Keyword arguments to pass to the target.
        """
        cfg = GENERATOR.generate_target_config(
            target,
            remove_first_arg=use_first_arg_as_pipe_obj,
            **kwargs_for_target,
        )
        config_name = config_name or target.__name__
        filename = f"{config_name}.yaml"
        config_root = config_root or global_hyfi.config_root
        config_path = Path(config_root) / config_path
        config_path.mkdir(parents=True, exist_ok=True)
        config_path /= filename
        CONFs.save(cfg, config_path)
        logger.info(f"Saved HyFI config for {target.__name__} to {config_path}")
        return config_name

    @staticmethod
    def generate_target_config(
        target: Callable,
        remove_first_arg: bool = False,
        **kwargs,
    ) -> Dict[str, Any]:
        """
        Generates a HyFI config for a given target.

        Args:
            target (Callable): The function or class to generate a config for.
            remove_first_arg (bool): Whether to remove the first argument from the config.
            **kwargs: Keyword arguments to pass to the target.
        """
        params = inspect.signature(target).parameters

        config_dict = {
            "_target_": f"{target.__module__}.{target.__qualname__}",
        }

        for i, (key, param) in enumerate(params.items()):
            if remove_first_arg and i == 0:
                continue
            if param.kind == inspect.Parameter.VAR_KEYWORD:
                continue
            if key in kwargs:
                value = kwargs[key]
            else:
                value = (
                    None if param.default == inspect.Parameter.empty else param.default
                )
            config_dict[key] = sanitized_default_value(value)
        return config_dict

generate_callable_config(target, config_name=None, config_path='run', config_root=None, use_first_arg_as_pipe_obj=False, **kwargs_for_target) staticmethod

Saves a HyFI config to a file.

Parameters:

Name Type Description Default
target Callable

The function or class to generate a config for.

 required
use_first_arg_as_pipe_obj bool

Whether to use the first argument as the pipe object.

 False
config_name Optional[str]

The name of the config. If not provided, the name of the target will be used.

 None
config_path Optional[str]

The path to save the config to (relative to the config root). Defaults to "run".

 'run'
config_root Optional[str]

The root of the config path. If not provided, the global hyfi config directory will be used.

 None
**kwargs_for_target

Keyword arguments to pass to the target.

 {}
Source code in hyfi/composer/generator.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@staticmethod
def generate_callable_config(
    target: Callable,
    config_name: Optional[str] = None,
    config_path: str = "run",
    config_root: Optional[str] = None,
    use_first_arg_as_pipe_obj: bool = False,
    **kwargs_for_target,
) -> str:
    """
    Saves a HyFI config to a file.

    Args:
        target (Callable): The function or class to generate a config for.
        use_first_arg_as_pipe_obj (bool): Whether to use the first argument as the pipe object.
        config_name (Optional[str]): The name of the config. If not provided, the name of the target will be used.
        config_path (Optional[str]): The path to save the config to (relative to the config root). Defaults to "run".
        config_root (Optional[str]): The root of the config path. If not provided, the global hyfi config directory will be used.
        **kwargs_for_target: Keyword arguments to pass to the target.
    """
    cfg = GENERATOR.generate_target_config(
        target,
        remove_first_arg=use_first_arg_as_pipe_obj,
        **kwargs_for_target,
    )
    config_name = config_name or target.__name__
    filename = f"{config_name}.yaml"
    config_root = config_root or global_hyfi.config_root
    config_path = Path(config_root) / config_path
    config_path.mkdir(parents=True, exist_ok=True)
    config_path /= filename
    CONFs.save(cfg, config_path)
    logger.info(f"Saved HyFI config for {target.__name__} to {config_path}")
    return config_name

generate_pipe_config(target, pipe_target_type=PipeTargetTypes.GENERAL_EXTERNAL_FUNCS, use_pipe_obj=True, pipe_obj_arg_name=None, return_pipe_obj=False, pipe_prefix=None, config_name=None, config_root=None, **kwargs_for_target) staticmethod

Generates HyFI pipe config for a given target.

Parameters:

Name Type Description Default
target Callable

Target function or class.

 required
pipe_target_type PipeTargetTypes

Type of target function or class.

 GENERAL_EXTERNAL_FUNCS
use_pipe_obj bool

Whether to use pipe object as the first argument.

 True
pipe_obj_arg_name Optional[str]

Name of the pipe object argument.

 None
return_pipe_obj bool

Whether to return pipe object.

 False
pipe_prefix Optional[str]

Prefix for pipe object argument.

 None
config_name Optional[str]

Name of the config.

 None
config_root Optional[str]

Root of the config.

 None
**kwargs_for_target

Keyword arguments for the target.

 {}
Source code in hyfi/composer/generator.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@staticmethod
def generate_pipe_config(
    target: Callable,
    pipe_target_type: PipeTargetTypes = PipeTargetTypes.GENERAL_EXTERNAL_FUNCS,
    use_pipe_obj: bool = True,
    pipe_obj_arg_name: Optional[str] = None,
    return_pipe_obj: bool = False,
    pipe_prefix: Optional[str] = None,
    config_name: Optional[str] = None,
    config_root: Optional[str] = None,
    **kwargs_for_target,
) -> str:
    """
    Generates HyFI pipe config for a given target.

    Args:
        target: Target function or class.
        pipe_target_type: Type of target function or class.
        use_pipe_obj: Whether to use pipe object as the first argument.
        pipe_obj_arg_name: Name of the pipe object argument.
        return_pipe_obj: Whether to return pipe object.
        pipe_prefix: Prefix for pipe object argument.
        config_name: Name of the config.
        config_root: Root of the config.
        **kwargs_for_target: Keyword arguments for the target.
    """
    use_first_arg_as_pipe_obj = not pipe_obj_arg_name and use_pipe_obj
    config_name = config_name or target.__name__
    config_name = f"{pipe_prefix}_{config_name}" if pipe_prefix else config_name
    config_root = config_root or global_hyfi.config_root

    run_config_name = GENERATOR.generate_callable_config(
        target,
        config_name=config_name,
        config_path="run",
        config_root=config_root,
        use_first_arg_as_pipe_obj=use_first_arg_as_pipe_obj,
        **kwargs_for_target,
    )

    cfg = {
        "defaults": [
            pipe_target_type.value,
            {"/run": run_config_name},
        ],
        "use_pipe_obj": use_pipe_obj,
        "pipe_obj_arg_name": pipe_obj_arg_name,
        "return_pipe_obj": return_pipe_obj,
    }

    filename = f"{config_name}.yaml"
    config_path = Path(config_root) / "pipe"
    config_path.mkdir(parents=True, exist_ok=True)
    config_path /= filename
    CONFs.save(cfg, config_path)
    logger.info(f"Saved HyFI pipe config for {target.__name__} to {config_path}")

    return config_name

generate_target_config(target, remove_first_arg=False, **kwargs) staticmethod

Generates a HyFI config for a given target.

Parameters:

Name Type Description Default
target Callable

The function or class to generate a config for.

 required
remove_first_arg bool

Whether to remove the first argument from the config.

 False
**kwargs

Keyword arguments to pass to the target.

 {}
Source code in hyfi/composer/generator.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@staticmethod
def generate_target_config(
    target: Callable,
    remove_first_arg: bool = False,
    **kwargs,
) -> Dict[str, Any]:
    """
    Generates a HyFI config for a given target.

    Args:
        target (Callable): The function or class to generate a config for.
        remove_first_arg (bool): Whether to remove the first argument from the config.
        **kwargs: Keyword arguments to pass to the target.
    """
    params = inspect.signature(target).parameters

    config_dict = {
        "_target_": f"{target.__module__}.{target.__qualname__}",
    }

    for i, (key, param) in enumerate(params.items()):
        if remove_first_arg and i == 0:
            continue
        if param.kind == inspect.Parameter.VAR_KEYWORD:
            continue
        if key in kwargs:
            value = kwargs[key]
        else:
            value = (
                None if param.default == inspect.Parameter.empty else param.default
            )
        config_dict[key] = sanitized_default_value(value)
    return config_dict

SpecialKeys

Bases: str, Enum

Special keys in configs used by HyFI.

Source code in hyfi/composer/composer.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class SpecialKeys(str, Enum):
    """Special keys in configs used by HyFI."""

    CALL = "_call_"
    CONFIG = "_config_"
    CONFIG_GROUP = "_config_group_"
    CONFIG_NAME = "_config_name_"
    DESCRIPTION = "_description_"
    EXEC = "_exec_"
    FUNC = "_func_"
    KWARGS = "_kwargs_"
    METHOD = "_method_"
    PARTIAL = "_partial_"
    PIPE_TARGET = "pipe_target"
    RECURSIVE = "_recursive_"
    RUN = "run"
    RUN_WITH = "run_with"
    TARGET = "_target_"
    TYPE = "_type_"
    USES = "uses"
    WITH = "with"