Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ packages = [
{include = "**/*.py", from = "src"},
]
readme = "README.md"
version = "0.27.0"
version = "0.28.0"

[tool.poetry.dependencies]
# For certifi, use ">=" instead of "^" since it upgrades its "major version" every year, not really following semver
Expand Down
230 changes: 214 additions & 16 deletions src/groundlight/cli.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,47 @@
import json
import logging
import sys
from datetime import date, datetime
from decimal import Decimal
from enum import Enum
from functools import wraps
from typing import Union
from importlib.metadata import version as importlib_version
from typing import Any, Union
from uuid import UUID

import typer
from groundlight_openapi_client.model_utils import OpenApiModel
from pydantic import BaseModel
from typing_extensions import get_origin

from groundlight import Groundlight
from groundlight import ExperimentalApi, Groundlight
from groundlight.client import ApiTokenError

logger = logging.getLogger(__name__)

cli_app = typer.Typer(
context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 800},
)


@cli_app.callback(invoke_without_command=True)
def _main(
ctx: typer.Context,
version: bool = typer.Option(False, "--version", "-v", is_eager=True, help="Show the SDK version and exit."),
):
if version:
print(importlib_version("groundlight"))
raise typer.Exit()
if ctx.invoked_subcommand is None:
print(ctx.get_help())


experimental_app = typer.Typer(
no_args_is_help=True,
help="Experimental commands — may change or be removed without notice.",
context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 800},
)
cli_app.add_typer(experimental_app, name="exp", rich_help_panel="Subcommands")


def is_cli_supported_type(annotation):
Expand All @@ -21,15 +52,66 @@ def is_cli_supported_type(annotation):
return annotation in (int, float, bool)


def class_func_to_cli(method):
def is_cli_representable(annotation) -> bool:
"""Returns True if the annotation is a type Typer can natively represent as a CLI argument.

Primitive scalar types, Enum subclasses, and Union types (handled separately) are considered
representable. Complex types like dict, list, bytes, and custom model classes are not.
"""
Given the class method, create a method with the identical signature to provide the help documentation and
but only instantiates the class when the method is actually called.
if annotation in (str, int, float, bool):
return True
if isinstance(annotation, type) and issubclass(annotation, Enum):
return True
if get_origin(annotation) is Union:
return True
return False


def _json_default(obj: Any) -> Any:
"""Fallback serializer for json.dumps for types the stdlib encoder doesn't handle.

Covers common types that appear in OpenAPI client to_dict() output. Unknown types
fall back to str() rather than raising, so CLI output is always usable.
"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, UUID):
return str(obj)
if isinstance(obj, Enum):
return obj.value
return str(obj)


def _format_result(result: Any) -> str:
"""Format a CLI result value as a human-readable, jq-compatible string.

Pydantic models and OpenAPI client objects are serialized to indented JSON.
Plain dicts and lists are also JSON. Everything else falls back to str().
"""
if isinstance(result, BaseModel):
return result.model_dump_json(indent=2)
if isinstance(result, OpenApiModel):
return json.dumps(result.to_dict(), indent=2, default=_json_default)
if isinstance(result, (dict, list)):
return json.dumps(result, indent=2, default=_json_default)
return str(result)


# We create a fake class and fake method so we have the correct annotations for typer to use
# When we wrap the fake method, we only use the fake method's name to access the real method
# and attach it to a Groundlight instance that we create at function call time
def class_func_to_cli(method, is_experimental: bool = False):
"""
Given a class method, return a wrapper function with the same signature that Typer can
register as a CLI command. The wrapper instantiates ExperimentalApi at call time (which
also provides all stable Groundlight methods via inheritance), so a single instantiation
path serves both stable and experimental commands.

