29761095343

Committed 20 Jul 2026 04:47PM UTC coverage: 79.991% (+1.4%) from 78.598%

Build # 29761095343

Build Type

Pull #201

github

Committed by

wyli

Commit Message

feat: add structured dry-run command plans

Co-authored-by: Codex <noreply@openai.com>

Pull Request Pull Request #201: feat: add JSON and shell dry-run plans for build-container

Coverage Stats

273 of 300 new or added lines in 7 files covered. (91.0%)

3 existing lines in 3 files now uncovered.

3386 of 4233 relevant lines covered (79.99%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.03

/src/holoscan_cli/cli.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# See README.md for command and flag reference.

import sys

# Python version check - must be before other imports that use Python 3.10+ features
PYTHON_MIN_VERSION = (3, 10, 0)
if sys.version_info < PYTHON_MIN_VERSION:
    sys_major, sys_minor, sys_micro = sys.version_info[:3]
    print(
        f"Error: Python {'.'.join(map(str, PYTHON_MIN_VERSION))} or higher required, "
        f"found {sys_major}.{sys_minor}.{sys_micro}",
        file=sys.stderr,
    )
    sys.exit(1)

# ruff: noqa: E402  # Imports after python version check
import argparse
import functools
import os
from contextlib import redirect_stdout
from io import StringIO
from pathlib import Path
from typing import List, Optional

import holoscan_cli.metadata.gather_metadata as metadata_util
from holoscan_cli.command_plan import CommandPlanError, PlanRecorder
from holoscan_cli.commands import registry as commands_registry
from holoscan_cli.container import HoloscanContainer
from holoscan_cli.container.parsers import get_build_argparse, get_run_argparse
from holoscan_cli.metadata.utils import (
    list_normalized_languages,
    normalize_language,
)
from holoscan_cli.utils.holohub import (
    get_component_search_paths,
    get_holohub_root,
    resolve_path_prefix,
)
from holoscan_cli.utils.io import Color, fatal, warn
from holoscan_cli.utils.text import levenshtein_distance, normalize_args_str


def in_container_cli_command() -> str:
    """Command used when the host CLI recurses into a container build/run/install.

    Returns the installed ``holoscan`` console script by default. Decoupled from
    ``HoloscanCLI.script_name`` so the in-container recursion is independent of how
    the user invoked the CLI on the host (e.g. via ``./holohub``, ``./i4h``, or
    ``python -m holoscan_cli``). Override via ``HOLOSCAN_CLI_IN_CONTAINER_CMD``
    when the container ships a different entry point (e.g. ``python3 -m
    holoscan_cli``).
    """
    return os.environ.get("HOLOSCAN_CLI_IN_CONTAINER_CMD", "holoscan")


class HoloscanCLI:
    """Command-line interface for Holoscan source-project workflows."""

    HOLOHUB_ROOT = get_holohub_root()
    DEFAULT_BUILD_PARENT_DIR = Path(
        os.environ.get("HOLOSCAN_CLI_BUILD_PARENT_DIR", HOLOHUB_ROOT / "build")
    )
    DEFAULT_DATA_DIR = Path(os.environ.get("HOLOSCAN_CLI_DATA_DIR", HOLOHUB_ROOT / "data"))
    DEFAULT_SDK_DIR = os.environ.get("HOLOSCAN_CLI_DEFAULT_HSDK_DIR", "/opt/nvidia/holoscan")
    # Allow overriding the default CTest script path via environment variable
    DEFAULT_CTEST_SCRIPT = os.environ.get(
        "HOLOSCAN_CLI_CTEST_SCRIPT",
        str(Path(__file__).resolve().parent / "testing" / "container.ctest"),
    )

    def __init__(self, script_name: Optional[str] = None):
        self.script_name = script_name or os.environ.get("HOLOSCAN_CLI_CMD_NAME", "holoscan")
        self.parser = self._create_parser()
        # Cache for resolved projects to avoid duplicate lookups
        self._project_data: dict[tuple[str, str], dict] = {}
        self.prefix = resolve_path_prefix(None)

    def _create_parser(self) -> argparse.ArgumentParser:
        """Create the argument parser with all supported commands.

        Subparser construction is delegated to per-command modules under
        :mod:`holoscan_cli.commands`; the wiring lives in
        :func:`holoscan_cli.commands.registry.register_all`. Help strings
        come from the registry so the top-level dispatch surface and the
        per-subparser help cannot drift.
        """
        parser = argparse.ArgumentParser(
            prog=self.script_name,
            description=(
                f"{self.script_name} CLI tool for managing Holoscan-based "
                "applications and containers"
            ),
        )
        subparsers = parser.add_subparsers(dest="command", required=True)

        # Cache subparsers so HoloscanCLI.run() can render targeted error
        # messages and Levenshtein-based suggestions.
        self.subparsers: dict[str, argparse.ArgumentParser] = commands_registry.register_all(
            self,
            subparsers,
            container_build=get_build_argparse(),
            container_run=get_run_argparse(),
        )

        return parser

    @functools.cached_property
    def projects(self) -> list[dict]:
        """All discovered source-project metadata, computed on first access.

        Walking the project search paths and parsing every ``metadata.json``
        costs tens-to-hundreds of small file reads on a populated HoloHub
        clone, so it's deferred off the ``__init__`` path. Commands that
        never touch project metadata (``status``, ``env-info``, ``env-check``,
        ``clear-cache``, ``setup``, ``create``, ``lint``) don't
        pay the cost.
        """
        # Known exceptions: templates that don't represent a standalone project.
        EXCLUDE_PATHS = ["applications/holoviz/template", "applications/template"]
        app_paths = get_component_search_paths(self.HOLOHUB_ROOT)
        return metadata_util.gather_metadata(app_paths, exclude_paths=EXCLUDE_PATHS)

    def find_project(self, project_name: str, language: Optional[str] = None) -> dict:
        """Find a project by name"""
        normalized_language = normalize_language(language)

        cache_key = (project_name, normalized_language)
        if cache_key in self._project_data:
            return self._project_data[cache_key]

        # Find all projects with the given name
        candidates = [p for p in self.projects if p.get("project_name") == project_name]
        if candidates:
            available_lang = []
            for p in candidates:
                for lang in list_normalized_languages(
                    p.get("metadata", {}).get("language", None), strict=True
                ):
                    available_lang.append(lang)
            available_lang = sorted(list(set(available_lang)))

            # Determine target language (if unspecified, prefer cpp then first available)
            if normalized_language:
                target_lang = normalized_language
            elif "python" in available_lang:
                target_lang = "python"
            else:
                target_lang = available_lang[0] if available_lang else ""
            # Warn if ambiguous and no language specified
            if not normalized_language and len(available_lang) > 1:
                msg = f"'{project_name}' has multiple languages: {', '.join(available_lang)}.\n"
                msg += f"Defaulting to '{target_lang}'. Use --language to select explicitly.\n\n"
                print(Color.green(msg))
            for p in candidates:
                if target_lang in list_normalized_languages(
                    p.get("metadata", {}).get("language", None), strict=True
                ):
                    self._project_data[cache_key] = p  # Return candidate matching target_lang
                    return p
            if normalized_language:  # If target_lang specified but not found
                fatal(
                    f"Project '{project_name}' (language: {normalized_language}) not found. "
                    f"Available: {', '.join(available_lang) if available_lang else 'unknown'}"
                )
            # No language info or no match found; return first candidate
            fallback_candidate = candidates[0]
            fallback_lang = fallback_candidate.get("metadata", {}).get("language", None)
            if not fallback_lang:
                msg = f"Returning '{project_name}' with missing or unknown language metadata.\n"
                msg += "Consider specifying --language for more consistent results.\n"
                warn(msg)
            self._project_data[cache_key] = fallback_candidate
            return self._project_data[cache_key]
        # If project not found, suggest similar names
        distances = [
            (
                p["project_name"],
                levenshtein_distance(project_name, p["project_name"]),
                p.get("source_folder", ""),
                p.get("metadata", {}).get("language", ""),
            )
            for p in self.projects
        ]
        distances.sort(key=lambda x: x[1])  # Sort by distance
        closest_matches = [
            (name, folder, lang) for name, dist, folder, lang in distances[:3] if dist <= 3
        ]  # Show up to 3 matches
        msg = f"Project '{project_name}' (language: {normalized_language}) not found."
        if closest_matches:
            msg += "\nDid you mean:"
            for name, folder, lang in closest_matches:
                details = []
                if lang:
                    details.append(f"language: {lang}")
                if folder:
                    details.append(f"source: {folder}")
                msg += f"\n  '{name}'" + (f" ({', '.join(details)})" if details else "")
        fatal(msg)

    def resolve_mode(self, project_data: dict, requested_mode: Optional[str] = None) -> tuple:
        """
        Resolve mode from metadata and validate
        Returns: (mode_name, mode_config) or (None, None) for legacy behavior
        """
        modes = project_data.get("metadata", {}).get("modes", {})
        if not modes:
            return None, None  # No modes defined - should use legacy behavior

        if requested_mode is None:
            # Validate that multiple modes have a default_mode specified
            application_metadata = project_data.get("metadata", {})
            if len(modes) > 1 and "default_mode" not in application_metadata:
                available = ", ".join(modes.keys())
                fatal(
                    f"Multiple modes found ({available}) but no 'default_mode' specified. "
                    f"Please add a 'default_mode' field to specify which mode to use by default."
                )

            if "default_mode" in application_metadata:
                requested_mode = application_metadata["default_mode"]
                # Validate that default_mode references an existing mode
                if requested_mode not in modes:
                    available = ", ".join(modes.keys())
                    fatal(f"Invalid default_mode '{requested_mode}' in metadata among {available}")
            else:
                requested_mode = list(modes.keys())[0]
        if requested_mode not in modes:
            available = ", ".join(modes.keys())
            fatal(f"Mode '{requested_mode}' not found. Available modes: {available}")
        return requested_mode, modes[requested_mode]

    def validate_mode(
        self,
        mode_name: Optional[str],
        mode_config: dict,
    ) -> None:
        """Validate mode configuration"""
        if not mode_config:
            return  # No mode configuration to validate

        # Define valid keys for mode configuration
        valid_top_level_keys = ["description", "requirements", "build", "run", "env"]
        valid_build_keys = ["depends", "docker_build_args", "cmake_options", "env"]
        valid_run_keys = ["command", "workdir", "docker_run_args", "env"]

        # Check top-level keys
        for key in mode_config.keys():
            if key not in valid_top_level_keys:
                suggestions = self._suggest_command(key, valid_top_level_keys)
                msg = f"Unknown key '{key}' in mode '{mode_name}'"
                if suggestions:
                    msg += f". Did you mean '{suggestions[0]}'?"
                warn(msg)

        # Check section keys (build and run)
        sections_to_validate = {"build": valid_build_keys, "run": valid_run_keys}
        for section_name, valid_keys in sections_to_validate.items():
            if section_name in mode_config and isinstance(mode_config[section_name], dict):
                for key in mode_config[section_name].keys():
                    if key not in valid_keys:
                        suggestions = self._suggest_command(key, valid_keys)
                        msg = f"Unknown key '{section_name}.{key}' in mode '{mode_name}'"
                        if suggestions:
                            msg += f". Did you mean '{suggestions[0]}'?"
                        warn(msg)

    def get_effective_build_config(
        self,
        args: argparse.Namespace,
        mode_config: dict,
    ) -> dict:
        """
        Get effective build configuration combining CLI args and mode config.
        """
        config = {
            "with_operators": getattr(args, "with_operators", None),
            "docker_opts": getattr(args, "docker_opts", ""),
            "build_args": getattr(args, "build_args", ""),
            "configure_args": getattr(args, "configure_args", None),
        }
        if not mode_config:
            return config

        # Apply build configuration - CLI parameters always override mode settings when provided
        if "build" in mode_config:
            build_config = mode_config["build"]

            if "depends" in build_config:
                if config["with_operators"]:
                    warn("CLI --build-with overrides mode build.depends")
                else:
                    mode_deps = [dep.strip() for dep in build_config["depends"] if dep.strip()]
                    config["with_operators"] = ";".join(mode_deps) if mode_deps else ""

            if "docker_build_args" in build_config:
                if config["build_args"]:
                    warn("CLI --build-args overrides mode build.docker_build_args")
                else:
                    config["build_args"] = normalize_args_str(build_config["docker_build_args"])

            if "cmake_options" in build_config:
                if config["configure_args"]:
                    warn("CLI --configure-args overrides mode build.cmake_options")
                else:
                    config["configure_args"] = build_config["cmake_options"]

        if "run" in mode_config and "docker_run_args" in mode_config["run"]:
            if getattr(args, "docker_opts", ""):
                warn("CLI --docker-opts overrides mode run.docker_run_args")
            else:
                config["docker_opts"] = normalize_args_str(mode_config["run"]["docker_run_args"])

        return config

    def get_effective_run_config(
        self,
        args: argparse.Namespace,
        mode_config: dict,
    ) -> dict:
        """Get effective run configuration combining CLI args and mode config without mutation"""
        config = {
            "run_args": getattr(args, "run_args", "") or "",
            "docker_opts": getattr(args, "docker_opts", ""),
        }

        if mode_config and "run" in mode_config:
            run_config = mode_config["run"]

            if "command" in run_config:
                config["command"] = run_config["command"]
            if "workdir" in run_config:
                config["workdir"] = run_config["workdir"]

            if "command" in run_config and getattr(args, "run_args", ""):
                warn("CLI --run-args will be appended to mode run.command")

            if "docker_run_args" in run_config:
                if getattr(args, "docker_opts", ""):
                    warn("CLI --docker-opts overrides mode run.docker_run_args")
                else:
                    config["docker_opts"] = normalize_args_str(run_config["docker_run_args"])
        return config

    def make_project_container(
        self, project_name: Optional[str] = None, language: Optional[str] = None
    ) -> HoloscanContainer:
        """Define a project container"""
        if not project_name:
            return HoloscanContainer(project_metadata=None)
        project_data = self.find_project(project_name=project_name, language=language)
        return HoloscanContainer(project_metadata=project_data, language=language)

    def collect_cache_dirs(self, patterns: list[str], default_dir=None) -> list:
        """Helper to collect cache directories matching patterns."""
        dirs = []
        if default_dir is not None:
            dirs.append(default_dir)
        for pattern in patterns:
            for path in HoloscanCLI.HOLOHUB_ROOT.glob(pattern):
                if path.is_dir() and path not in dirs:
                    dirs.append(path)
        return dirs

    def _suggest_command(self, invalid_value: str, valid_options: list[str]) -> list[str]:
        """Suggest similar values using Levenshtein distance."""
        distances = [
            (option, levenshtein_distance(invalid_value, option)) for option in valid_options
        ]
        distances.sort(key=lambda x: x[1])
        return [option for option, dist in distances[:2] if dist <= 2]  # Show up to 2 matches

    def _check_for_dash_prefix_issue(self, cmd_args: List[str]) -> Optional[str]:
        """
        Check if the parsing error is likely due to dash-prefixed arguments
        """
        DASH_VALUE_ARGS = ["--run-args", "--build-args", "--docker-opts", "--configure-args"]
        for i, arg in enumerate(cmd_args):
            if arg in DASH_VALUE_ARGS and "=" not in arg:
                if i + 1 < len(cmd_args) and cmd_args[i + 1].startswith("-"):
                    next_arg = cmd_args[i + 1]
                    return (
                        f"💡 Tip: ambiguous dash-prefixed arguments, use the equals format:\n"
                        f"   Instead of: {arg} {next_arg}\n"
                        f"   Use: {arg}={next_arg}"
                    )
        return None

    def run(self, argv: Optional[List[str]] = None) -> None:
        """Main entry point for the CLI"""

        trailing_docker_args = []  # Handle " -- " separator for run-container command forwarding
        if argv is None:
            argv = sys.argv
        argv = list(argv)
        cmd_args = argv[1:]  # Skip script name, return a copy of the args
        if len(cmd_args) >= 2 and cmd_args[0] == "run-container" and "--" in cmd_args:
            sep = cmd_args.index("--")
            cmd_args, trailing_docker_args = cmd_args[:sep], cmd_args[sep + 1 :]

        potential_command = cmd_args[0] if cmd_args else None
        dash_suggestion = None
        if potential_command and potential_command in self.subparsers:
            dash_suggestion = self._check_for_dash_prefix_issue(cmd_args)

        try:
            args = self.parser.parse_args(cmd_args)
            if trailing_docker_args:
                args._trailing_args = trailing_docker_args  # " -- " used for run-container command
        except SystemExit as e:
            if len(cmd_args) > 0 and e.code != 0:  # exit code is 0 => help was successfully shown
                if dash_suggestion:
                    print(f"\n{dash_suggestion}\n", file=sys.stderr)

                if potential_command and not potential_command.startswith("-"):
                    if potential_command in self.subparsers:
                        # Valid subcommand but parsing failed
                        print(f"\n💡 For more help with '{potential_command}':", file=sys.stderr)
                        print(f"  {self.script_name} {potential_command} --help\n", file=sys.stderr)
                        sys.exit(e.code if e.code is not None else 1)
                    else:  # Invalid subcommand - suggest similar ones
                        suggestions = self._suggest_command(
                            potential_command, list(self.subparsers.keys())
                        )
                        if suggestions:
                            print("\n💡 Did you mean:", file=sys.stderr)
                            for cmd in suggestions:
                                print(f"  {self.script_name} {cmd}", file=sys.stderr)
                            print(file=sys.stderr)
                        sys.exit(1)
            raise
        plan_format = getattr(args, "plan_format", None)
        if plan_format is not None:
            if not getattr(args, "_supports_plan", False):
                self.subparsers[args.command].error(
                    "structured command plans are not supported for this command"
                )
            if not getattr(args, "dryrun", False):
                self.subparsers[args.command].error(f"--{plan_format} requires --dryrun")
            self._run_structured_plan(args, plan_format)
        elif hasattr(args, "func"):
            args.func(args)
        else:
            self.parser.print_help()
            sys.exit(1)

    @staticmethod
    def _run_structured_plan(args: argparse.Namespace, plan_format: str) -> None:
        """Run one audited dry-run and write its artifact atomically to stdout."""

        recorder = PlanRecorder()
        legacy_stdout = StringIO()
        try:
            try:
                with recorder.activate(), redirect_stdout(legacy_stdout):
                    args.func(args)
            finally:
                notices = legacy_stdout.getvalue()
                if notices:
                    sys.stderr.write(notices)

            output = recorder.json_text() if plan_format == "json" else recorder.shell_text()
        except CommandPlanError as exc:
            print(f"Command planning failed: {exc}", file=sys.stderr)
            raise SystemExit(1) from exc

        if plan_format == "json":
            output += "\n"
        sys.stdout.write(output)


def main(argv: Optional[List[str]] = None):
    script_name = None
    if argv and not os.environ.get("HOLOSCAN_CLI_CMD_NAME"):
        executable = Path(argv[0]).name
        script_name = "holoscan" if executable == "__main__.py" else executable
    cli = HoloscanCLI(script_name=script_name)
    cli.run(argv)


if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2	# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
3	# SPDX-License-Identifier: Apache-2.0
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License");
6	# you may not use this file except in compliance with the License.
7	# You may obtain a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS,
13	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	# See the License for the specific language governing permissions and
15	# limitations under the License.
16
17	# See README.md for command and flag reference.
18
19	import sys	1✔
20
21	# Python version check - must be before other imports that use Python 3.10+ features
22	PYTHON_MIN_VERSION = (3, 10, 0)	1✔
23	if sys.version_info < PYTHON_MIN_VERSION:	1✔
24	sys_major, sys_minor, sys_micro = sys.version_info[:3]	×
25	print(	×
26	f"Error: Python {'.'.join(map(str, PYTHON_MIN_VERSION))} or higher required, "
27	f"found {sys_major}.{sys_minor}.{sys_micro}",
28	file=sys.stderr,
29	)
30	sys.exit(1)	×
31
32	# ruff: noqa: E402 # Imports after python version check
33	import argparse	1✔
34	import functools	1✔
35	import os	1✔
36	from contextlib import redirect_stdout	1✔
37	from io import StringIO	1✔
38	from pathlib import Path	1✔
39	from typing import List, Optional	1✔
40
41	import holoscan_cli.metadata.gather_metadata as metadata_util	1✔
42	from holoscan_cli.command_plan import CommandPlanError, PlanRecorder	1✔
43	from holoscan_cli.commands import registry as commands_registry	1✔
44	from holoscan_cli.container import HoloscanContainer	1✔
45	from holoscan_cli.container.parsers import get_build_argparse, get_run_argparse	1✔
46	from holoscan_cli.metadata.utils import (	1✔
47	list_normalized_languages,
48	normalize_language,
49	)
50	from holoscan_cli.utils.holohub import (	1✔
51	get_component_search_paths,
52	get_holohub_root,
53	resolve_path_prefix,
54	)
55	from holoscan_cli.utils.io import Color, fatal, warn	1✔
56	from holoscan_cli.utils.text import levenshtein_distance, normalize_args_str	1✔
57
58
59	def in_container_cli_command() -> str:	1✔
60	"""Command used when the host CLI recurses into a container build/run/install.
61
62	Returns the installed ``holoscan`` console script by default. Decoupled from
63	``HoloscanCLI.script_name`` so the in-container recursion is independent of how
64	the user invoked the CLI on the host (e.g. via ``./holohub``, ``./i4h``, or
65	``python -m holoscan_cli``). Override via ``HOLOSCAN_CLI_IN_CONTAINER_CMD``
66	when the container ships a different entry point (e.g. ``python3 -m
67	holoscan_cli``).
68	"""
69	return os.environ.get("HOLOSCAN_CLI_IN_CONTAINER_CMD", "holoscan")	1✔
70
71
72	class HoloscanCLI:	1✔
73	"""Command-line interface for Holoscan source-project workflows."""
74
75	HOLOHUB_ROOT = get_holohub_root()	1✔
76	DEFAULT_BUILD_PARENT_DIR = Path(	1✔
77	os.environ.get("HOLOSCAN_CLI_BUILD_PARENT_DIR", HOLOHUB_ROOT / "build")
78	)
79	DEFAULT_DATA_DIR = Path(os.environ.get("HOLOSCAN_CLI_DATA_DIR", HOLOHUB_ROOT / "data"))	1✔
80	DEFAULT_SDK_DIR = os.environ.get("HOLOSCAN_CLI_DEFAULT_HSDK_DIR", "/opt/nvidia/holoscan")	1✔
81	# Allow overriding the default CTest script path via environment variable
82	DEFAULT_CTEST_SCRIPT = os.environ.get(	1✔
83	"HOLOSCAN_CLI_CTEST_SCRIPT",
84	str(Path(__file__).resolve().parent / "testing" / "container.ctest"),
85	)
86
87	def __init__(self, script_name: Optional[str] = None):	1✔
88	self.script_name = script_name or os.environ.get("HOLOSCAN_CLI_CMD_NAME", "holoscan")	1✔
89	self.parser = self._create_parser()	1✔
90	# Cache for resolved projects to avoid duplicate lookups
91	self._project_data: dict[tuple[str, str], dict] = {}	1✔
92	self.prefix = resolve_path_prefix(None)	1✔
93
94	def _create_parser(self) -> argparse.ArgumentParser:	1✔
95	"""Create the argument parser with all supported commands.
96
97	Subparser construction is delegated to per-command modules under
98	:mod:`holoscan_cli.commands`; the wiring lives in
99	:func:`holoscan_cli.commands.registry.register_all`. Help strings
100	come from the registry so the top-level dispatch surface and the
101	per-subparser help cannot drift.
102	"""
103	parser = argparse.ArgumentParser(	1✔
104	prog=self.script_name,
105	description=(
106	f"{self.script_name} CLI tool for managing Holoscan-based "
107	"applications and containers"
108	),
109	)
110	subparsers = parser.add_subparsers(dest="command", required=True)	1✔
111
112	# Cache subparsers so HoloscanCLI.run() can render targeted error
113	# messages and Levenshtein-based suggestions.
114	self.subparsers: dict[str, argparse.ArgumentParser] = commands_registry.register_all(	1✔
115	self,
116	subparsers,
117	container_build=get_build_argparse(),
118	container_run=get_run_argparse(),
119	)
120
121	return parser	1✔
122
123	@functools.cached_property	1✔
124	def projects(self) -> list[dict]:	1✔
125	"""All discovered source-project metadata, computed on first access.
126
127	Walking the project search paths and parsing every ``metadata.json``
128	costs tens-to-hundreds of small file reads on a populated HoloHub
129	clone, so it's deferred off the ``__init__`` path. Commands that
130	never touch project metadata (``status``, ``env-info``, ``env-check``,
131	``clear-cache``, ``setup``, ``create``, ``lint``) don't
132	pay the cost.
133	"""
134	# Known exceptions: templates that don't represent a standalone project.
135	EXCLUDE_PATHS = ["applications/holoviz/template", "applications/template"]	1✔
136	app_paths = get_component_search_paths(self.HOLOHUB_ROOT)	1✔
137	return metadata_util.gather_metadata(app_paths, exclude_paths=EXCLUDE_PATHS)	1✔
138
139	def find_project(self, project_name: str, language: Optional[str] = None) -> dict:	1✔
140	"""Find a project by name"""
141	normalized_language = normalize_language(language)	1✔
142
143	cache_key = (project_name, normalized_language)	1✔
144	if cache_key in self._project_data:	1✔
145	return self._project_data[cache_key]	×
146
147	# Find all projects with the given name
148	candidates = [p for p in self.projects if p.get("project_name") == project_name]	1✔
149	if candidates:	1✔
150	available_lang = []	1✔
151	for p in candidates:	1✔
152	for lang in list_normalized_languages(	1✔
153	p.get("metadata", {}).get("language", None), strict=True
154	):
155	available_lang.append(lang)	1✔
156	available_lang = sorted(list(set(available_lang)))	1✔
157
158	# Determine target language (if unspecified, prefer cpp then first available)
159	if normalized_language:	1✔
160	target_lang = normalized_language	×
161	elif "python" in available_lang:	1✔
162	target_lang = "python"	1✔
163	else:
164	target_lang = available_lang[0] if available_lang else ""	×
165	# Warn if ambiguous and no language specified
166	if not normalized_language and len(available_lang) > 1:	1✔
167	msg = f"'{project_name}' has multiple languages: {', '.join(available_lang)}.\n"	×
168	msg += f"Defaulting to '{target_lang}'. Use --language to select explicitly.\n\n"	×
169	print(Color.green(msg))	×
170	for p in candidates:	1✔
171	if target_lang in list_normalized_languages(	1✔
172	p.get("metadata", {}).get("language", None), strict=True
173	):
174	self._project_data[cache_key] = p # Return candidate matching target_lang	1✔
175	return p	1✔
176	if normalized_language: # If target_lang specified but not found	×
177	fatal(	×
178	f"Project '{project_name}' (language: {normalized_language}) not found. "
179	f"Available: {', '.join(available_lang) if available_lang else 'unknown'}"
180	)
181	# No language info or no match found; return first candidate
182	fallback_candidate = candidates[0]	×
183	fallback_lang = fallback_candidate.get("metadata", {}).get("language", None)	×
184	if not fallback_lang:	×
185	msg = f"Returning '{project_name}' with missing or unknown language metadata.\n"	×
186	msg += "Consider specifying --language for more consistent results.\n"	×
187	warn(msg)	×
188	self._project_data[cache_key] = fallback_candidate	×
189	return self._project_data[cache_key]	×
190	# If project not found, suggest similar names
191	distances = [	1✔
192	(
193	p["project_name"],
194	levenshtein_distance(project_name, p["project_name"]),
195	p.get("source_folder", ""),
196	p.get("metadata", {}).get("language", ""),
197	)
198	for p in self.projects
199	]
200	distances.sort(key=lambda x: x[1]) # Sort by distance	1✔
201	closest_matches = [	1✔
202	(name, folder, lang) for name, dist, folder, lang in distances[:3] if dist <= 3
203	] # Show up to 3 matches
204	msg = f"Project '{project_name}' (language: {normalized_language}) not found."	1✔
205	if closest_matches:	1✔
206	msg += "\nDid you mean:"	×
207	for name, folder, lang in closest_matches:	×
208	details = []	×
209	if lang:	×
210	details.append(f"language: {lang}")	×
211	if folder:	×
212	details.append(f"source: {folder}")	×
213	msg += f"\n '{name}'" + (f" ({', '.join(details)})" if details else "")	×
214	fatal(msg)	1✔
215
216	def resolve_mode(self, project_data: dict, requested_mode: Optional[str] = None) -> tuple:	1✔
217	"""
218	Resolve mode from metadata and validate
219	Returns: (mode_name, mode_config) or (None, None) for legacy behavior
220	"""
221	modes = project_data.get("metadata", {}).get("modes", {})	1✔
222	if not modes:	1✔
223	return None, None # No modes defined - should use legacy behavior	1✔
224
225	if requested_mode is None:	1✔
226	# Validate that multiple modes have a default_mode specified
227	application_metadata = project_data.get("metadata", {})	1✔
228	if len(modes) > 1 and "default_mode" not in application_metadata:	1✔
229	available = ", ".join(modes.keys())	1✔
230	fatal(	1✔
231	f"Multiple modes found ({available}) but no 'default_mode' specified. "
232	f"Please add a 'default_mode' field to specify which mode to use by default."
233	)
234
235	if "default_mode" in application_metadata:	×
236	requested_mode = application_metadata["default_mode"]	×
237	# Validate that default_mode references an existing mode
238	if requested_mode not in modes:	×
239	available = ", ".join(modes.keys())	×
240	fatal(f"Invalid default_mode '{requested_mode}' in metadata among {available}")	×
241	else:
242	requested_mode = list(modes.keys())[0]	×
243	if requested_mode not in modes:	×
244	available = ", ".join(modes.keys())	×
245	fatal(f"Mode '{requested_mode}' not found. Available modes: {available}")	×
246	return requested_mode, modes[requested_mode]	×
247
248	def validate_mode(	1✔
249	self,
250	mode_name: Optional[str],
251	mode_config: dict,
252	) -> None:
253	"""Validate mode configuration"""
254	if not mode_config:	1✔
255	return # No mode configuration to validate	1✔
256
257	# Define valid keys for mode configuration
258	valid_top_level_keys = ["description", "requirements", "build", "run", "env"]	×
259	valid_build_keys = ["depends", "docker_build_args", "cmake_options", "env"]	×
260	valid_run_keys = ["command", "workdir", "docker_run_args", "env"]	×
261
262	# Check top-level keys
263	for key in mode_config.keys():	×
264	if key not in valid_top_level_keys:	×
265	suggestions = self._suggest_command(key, valid_top_level_keys)	×
266	msg = f"Unknown key '{key}' in mode '{mode_name}'"	×
267	if suggestions:	×
268	msg += f". Did you mean '{suggestions[0]}'?"	×
269	warn(msg)	×
270
271	# Check section keys (build and run)
272	sections_to_validate = {"build": valid_build_keys, "run": valid_run_keys}	×
273	for section_name, valid_keys in sections_to_validate.items():	×
274	if section_name in mode_config and isinstance(mode_config[section_name], dict):	×
275	for key in mode_config[section_name].keys():	×
276	if key not in valid_keys:	×
277	suggestions = self._suggest_command(key, valid_keys)	×
278	msg = f"Unknown key '{section_name}.{key}' in mode '{mode_name}'"	×
279	if suggestions:	×
280	msg += f". Did you mean '{suggestions[0]}'?"	×
281	warn(msg)	×
282
283	def get_effective_build_config(	1✔
284	self,
285	args: argparse.Namespace,
286	mode_config: dict,
287	) -> dict:
288	"""
289	Get effective build configuration combining CLI args and mode config.
290	"""
291	config = {	1✔
292	"with_operators": getattr(args, "with_operators", None),
293	"docker_opts": getattr(args, "docker_opts", ""),
294	"build_args": getattr(args, "build_args", ""),
295	"configure_args": getattr(args, "configure_args", None),
296	}
297	if not mode_config:	1✔
298	return config	1✔
299
300	# Apply build configuration - CLI parameters always override mode settings when provided
301	if "build" in mode_config:	1✔
302	build_config = mode_config["build"]	1✔
303
304	if "depends" in build_config:	1✔
305	if config["with_operators"]:	1✔
306	warn("CLI --build-with overrides mode build.depends")	1✔
307	else:
308	mode_deps = [dep.strip() for dep in build_config["depends"] if dep.strip()]	1✔
309	config["with_operators"] = ";".join(mode_deps) if mode_deps else ""	1✔
310
311	if "docker_build_args" in build_config:	1✔
312	if config["build_args"]:	1✔
313	warn("CLI --build-args overrides mode build.docker_build_args")	1✔
314	else:
315	config["build_args"] = normalize_args_str(build_config["docker_build_args"])	1✔
316
317	if "cmake_options" in build_config:	1✔
318	if config["configure_args"]:	1✔
319	warn("CLI --configure-args overrides mode build.cmake_options")	1✔
320	else:
321	config["configure_args"] = build_config["cmake_options"]	1✔
322
323	if "run" in mode_config and "docker_run_args" in mode_config["run"]:	1✔
324	if getattr(args, "docker_opts", ""):	1✔
325	warn("CLI --docker-opts overrides mode run.docker_run_args")	1✔
326	else:
327	config["docker_opts"] = normalize_args_str(mode_config["run"]["docker_run_args"])	1✔
328
329	return config	1✔
330
331	def get_effective_run_config(	1✔
332	self,
333	args: argparse.Namespace,
334	mode_config: dict,
335	) -> dict:
336	"""Get effective run configuration combining CLI args and mode config without mutation"""
337	config = {	1✔
338	"run_args": getattr(args, "run_args", "") or "",
339	"docker_opts": getattr(args, "docker_opts", ""),
340	}
341
342	if mode_config and "run" in mode_config:	1✔
343	run_config = mode_config["run"]	1✔
344
345	if "command" in run_config:	1✔
346	config["command"] = run_config["command"]	1✔
347	if "workdir" in run_config:	1✔
348	config["workdir"] = run_config["workdir"]	1✔
349
350	if "command" in run_config and getattr(args, "run_args", ""):	1✔
351	warn("CLI --run-args will be appended to mode run.command")	1✔
352
353	if "docker_run_args" in run_config:	1✔
354	if getattr(args, "docker_opts", ""):	1✔
355	warn("CLI --docker-opts overrides mode run.docker_run_args")	1✔
356	else:
357	config["docker_opts"] = normalize_args_str(run_config["docker_run_args"])	1✔
358	return config	1✔
359
360	def make_project_container(	1✔
361	self, project_name: Optional[str] = None, language: Optional[str] = None
362	) -> HoloscanContainer:
363	"""Define a project container"""
364	if not project_name:	1✔
365	return HoloscanContainer(project_metadata=None)	1✔
366	project_data = self.find_project(project_name=project_name, language=language)	×
367	return HoloscanContainer(project_metadata=project_data, language=language)	×
368
369	def collect_cache_dirs(self, patterns: list[str], default_dir=None) -> list:	1✔
370	"""Helper to collect cache directories matching patterns."""
371	dirs = []	1✔
372	if default_dir is not None:	1✔
373	dirs.append(default_dir)	1✔
374	for pattern in patterns:	1✔
375	for path in HoloscanCLI.HOLOHUB_ROOT.glob(pattern):	1✔
376	if path.is_dir() and path not in dirs:	1✔
377	dirs.append(path)	1✔
378	return dirs	1✔
379
380	def _suggest_command(self, invalid_value: str, valid_options: list[str]) -> list[str]:	1✔
381	"""Suggest similar values using Levenshtein distance."""
382	distances = [	×
383	(option, levenshtein_distance(invalid_value, option)) for option in valid_options
384	]
385	distances.sort(key=lambda x: x[1])	×
386	return [option for option, dist in distances[:2] if dist <= 2] # Show up to 2 matches	×
387
388	def _check_for_dash_prefix_issue(self, cmd_args: List[str]) -> Optional[str]:	1✔
389	"""
390	Check if the parsing error is likely due to dash-prefixed arguments
391	"""
392	DASH_VALUE_ARGS = ["--run-args", "--build-args", "--docker-opts", "--configure-args"]	1✔
393	for i, arg in enumerate(cmd_args):	1✔
394	if arg in DASH_VALUE_ARGS and "=" not in arg:	1✔
395	if i + 1 < len(cmd_args) and cmd_args[i + 1].startswith("-"):	1✔
396	next_arg = cmd_args[i + 1]	1✔
397	return (	1✔
398	f"💡 Tip: ambiguous dash-prefixed arguments, use the equals format:\n"
399	f" Instead of: {arg} {next_arg}\n"
400	f" Use: {arg}={next_arg}"
401	)
402	return None	1✔
403
404	def run(self, argv: Optional[List[str]] = None) -> None:	1✔
405	"""Main entry point for the CLI"""
406
407	trailing_docker_args = [] # Handle " -- " separator for run-container command forwarding	1✔
408	if argv is None:	1✔
409	argv = sys.argv	×
410	argv = list(argv)	1✔
411	cmd_args = argv[1:] # Skip script name, return a copy of the args	1✔
412	if len(cmd_args) >= 2 and cmd_args[0] == "run-container" and "--" in cmd_args:	1✔
413	sep = cmd_args.index("--")	1✔
414	cmd_args, trailing_docker_args = cmd_args[:sep], cmd_args[sep + 1 :]	1✔
415
416	potential_command = cmd_args[0] if cmd_args else None	1✔
417	dash_suggestion = None	1✔
418	if potential_command and potential_command in self.subparsers:	1✔
419	dash_suggestion = self._check_for_dash_prefix_issue(cmd_args)	1✔
420
421	try:	1✔
422	args = self.parser.parse_args(cmd_args)	1✔
423	if trailing_docker_args:	1✔
424	args._trailing_args = trailing_docker_args # " -- " used for run-container command	1✔
425	except SystemExit as e:	×
426	if len(cmd_args) > 0 and e.code != 0: # exit code is 0 => help was successfully shown	×
427	if dash_suggestion:	×
428	print(f"\n{dash_suggestion}\n", file=sys.stderr)	×
429
430	if potential_command and not potential_command.startswith("-"):	×
431	if potential_command in self.subparsers:	×
432	# Valid subcommand but parsing failed
433	print(f"\n💡 For more help with '{potential_command}':", file=sys.stderr)	×
434	print(f" {self.script_name} {potential_command} --help\n", file=sys.stderr)	×
435	sys.exit(e.code if e.code is not None else 1)	×
436	else: # Invalid subcommand - suggest similar ones
437	suggestions = self._suggest_command(	×
438	potential_command, list(self.subparsers.keys())
439	)
440	if suggestions:	×
441	print("\n💡 Did you mean:", file=sys.stderr)	×
442	for cmd in suggestions:	×
443	print(f" {self.script_name} {cmd}", file=sys.stderr)	×
444	print(file=sys.stderr)	×
445	sys.exit(1)	×
446	raise	×
447	plan_format = getattr(args, "plan_format", None)	1✔
448	if plan_format is not None:	1✔
449	if not getattr(args, "_supports_plan", False):	1✔
NEW 450	self.subparsers[args.command].error(	×
451	"structured command plans are not supported for this command"
452	)
453	if not getattr(args, "dryrun", False):	1✔
454	self.subparsers[args.command].error(f"--{plan_format} requires --dryrun")	1✔
455	self._run_structured_plan(args, plan_format)	1✔
456	elif hasattr(args, "func"):	1✔
457	args.func(args)	1✔
458	else:
459	self.parser.print_help()	×
460	sys.exit(1)	×
461
462	@staticmethod	1✔
463	def _run_structured_plan(args: argparse.Namespace, plan_format: str) -> None:	1✔
464	"""Run one audited dry-run and write its artifact atomically to stdout."""
465
466	recorder = PlanRecorder()	1✔
467	legacy_stdout = StringIO()	1✔
468	try:	1✔
469	try:	1✔
470	with recorder.activate(), redirect_stdout(legacy_stdout):	1✔
471	args.func(args)	1✔
472	finally:
473	notices = legacy_stdout.getvalue()	1✔
474	if notices:	1✔
475	sys.stderr.write(notices)	1✔
476
477	output = recorder.json_text() if plan_format == "json" else recorder.shell_text()	1✔
478	except CommandPlanError as exc:	1✔
479	print(f"Command planning failed: {exc}", file=sys.stderr)	1✔
480	raise SystemExit(1) from exc	1✔
481
482	if plan_format == "json":	1✔
483	output += "\n"	1✔
484	sys.stdout.write(output)	1✔
485
486
487	def main(argv: Optional[List[str]] = None):	1✔
488	script_name = None	×
489	if argv and not os.environ.get("HOLOSCAN_CLI_CMD_NAME"):	×
490	executable = Path(argv[0]).name	×
491	script_name = "holoscan" if executable == "__main__.py" else executable	×
492	cli = HoloscanCLI(script_name=script_name)	×
493	cli.run(argv)	×
494
495
496	if __name__ == "__main__":	1✔
497	main()	×

nvidia-holoscan / holoscan-cli / 29761095343

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous