hyfi.main

This module serves as the entry point for the HyFI application.

It imports the necessary classes and objects from the 'config' and 'main' modules, and defines the 'all' list to specify the public interface of this module.

Classes: - GlobalConfig: Represents the global configuration for the HyFI application. - GlobalConfigResolver: Resolves the global configuration for the HyFI application. - HyFIConfig: Represents the configuration for the HyFI application. - HyFI: Represents the main class of the HyFI application.

Objects: - global_config: An instance of the GlobalConfig class representing the global configuration.

GlobalConfig

Bases: UTILs

HyFI global config class. This class is used to store the global configuration

Source code in hyfi/main/config.py

class GlobalConfig(UTILs):
    """HyFI global config class.  This class is used to store the global configuration"""

    __config__: Optional[HyFIConfig] = None

    _about_: Optional[About] = None
    _project_: Optional[Project] = None
    _version_: str = PrivateAttr(global_hyfi.version)

    def __init__(self, **config_kwargs):
        if config_kwargs:
            self.__config__ = HyFIConfig(**config_kwargs)

    @property
    def about(self) -> About:
        """
        Returns the About object associated with the configuration.

        If the About object is not yet created, it will be created using the
        configuration's package name.

        Returns:
            The About object associated with the configuration.
        """
        if self._about_ is None:
            config_name = (
                "__init__"
                if global_hyfi.package_name == "hyfi"
                else global_hyfi.package_name
            )
            kwargs = {"_config_name_": config_name}
            self._about_ = About(**kwargs)
        return self._about_

    @property
    def project(self) -> Optional[Project]:
        """
        Returns the project associated with the configuration.

        Returns:
            Optional[Project]: The project associated with the configuration, or None if no project is set.
        """
        return self._project_

    def inititialize(
        self,
        project_name: Optional[str] = None,
        project_description: Optional[str] = None,
        project_root: Optional[str] = None,
        project_workspace_name: Optional[str] = None,
        global_hyfi_root: Optional[str] = None,
        global_workspace_name: Optional[str] = None,
        num_workers: Optional[int] = None,
        logging_level: Optional[str] = None,
        plugins: Optional[List[str]] = None,
        user_config_path: Optional[str] = None,
        dotenv_file: Optional[str] = None,
        secrets_dir: Optional[str] = None,
        reinit: bool = True,
        autotime: bool = True,
        retina: bool = True,
        verbose: Union[bool, int] = False,
        **project_kwargs,
    ):
        """
        Initialize and start hyfi.

        Args:
            project_name: Name of the project to use.
            project_description: Description of the project that will be used.
            project_root: Root directory of the project.
            project_workspace_name: Name of the project's workspace directory.
            global_hyfi_root: Root directory of the global hyfi.
            global_workspace_name: Name of the global hierachical workspace directory.
            num_workers: Number of workers to run.
            log_level: Log level for the log.
            plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
            user_config_path: Path to the user configuration directory. e.g. `./config`
            config_dirname: Name of the configuration directory. e.g. `conf`
            dotenv_file: Name of the dotenv file. e.g. `.env`
            secrets_dir: Name of the secrets directory. e.g. `secrets`
            reinit: Whether to reinitialize the global config.
            autotime: Whether to automatically set time and / or keep track of run times.
            retina: Whether to use retina or not.
            verbose: Enables or disables logging
        """
        if self.project and not reinit:
            return
        # Set the log level to the given log level.
        if logging_level:
            GlobalConfig.setLogger(logging_level)
            logger.setLevel(logging_level)

        global_hyfi.reinitialize(
            plugins=plugins,
            user_config_path=user_config_path,
            dotenv_file=dotenv_file,
            secrets_dir=secrets_dir,
        )

        # Load the extentions for the autotime extension.
        if autotime:
            GlobalConfig.load_extentions(exts=["autotime"])
        # Set the retina matplotlib formats.
        if retina:
            GlobalConfig.set_matplotlib_formats("retina")
        if project_name:
            project_kwargs["project_name"] = project_name
        if project_description:
            project_kwargs["project_description"] = project_description
        if project_root:
            project_kwargs["project_root"] = ENVs.expand_posix_vars(project_root)
        if project_workspace_name:
            project_kwargs["project_workspace_name"] = project_workspace_name
        # Expand the hyfi_root environment variable.
        if global_hyfi_root:
            project_kwargs["global_hyfi_root"] = ENVs.expand_posix_vars(
                global_hyfi_root
            )
        if global_workspace_name:
            project_kwargs["global_workspace_name"] = global_workspace_name
        if num_workers:
            project_kwargs["num_workers"] = num_workers
        project_kwargs["verbose"] = verbose

        self._project_ = Project(**project_kwargs)
        logger.info("HyFi project [%s] initialized", self._project_.project_name)

    def terminate(self) -> None:
        """
        Terminate hyfi config by stopping joblib

        Returns:
            True if successful False
        """
        # Stop the backend if the joblib is running.
        if self.project and self.project.joblib:
            self.project.joblib.stop_backend()

    def __repr__(self):
        """
        Returns a string representation of GlobalConfig.
        """
        return f"GlobalConfig(project={self.project})"

    def __str__(self):
        """
        Returns a string representation of the object.
        """
        return self.__repr__()

    @property
    def app_version(self):
        """
        Get the version of the application.
        """
        return global_hyfi.version

    @property
    def app_name(self):
        """
        Get the name of the application.
        """
        return self.about.name

    @property
    def package_name(self):
        """
        Get the name of the package.
        """
        return global_hyfi.package_name

    def print_about(self, **kwargs):
        if not kwargs:
            config_name = (
                "__init__"
                if global_hyfi.package_name == "hyfi"
                else global_hyfi.package_name
            )
            kwargs = {"_config_name_": config_name}
        self._about_ = About(**kwargs)
        pkg_name = self.package_name
        name = self.app_name
        print()
        for k, v in self.about.model_dump().items():
            if k.startswith("_") or k == "version":
                continue
            print(f"{k:11} : {v}")
        print(f"{'version':11} : {self.app_version}")
        if pkg_name:
            print(f"\nExecute `{pkg_name} --help` to see what you can do with {name}")
        if kwargs.get("_open_link_", False) and kwargs.get("homepage"):
            webbrowser.open_new_tab(kwargs["homepage"])

    @property
    def project_dir(self) -> Path:
        """Get the project root directory."""
        return (
            self.project.root_dir
            if self.project
            else Path(__default_project_root__).absolute()
        )

    @property
    def project_workspace_dir(self) -> Path:
        """Get the project workspace directory."""
        return (
            self.project.workspace_dir
            if self.project
            else self.project_dir / __default_workspace_name__
        )

    def get_path(
        self,
        path_name: str,
        base_dir: Optional[Union[Path, str]] = None,
        ensure_exists: bool = False,
    ) -> Optional[Path]:
        """
        Get the path to a directory or file.
        """
        return (
            self.project.get_path(
                path_name, base_dir=base_dir, ensure_exists=ensure_exists
            )
            if self.project
            else None
        )

