CLI

kedro_aim.framework.cli.cli

kedro_aim.framework.cli.cli.KedroClickGroup

Bases: click.Group

The main entry point for the Kedro CLI.

Source code in kedro_aim/framework/cli/cli.py

class KedroClickGroup(click.Group):
    """The main entry point for the Kedro CLI."""

    def reset_commands(self) -> None:
        """Reset the commands to the default ones."""
        self.commands: Dict[str, Command] = {}

        # add commands on the fly based on conditions
        if _is_project(Path.cwd()):
            self.add_command(init)  # type: ignore
            self.add_command(ui)  # type: ignore

    def list_commands(self, ctx: Context) -> List[str]:
        """List the names of all commands.

        Args:
            ctx: The click context.

        Returns:
            A list of command names.
        """
        self.reset_commands()
        commands_list = sorted(self.commands)
        return commands_list

    def get_command(self, ctx: Context, cmd_name: str) -> Optional[click.Command]:
        """Get a click command by name.

        Args:
            ctx: The click context.
            cmd_name: The name of the command.

        Returns:
            The click command with the given name.
        """
        self.reset_commands()
        return self.commands.get(cmd_name)

kedro_aim.framework.cli.cli.KedroClickGroup.get_command(ctx, cmd_name)

Get a click command by name.

Parameters:

Name	Type	Description	Default
ctx	Context	The click context.	required
cmd_name	str	The name of the command.	required

Returns:

Type	Description
Optional[click.Command]	The click command with the given name.

Source code in kedro_aim/framework/cli/cli.py

def get_command(self, ctx: Context, cmd_name: str) -> Optional[click.Command]:
    """Get a click command by name.

    Args:
        ctx: The click context.
        cmd_name: The name of the command.

    Returns:
        The click command with the given name.
    """
    self.reset_commands()
    return self.commands.get(cmd_name)

kedro_aim.framework.cli.cli.KedroClickGroup.list_commands(ctx)

List the names of all commands.

Parameters:

Name	Type	Description	Default
ctx	Context	The click context.	required

Returns:

Type	Description
List[str]	A list of command names.

Source code in kedro_aim/framework/cli/cli.py

def list_commands(self, ctx: Context) -> List[str]:
    """List the names of all commands.

    Args:
        ctx: The click context.

    Returns:
        A list of command names.
    """
    self.reset_commands()
    commands_list = sorted(self.commands)
    return commands_list

kedro_aim.framework.cli.cli.KedroClickGroup.reset_commands()

Reset the commands to the default ones.

Source code in kedro_aim/framework/cli/cli.py

def reset_commands(self) -> None:
    """Reset the commands to the default ones."""
    self.commands: Dict[str, Command] = {}

    # add commands on the fly based on conditions
    if _is_project(Path.cwd()):
        self.add_command(init)  # type: ignore
        self.add_command(ui)  # type: ignore

kedro_aim.framework.cli.cli.aim_commands()

Use aim-specific commands inside kedro project.

Source code in kedro_aim/framework/cli/cli.py

@commands.command(name="aim", cls=KedroClickGroup)
def aim_commands() -> None:
    """Use aim-specific commands inside kedro project."""
    pass  # pragma: no cover

kedro_aim.framework.cli.cli.commands()

Kedro plugin for interactions with aim.

Source code in kedro_aim/framework/cli/cli.py

@click.group(name="Aim")
def commands() -> None:
    """Kedro plugin for interactions with aim."""
    pass  # pragma: no cover

kedro_aim.framework.cli.cli.init(env, force, silent)

Updates the template of a kedro project.

Running this command is mandatory to use kedro-aim. This adds "conf/base/aim.yml": This is a configuration file used for run parametrization when calling "kedro run" command.

Source code in kedro_aim/framework/cli/cli.py

@aim_commands.command()  # type: ignore
@click.option(
    "--env",
    "-e",
    default="local",
    help=(
        "The name of the kedro environment where the 'aim.yml' should be created. "
        "Default to 'local'"
    ),
)
@click.option(
    "--force",
    "-f",
    is_flag=True,
    default=False,
    help="Update the template without any checks.",
)
@click.option(
    "--silent",
    "-s",
    is_flag=True,
    default=False,
    help="Should message be logged when files are modified?",
)
def init(env: str, force: bool, silent: bool) -> None:
    """Updates the template of a kedro project.

    Running this command is mandatory to use kedro-aim.
    This adds "conf/base/aim.yml": This is a configuration file
    used for run parametrization when calling "kedro run" command.
    """
    # get constants
    aim_yml = "aim.yml"
    project_path = Path().cwd()
    project_metadata = bootstrap_project(project_path)
    aim_yml_path: Path = (
        project_path / settings.CONF_SOURCE / env / aim_yml  # type: ignore
    )

    # aim.yml is just a static file,
    # but the name of the experiment is set to be the same as the project
    if aim_yml_path.is_file() and not force:
        click.secho(
            click.style(
                (
                    f"A 'aim.yml' already exists at '{aim_yml_path}'. "
                    "You can use the ``--force`` option to override it."
                ),
                fg="red",
            )
        )
    else:
        try:
            write_jinja_template(
                src=TEMPLATE_FOLDER_PATH / aim_yml,
                is_cookiecutter=False,
                dst=aim_yml_path,
                python_package=project_metadata.package_name,
            )
            if not silent:
                click.secho(
                    click.style(
                        (
                            f"'{settings.CONF_SOURCE}/{env}/{aim_yml}' "
                            "successfully updated."
                        ),
                        fg="green",
                    )
                )
        except FileNotFoundError:
            click.secho(
                click.style(
                    (
                        f"No env '{env}' found. "
                        "Please check this folder exists inside "
                        f"'{settings.CONF_SOURCE}' folder.",
                    ),
                    fg="red",
                )
            )

