fovi.models.loading

Configuration and checkpoint loading without dataset or trainer imports.

class fovi.models.loading.Sequence

Bases: Reversible, Collection

All the operations on a read-only sequence.

Concrete subclasses must override __new__ or __init__, __getitem__, and __len__.

count(value) → integer -- return number of occurrences of value
index(value[, start[, stop]]) → integer -- return first index of value.

Raises ValueError if the value is not present.

Supporting start and stop arguments is optional, but recommended.

class fovi.models.loading.Path(*args, **kwargs)[source]

Bases: PurePath

PurePath subclass that can make system calls.

Path represents a filesystem path but unlike PurePath, also offers methods to do system calls on path objects. Depending on your system, instantiating a Path will return either a PosixPath or a WindowsPath object. You can also instantiate a PosixPath or WindowsPath directly, but cannot instantiate a WindowsPath on a POSIX system or vice versa.

classmethod cwd()[source]

Return a new path pointing to the current working directory (as returned by os.getcwd()).

classmethod home()[source]

Return a new path pointing to the user’s home directory (as returned by os.path.expanduser(‘~’)).

samefile(other_path)[source]

Return whether other_path is the same or not as this file (as returned by os.path.samefile()).

iterdir()[source]

Iterate over the files in this directory. Does not yield any result for the special paths ‘.’ and ‘..’.

glob(pattern)[source]

Iterate over this subtree and yield all existing files (of any kind, including directories) matching the given relative pattern.

rglob(pattern)[source]

Recursively yield all existing files (of any kind, including directories) matching the given relative pattern, anywhere in this subtree.

absolute()[source]

Return an absolute version of this path by prepending the current working directory. No normalization or symlink resolution is performed.

Use resolve() to get the canonical path to a file.

resolve(strict=False)[source]

Make the path absolute, resolving all symlinks on the way and also normalizing it.

stat(*, follow_symlinks=True)[source]

Return the result of the stat() system call on this path, like os.stat() does.

owner()[source]

Return the login name of the file owner.

group()[source]

Return the group name of the file gid.

open(mode='r', buffering=-1, encoding=None, errors=None, newline=None)[source]

Open the file pointed by this path and return a file object, as the built-in open() function does.

read_bytes()[source]

Open the file in bytes mode, read it, and close the file.

read_text(encoding=None, errors=None)[source]

Open the file in text mode, read it, and close the file.

write_bytes(data)[source]

Open the file in bytes mode, write to it, and close the file.

write_text(data, encoding=None, errors=None, newline=None)[source]

Open the file in text mode, write to it, and close the file.

Return the path to which the symbolic link points.

touch(mode=0o666, exist_ok=True)[source]

Create this file with the given access mode, if it doesn’t exist.

mkdir(mode=0o777, parents=False, exist_ok=False)[source]

Create a new directory at this given path.

chmod(mode, *, follow_symlinks=True)[source]

Change the permissions of the path, like os.chmod().

lchmod(mode)[source]

Like chmod(), except if the path points to a symlink, the symlink’s permissions are changed, rather than its target’s.

Remove this file or link. If the path is a directory, use rmdir() instead.

rmdir()[source]

Remove this directory. The directory must be empty.

lstat()[source]

Like stat(), except if the path points to a symlink, the symlink’s status information is returned, rather than its target’s.

rename(target)[source]

Rename this path to the target path.

The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, not the directory of the Path object.

Returns the new Path instance pointing to the target path.

replace(target)[source]

Rename this path to the target path, overwriting if that path exists.

The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, not the directory of the Path object.

Returns the new Path instance pointing to the target path.

Make this path a symlink pointing to the target path. Note the order of arguments (link, target) is the reverse of os.symlink.

Make this path a hard link pointing to the same file as target.

Note the order of arguments (self, target) is the reverse of os.link’s.

Make the target path a hard link pointing to this path.

Note this function does not make this path a hard link to target, despite the implication of the function and argument names. The order of arguments (target, link) is the reverse of Path.symlink_to, but matches that of os.link.

Deprecated since Python 3.10 and scheduled for removal in Python 3.12. Use hardlink_to() instead.

exists()[source]

Whether this path exists.

is_dir()[source]

Whether this path is a directory.

is_file()[source]

Whether this path is a regular file (also True for symlinks pointing to regular files).

is_mount()[source]

Check if this path is a POSIX mount point

Whether this path is a symbolic link.

is_block_device()[source]

Whether this path is a block device.

is_char_device()[source]

Whether this path is a character device.

is_fifo()[source]

Whether this path is a FIFO.