about: About property

Returns the About object associated with the configuration.

If the About object is not yet created, it will be created using the configuration's package name.

Returns:

Type	Description
About	The About object associated with the configuration.

app_name property

Get the name of the application.

app_version property

Get the version of the application.

package_name property

Get the name of the package.

project: Optional[Project] property

Returns the project associated with the configuration.

Returns:

Type	Description
Optional[Project]	Optional[Project]: The project associated with the configuration, or None if no project is set.

project_dir: Path property

Get the project root directory.

project_workspace_dir: Path property

Get the project workspace directory.

__repr__()

Returns a string representation of GlobalConfig.

Source code in hyfi/main/config.py

def __repr__(self):
    """
    Returns a string representation of GlobalConfig.
    """
    return f"GlobalConfig(project={self.project})"

__str__()

Returns a string representation of the object.

Source code in hyfi/main/config.py

def __str__(self):
    """
    Returns a string representation of the object.
    """
    return self.__repr__()

get_path(path_name, base_dir=None, ensure_exists=False)

Get the path to a directory or file.

Source code in hyfi/main/config.py

def get_path(
    self,
    path_name: str,
    base_dir: Optional[Union[Path, str]] = None,
    ensure_exists: bool = False,
) -> Optional[Path]:
    """
    Get the path to a directory or file.
    """
    return (
        self.project.get_path(
            path_name, base_dir=base_dir, ensure_exists=ensure_exists
        )
        if self.project
        else None
    )

inititialize(project_name=None, project_description=None, project_root=None, project_workspace_name=None, global_hyfi_root=None, global_workspace_name=None, num_workers=None, logging_level=None, plugins=None, user_config_path=None, dotenv_file=None, secrets_dir=None, reinit=True, autotime=True, retina=True, verbose=False, **project_kwargs)

Initialize and start hyfi.

Parameters:

Name	Type	Description	Default
project_name	Optional[str]	Name of the project to use.	None
project_description	Optional[str]	Description of the project that will be used.	None
project_root	Optional[str]	Root directory of the project.	None
project_workspace_name	Optional[str]	Name of the project's workspace directory.	None
global_hyfi_root	Optional[str]	Root directory of the global hyfi.	None
global_workspace_name	Optional[str]	Name of the global hierachical workspace directory.	None
num_workers	Optional[int]	Number of workers to run.	None
log_level		Log level for the log.	required
plugins	Optional[List[str]]	A list of plugins to load. e.g. ["hyfi.conf"]	None
user_config_path	Optional[str]	Path to the user configuration directory. e.g. ./config	None
config_dirname		Name of the configuration directory. e.g. conf	required
dotenv_file	Optional[str]	Name of the dotenv file. e.g. .env	None
secrets_dir	Optional[str]	Name of the secrets directory. e.g. secrets	None
reinit	bool	Whether to reinitialize the global config.	True
autotime	bool	Whether to automatically set time and / or keep track of run times.	True
retina	bool	Whether to use retina or not.	True
verbose	Union[bool, int]	Enables or disables logging	False

Source code in hyfi/main/config.py

