Skip to content

utils

canari_ml.hydra.utils

canari_ml.hydra.utils.logger = logging.getLogger(__name__) module-attribute

canari_ml.hydra.utils.IndentDumper

Bases: Dumper

canari_ml.hydra.utils.IndentDumper.increase_indent(flow=False, indentless=False)

Source code in src/canari_ml/hydra/utils.py 
64
65
def increase_indent(self, flow=False, indentless=False):
    return super(IndentDumper, self).increase_indent(flow, False)

canari_ml.hydra.utils.get_hydra_config_root_path()

Source code in src/canari_ml/hydra/utils.py 
11
12
13
def get_hydra_config_root_path() -> str:
    config_path = importlib.resources.files("canari_ml").joinpath("conf")
    return str(config_path)

canari_ml.hydra.utils.run_command(command)

Run a command and log its output.

Parameters:

Name Type Description Default
command list

List of strings representing the command to execute.

 required

Raises:

Type Description
RuntimeError

If the command exits with a non-zero return code.
Source code in src/canari_ml/hydra/utils.py 
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
def run_command(command: list) -> None:
    """
    Run a command and log its output.

    Args:
        command: List of strings representing the command to execute.

    Raises:
        RuntimeError: If the command exits with a non-zero return code.
    """
    command = [str(v) for v in command]
    cmd_str = f"\n\nRunning command: {' '.join(command)}\n{'_' * 75}\n\n"
    logger.info(cmd_str)


    process = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True
    )
    for line in process.stdout:  # type: ignore
        logger.info(line.strip())

    process.wait()

    if process.returncode != 0:
        raise RuntimeError("Command failed with exit code %d" % process.returncode)

canari_ml.hydra.utils.dynamic_import(path)

Dynamically import a class or function from a module.

Takes a fully qualified name (e.g., 'module.class') and imports the specified class or function.

Parameters:

Name Type Description Default
path str

String containing the fully qualified name of the object to import, in the format 'module.class'.

 required

Returns:

Type Description
None

The imported class or function.
Source code in src/canari_ml/hydra/utils.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def dynamic_import(path: str) -> None:
    """
    Dynamically import a class or function from a module.

    Takes a fully qualified name (e.g., 'module.class') and imports the
    specified class or function.

    Args:
        path: String containing the fully qualified name of the object
            to import, in the format 'module.class'.

    Returns:
        The imported class or function.
    """
    # Split into module and class name
    module_name, class_name = path.rsplit(".", 1)
    module = importlib.import_module(module_name)
    return getattr(module, class_name)

canari_ml.hydra.utils.print_omega_config(cfg, resolve=False)

Print a HYDRA configuration as YAML.

This function converts the given OmegaConf DictionaryConfig to YAML format and logs it to stdout.

Parameters:

Name Type Description Default
cfg DictConfig

The Hydra configuration to print.

 required
Source code in src/canari_ml/hydra/utils.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def print_omega_config(cfg: DictConfig, resolve: bool = False) -> None:
    """Print a HYDRA configuration as YAML.

    This function converts the given OmegaConf DictionaryConfig to
    YAML format and logs it to stdout.

    Args:
        cfg: The Hydra configuration to print.
    """
    # cfg_yaml = OmegaConf.to_yaml(cfg)
    cfg_dict = OmegaConf.to_container(cfg, resolve=resolve, enum_to_str=True)
    cfg_yaml = yaml.dump(cfg_dict, Dumper=IndentDumper)

    logger.info("Loaded HYDRA Configuration YAML")
    logger.info(f"\n{cfg_yaml}")