kedro_aim.framework.cli.cli.ui(env, port, host)

Start the aim UI.

Opens the aim user interface with the project-specific settings of aim.yml. This interface enables to browse and compares runs.

Source code in kedro_aim/framework/cli/cli.py

@aim_commands.command()  # type: ignore
@click.option(
    "--env",
    "-e",
    required=False,
    default="local",
    help="The environment within conf folder we want to retrieve.",
)
@click.option(
    "--port",
    "-p",
    required=False,
    type=int,
    help="The port to listen on",
)
@click.option(
    "--host",
    "-h",
    required=False,
    help=(
        "The network address to listen on (default: 127.0.0.1). "
        "Use 0.0.0.0 to bind to all addresses if you want to access the tracking "
        "server from other machines."
    ),
)
def ui(env: str, port: int, host: str) -> None:
    """Start the aim UI.

    Opens the aim user interface with the project-specific settings of aim.yml.
    This interface enables to browse and compares runs.
    """
    project_path = Path().cwd()
    bootstrap_project(project_path)
    with KedroSession.create(
        project_path=project_path,
        env=env,
    ) as session:

        context = session.load_context()
        aim_config: KedroAimConfig = context.aim  # type: ignore
        host = host or aim_config.ui.host
        port = port or aim_config.ui.port

        subprocess.call(
            [
                "aim",
                "up",
                "--port",
                str(port),
                "--host",
                host,
            ]
        )

kedro_aim.framework.cli.cli_utils

kedro_aim.framework.cli.cli_utils.render_jinja_template(src, is_cookiecutter=False, **kwargs)

Render a template file and replace tis jinja's tags.

This functions enable to copy a file and render the tags (identified by {{ my_tag }}) with the values provided in kwargs.

Parameters:

Name	Type	Description	Default
src	Union[str, Path]	The path to the template which should be rendered	required
is_cookiecutter	bool	If the template is a cookiecutter template	False
**kwargs	Any	Extra arguments to be passed to the template	{}

Returns:

Type	Description
str	A string that contains all the files with replaced tags.

Source code in kedro_aim/framework/cli/cli_utils.py

def render_jinja_template(
    src: Union[str, Path], is_cookiecutter: bool = False, **kwargs: Any
) -> str:  # pragma: no cover
    """Render a template file and replace tis jinja's tags.

    This functions enable to copy a file and render the tags (identified by
    `{{ my_tag }}`) with the values provided in kwargs.

    Args:
        src:  The path to the template which should be rendered
        is_cookiecutter: If the template is a cookiecutter template
        **kwargs: Extra arguments to be passed to the template

    Returns:
        A string that contains all the files with replaced tags.
    """
    src = Path(src)

    template_loader = FileSystemLoader(searchpath=src.parent.as_posix())
    # the keep_trailing_new_line option is mandatory to
    # make sure that black formatting will be preserved
    template_env = Environment(loader=template_loader, keep_trailing_newline=True)
    template = template_env.get_template(src.name)
    if is_cookiecutter:
        # we need to match tags from a cookiecutter object
        # but cookiecutter only deals with folder, not file
        # thus we need to create an object with all necessary attributes
        class FalseCookieCutter:
            def __init__(self, **kwargs: Any) -> None:
                self.__dict__.update(kwargs)

        parsed_template = template.render(cookiecutter=FalseCookieCutter(**kwargs))
    else:
        parsed_template = template.render(**kwargs)

    return parsed_template

kedro_aim.framework.cli.cli_utils.write_jinja_template(src, dst, **kwargs)

Write a template file and replace tis jinja's tags.

Write a template file and replace tis jinja's tags (identified by {{ my_tag }}) with the values provided in kwargs.

Parameters:

Name	Type	Description	Default
src	Union[str, Path]	Path to the template which should be rendered	required
dst	Union[str, Path]	Path where the rendered template should be saved	required
**kwargs	Any	Extra arguments to be passed to the template	{}

Source code in kedro_aim/framework/cli/cli_utils.py

def write_jinja_template(
    src: Union[str, Path], dst: Union[str, Path], **kwargs: Any
) -> None:
    """Write a template file and replace tis jinja's tags.

    Write a template file and replace tis jinja's tags (identified by `{{ my_tag }}`)
    with the values provided in kwargs.

    Args:
        src: Path to the template which should be rendered
        dst: Path where the rendered template should be saved
        **kwargs: Extra arguments to be passed to the template
    """
    dst = Path(dst)
    parsed_template = render_jinja_template(src, **kwargs)
    with open(dst, "w") as file_handler:
        file_handler.write(parsed_template)