def inititialize(
    self,
    project_name: Optional[str] = None,
    project_description: Optional[str] = None,
    project_root: Optional[str] = None,
    project_workspace_name: Optional[str] = None,
    global_hyfi_root: Optional[str] = None,
    global_workspace_name: Optional[str] = None,
    num_workers: Optional[int] = None,
    logging_level: Optional[str] = None,
    plugins: Optional[List[str]] = None,
    user_config_path: Optional[str] = None,
    dotenv_file: Optional[str] = None,
    secrets_dir: Optional[str] = None,
    reinit: bool = True,
    autotime: bool = True,
    retina: bool = True,
    verbose: Union[bool, int] = False,
    **project_kwargs,
):
    """
    Initialize and start hyfi.

    Args:
        project_name: Name of the project to use.
        project_description: Description of the project that will be used.
        project_root: Root directory of the project.
        project_workspace_name: Name of the project's workspace directory.
        global_hyfi_root: Root directory of the global hyfi.
        global_workspace_name: Name of the global hierachical workspace directory.
        num_workers: Number of workers to run.
        log_level: Log level for the log.
        plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
        user_config_path: Path to the user configuration directory. e.g. `./config`
        config_dirname: Name of the configuration directory. e.g. `conf`
        dotenv_file: Name of the dotenv file. e.g. `.env`
        secrets_dir: Name of the secrets directory. e.g. `secrets`
        reinit: Whether to reinitialize the global config.
        autotime: Whether to automatically set time and / or keep track of run times.
        retina: Whether to use retina or not.
        verbose: Enables or disables logging
    """
    if self.project and not reinit:
        return
    # Set the log level to the given log level.
    if logging_level:
        GlobalConfig.setLogger(logging_level)
        logger.setLevel(logging_level)

    global_hyfi.reinitialize(
        plugins=plugins,
        user_config_path=user_config_path,
        dotenv_file=dotenv_file,
        secrets_dir=secrets_dir,
    )

    # Load the extentions for the autotime extension.
    if autotime:
        GlobalConfig.load_extentions(exts=["autotime"])
    # Set the retina matplotlib formats.
    if retina:
        GlobalConfig.set_matplotlib_formats("retina")
    if project_name:
        project_kwargs["project_name"] = project_name
    if project_description:
        project_kwargs["project_description"] = project_description
    if project_root:
        project_kwargs["project_root"] = ENVs.expand_posix_vars(project_root)
    if project_workspace_name:
        project_kwargs["project_workspace_name"] = project_workspace_name
    # Expand the hyfi_root environment variable.
    if global_hyfi_root:
        project_kwargs["global_hyfi_root"] = ENVs.expand_posix_vars(
            global_hyfi_root
        )
    if global_workspace_name:
        project_kwargs["global_workspace_name"] = global_workspace_name
    if num_workers:
        project_kwargs["num_workers"] = num_workers
    project_kwargs["verbose"] = verbose

    self._project_ = Project(**project_kwargs)
    logger.info("HyFi project [%s] initialized", self._project_.project_name)

terminate()

Terminate hyfi config by stopping joblib

Returns:

Type	Description
None	True if successful False

Source code in hyfi/main/config.py

def terminate(self) -> None:
    """
    Terminate hyfi config by stopping joblib

    Returns:
        True if successful False
    """
    # Stop the backend if the joblib is running.
    if self.project and self.project.joblib:
        self.project.joblib.stop_backend()

GlobalConfigResolver

A class that resolves global configuration paths for HyFI.

Source code in hyfi/main/config.py

class GlobalConfigResolver:
    """
    A class that resolves global configuration paths for HyFI.
    """

    @staticmethod
    def __project_root_path__() -> str:
        """Global HyFI config path for the project root."""
        return str(global_config.project_dir or "")

    @staticmethod
    def __project_workspace_path__() -> str:
        """Global HyFI config path for the project workspace directory."""
        return str(global_config.project_workspace_dir or "")

    @staticmethod
    def __get_path__(path_name: str, base_dir: Optional[str] = None) -> str:
        """
        Get the path to a directory or file.

        Args:
            path_name (str): The name of the path.
            base_dir (Optional[str], optional): The base directory to resolve the path from. Defaults to None.

        Returns:
            str: The resolved path.
        """
        return str(global_config.get_path(path_name, base_dir=base_dir) or "")

__get_path__(path_name, base_dir=None) staticmethod

Get the path to a directory or file.

Parameters:

Name	Type	Description	Default
path_name	str	The name of the path.	required
base_dir	Optional[str]	The base directory to resolve the path from. Defaults to None.	None

Returns:

Name	Type	Description
str	str	The resolved path.

Source code in hyfi/main/config.py

@staticmethod
def __get_path__(path_name: str, base_dir: Optional[str] = None) -> str:
    """
    Get the path to a directory or file.

    Args:
        path_name (str): The name of the path.
        base_dir (Optional[str], optional): The base directory to resolve the path from. Defaults to None.

    Returns:
        str: The resolved path.
    """
    return str(global_config.get_path(path_name, base_dir=base_dir) or "")