is_socket()[source]

Whether this path is a socket.

expanduser()[source]

Return a new path with expanded ~ and ~user constructs (as returned by os.path.expanduser)

class fovi.models.loading.GlobalHydra(*args: Any, **kwargs: Any)[source]

Bases: object

__init__() → None[source]
initialize(hydra: Hydra) → None[source]
config_loader() → ConfigLoader[source]
is_initialized() → bool[source]
clear() → None[source]
static instance(*args: Any, **kwargs: Any) → GlobalHydra[source]
static set_instance(instance: GlobalHydra) → None[source]
class fovi.models.loading.DictConfig(content: Dict[str | bytes | int | Enum | float | bool, Any] | DictConfig | Any, key: Any = None, parent: Box | None = None, ref_type: Any | Type[Any] = Any, key_type: Any | Type[Any] = Any, element_type: Any | Type[Any] = Any, is_optional: bool = True, flags: Dict[str, bool] | None = None)[source]

Bases: BaseContainer, MutableMapping[Any, Any]

__init__(content: Dict[str | bytes | int | Enum | float | bool, Any] | DictConfig | Any, key: Any = None, parent: Box | None = None, ref_type: Any | Type[Any] = Any, key_type: Any | Type[Any] = Any, element_type: Any | Type[Any] = Any, is_optional: bool = True, flags: Dict[str, bool] | None = None) → None[source]
copy() → DictConfig[source]
get(key: str | bytes | int | Enum | float | bool, default_value: Any = None) → Any[source]

Return the value for key if key is in the dictionary, else default_value (defaulting to None).

pop(k[, d]) → v, remove specified key and return the corresponding value.[source]

If key is not found, d is returned if given, otherwise KeyError is raised.

keys() → a set-like object providing a view on D's keys[source]
items() → a set-like object providing a view on D's items[source]
setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D[source]
items_ex(resolve: bool = True, keys: Sequence[str | bytes | int | Enum | float | bool] | None = None) → List[Tuple[str | bytes | int | Enum | float | bool, Any]][source]
_promote(type_or_prototype: Type[Any] | None) → None[source]

Retypes a node. This should only be used in rare circumstances, where you want to dynamically change the runtime structured-type of a DictConfig. It will change the type and add the additional fields based on the input class or object

_to_object() → Any[source]

Instantiate an instance of self._metadata.object_type. This requires self to be a structured config. Nested subconfigs are converted by calling OmegaConf.to_object.

class fovi.models.loading.OmegaConf[source]

Bases: object

OmegaConf primary class

