Skip to content
Draft
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
218 changes: 203 additions & 15 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.
"""
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 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)


def class_func_to_cli(method, is_experimental: bool = False):
"""
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.
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 access the real method
# and attach it to a Groundlight instance that we create at function call time
# 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,26 @@ 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()
# Typer sees the fake method's annotations (for correct CLI argument types), but the
# actual call goes to the real method on a live ExperimentalApi instance. The fake
# method's name is identical to the real one, so getattr resolves to the correct
# implementation, including inherited Groundlight methods.
bound_method = getattr(gl, fake_method.__name__)
result = bound_method(*args, **kwargs)
if result is not None:
print(_format_result(result))
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Heavily biased as the one who wrote the original, but there should be more comments around this one line. This is the magic line that makes the whole thing happen

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. I added a comment.


# 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 +154,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 +170,102 @@ def wrapper(*args, **kwargs):
return wrapper


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

# 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] = {
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this adds maintenance overhead, new methods need to belong to a group. The original design tried to avoid that additional overhead, but in the age of AI maybe we don't care

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added a test to ensure that each method has a CLI group. It will be an extra step, but a worthwhile extra step that cannot be forgotten.

# 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, 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():
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("_"):
cli_func = class_func_to_cli(method)
cli_app.command()(cli_func)
cli_app.command(rich_help_panel=_COMMAND_GROUPS[name])(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:
continue
try:
cli_func = class_func_to_cli(method, is_experimental=True)
experimental_app.command(rich_help_panel=_COMMAND_GROUPS[name])(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: 0 additions & 36 deletions src/groundlight/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1277,42 +1277,6 @@ def add_label(
request_params = LabelValueRequest(label=label, image_query_id=image_query_id, rois=roi_requests)
self.labels_api.create_label(request_params)

def start_inspection(self) -> str:
"""
**NOTE:** For users with Inspection Reports enabled only.
Starts an inspection report and returns the id of the inspection.

:return: The unique identifier of the inspection.
"""
return self.api_client.start_inspection()

def update_inspection_metadata(self, inspection_id: str, user_provided_key: str, user_provided_value: str) -> None:
"""
**NOTE:** For users with Inspection Reports enabled only.
Add/update inspection metadata with the user_provided_key and user_provided_value.

:param inspection_id: The unique identifier of the inspection.

:param user_provided_key: the key in the key/value pair for the inspection metadata.

:param user_provided_value: the value in the key/value pair for the inspection metadata.

:return: None
"""
self.api_client.update_inspection_metadata(inspection_id, user_provided_key, user_provided_value)

def stop_inspection(self, inspection_id: str) -> str:
"""
**NOTE:** For users with Inspection Reports enabled only.
Stops an inspection and raises an exception if the response from the server
indicates that the inspection was not successfully stopped.

:param inspection_id: The unique identifier of the inspection.

:return: "PASS" or "FAIL" depending on the result of the inspection.
"""
return self.api_client.stop_inspection(inspection_id)

def update_detector_confidence_threshold(self, detector: Union[str, Detector], confidence_threshold: float) -> None:
"""
Updates the confidence threshold for the given detector
Expand Down
Loading