__project_root_path__() staticmethod

Global HyFI config path for the project root.

Source code in hyfi/main/config.py

@staticmethod
def __project_root_path__() -> str:
    """Global HyFI config path for the project root."""
    return str(global_config.project_dir or "")

__project_workspace_path__() staticmethod

Global HyFI config path for the project workspace directory.

Source code in hyfi/main/config.py

@staticmethod
def __project_workspace_path__() -> str:
    """Global HyFI config path for the project workspace directory."""
    return str(global_config.project_workspace_dir or "")

HyFI

Bases: BATCHER, Composer, GRAPHICs, PIPELINEs

Primary class for the hyfi config package

Source code in hyfi/main/main.py

class HyFI(
    BATCHER,
    Composer,
    GRAPHICs,
    PIPELINEs,
):
    """Primary class for the hyfi config package"""

    __config__: Optional[HyFIConfig] = None
    __variables__: Optional[Variables] = None

    __version__ = GlobalHyFIResolver.__hyfi_version__()
    __hyfi_path__ = GlobalHyFIResolver.__hyfi_path__()
    __home_path__ = GlobalHyFIResolver.__home_path__()
    __package_name__ = GlobalHyFIResolver.__package_name__()
    __package_path__ = GlobalHyFIResolver.__package_path__()
    __app_version__ = GlobalHyFIResolver.__app_version__()

    def __init__(self, **config_kwargs):
        if config_kwargs:
            self.__config__ = HyFIConfig(**config_kwargs)
            if self.__config__.project:
                self.initialize(**self.__config__.project)

    @property
    def config(self) -> HyFIConfig:
        """Get the global config."""
        return self.__config__

    @property
    def project(self) -> Optional[Project]:
        """Get the project."""
        if global_config.project:
            return global_config.project
        else:
            raise ValueError("Project not initialized.")

    @project.setter
    def project(self, project: Project) -> None:
        """Set the project."""
        global_config.project = project

    @staticmethod
    def set_project(project: Project) -> None:
        """
        Set the project.

        Args:
            project: Project to set.
        """
        logger.info(f"Setting the global project to {project.project_name}")
        global_config.project = project

    @staticmethod
    def initialize(
        project_name: Optional[str] = None,
        project_description: Optional[str] = None,
        project_root: Optional[str] = None,
        project_workspace_name: Optional[str] = None,
        global_hyfi_root: Optional[str] = None,
        global_workspace_name: Optional[str] = None,
        num_workers: Optional[int] = None,
        logging_level: Optional[str] = None,
        plugins: Optional[List[str]] = None,
        user_config_path: Optional[str] = None,
        dotenv_file: Optional[str] = None,
        secrets_dir: Optional[str] = None,
        reinit: bool = True,
        autotime: bool = True,
        retina: bool = True,
        verbose: Union[bool, int] = False,
        **project_kwargs,
    ) -> "HyFI":
        """
        Initialize and start hyfi.

        Args:
            project_name: Name of the project to use.
            project_description: Description of the project that will be used.
            project_root: Root directory of the project.
            project_workspace_name: Name of the project's workspace directory.
            global_hyfi_root: Root directory of the global hyfi.
            global_workspace_name: Name of the global hierachical workspace directory.
            num_workers: Number of workers to run.
            logging_level: Log level for the log.
            plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
            user_config_path: Path to the user configuration directory. e.g. `./config`
            config_dirname: Name of the configuration directory. e.g. `conf`
            dotenv_file: Name of the dotenv file. e.g. `.env`
            secrets_dir: Name of the secrets directory. e.g. `secrets`
            reinit: Whether to reinitialize the global config.
            autotime: Whether to automatically set time and / or keep track of run times.
            retina: Whether to use retina or not.
            verbose: Enables or disables logging
        """
        global_config.inititialize(
            project_name=project_name,
            project_description=project_description,
            project_root=project_root,
            project_workspace_name=project_workspace_name,
            global_hyfi_root=global_hyfi_root,
            global_workspace_name=global_workspace_name,
            num_workers=num_workers,
            logging_level=logging_level,
            plugins=plugins,
            user_config_path=user_config_path,
            dotenv_file=dotenv_file,
            secrets_dir=secrets_dir,
            reinit=reinit,
            autotime=autotime,
            retina=retina,
            verbose=verbose,
            **HyFI.to_dict(project_kwargs),
        )
        return HyFI()

    @staticmethod
    def terminate() -> None:
        """
        Terminate the global config.

        Returns:
            bool: True if termination was successful, False otherwise.
        """
        global_config.terminate()

    @staticmethod
    def initialize_global_hyfi(
        package_path: str,
        version: str,
        plugins: Optional[List[str]] = None,
        user_config_path: Optional[str] = None,
        config_dirname: Optional[str] = None,
        dotenv_file: Optional[str] = None,
        secrets_dir: Optional[str] = None,
        **kwargs,
    ) -> None:
        """
        Initializes the global HyFI instance.
        This function should be called before any other HyFI function.
        A plugin is a python module which contains a configuration module.

        Args:
            package_path: Path to the package root folder. e.g. `./src/hyfi`
            version: Version of the package. e.g. `0.1.0`
            plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
            user_config_path: Path to the user configuration directory. e.g. `./config`
            config_dirname: Name of the configuration directory. e.g. `conf`
            dotenv_file: Name of the dotenv file. e.g. `.env`
            secrets_dir: Name of the secrets directory. e.g. `secrets`
            **kwargs: Additional arguments to be set as attributes.
        """
        global_hyfi.initialize(
            package_path=package_path,
            version=version,
            plugins=plugins,
            user_config_path=user_config_path,
            config_dirname=config_dirname,
            dotenv_file=dotenv_file,
            secrets_dir=secrets_dir,
            **kwargs,
        )

    @property
    def dryrun(self) -> bool:
        """Get the dryrun flag."""
        return self.config.dryrun or self.config.noop

    @property
    def variables(self) -> Variables:
        """Get the global variables."""
        if self.__variables__ is None:
            self.__variables__ = Variables()
        return self.__variables__

    @property
    def resolve(self) -> bool:
        """Get the resolve flag."""
        return self.config.resolve

    @property
    def verbose(self) -> bool:
        """Get the verbose flag."""
        return self.config.verbose or self.dryrun

    @property
    def app_name(self):
        """
        Get the name of the application.
        """
        return global_config.app_name

    @property
    def app_version(self):
        """
        Get the version of the application.
        """
        return global_config.app_version

    @staticmethod
    def print_about(**args) -> None:
        """Print the about information"""
        global_config.print_about(**args)

    @staticmethod
    def Variables(**kwargs) -> Variables:
        """
        Return the Variables.

        Args:
            **kwargs: Additional keyword arguments to pass to the Variables constructor.

        Returns:
            Variables: An instance of the Variables class.
        """
        return Variables(**kwargs)

    @staticmethod
    def JobLib(**kwargs) -> JobLib:
        """
        Return the joblib pipe.

        Args:
            **kwargs: Additional keyword arguments to pass to the JobLibConfig constructor.

        Returns:
            JobLibConfig: An instance of the JobLibConfig class.
        """
        return JobLib(**kwargs)

    @staticmethod
    def Env(**kwargs) -> Env:
        """
        Return the Env.

        Args:
            **kwargs: Additional keyword arguments to pass to the Env constructor.

        Returns:
            Env: An instance of the Env class.
        """
        return Env(**kwargs)

    @staticmethod
    def Task(**kwargs) -> Task:
        """
        Return the TaskConfig.

        Args:
            **kwargs: Additional keyword arguments to pass to the TaskConfig constructor.

        Returns:
            TaskConfig: An instance of the TaskConfig class.
        """
        return Task(**kwargs)

    @staticmethod
    def Workflow(**kwargs) -> Workflow:
        """
        Return the WorkflowConfig.

        Args:
            **kwargs: Additional keyword arguments to pass to the WorkflowConfig constructor.

        Returns:
            WorkflowConfig: An instance of the WorkflowConfig class.
        """
        return Workflow(**kwargs)

    ###############################
    # Pipeline related functions
    ###############################
    @staticmethod
    def run_command(**config):
        """Run a command"""
        return HyFI.run_config(config)

    @staticmethod
    def run(
        config_group: Optional[str] = None,
        overrides: Optional[List[str]] = None,
        config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
        global_package=False,
        dryrun=False,
        **kwargs,
    ):
        """Run the config by composing it and running it"""
        if config_group:
            logger.info("Composing the HyFI config from config group %s", config_group)
            config = HyFI.compose_as_dict(
                config_group=config_group,
                overrides=overrides,
                config_data=config_data,
                global_package=global_package,
            )
        HyFI.run_config(config, dryrun=dryrun)

    @staticmethod
    def run_intantiatable(
        config: Dict[str, Any],
        dryrun=False,
    ):
        """Run the config by composing it and running it"""
        logger.info("Instantiating the HyFI config")
        if dryrun:
            print("\nDryrun is enabled, not running the HyFI config\n")
            return
        task = HyFI.instantiate(config)
        if task and getattr(task, "__call__", None):
            logger.info("The HyFI config is callable, running it")
            task()

    @staticmethod
    def run_config(
        config: Union[Dict[str, Any], DictConfig],
        dryrun=False,
    ):
        """Run the provided config"""
        config = HyFI.to_dict(config) if config else {}
        # Check if the config is instantiatable
        if HyFI.is_instantiatable(config):
            HyFI.run_intantiatable(config, dryrun=dryrun)
            return

        logger.info(
            "The HyFI config is not instantiatable, running HyFI task with the config"
        )
        # Run the HyFI task
        cmd_name = config.get("cmd_name")
        config_group = config.get("_config_group_", "")
        if not cmd_name:
            if config_group == "/workflow":
                cmd_name = "run_workflow"
            elif "task" in config:
                cmd_name = "run_task"
            elif "copier" in config:
                cmd_name = "copy_conf"

        if cmd_name == "run_workflow":
            workflow = HyFI.Workflow(**config)
            HyFI.run_workflow(workflow, dryrun=dryrun)
        elif cmd_name == "run_task":
            if "project" in config:
                HyFI.initialize(**config["project"])
            task = HyFI.Task(**config["task"])
            HyFI.run_task(task, dryrun=dryrun)
        elif cmd_name == "copy_conf":
            copier_cfg = config["copier"]
            copier_cfg["dryrun"] = dryrun
            with Copier(**copier_cfg) as worker:
                worker.run_copy()
        else:
            for _, cfg in config.items():
                if HyFI.is_instantiatable(cfg):
                    HyFI.run_intantiatable(cfg, dryrun)
                    return
            if not dryrun:
                HyFI.print_about(**config.get("about", {}))