__init__() → None[source]
static structured(obj: Any, parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → Any[source]
static create(obj: str, parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → DictConfig | ListConfig[source]
static create(obj: List[Any] | Tuple[Any, ...], parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → ListConfig
static create(obj: DictConfig, parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → DictConfig
static create(obj: ListConfig, parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → ListConfig
static create(obj: Dict[Any, Any] | None = None, parent: BaseContainer | None = None, flags: Dict[str, bool] | None = None) → DictConfig
static load(file_: str | Path | IO[Any]) → DictConfig | ListConfig[source]
static save(config: Any, f: str | Path | IO[Any], resolve: bool = False) → None[source]

Save as configuration object to a file

Parameters:

  • config – omegaconf.Config object (DictConfig or ListConfig).

  • f – filename or file object

  • resolve – True to save a resolved config (defaults to False)

static from_cli(args_list: List[str] | None = None) → DictConfig[source]
static from_dotlist(dotlist: List[str]) → DictConfig[source]

Creates config from the content sys.argv or from the specified args list of not None

Parameters:

dotlist – A list of dotlist-style strings, e.g. ["foo.bar=1", "baz=qux"].

Returns:

A DictConfig object created from the dotlist.

static merge(*configs: DictConfig | ListConfig | Dict[str | bytes | int | Enum | float | bool, Any] | List[Any] | Tuple[Any, ...] | Any) → ListConfig | DictConfig[source]

Merge a list of previously created configs into a single one

Parameters:

configs – Input configs

Returns:

the merged config object.

static unsafe_merge(*configs: DictConfig | ListConfig | Dict[str | bytes | int | Enum | float | bool, Any] | List[Any] | Tuple[Any, ...] | Any) → ListConfig | DictConfig[source]

Merge a list of previously created configs into a single one This is much faster than OmegaConf.merge() as the input configs are not copied. However, the input configs must not be used after this operation as will become inconsistent.

Parameters:

configs – Input configs

Returns:

the merged config object.

static register_resolver(name: str, resolver: Callable[[...], Any]) → None[source]
static legacy_register_resolver(name: str, resolver: Callable[[...], Any]) → None[source]
static register_new_resolver(name: str, resolver: Callable[[...], Any], *, replace: bool = False, use_cache: bool = False) → None[source]

Register a resolver.

Parameters:

  • name – Name of the resolver.

  • resolver – Callable whose arguments are provided in the interpolation, e.g., with ${foo:x,0,${y.z}} these arguments are respectively “x” (str), 0 (int) and the value of y.z.

  • replace – If set to False (default), then a ValueError is raised if an existing resolver has already been registered with the same name. If set to True, then the new resolver replaces the previous one. NOTE: The cache on existing config objects is not affected, use OmegaConf.clear_cache(cfg) to clear it.

  • use_cache – Whether the resolver’s outputs should be cached. The cache is based only on the string literals representing the resolver arguments, e.g., ${foo:${bar}} will always return the same value regardless of the value of bar if the cache is enabled for foo.

classmethod has_resolver(name: str) → bool[source]
static clear_resolvers() → None[source]

Clear(remove) all OmegaConf resolvers, then re-register OmegaConf’s default resolvers.

classmethod clear_resolver(name: str) → bool[source]

Clear(remove) any resolver only if it exists.

Returns a bool: True if resolver is removed and False if not removed.

Parameters:

name – Name of the resolver.

Returns:

A bool (True if resolver is removed, False if not found before removing).

static get_cache(conf: BaseContainer) → Dict[str, Any][source]
static set_cache(conf: BaseContainer, cache: Dict[str, Any]) → None[source]
static clear_cache(conf: BaseContainer) → None[source]
static copy_cache(from_config: BaseContainer, to_config: BaseContainer) → None[source]
static set_readonly(conf: Node, value: bool | None) → None[source]
static is_readonly(conf: Node) → bool | None[source]
static set_struct(conf: Container, value: bool | None) → None[source]
static is_struct(conf: Container) → bool | None[source]
static masked_copy(conf: DictConfig, keys: str | List[str]) → DictConfig[source]

Create a masked copy of of this config that contains a subset of the keys

Parameters:

  • conf – DictConfig object

  • keys – keys to preserve in the copy

Returns:

The masked DictConfig object.

static to_container(cfg: Any, *, resolve: bool = False, throw_on_missing: bool = False, enum_to_str: bool = False, structured_config_mode: SCMode = SCMode.DICT) → Dict[str | bytes | int | Enum | float | bool, Any] | List[Any] | None | str | Any[source]

Resursively converts an OmegaConf config to a primitive container (dict or list).

Parameters:

  • cfg – the config to convert

  • resolve – True to resolve all values

  • throw_on_missing – When True, raise MissingMandatoryValue if any missing values are present. When False (the default), replace missing values with the string “???” in the output container.

  • enum_to_str – True to convert Enum keys and values to strings

  • structured_config_mode –

    Specify how Structured Configs (DictConfigs backed by a dataclass) are handled.

    • By default (structured_config_mode=SCMode.DICT) structured configs are converted to plain dicts.

    • If structured_config_mode=SCMode.DICT_CONFIG, structured config nodes will remain as DictConfig.

    • If structured_config_mode=SCMode.INSTANTIATE, this function will instantiate structured configs (DictConfigs backed by a dataclass), by creating an instance of the underlying dataclass.

    See also OmegaConf.to_object.

Returns:

A dict or a list representing this config as a primitive container.

static to_object(cfg: Any) → Dict[str | bytes | int | Enum | float | bool, Any] | List[Any] | None | str | Any[source]

Resursively converts an OmegaConf config to a primitive container (dict or list). Any DictConfig objects backed by dataclasses or attrs classes are instantiated as instances of those backing classes.

This is an alias for OmegaConf.to_container(…, resolve=True, throw_on_missing=True,

structured_config_mode=SCMode.INSTANTIATE)

Parameters:

cfg – the config to convert

Returns:

A dict or a list or dataclass representing this config.

static is_missing(cfg: Any, key: str | bytes | int | Enum | float | bool) → bool[source]
static is_interpolation(node: Any, key: int | str | None = None) → bool[source]
static is_list(obj: Any) → bool[source]
static is_dict(obj: Any) → bool[source]
static is_config(obj: Any) → bool[source]
static get_type(obj: Any, key: str | None = None) → Type[Any] | None[source]
static select(cfg: Container, key: str, *, default: Any = _DEFAULT_MARKER_, throw_on_resolution_failure: bool = True, throw_on_missing: bool = False) → Any[source]
Parameters:

  • cfg – Config node to select from

  • key – Key to select

  • default – Default value to return if key is not found

  • throw_on_resolution_failure – Raise an exception if an interpolation resolution error occurs, otherwise return None

  • throw_on_missing – Raise an exception if an attempt to select a missing key (with the value ‘???’) is made, otherwise return None

Returns:

selected value or None if not found.

static update(cfg: Container, key: str, value: Any = None, *, merge: bool = True, force_add: bool = False) → None[source]

Updates a dot separated key sequence to a value

Parameters:

  • cfg – input config to update

  • key – key to update (can be a dot separated path)

  • value – value to set, if value if a list or a dict it will be merged or set depending on merge_config_values

  • merge – If value is a dict or a list, True (default) to merge into the destination, False to replace the destination.

  • force_add – insert the entire path regardless of Struct flag or Structured Config nodes.

static to_yaml(cfg: Any, *, resolve: bool = False, sort_keys: bool = False) → str[source]

returns a yaml dump of this config object.

Parameters:

  • cfg – Config object, Structured Config type or instance

  • resolve – if True, will return a string with the interpolations resolved, otherwise interpolations are preserved

  • sort_keys – If True, will print dict keys in sorted order. default False.

Returns:

A string containing the yaml representation.

static resolve(cfg: Container) → None[source]

Resolves all interpolations in the given config object in-place.

Parameters:

cfg – An OmegaConf container (DictConfig, ListConfig) Raises a ValueError if the input object is not an OmegaConf container.

static missing_keys(cfg: Any) → Set[str][source]

Returns a set of missing keys in a dotlist style.

Parameters:

cfg – An OmegaConf.Container, or a convertible object via OmegaConf.create (dict, list, …).

Returns:

set of strings of the missing keys.

Raises:

ValueError – On input not representing a config.

fovi.models.loading.open_dict(config: Container) → Generator[Container, None, None][source]
class fovi.models.loading.HiddenPrints(enabled=True)[source]

Bases: object

Context manager to suppress stdout output.

Temporarily redirects stdout to devnull to hide print statements from called functions.

Parameters:

enabled (bool, optional) – Whether to suppress prints. If False, acts as a no-op. Defaults to True.

Example

>>> with HiddenPrints():
...     print("This won't be shown")
>>> print("This will be shown")
This will be shown
__init__(enabled=True)[source]
fovi.models.loading.default_model_dirs() → list[str][source]

Include research log directories only when explicitly configured.

fovi.models.loading.load_sharded_state_dict(model_dir: str | Path, base_name: str = 'state_dict', device: str | device = 'cuda') → dict[str, dict[str, Tensor]][source]

Load weights from the shard files named by a checkpoint index.

fovi.models.loading.load_config(base_fn: str, load: bool, folder: str | Path, device: str | device = 'cuda') → tuple[DictConfig, dict[str, dict[str, Tensor]] | None, str | None][source]

Load a local configuration and optionally its checkpoint.

Parameters:

  • base_fn – Model directory name or standalone Hydra config stem.

  • load – Whether to load weights.

  • folder – Parent directory containing the model/configuration.

  • device – Device receiving checkpoint tensors.

Returns:

Configuration, checkpoint (or None), and model state key (or None).

Raises:
fovi.models.loading.find_config(base_fn: str, load: bool, model_dirs: Sequence[str | Path] | None = None, device: str | device = 'cuda') → tuple[DictConfig, dict[str, dict[str, Tensor]] | None, str | None][source]

Search explicit local locations, then download a model from the Hub.

An existing local model with malformed configuration or missing weights raises immediately. It must not silently select a different checkpoint from the Hub.

fovi.models.loading.get_model_from_base_fn(base_fn: str, load: bool = True, load_strict: bool = True, quiet: bool = False, device: str | device = 'cuda', model_dirs: Sequence[str | Path] = ('../models',), fovinet_cls: type[Module] | None = None, **kwargs: str | float | bool | None) → Module[source]

Construct a model and restore weights without creating a Trainer.

Parameters:

  • base_fn – Local model name or HuggingFace repository identifier.

  • load – Whether to restore weights.

  • load_strict – Passed to the model’s state-dict loader.

  • quiet – Suppress model construction output.

  • device – Device for model construction and checkpoint tensors.

  • model_dirs – Local search locations, in priority order.

  • fovinet_cls – Model constructor; defaults to FoviNet.

  • **kwargs – Dotted configuration overrides.

Returns:

Constructed model with optional checkpoint weights.