"""Kelvin CLI group definition and command registration.
Separated from main.py to avoid circular imports: configuration.py
needs to import kelvin_cli for shell completion generation, while
main.py imports configuration.py to register the command.
"""
import sys
import time
import webbrowser
from typing import Optional, Union
import rich_click as click
from rich.console import Console
from rich.tree import Tree
from kelvin.sdk.container import Services, create_services
from kelvin.sdk.logging import setup_logging
from kelvin.sdk.services.config import ConfigService
from kelvin.sdk.version import version
KELVIN_DOCS_DEFAULT_URL = "https://docs.kelvin.ai"
def _open_docs(ctx: click.Context, _param: Union[click.Option, click.Parameter], value: Optional[str]) -> None:
"""Open Kelvin documentation in the browser."""
if not value or ctx.resilient_parsing:
return
from kelvin.sdk.services.session import SessionService
docs_url = KELVIN_DOCS_DEFAULT_URL
try:
session_service = SessionService()
platform_url = session_service.get_documentation_url()
if platform_url:
docs_url = platform_url
except Exception:
pass
_ = webbrowser.open(docs_url)
ctx.exit(0)
def _get_config_defaults() -> dict[str, object]:
"""Load configuration defaults from config file.
Returns config values to use as CLI option defaults.
"""
try:
config_service = ConfigService()
config = config_service.get_config()
return {
"json_mode": config.json_output,
"assume_yes": config.no_prompt,
"verbose": config.verbose,
}
except (OSError, ValueError):
# OSError: file permission issues; ValueError: corrupt YAML/invalid config
return {}
def _build_command_tree(
group: click.Group,
tree: Tree,
ctx: click.Context,
*,
parent_deprecated: bool = False,
) -> None:
"""Recursively build a tree of commands with their summaries."""
commands = sorted(group.list_commands(ctx) or [])
# Resolve commands once to separate groups from leaves
resolved: list[tuple[str, click.Command]] = []
for name in commands:
cmd = group.get_command(ctx, name)
if cmd is not None:
resolved.append((name, cmd))
for name, cmd in resolved:
cur_is_group = isinstance(cmd, click.Group)
short_help = cmd.get_short_help_str(limit=60)
deprecated = bool(cmd.deprecated) or parent_deprecated
# Groups bold green, leaves green; deprecated commands are dimmed
label = f"[bold green]{name}[/]" if cur_is_group else f"[green]{name}[/]"
if short_help:
label += f": {short_help}"
if deprecated:
label = f"[dim]{label}[/]"
branch = tree.add(label)
# Recurse into subgroups
if isinstance(cmd, click.Group):
_build_command_tree(cmd, branch, ctx, parent_deprecated=deprecated)
# Configure rich-click styling
click.rich_click.TEXT_MARKUP = "rich"
click.rich_click.SHOW_ARGUMENTS = True
click.rich_click.GROUP_ARGUMENTS_OPTIONS = True
# Module-level flags set by kelvin_cli() for exception handler access.
# This avoids re-parsing sys.argv with fragile string matching.
_resolved_json_mode: bool = False
_resolved_verbose: bool = False
[docs]
def get_resolved_json_mode() -> bool:
"""Return the resolved --json flag value."""
return _resolved_json_mode
[docs]
def get_resolved_verbose() -> bool:
"""Return the resolved --verbose flag value."""
return _resolved_verbose
[docs]
def set_resolved_json_mode(value: bool) -> None:
"""Update the resolved --json flag (called by @global_options)."""
global _resolved_json_mode # noqa: PLW0603
_resolved_json_mode = value
[docs]
def set_resolved_verbose(value: bool) -> None:
"""Update the resolved --verbose flag (called by @global_options)."""
global _resolved_verbose # noqa: PLW0603
_resolved_verbose = value
# ============ Telemetry ============
def _get_passed_flags(ctx: click.Context) -> list[str]:
"""Return the long names of the *options* the user explicitly passed.
Captures flag/option *names* only (e.g. ``--browser``, ``--device``) so we
learn which sub-features of a command are used. Deliberately:
- includes only parameters whose source is the command line (not defaults,
env vars, prompts, or config), and
- excludes positional *arguments* (which carry values like URLs/paths),
so we never record a flag's *value* — only that the flag was present. Never
raises; returns an empty list on any problem.
"""
from click.core import Argument, ParameterSource
flags: list[str] = []
try:
command = ctx.command
for param in command.params:
if isinstance(param, Argument):
continue # arguments carry values (URLs, paths, …) — skip entirely
name = param.name
if not name:
continue
try:
source = ctx.get_parameter_source(name)
except Exception: # noqa: BLE001 - per-param best effort
continue
if source is not ParameterSource.COMMANDLINE:
continue
# Prefer the long option form (e.g. "--browser"); fall back to any.
opts = list(getattr(param, "opts", []) or [])
long_opts = [o for o in opts if o.startswith("--")]
chosen = (long_opts or opts or [name])[0]
flags.append(chosen)
except Exception: # noqa: BLE001 - never let flag gathering break tracking
return []
return sorted(set(flags))
def _gather_telemetry_properties(
services: Services,
command_path: str,
flags: Optional[list[str]] = None,
) -> tuple[dict[str, object], Optional[str]]:
"""Collect properties for the ``command_run`` event and resolve the identity.
Sends the *command path* (e.g. ``app build``), the *names* of any options
passed (e.g. ``--browser``), coarse environment data, and — when the user is
logged in — their username; never raw args, flag values, file paths, error
messages, or payloads. The instance is reduced to its hostname.
Returns the properties dict and the identity to attribute the event to: the
logged-in user's identity when available, else ``None`` (the telemetry
service then falls back to the anonymous device id). Reads the session file
only — never the keyring — to stay off the hot path.
"""
import platform
from urllib.parse import urlparse
props: dict[str, object] = {
"command": command_path,
"cli_version": version,
"python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
"os": platform.system().lower(),
"arch": platform.machine(),
"json_mode": services.output.json_mode,
}
if flags:
# Space-joined list of flag names (Heap properties must be scalars).
props["flags"] = " ".join(flags)
identity: Optional[str] = None
try:
session = services.session.get_current_session()
if session:
if session.url:
host = urlparse(session.url).hostname
if host:
props["instance_host"] = host
if session.username:
props["username"] = session.username
# The username doubles as the analytics identity.
identity = session.username
except Exception: # noqa: BLE001 - never let property gathering break tracking
pass
return props, identity
[docs]
def register_command_telemetry(ctx: click.Context, services: Services) -> None:
"""Register a close hook that emits one usage event per command invocation.
The Heap event name *is* the command that was run (e.g. ``app build``), so
each command shows up as its own event in Heap. Called from
``@global_options`` so the full command path is available via
``ctx.command_path``. Uses ``ctx.call_on_close``, which Click runs on both
success and exception, so a single seam captures the outcome of every
command. The exception state is read from ``sys.exc_info()`` inside the
close callback. Never raises.
The command path is normalised to drop the ``kelvin`` prog name so events
read as ``app build`` rather than ``kelvin app build``.
"""
try:
command_path = ctx.command_path
except Exception: # noqa: BLE001 - fall back to a generic label
command_path = ""
# The CLI ships both ``kelvin`` and ``ksdk`` console scripts; Click prefixes
# ``command_path`` with whichever was invoked. Strip either so the same
# command reports under one event name regardless of entry point.
command_path = command_path.removeprefix("kelvin ").removeprefix("ksdk ").strip() or "kelvin"
# Capture which options were passed now, while the context is live; the
# close callback runs after the command and may see a different context.
flags = _get_passed_flags(ctx)
start = time.monotonic()
def _emit() -> None:
try:
exc_type, exc_val = sys.exc_info()[:2]
props, identity = _gather_telemetry_properties(services, command_path, flags)
props["duration_ms"] = round((time.monotonic() - start) * 1000)
# A zero-code SystemExit is an intentional non-error outcome (e.g.
# declining a confirmation, cancelling reset); Click leaves it in
# sys.exc_info() during close, so treat it as success.
is_clean_exit = isinstance(exc_val, SystemExit) and exc_val.code in (0, None)
if exc_type is None or is_clean_exit:
props["status"] = "success"
else:
props["status"] = "error"
props["error_type"] = exc_type.__name__
# The event name is the command itself (e.g. "app build").
services.telemetry.track(command_path, props, identity=identity)
# Give the background send a brief chance to land before exit.
services.telemetry.flush()
except Exception: # noqa: BLE001 - telemetry must never break the CLI
pass
_ = ctx.call_on_close(_emit)
@click.group(invoke_without_command=True)
@click.option(
"--json",
"json_mode",
is_flag=True,
default=None,
is_eager=True,
help="Output as JSON (for scripting).",
)
@click.option(
"-y",
"--yes",
"assume_yes",
is_flag=True,
default=None,
is_eager=True,
help="Skip confirmation prompts.",
)
@click.option(
"-v",
"--verbose",
is_flag=True,
is_eager=True,
help="Enable debug logging.",
)
@click.option(
"--docs",
is_flag=True,
help="Open the Kelvin SDK documentation webpage.",
callback=_open_docs,
expose_value=False,
is_eager=True,
)
@click.option(
"--tree",
"show_tree",
is_flag=True,
help="Show command tree with descriptions.",
)
@click.version_option(version=version, prog_name="Kelvin CLI")
@click.pass_context
def kelvin_cli(
ctx: click.Context,
json_mode: Optional[bool],
assume_yes: Optional[bool],
verbose: bool,
show_tree: bool,
) -> None:
"""Kelvin CLI - The complete tool to interact with the Kelvin Ecosystem.
Use --json for machine-readable output.
Use --yes to skip confirmations in scripts.
Use --verbose for debug information.
Use --tree to see all available commands.
"""
# Handle --tree before setting up services
if show_tree:
console = Console()
tree = Tree(f"[bold]kelvin[/] [dim]v{version}[/]", guide_style="dim")
_build_command_tree(kelvin_cli, tree, ctx)
console.print(tree)
ctx.exit(0)
# Show help if no subcommand is provided
if ctx.invoked_subcommand is None:
click.echo(ctx.get_help())
ctx.exit(0)
# Load config defaults (deferred to avoid filesystem access on --help/--version)
config_defaults = _get_config_defaults()
# Merge CLI options with config defaults
# CLI options take precedence when explicitly provided
effective_json = json_mode if json_mode is not None else config_defaults.get("json_mode", False)
effective_yes = assume_yes if assume_yes is not None else config_defaults.get("assume_yes", False)
effective_verbose = verbose or bool(config_defaults.get("verbose", False))
# Ensure types are correct
effective_json = bool(effective_json)
effective_yes = bool(effective_yes)
setup_logging(verbose=effective_verbose, json_mode=effective_json)
# Store resolved flags at module level for exception handler access
set_resolved_json_mode(effective_json)
set_resolved_verbose(effective_verbose)
# Store verbose flag for exception handler
_ = ctx.ensure_object(dict) # pyright: ignore[reportUnknownVariableType]
services = create_services(
json_mode=effective_json,
assume_yes=effective_yes,
verbose=effective_verbose,
)
ctx.obj = services
# Store flags in context for exception handler access
ctx.meta["json_mode"] = effective_json
ctx.meta["verbose"] = effective_verbose