If is_experimental is True, a warning is printed to stderr before the method runs.
"""

# We create a fake class and fake method so we have the correct annotations for typer to use.
# When we wrap the fake method, we only use the fake method's name to look up and call the
# real method on an ExperimentalApi instance created at call time.
class FakeClass:
pass

Expand All @@ -38,14 +120,22 @@ class FakeClass:

@wraps(fake_method)
def wrapper(*args, **kwargs):
gl = Groundlight()
gl_method = vars(Groundlight)[fake_method.__name__]
gl_bound_method = gl_method.__get__(gl, Groundlight) # pylint: disable=all
print(gl_bound_method(*args, **kwargs)) # this is where we output to the console
if is_experimental:
print(
f"Warning: '{fake_method.__name__}' is an experimental command and may change without notice.",
file=sys.stderr,
)
gl = ExperimentalApi()
bound_method = getattr(gl, fake_method.__name__)
result = bound_method(*args, **kwargs)
if result is not None:
print(_format_result(result))

# not recommended practice to directly change annotations, but gets around Typer not supporting Union types
cli_unsupported_params = []
for name, annotation in method.__annotations__.items():
if name == "return":
continue
if get_origin(annotation) is Union:
# If we can submit a string, we take the string from the cli
if str in annotation.__args__:
Expand All @@ -60,6 +150,11 @@ def wrapper(*args, **kwargs):
break
if not found_supported_type:
cli_unsupported_params.append(name)
elif is_experimental and not is_cli_representable(annotation):
# For experimental methods only: proactively flag non-Union types that Typer cannot
# represent (e.g. dict, list, custom models) so the caller can skip them gracefully
# before Typer raises a deferred RuntimeError at cli_app() invocation time.
cli_unsupported_params.append(name)
# Ideally we could just not list the unsupported params, but it doesn't seem natively supported by Typer
# and requires more metaprogamming than makes sense at the moment. For now, we require methods to support str
for param in cli_unsupported_params:
Expand All @@ -71,13 +166,116 @@ def wrapper(*args, **kwargs):
return wrapper


# Methods to exclude from the CLI entirely
_CLI_EXCLUDED_METHODS = {
"make_action",
"create_rule",
"get_rule",
"delete_rule",
"list_rules",
"delete_all_rules",
"start_inspection",
"update_inspection_metadata",
"stop_inspection",
}

# Desired display order of command groups in the CLI help output.
_GROUP_ORDER = [
"Account",
"Detectors",
"Image Queries",
"ML Pipelines & Priming",
"Notes",
"Utilities",
"Other",
]

# Maps method names to their rich_help_panel group label for the CLI help output.
# Applies to both stable and experimental commands.
_COMMAND_GROUPS: dict[str, str] = {
# Account
"whoami": "Account",
"get_month_to_date_usage": "Account",
# Detectors
"get_detector": "Detectors",
"get_detector_by_name": "Detectors",
"list_detectors": "Detectors",
"create_detector": "Detectors",
"get_or_create_detector": "Detectors",
"delete_detector": "Detectors",
"create_binary_detector": "Detectors",
"create_counting_detector": "Detectors",
"create_multiclass_detector": "Detectors",
"create_bounding_box_detector": "Detectors",
"create_detector_group": "Detectors",
"list_detector_groups": "Detectors",
"create_roi": "Detectors",
"update_detector_confidence_threshold": "Detectors",
"update_detector_status": "Detectors",
"update_detector_escalation_type": "Detectors",
"reset_detector": "Detectors",
"update_detector_name": "Detectors",
"create_text_recognition_detector": "Detectors",
"get_detector_evaluation": "Detectors",
"get_detector_metrics": "Detectors",
"download_mlbinary": "Detectors",
# Image Queries
"get_image_query": "Image Queries",
"list_image_queries": "Image Queries",
"submit_image_query": "Image Queries",
"ask_confident": "Image Queries",
"ask_ml": "Image Queries",
"ask_async": "Image Queries",
"wait_for_confident_result": "Image Queries",
"wait_for_ml_result": "Image Queries",
"get_image": "Image Queries",
"add_label": "Image Queries",
# Notes
"get_notes": "Notes",
"create_note": "Notes",
# ML Pipelines & Priming
"list_detector_pipelines": "ML Pipelines & Priming",
"list_priming_groups": "ML Pipelines & Priming",
"create_priming_group": "ML Pipelines & Priming",
"get_priming_group": "ML Pipelines & Priming",
"delete_priming_group": "ML Pipelines & Priming",
# Utilities
"edge_base_url": "Utilities",
"get_raw_headers": "Utilities",
}


def _cli_sort_key(item: tuple) -> tuple:
"""Sort key for CLI command registration that controls group and within-group ordering.

