omegaconf

optimus_dl.core.omegaconf

OmegaConf custom resolvers for configuration.

This module registers custom resolvers for OmegaConf that enable advanced configuration features like Python expression evaluation and environment variable access.

It also provides a custom non-resolving instantiation system that allows Hydra-style object creation while preserving OmegaConf reactivity and interpolation integrity.

Lazy reactive Hydra instantiation for OmegaConf trees.

Goal

Hydra's hydra.utils.instantiate(...) normally resolves interpolations eagerly. That breaks OmegaConf reactivity in some workflows.

This module implements a lazy alternative:

• Every DictConfig node with _target_ is replaced by ${_lazy_inst:<id>}
• Real target config is stored in a resolver-side cache
• On access, _lazy_inst: 1) checks whether external dependencies changed 2) returns cached instance if not changed 3) re-instantiates if changed

Key guarantees

1. Internal references inside a target node remain relative/portable.
2. External references are tracked for cache invalidation.
3. Nested lazy targets continue to work after instantiation.
4. Circular lazy recursion is detected.

Important

• This relies on OmegaConf internals (_set_parent, resolver cache usage).
• Not thread-safe by design (shared mutable cache in config object).

conf_hash_resolver(*args, _root_)

Resolver for computing hash of a root config.

Source code in optimus_dl/core/omegaconf.py

def conf_hash_resolver(*args, _root_):
    """Resolver for computing hash of a root config."""
    max_len = 16
    if len(args) > 0:
        assert len(args) == 1, "Only one argument is allowed"
        max_len = int(args[0])
    return hash_resolver(_root_, max_len=max_len)

hash_resolver(x, max_len=16)

Resolver for computing hash of a value repr.

Source code in optimus_dl/core/omegaconf.py

def hash_resolver(x, max_len=16):
    """Resolver for computing hash of a value repr."""
    x = repr(x)
    return hashlib.sha256(x.encode("utf-8")).hexdigest()[:max_len]

non_resolving_instantiate(config, lazy=True, **kwargs)

Custom instantiate function that preserves OmegaConf interpolations.

It selectively instantiates nodes with a 'target' using hydra.utils.instantiate, without eagerly resolving the entire configuration tree.

Source code in optimus_dl/core/omegaconf.py

def non_resolving_instantiate(config: Any, lazy: bool = True, **kwargs: Any) -> Any:
    """Custom instantiate function that preserves OmegaConf interpolations.

    It selectively instantiates nodes with a '_target_' using hydra.utils.instantiate,
    without eagerly resolving the entire configuration tree.
    """
    if isinstance(config, DictConfig | ListConfig):
        config_copy = copy.deepcopy(config)
        OmegaConf.set_struct(config_copy, False)
        OmegaConf.set_readonly(config_copy, False)
        config_copy._set_flag("allow_objects", True)

        if not lazy:

            def _walk_and_instantiate(node: Any) -> Any:
                if isinstance(node, DictConfig):
                    if "_target_" in node:
                        return hydra.utils.instantiate(node, **kwargs)
                    for key in list(node.keys()):
                        if not OmegaConf.is_interpolation(node, key):
                            node[key] = _walk_and_instantiate(node[key])
                    return node
                elif isinstance(node, ListConfig):
                    for i in range(len(node)):
                        if not OmegaConf.is_interpolation(node, i):
                            node[i] = _walk_and_instantiate(node[i])
                    return node
                return node

            return _walk_and_instantiate(config_copy)

        def _walk_and_ghost(
            node: Any,
            path: str,
            parent_node: Any,
            local_key: str | int,
            target_deps: list[str] | None,
        ) -> Any:
            if isinstance(node, DictConfig):
                if "_target_" in node:
                    # Bottom-Up
                    my_deps: list[str] = []
                    node_copy = copy.deepcopy(node)
                    for k in list(node_copy.keys()):
                        if not OmegaConf.is_interpolation(node_copy, k):
                            node_copy[k] = _walk_and_ghost(
                                node_copy[k],
                                f"{path}.{k}" if path else str(k),
                                node_copy,
                                k,
                                my_deps,
                            )

                    raw_data = OmegaConf.to_container(node_copy, resolve=False)
                    norm_data = _normalize_and_collect_deps(
                        raw_data, path, path, my_deps
                    )
                    norm_node = OmegaConf.create(norm_data)

                    # Store definition in the parent container's resolver cache
                    cache = OmegaConf.get_cache(parent_node)
                    lazy_configs = cache.setdefault("lazy_configs", {})

                    # Propagate caches from node_copy to norm_node to preserve nested targets
                    _copy_lazy_state(node_copy, norm_node)

                    # Deduplicate dependencies while preserving order
                    unique_deps = list(dict.fromkeys(my_deps))
                    lazy_configs[str(local_key)] = (norm_node, unique_deps, kwargs)

                    if target_deps is not None:
                        target_deps.extend(unique_deps)

                    return f"${{_lazy_inst:{local_key}}}"

                for key in list(node.keys()):
                    if OmegaConf.is_interpolation(node, key):
                        continue
                    child_path = f"{path}.{key}" if path else str(key)
                    node[key] = _walk_and_ghost(
                        node[key], child_path, node, key, target_deps
                    )
                return node
            elif isinstance(node, ListConfig):
                for i in range(len(node)):
                    if OmegaConf.is_interpolation(node, i):
                        continue
                    child_path = f"{path}[{i}]"
                    node[i] = _walk_and_ghost(node[i], child_path, node, i, target_deps)
                return node
            return node

        if isinstance(config_copy, DictConfig) and "_target_" in config_copy:
            # Eagerly instantiate the root node if it's a target
            for k in list(config_copy.keys()):
                if k != "_target_" and not OmegaConf.is_interpolation(config_copy, k):
                    config_copy[k] = _walk_and_ghost(
                        config_copy[k], str(k), config_copy, k, None
                    )
            return hydra.utils.instantiate(config_copy, **kwargs)
        else:
            _walk_and_ghost(config_copy, "", None, "", None)
            return config_copy

    elif isinstance(config, dict):
        if "_target_" in config:
            return hydra.utils.instantiate(config, **kwargs)
        return {
            k: non_resolving_instantiate(v, lazy=lazy, **kwargs)
            for k, v in config.items()
        }
    elif isinstance(config, list):
        return [non_resolving_instantiate(v, lazy=lazy, **kwargs) for v in config]
    return config