feat(worker): add --pipeline filter to register, --project filter to run, and ami worker run alias

- `ami worker register --pipeline <slug>` now only advertises the
  specified pipelines instead of all. Filters the pipeline_configs
  list before POSTing to Antenna.
- `ami worker [run] --project <id>` limits which project jobs the
  worker pulls via `project__id__in` query param on the /jobs endpoint.
- `ami worker run` is now an explicit subcommand alias for `ami worker`.
- Registration logging now reports processing service ID, created vs
  updated pipeline counts, and specific slugs.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mihow

trapdata/antenna/client.py

@@ -32,6 +32,7 @@ def get_jobs(
     base_url: str,
     auth_token: str,
     pipeline_slugs: list[str],
+    project_ids: list[int] | None = None,
 ) -> list[tuple[int, str]]:
     """Fetch job ids from the API for the given pipelines in a single request.
 
@@ -41,6 +42,8 @@ def get_jobs(
         base_url: Antenna API base URL (e.g., "http://localhost:8000/api/v2")
         auth_token: API authentication token
         pipeline_slugs: List of pipeline slugs to filter jobs
+        project_ids: Optional list of project IDs to limit jobs to.
+            If empty or None, pulls jobs for all projects the token has access to.
 
     Returns:
         List of (job_id, pipeline_slug) tuples (possibly empty) on success or error.
@@ -50,12 +53,14 @@ def get_jobs(
             if not pipeline_slugs:
                 return []
             url = f"{base_url.rstrip('/')}/jobs"
-            params = {
+            params: dict[str, str | int] = {
                 "pipeline__slug__in": ",".join(pipeline_slugs),
                 "ids_only": 1,
                 "incomplete_only": 1,
                 "dispatch_mode": JobDispatchMode.ASYNC_API,  # Only fetch async_api jobs
             }
+            if project_ids:
+                params["project__id__in"] = ",".join(str(pid) for pid in project_ids)
 
             resp = session.get(url, params=params, timeout=30)
             resp.raise_for_status()

trapdata/antenna/registration.py

@@ -7,7 +7,7 @@
     AsyncPipelineRegistrationRequest,
     AsyncPipelineRegistrationResponse,
 )
-from trapdata.api.api import PIPELINE_CHOICES, initialize_service_info
+from trapdata.api.api import initialize_service_info
 from trapdata.api.utils import get_http_session
 from trapdata.common.logs import logger
 from trapdata.settings import Settings, read_settings
@@ -48,7 +48,24 @@ def register_pipelines_for_project(
             response.raise_for_status()
 
             result = AsyncPipelineRegistrationResponse.model_validate(response.json())
-            return True, f"Created {len(result.pipelines_created)} new pipelines"
+            parts = []
+            if result.processing_service_id:
+                parts.append(f"Processing service ID {result.processing_service_id}")
+            if result.pipelines_created:
+                parts.append(
+                    f"created {len(result.pipelines_created)} pipelines "
+                    f"({', '.join(result.pipelines_created)})"
+                )
+            if result.pipelines_updated:
+                parts.append(
+                    f"updated {len(result.pipelines_updated)} pipelines "
+                    f"({', '.join(result.pipelines_updated)})"
+                )
+            if not result.pipelines_created and not result.pipelines_updated:
+                parts.append(
+                    f"all {len(pipeline_configs)} pipelines already registered"
+                )
+            return True, "; ".join(parts)
 
         except requests.RequestException as e:
             if (
@@ -72,6 +89,7 @@ def register_pipelines(
     project_ids: list[int],
     service_name: str,
     settings: Settings | None = None,
+    pipeline_slugs: list[str] | None = None,
 ) -> None:
     """
     Register pipelines for specified projects or all accessible projects.