app_name property

Get the name of the application.

app_version property

Get the version of the application.

config: HyFIConfig property

Get the global config.

dryrun: bool property

Get the dryrun flag.

project: Optional[Project] property writable

Get the project.

resolve: bool property

Get the resolve flag.

variables: Variables property

Get the global variables.

verbose: bool property

Get the verbose flag.

Env(**kwargs) staticmethod

Return the Env.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments to pass to the Env constructor.	{}

Returns:

Name	Type	Description
Env	Env	An instance of the Env class.

Source code in hyfi/main/main.py

@staticmethod
def Env(**kwargs) -> Env:
    """
    Return the Env.

    Args:
        **kwargs: Additional keyword arguments to pass to the Env constructor.

    Returns:
        Env: An instance of the Env class.
    """
    return Env(**kwargs)

JobLib(**kwargs) staticmethod

Return the joblib pipe.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments to pass to the JobLibConfig constructor.	{}

Returns:

Name	Type	Description
JobLibConfig	JobLib	An instance of the JobLibConfig class.

Source code in hyfi/main/main.py

@staticmethod
def JobLib(**kwargs) -> JobLib:
    """
    Return the joblib pipe.

    Args:
        **kwargs: Additional keyword arguments to pass to the JobLibConfig constructor.

    Returns:
        JobLibConfig: An instance of the JobLibConfig class.
    """
    return JobLib(**kwargs)