Commands are ordered first by their group's position in _GROUP_ORDER (ungrouped last),
then alphabetically by method name within each group.
"""
name, _ = item
group = _COMMAND_GROUPS.get(name)
group_rank = _GROUP_ORDER.index(group) if group in _GROUP_ORDER else len(_GROUP_ORDER)
return (group_rank, name)


def groundlight():
"""Entry point for the groundlight CLI."""
try:
# For each method in the Groundlight class, create a function that can be called from the command line
for name, method in vars(Groundlight).items():
if callable(method) and not name.startswith("_"):
stable_names = {n for n, m in vars(Groundlight).items() if callable(m) and not n.startswith("_")}

for name, method in sorted(vars(Groundlight).items(), key=_cli_sort_key):
if callable(method) and not name.startswith("_") and name not in _CLI_EXCLUDED_METHODS:
cli_func = class_func_to_cli(method)
cli_app.command()(cli_func)
cli_app.command(rich_help_panel=_COMMAND_GROUPS.get(name, "Other"))(cli_func)

for name, method in sorted(vars(ExperimentalApi).items(), key=_cli_sort_key):
if not callable(method) or name.startswith("_") or name in stable_names or name in _CLI_EXCLUDED_METHODS:
continue
try:
cli_func = class_func_to_cli(method, is_experimental=True)
experimental_app.command(rich_help_panel=_COMMAND_GROUPS.get(name, "Other"))(cli_func)
except Exception as e: # pylint: disable=broad-except
logger.debug("Skipping experimental CLI command '%s': %s", name, e)

cli_app()
except ApiTokenError as e:
print(e)
Expand Down
36 changes: 27 additions & 9 deletions test/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,16 +41,9 @@ def test_detector_and_image_queries(detector_name: Callable):
check=False,
)
assert completed_process.returncode == 0
match = re.search("id='([^']+)'", completed_process.stdout)
match = re.search(r'"id":\s*"([^"]+)"', completed_process.stdout)
assert match is not None
det_id_on_create = match.group(1)
# The output of the create-detector command looks something like:
# id='det_abc123'
# type=<DetectorTypeEnum.detector: 'detector'>
# created_at=datetime.datetime(2023, 8, 30, 18, 3, 9, 489794,
# tzinfo=datetime.timezone(datetime.timedelta(days=-1, seconds=61200)))
# name='testdetector 2023-08-31 01:03:09.039448' query='testdetector'
# group_name='__DEFAULT' confidence_threshold=0.9

# test getting detectors
completed_process = subprocess.run(
Expand All @@ -61,7 +54,7 @@ def test_detector_and_image_queries(detector_name: Callable):
check=False,
)
assert completed_process.returncode == 0
match = re.search("id='([^']+)'", completed_process.stdout)
match = re.search(r'"id":\s*"([^"]+)"', completed_process.stdout)
assert match is not None
det_id_on_get = match.group(1)
assert det_id_on_create == det_id_on_get
Expand Down Expand Up @@ -110,6 +103,31 @@ def test_help():
assert completed_process.returncode == 0


def test_version():
for flag in ("--version", "-v"):
completed_process = subprocess.run(
["groundlight", flag],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
assert completed_process.returncode == 0
assert re.match(r"\d+\.\d+\.\d+", completed_process.stdout.strip())


def test_experimental_subcommand():
completed_process = subprocess.run(
["groundlight", "exp", "--help"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
assert completed_process.returncode == 0
assert "list-priming-groups" in completed_process.stdout


def test_bad_commands():
completed_process = subprocess.run(
["groundlight", "wat"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
Expand Down
Loading