inference_cli API Reference¶

def terminate_running_containers(
    containers: List[Container], interactive_mode: bool = True
) -> bool:
    """
    Args:
        containers (List[Container]): List of containers to handle
        interactive_mode (bool): Flag to determine if user prompt should decide on container termination

    Returns: boolean value that informs if there are containers that have not received SIGKILL
        as a result of procedure.
    """
    running_inference_containers = [
        c for c in containers if is_container_running(container=c)
    ]
    containers_to_kill = running_inference_containers
    if interactive_mode:
        containers_to_kill = [
            c for c in running_inference_containers if ask_user_to_kill_container(c)
        ]
    kill_containers(containers=containers_to_kill)
    return len(containers_to_kill) < len(running_inference_containers)

def create_images_batch_from_cloud_storage(
    bucket_path: str,
    batch_id: str,
    api_key: str,
    batch_name: Optional[str] = None,
    ingest_id: Optional[str] = None,
    notifications_url: Optional[str] = None,
    notification_categories: Optional[List[str]] = None,
    presign_expiration_seconds: int = 86400,
) -> None:
    """
    Create image batch from cloud storage by generating presigned URLs.

    Args:
        bucket_path: Cloud path with optional glob pattern (e.g., 's3://bucket/**/*.jpg')
        batch_id: Batch identifier
        api_key: Roboflow API key
        presign_expiration_seconds: Presigned URL expiration time (default: 24 hours)

    Internally calls trigger_images_references_ingest with generated presigned URLs.
    """
    try:
        import fsspec
    except ImportError:
        raise ImportError(
            "Cloud storage support requires additional dependencies. "
            "Install with: pip install 'inference-cli[cloud-storage]'"
        )

    base_path, glob_pattern = _parse_bucket_path(bucket_path)
    protocol = base_path.split("://")[0]
    fs = fsspec.filesystem(protocol, **_get_fs_kwargs(protocol))

    # Stream and filter image files with progress
    image_files_generator = _list_and_filter_files_streaming(
        fs, base_path, glob_pattern, IMAGES_EXTENSIONS
    )

    # Generate presigned URLs in parallel (consumes generator and shows progress)
    references = _generate_presigned_urls_parallel(
        fs, image_files_generator, base_path, presign_expiration_seconds
    )

    if len(references) == 0:
        pattern_desc = glob_pattern if glob_pattern else "all image files"
        raise ValueError(
            f"No image files found matching pattern: {pattern_desc} in {base_path}\n"
            f"Supported extensions: {', '.join(IMAGES_EXTENSIONS)}\n"
            f"Note: If you're getting connection errors, check your cloud credentials and network access."
        )

    if len(references) > MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST:
        num_chunks = (
            len(references) + MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST - 1
        ) // MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST
        print(
            f"Files will be split into {num_chunks} chunks of up to {MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST} files each"
        )

    workspace = get_workspace(api_key=api_key)

    # Split into batches if needed
    ingest_parts = list(
        create_batches(
            sequence=references, batch_size=MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST
        )
    )
    if len(ingest_parts) > 1:
        print(
            f"Your ingest exceeds {MAX_IMAGE_REFERENCES_IN_INGEST_REQUEST} files - we split the ingest "
            f"into {len(ingest_parts)} chunks."
        )

    # Trigger ingest for each batch
    for batch_references in ingest_parts:
        response = trigger_images_references_ingest(
            workspace=workspace,
            batch_id=batch_id,
            references=batch_references,
            api_key=api_key,
            ingest_id=ingest_id,
            batch_name=batch_name,
            notifications_url=notifications_url,
            notification_categories=notification_categories,
        )
        print(f"Ingest triggered. Ingest ID: {response.ingest_id}")

    if notifications_url:
        print(f"Monitor updates that will be sent to: {notifications_url}")
        print(
            f"You can also use `inference rf-cloud data-staging list-ingest-details --batch-id {batch_id}` command "
            f"to check progress."
        )
    else:
        print(
            f"Use `inference rf-cloud data-staging list-ingest-details --batch-id {batch_id}` "
            "command to watch the ingest progress. If you want automated updates - use `--notifications-url` option "
            "of this command."
        )

def create_videos_batch_from_cloud_storage(
    bucket_path: str,
    batch_id: str,
    api_key: str,
    batch_name: Optional[str] = None,
    ingest_id: Optional[str] = None,
    notifications_url: Optional[str] = None,
    notification_categories: Optional[List[str]] = None,
    presign_expiration_seconds: int = 86400,
) -> None:
    """
    Create video batch from cloud storage by generating presigned URLs.

    Args:
        bucket_path: Cloud path with optional glob pattern (e.g., 's3://bucket/**/*.mp4')
        batch_id: Batch identifier
        api_key: Roboflow API key
        presign_expiration_seconds: Presigned URL expiration time (default: 24 hours)

    Internally calls trigger_videos_references_ingest with generated presigned URLs.
    """
    try:
        import fsspec
    except ImportError:
        raise ImportError(
            "Cloud storage support requires additional dependencies. "
            "Install with: pip install 'inference-cli[cloud-storage]'"
        )

    base_path, glob_pattern = _parse_bucket_path(bucket_path)
    protocol = base_path.split("://")[0]
    fs = fsspec.filesystem(protocol, **_get_fs_kwargs(protocol))

    # Stream and filter video files with progress
    video_files_generator = _list_and_filter_files_streaming(
        fs, base_path, glob_pattern, VIDEOS_EXTENSIONS
    )

    # Generate presigned URLs in parallel (consumes generator and shows progress)
    references = _generate_presigned_urls_parallel(
        fs, video_files_generator, base_path, presign_expiration_seconds
    )

    if len(references) == 0:
        pattern_desc = glob_pattern if glob_pattern else "all video files"
        raise ValueError(
            f"No video files found matching pattern: {pattern_desc} in {base_path}\n"
            f"Supported extensions: {', '.join(VIDEOS_EXTENSIONS)}\n"
            f"Note: If you're getting connection errors, check your cloud credentials and network access."
        )

    print(f"Found {len(references)} video files")
    if len(references) > SUGGESTED_MAX_VIDEOS_IN_BATCH:
        print(
            f"Warning: Found {len(references)} videos. Suggested max is {SUGGESTED_MAX_VIDEOS_IN_BATCH} videos per batch."
        )

    workspace = get_workspace(api_key=api_key)

    # Trigger ingest directly with the list of references
    response = trigger_videos_references_ingest(
        workspace=workspace,
        batch_id=batch_id,
        references=references,
        api_key=api_key,
        ingest_id=ingest_id,
        batch_name=batch_name,
        notifications_url=notifications_url,
        notification_categories=notification_categories,
    )
    print(f"Ingest triggered. Ingest ID: {response.ingest_id}")

    if notifications_url:
        print(f"Monitor updates that will be sent to: {notifications_url}")
        print(
            f"You can also use `inference rf-cloud data-staging list-ingest-details --batch-id {batch_id}` command "
            f"to check progress."
        )
    else:
        print(
            f"Use `inference rf-cloud data-staging list-ingest-details --batch-id {batch_id}` "
            "command to watch the ingest progress. If you want automated updates - use `--notifications-url` option "
            "of this command."
        )