Task(**kwargs) staticmethod

Return the TaskConfig.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments to pass to the TaskConfig constructor.	{}

Returns:

Name	Type	Description
TaskConfig	Task	An instance of the TaskConfig class.

Source code in hyfi/main/main.py

@staticmethod
def Task(**kwargs) -> Task:
    """
    Return the TaskConfig.

    Args:
        **kwargs: Additional keyword arguments to pass to the TaskConfig constructor.

    Returns:
        TaskConfig: An instance of the TaskConfig class.
    """
    return Task(**kwargs)

Variables(**kwargs) staticmethod

Return the Variables.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments to pass to the Variables constructor.	{}

Returns:

Name	Type	Description
Variables	Variables	An instance of the Variables class.

Source code in hyfi/main/main.py

@staticmethod
def Variables(**kwargs) -> Variables:
    """
    Return the Variables.

    Args:
        **kwargs: Additional keyword arguments to pass to the Variables constructor.

    Returns:
        Variables: An instance of the Variables class.
    """
    return Variables(**kwargs)

Workflow(**kwargs) staticmethod

Return the WorkflowConfig.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments to pass to the WorkflowConfig constructor.	{}

Returns:

Name	Type	Description
WorkflowConfig	Workflow	An instance of the WorkflowConfig class.

Source code in hyfi/main/main.py

@staticmethod
def Workflow(**kwargs) -> Workflow:
    """
    Return the WorkflowConfig.

    Args:
        **kwargs: Additional keyword arguments to pass to the WorkflowConfig constructor.

    Returns:
        WorkflowConfig: An instance of the WorkflowConfig class.
    """
    return Workflow(**kwargs)

initialize(project_name=None, project_description=None, project_root=None, project_workspace_name=None, global_hyfi_root=None, global_workspace_name=None, num_workers=None, logging_level=None, plugins=None, user_config_path=None, dotenv_file=None, secrets_dir=None, reinit=True, autotime=True, retina=True, verbose=False, **project_kwargs) staticmethod

Initialize and start hyfi.

Parameters:

Name	Type	Description	Default
project_name	Optional[str]	Name of the project to use.	None
project_description	Optional[str]	Description of the project that will be used.	None
project_root	Optional[str]	Root directory of the project.	None
project_workspace_name	Optional[str]	Name of the project's workspace directory.	None
global_hyfi_root	Optional[str]	Root directory of the global hyfi.	None
global_workspace_name	Optional[str]	Name of the global hierachical workspace directory.	None
num_workers	Optional[int]	Number of workers to run.	None
logging_level	Optional[str]	Log level for the log.	None
plugins	Optional[List[str]]	A list of plugins to load. e.g. ["hyfi.conf"]	None
user_config_path	Optional[str]	Path to the user configuration directory. e.g. ./config	None
config_dirname		Name of the configuration directory. e.g. conf	required
dotenv_file	Optional[str]	Name of the dotenv file. e.g. .env	None
secrets_dir	Optional[str]	Name of the secrets directory. e.g. secrets	None
reinit	bool	Whether to reinitialize the global config.	True
autotime	bool	Whether to automatically set time and / or keep track of run times.	True
retina	bool	Whether to use retina or not.	True
verbose	Union[bool, int]	Enables or disables logging	False