@@ -80,6 +98,8 @@ def register_pipelines(
         project_ids: List of specific project IDs to register for. If empty, registers for all accessible projects.
         service_name: Name of the processing service
         settings: Settings object with antenna_api_* configuration (defaults to read_settings())
+        pipeline_slugs: Optional list of pipeline slugs to register. If None or empty,
+            registers all available pipelines.
     """
     # Import here to avoid circular import
     from trapdata.antenna.client import get_user_projects
@@ -128,13 +148,22 @@ def register_pipelines(
     logger.info("Initializing pipeline configurations...")
     service_info = initialize_service_info()
     pipeline_configs = service_info.pipelines
-    logger.info(f"Generated {len(pipeline_configs)} pipeline configurations")
+
+    # Filter to requested pipelines if specified
+    if pipeline_slugs:
+        slug_set = set(pipeline_slugs)
+        pipeline_configs = [p for p in pipeline_configs if p.slug in slug_set]
+        logger.info(
+            f"Filtered to {len(pipeline_configs)} pipelines: {', '.join(pipeline_slugs)}"
+        )
+    else:
+        logger.info(f"Registering all {len(pipeline_configs)} pipeline configurations")
 
     # Register pipelines for each project
     successful_registrations = []
     failed_registrations = []
 
-    logger.info(f"Available pipelines to register: {list(PIPELINE_CHOICES.keys())}")
+    logger.info(f"Pipelines to register: {[p.slug for p in pipeline_configs]}")
 
     for project in projects_to_process:
         project_id = project["id"]
@@ -164,10 +193,12 @@ def register_pipelines(
 
     # Summary report
     logger.info("\n=== Registration Summary ===")
-    logger.info(f"Service name: {full_service_name}")
-    logger.info(f"Total projects processed: {len(projects_to_process)}")
-    logger.info(f"Successful registrations: {len(successful_registrations)}")
-    logger.info(f"Failed registrations: {len(failed_registrations)}")
+    logger.info(f"Processing service: {full_service_name}")
+    logger.info(f"Pipelines advertised: {len(pipeline_configs)}")
+    logger.info(f"Projects processed: {len(projects_to_process)}")
+    logger.info(
+        f"Successful: {len(successful_registrations)}, Failed: {len(failed_registrations)}"
+    )
 
     if successful_registrations:
         logger.info("\nSuccessful registrations:")

trapdata/antenna/worker.py

@@ -30,13 +30,19 @@
 SLEEP_TIME_SECONDS = 5
 
 
-def run_worker(pipelines: list[str]):
+def run_worker(pipelines: list[str], project_ids: list[int] | None = None):
     """Run the worker to process images from the REST API queue.
 
     Automatically spawns one AMI worker instance process per available GPU.
     On single-GPU or CPU-only machines, runs in-process (no overhead).
+
+    Args:
+        pipelines: Pipeline slugs to poll for jobs.
+        project_ids: Optional project IDs to limit jobs to. If empty/None,
+            pulls jobs for all projects the auth token has access to.
     """
     settings = read_settings()
+    project_ids = project_ids or []
 
     # Validate auth token
     if not settings.antenna_api_auth_token:
@@ -59,7 +65,7 @@ def run_worker(pipelines: list[str]):
         # can't be pickled. Each child process calls read_settings() itself.
         mp.spawn(
             _worker_loop,
-            args=(pipelines,),
+            args=(pipelines, project_ids),
             nprocs=gpu_count,
             join=True,
         )
@@ -68,17 +74,21 @@ def run_worker(pipelines: list[str]):
             logger.info(f"Found 1 GPU: {torch.cuda.get_device_name(0)}")
         else:
             logger.info("No GPUs found, running on CPU")
-        _worker_loop(0, pipelines)
+        _worker_loop(0, pipelines, project_ids)
 
 
-def _worker_loop(gpu_id: int, pipelines: list[str]):
+def _worker_loop(
+    gpu_id: int, pipelines: list[str], project_ids: list[int] | None = None
+):
     """Main polling loop for a single AMI worker instance, pinned to a specific GPU.
 
     Args:
         gpu_id: GPU index to pin this AMI worker instance to (0 for CPU-only).
         pipelines: List of pipeline slugs to poll for jobs.
+        project_ids: Optional project IDs to limit jobs to.
     """
     settings = read_settings()
+    project_ids = project_ids or []
     device = torch.device(f"cuda:{gpu_id}" if torch.cuda.is_available() else "cpu")
     if torch.cuda.is_available() and torch.cuda.device_count() > 0:
         torch.cuda.set_device(gpu_id)
@@ -89,6 +99,8 @@ def _worker_loop(gpu_id: int, pipelines: list[str]):
     # Build full service name with hostname
     full_service_name = get_full_service_name(settings.antenna_service_name)
     logger.info(f"Running worker as: {full_service_name}")
+    if project_ids:
+        logger.info(f"Filtering jobs to projects: {project_ids}")
 
     while True:
         # TODO CGJS: Support pulling and prioritizing single image tasks, which are used in interactive testing
@@ -102,6 +114,7 @@ def _worker_loop(gpu_id: int, pipelines: list[str]):
             base_url=settings.antenna_api_base_url,
             auth_token=settings.antenna_api_auth_token,
             pipeline_slugs=pipelines,
+            project_ids=project_ids if project_ids else None,
         )
         for job_id, pipeline in jobs:
             logger.info(

trapdata/cli/worker.py

@@ -8,74 +8,121 @@
 
 cli = typer.Typer(help="Antenna worker commands for remote processing")
 
+_PIPELINE_HELP = (
+    "Pipeline to use for processing (e.g., moth_binary, panama_moths_2024). "
+    "Can be specified multiple times. Defaults to all pipelines if not specified."
+)
+_PROJECT_HELP = (
+    "Limit to jobs for specific Antenna project IDs. "
+    "Can be specified multiple times. Defaults to all projects the auth token has access to."
+)
+
+
+def _validate_pipelines(pipelines: list[str] | None) -> list[str]:
+    """Resolve and validate the --pipeline option."""
+    if not pipelines:
+        return list(PIPELINE_CHOICES.keys())
+
+    invalid = [p for p in pipelines if p not in PIPELINE_CHOICES]
+    if invalid:
+        raise typer.BadParameter(
+            f"Invalid pipeline(s): {', '.join(invalid)}. "
+            f"Must be one of: {', '.join(PIPELINE_CHOICES.keys())}"
+        )
+    return pipelines
+
+
+def _start_worker(pipelines: list[str] | None, project: list[int] | None) -> None:
+    """Shared implementation for ``ami worker`` and ``ami worker run``."""
+    validated = _validate_pipelines(pipelines)
+    project_ids = project or []
+
+    from trapdata.antenna.worker import run_worker
+
+    run_worker(pipelines=validated, project_ids=project_ids)
+
 
 @cli.callback(invoke_without_command=True)
-def run(
+def worker_callback(
     ctx: typer.Context,
     pipelines: Annotated[
         list[str] | None,
-        typer.Option(
-            "--pipeline",
-            help="Pipeline to use for processing (e.g., moth_binary, panama_moths_2024). Can be specified multiple times. "
-            "Defaults to all pipelines if not specified.",
-        ),
+        typer.Option("--pipeline", help=_PIPELINE_HELP),
+    ] = None,
+    project: Annotated[
+        list[int] | None,
+        typer.Option("--project", help=_PROJECT_HELP),
     ] = None,
 ):
     """
     Run the worker to process images from the Antenna API queue.
 
     Can be invoked as 'ami worker' or 'ami worker run'.
     """
-    # Only run the worker if no subcommand was invoked
     if ctx.invoked_subcommand is not None:
         return
+    _start_worker(pipelines, project)
 
-    if not pipelines:
-        pipelines = list(PIPELINE_CHOICES.keys())
-
-    # Validate that each pipeline is in PIPELINE_CHOICES
-    invalid_pipelines = [
-        pipeline for pipeline in pipelines if pipeline not in PIPELINE_CHOICES.keys()
-    ]
 
-    if invalid_pipelines:
-        raise typer.BadParameter(
-            f"Invalid pipeline(s): {', '.join(invalid_pipelines)}. Must be one of: {', '.join(PIPELINE_CHOICES.keys())}"
-        )
-
-    from trapdata.antenna.worker import run_worker
+@cli.command("run")
+def run_cmd(
+    pipelines: Annotated[
+        list[str] | None,
+        typer.Option("--pipeline", help=_PIPELINE_HELP),
+    ] = None,
+    project: Annotated[
+        list[int] | None,
+        typer.Option("--project", help=_PROJECT_HELP),
+    ] = None,
+):
+    """
+    Run the worker to process images from the Antenna API queue.
 
-    run_worker(pipelines=pipelines)
+    Alias for 'ami worker' — both forms are identical.
+    """
+    _start_worker(pipelines, project)
 
 
 @cli.command("register")
 def register(
     project: Annotated[
         list[int] | None,
         typer.Option(
+            "--project",
             help="Specific project IDs to register pipelines for. "
             "If not specified, registers for all accessible projects.",
         ),
     ] = None,
+    pipelines: Annotated[
+        list[str] | None,
+        typer.Option("--pipeline", help=_PIPELINE_HELP),
+    ] = None,
 ):
     """
     Register available pipelines with the Antenna platform for specified projects.
 
-    This command registers all available pipeline configurations with the Antenna platform
-    for the specified projects (or all accessible projects if none specified).
+    This command registers the processing service and its pipeline configurations
+    with the Antenna platform for the specified projects (or all accessible projects
+    if none specified).
+
+    When --pipeline is specified, only those pipelines are advertised (instead of all).
 
     The service name is read from the AMI_ANTENNA_SERVICE_NAME configuration setting.
     Hostname will be added automatically to the service name.
 
     Examples:
         ami worker register --project 1 --project 2
-        ami worker register  # registers for all accessible projects
+        ami worker register --pipeline mothbot_insect_orders_2025
+        ami worker register  # registers all pipelines for all accessible projects
     """
     from trapdata.antenna.registration import register_pipelines
     from trapdata.settings import read_settings
 
     settings = read_settings()
     project_ids = project if project else []
+    validated_pipelines = _validate_pipelines(pipelines)
     register_pipelines(
-        project_ids=project_ids, service_name=settings.antenna_service_name
+        project_ids=project_ids,
+        service_name=settings.antenna_service_name,
+        pipeline_slugs=validated_pipelines,
     )