Source code in hyfi/main/main.py

@staticmethod
def initialize(
    project_name: Optional[str] = None,
    project_description: Optional[str] = None,
    project_root: Optional[str] = None,
    project_workspace_name: Optional[str] = None,
    global_hyfi_root: Optional[str] = None,
    global_workspace_name: Optional[str] = None,
    num_workers: Optional[int] = None,
    logging_level: Optional[str] = None,
    plugins: Optional[List[str]] = None,
    user_config_path: Optional[str] = None,
    dotenv_file: Optional[str] = None,
    secrets_dir: Optional[str] = None,
    reinit: bool = True,
    autotime: bool = True,
    retina: bool = True,
    verbose: Union[bool, int] = False,
    **project_kwargs,
) -> "HyFI":
    """
    Initialize and start hyfi.

    Args:
        project_name: Name of the project to use.
        project_description: Description of the project that will be used.
        project_root: Root directory of the project.
        project_workspace_name: Name of the project's workspace directory.
        global_hyfi_root: Root directory of the global hyfi.
        global_workspace_name: Name of the global hierachical workspace directory.
        num_workers: Number of workers to run.
        logging_level: Log level for the log.
        plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
        user_config_path: Path to the user configuration directory. e.g. `./config`
        config_dirname: Name of the configuration directory. e.g. `conf`
        dotenv_file: Name of the dotenv file. e.g. `.env`
        secrets_dir: Name of the secrets directory. e.g. `secrets`
        reinit: Whether to reinitialize the global config.
        autotime: Whether to automatically set time and / or keep track of run times.
        retina: Whether to use retina or not.
        verbose: Enables or disables logging
    """
    global_config.inititialize(
        project_name=project_name,
        project_description=project_description,
        project_root=project_root,
        project_workspace_name=project_workspace_name,
        global_hyfi_root=global_hyfi_root,
        global_workspace_name=global_workspace_name,
        num_workers=num_workers,
        logging_level=logging_level,
        plugins=plugins,
        user_config_path=user_config_path,
        dotenv_file=dotenv_file,
        secrets_dir=secrets_dir,
        reinit=reinit,
        autotime=autotime,
        retina=retina,
        verbose=verbose,
        **HyFI.to_dict(project_kwargs),
    )
    return HyFI()

initialize_global_hyfi(package_path, version, plugins=None, user_config_path=None, config_dirname=None, dotenv_file=None, secrets_dir=None, **kwargs) staticmethod

Initializes the global HyFI instance. This function should be called before any other HyFI function. A plugin is a python module which contains a configuration module.

Parameters:

Name	Type	Description	Default
package_path	str	Path to the package root folder. e.g. ./src/hyfi	required
version	str	Version of the package. e.g. 0.1.0	required
plugins	Optional[List[str]]	A list of plugins to load. e.g. ["hyfi.conf"]	None
user_config_path	Optional[str]	Path to the user configuration directory. e.g. ./config	None
config_dirname	Optional[str]	Name of the configuration directory. e.g. conf	None
dotenv_file	Optional[str]	Name of the dotenv file. e.g. .env	None
secrets_dir	Optional[str]	Name of the secrets directory. e.g. secrets	None
**kwargs		Additional arguments to be set as attributes.	{}

Source code in hyfi/main/main.py

@staticmethod
def initialize_global_hyfi(
    package_path: str,
    version: str,
    plugins: Optional[List[str]] = None,
    user_config_path: Optional[str] = None,
    config_dirname: Optional[str] = None,
    dotenv_file: Optional[str] = None,
    secrets_dir: Optional[str] = None,
    **kwargs,
) -> None:
    """
    Initializes the global HyFI instance.
    This function should be called before any other HyFI function.
    A plugin is a python module which contains a configuration module.

    Args:
        package_path: Path to the package root folder. e.g. `./src/hyfi`
        version: Version of the package. e.g. `0.1.0`
        plugins: A list of plugins to load. e.g. `["hyfi.conf"]`
        user_config_path: Path to the user configuration directory. e.g. `./config`
        config_dirname: Name of the configuration directory. e.g. `conf`
        dotenv_file: Name of the dotenv file. e.g. `.env`
        secrets_dir: Name of the secrets directory. e.g. `secrets`
        **kwargs: Additional arguments to be set as attributes.
    """
    global_hyfi.initialize(
        package_path=package_path,
        version=version,
        plugins=plugins,
        user_config_path=user_config_path,
        config_dirname=config_dirname,
        dotenv_file=dotenv_file,
        secrets_dir=secrets_dir,
        **kwargs,
    )

print_about(**args) staticmethod

Print the about information

Source code in hyfi/main/main.py

@staticmethod
def print_about(**args) -> None:
    """Print the about information"""
    global_config.print_about(**args)

run(config_group=None, overrides=None, config_data=None, global_package=False, dryrun=False, **kwargs) staticmethod

Run the config by composing it and running it

Source code in hyfi/main/main.py

@staticmethod
def run(
    config_group: Optional[str] = None,
    overrides: Optional[List[str]] = None,
    config_data: Optional[Union[Dict[str, Any], DictConfig]] = None,
    global_package=False,
    dryrun=False,
    **kwargs,
):
    """Run the config by composing it and running it"""
    if config_group:
        logger.info("Composing the HyFI config from config group %s", config_group)
        config = HyFI.compose_as_dict(
            config_group=config_group,
            overrides=overrides,
            config_data=config_data,
            global_package=global_package,
        )
    HyFI.run_config(config, dryrun=dryrun)

run_command(**config) staticmethod

Run a command

Source code in hyfi/main/main.py

@staticmethod
def run_command(**config):
    """Run a command"""
    return HyFI.run_config(config)

run_config(config, dryrun=False) staticmethod

Run the provided config

Source code in hyfi/main/main.py

@staticmethod
def run_config(
    config: Union[Dict[str, Any], DictConfig],
    dryrun=False,
):
    """Run the provided config"""
    config = HyFI.to_dict(config) if config else {}
    # Check if the config is instantiatable
    if HyFI.is_instantiatable(config):
        HyFI.run_intantiatable(config, dryrun=dryrun)
        return

    logger.info(
        "The HyFI config is not instantiatable, running HyFI task with the config"
    )
    # Run the HyFI task
    cmd_name = config.get("cmd_name")
    config_group = config.get("_config_group_", "")
    if not cmd_name:
        if config_group == "/workflow":
            cmd_name = "run_workflow"
        elif "task" in config:
            cmd_name = "run_task"
        elif "copier" in config:
            cmd_name = "copy_conf"

    if cmd_name == "run_workflow":
        workflow = HyFI.Workflow(**config)
        HyFI.run_workflow(workflow, dryrun=dryrun)
    elif cmd_name == "run_task":
        if "project" in config:
            HyFI.initialize(**config["project"])
        task = HyFI.Task(**config["task"])
        HyFI.run_task(task, dryrun=dryrun)
    elif cmd_name == "copy_conf":
        copier_cfg = config["copier"]
        copier_cfg["dryrun"] = dryrun
        with Copier(**copier_cfg) as worker:
            worker.run_copy()
    else:
        for _, cfg in config.items():
            if HyFI.is_instantiatable(cfg):
                HyFI.run_intantiatable(cfg, dryrun)
                return
        if not dryrun:
            HyFI.print_about(**config.get("about", {}))

run_intantiatable(config, dryrun=False) staticmethod

Run the config by composing it and running it

Source code in hyfi/main/main.py

@staticmethod
def run_intantiatable(
    config: Dict[str, Any],
    dryrun=False,
):
    """Run the config by composing it and running it"""
    logger.info("Instantiating the HyFI config")
    if dryrun:
        print("\nDryrun is enabled, not running the HyFI config\n")
        return
    task = HyFI.instantiate(config)
    if task and getattr(task, "__call__", None):
        logger.info("The HyFI config is callable, running it")
        task()

set_project(project) staticmethod

Set the project.

Parameters:

Name	Type	Description	Default
project	Project	Project to set.	required

Source code in hyfi/main/main.py

@staticmethod
def set_project(project: Project) -> None:
    """
    Set the project.

    Args:
        project: Project to set.
    """
    logger.info(f"Setting the global project to {project.project_name}")
    global_config.project = project

terminate() staticmethod

Terminate the global config.

Returns:

Name	Type	Description
bool	None	True if termination was successful, False otherwise.

Source code in hyfi/main/main.py

@staticmethod
def terminate() -> None:
    """
    Terminate the global config.

    Returns:
        bool: True if termination was successful, False otherwise.
    """
    global_config.terminate()

HyFIConfig

Bases: BaseModel

HyFI root config class. This class is used to store the configuration

Source code in hyfi/main/config.py

class HyFIConfig(BaseModel):
    """HyFI root config class.  This class is used to store the configuration"""

    _config_name_: str = "config"
    _config_group_: str = "/"

    debug_mode: bool = False
    noop: bool = False
    dryrun: bool = False
    resolve: bool = False
    verbose: bool = False
    logging_level: str = "WARNING"

    hydra: Optional[ConfigType] = None

    about: Optional[ConfigType] = None
    copier: Optional[ConfigType] = None
    project: Optional[ConfigType] = None
    pipeline: Optional[ConfigType] = None
    task: Optional[ConfigType] = None
    variables: Optional[ConfigType] = None
    workflow: Optional[ConfigType] = None
    tasks: Optional[List[str]] = None
    workflows: Optional[List[str]] = None

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

    @field_validator("logging_level")
    def _validate_logging_level(cls, v, info: ValidationInfo):
        """
        Validate and set the logging level
        """
        verbose = info.data.get("verbose", False)
        # Set verbose to INFO.
        if verbose and v == "WARNING":
            v = "INFO"
        logger.setLevel(v)
        return v