biopb.image.cli

biopb.image.cli

CLI client for ProcessImage gRPC services.

Commands

ops List available operations from a ProcessImage server process Execute an operation on input image data

ops

ops(
    server: str = typer.Option(
        "grpc://localhost:50051",
        "--server",
        "-s",
        envvar="BIOPB_IMAGE_SERVER",
        help="ProcessImage server URI",
    ),
    token: Optional[str] = typer.Option(
        None,
        "--token",
        "-t",
        envvar="BIOPB_IMAGE_TOKEN",
        help="Bearer token for server authentication",
    ),
) -> None

List available operations from a ProcessImage server.

Example

biopb image ops --server grpc://localhost:50051 biopb image ops -s grpc://myhost:9000 --token mytoken123 BIOPB_IMAGE_TOKEN=mytoken123 biopb image ops

Source code in src/main/python/biopb/image/cli.py

@app.command()
def ops(
    server: str = typer.Option(
        "grpc://localhost:50051",
        "--server",
        "-s",
        envvar="BIOPB_IMAGE_SERVER",
        help="ProcessImage server URI",
    ),
    token: Optional[str] = typer.Option(
        None,
        "--token",
        "-t",
        envvar="BIOPB_IMAGE_TOKEN",
        help="Bearer token for server authentication",
    ),
) -> None:
    """List available operations from a ProcessImage server.

    Example:
        biopb image ops --server grpc://localhost:50051
        biopb image ops -s grpc://myhost:9000 --token mytoken123
        BIOPB_IMAGE_TOKEN=mytoken123 biopb image ops
    """
    start_time = time.time()
    channel = _create_grpc_channel(server)
    metadata = [("authorization", f"Bearer {token}")] if token else None
    try:
        stub = ProcessImageStub(channel)
        response: OpNames = stub.GetOpNames(
            empty_pb2.Empty(), metadata=metadata, timeout=10
        )

        if not response.names:
            stderr_console.print(f"[yellow]No operations found on {server}[/yellow]")
            _log_timing(start_time)
            return

        table = Table(title="Available Operations")
        table.add_column("Name", style="cyan")
        table.add_column("Description", style="green")
        table.add_column("Labels", style="magenta")
        table.add_column("Input Hint", style="blue")

        for name in response.names:
            schema: OpSchema = response.op_schemas.get(name)
            if schema:
                labels_str = ", ".join(schema.labels) if schema.labels else "-"
                hint_parts = []
                if schema.input_shape_hint:
                    if schema.input_shape_hint.expected_singletons:
                        hint_parts.append(
                            f"singleton: {','.join(schema.input_shape_hint.expected_singletons)}"
                        )
                    if schema.input_shape_hint.required_multivalue:
                        hint_parts.append(
                            f"multi: {','.join(schema.input_shape_hint.required_multivalue)}"
                        )
                hint_str = "; ".join(hint_parts) if hint_parts else "-"
                table.add_row(name, schema.description or "-", labels_str, hint_str)
            else:
                table.add_row(name, "-", "-", "-")

        console.print(table)
        stderr_console.print(
            f"\n[green]Server:[/green] {server}  [green]Operations:[/green] {len(response.names)}"
        )
        _log_timing(start_time)

    except grpc.RpcError as exc:
        stderr_console.print(f"[red]gRPC error:[/red] {exc.code()} - {exc.details()}")
        raise typer.Exit(1)
    except Exception as exc:
        stderr_console.print(f"[red]Error querying server:[/red] {exc}")
        raise typer.Exit(1)
    finally:
        channel.close()

process

process(
    input: Optional[str] = typer.Argument(
        None,
        help="Input file path or '-' for stdin. If omitted, reads from stdin.",
    ),
    op: Optional[str] = typer.Option(
        None,
        "--op",
        "-o",
        help="Operation name (optional if server has single/default op)",
    ),
    output: str = typer.Option(
        "-",
        "--output",
        "-O",
        help="Output path. Use '-' for stdout. Eager data requires filename.",
    ),
    format: Optional[str] = typer.Option(
        None,
        "--format",
        "-f",
        help="Output format for lazy data: pb (default) or pickle.",
    ),
    server: str = typer.Option(
        "grpc://localhost:50051",
        "--server",
        "-s",
        envvar="BIOPB_IMAGE_SERVER",
        help="ProcessImage server URI",
    ),
    token: Optional[str] = typer.Option(
        None,
        "--token",
        "-t",
        envvar="BIOPB_IMAGE_TOKEN",
        help="Bearer token for server authentication",
    ),
) -> None

Execute an image processing operation.

Input can be: - An image file (png, tiff, etc.) read via imageio - A protobuf SerializedTensor file (.pb) - Stdin containing protobuf or image bytes

Output depends on server response: - Eager data: saved as image file (stdout not allowed) - Lazy data: protobuf (.pb) or pickle (.pkl) format

Examples:

biopb image process input.png --op mock_echo --output output.png biopb image process input.pb --op mock_echo -O output.pb biopb tensor get my-source -o - | biopb image process --op segment -O - biopb image process input.png --op segment --token mytoken123 -O output.pb

Source code in src/main/python/biopb/image/cli.py

@app.command()
def process(
    input: Optional[str] = typer.Argument(
        None,
        help="Input file path or '-' for stdin. If omitted, reads from stdin.",
    ),
    op: Optional[str] = typer.Option(
        None,
        "--op",
        "-o",
        help="Operation name (optional if server has single/default op)",
    ),
    output: str = typer.Option(
        "-",
        "--output",
        "-O",
        help="Output path. Use '-' for stdout. Eager data requires filename.",
    ),
    format: Optional[str] = typer.Option(
        None,
        "--format",
        "-f",
        help="Output format for lazy data: pb (default) or pickle.",
    ),
    server: str = typer.Option(
        "grpc://localhost:50051",
        "--server",
        "-s",
        envvar="BIOPB_IMAGE_SERVER",
        help="ProcessImage server URI",
    ),
    token: Optional[str] = typer.Option(
        None,
        "--token",
        "-t",
        envvar="BIOPB_IMAGE_TOKEN",
        help="Bearer token for server authentication",
    ),
) -> None:
    """Execute an image processing operation.

    Input can be:
    - An image file (png, tiff, etc.) read via imageio
    - A protobuf SerializedTensor file (.pb)
    - Stdin containing protobuf or image bytes

    Output depends on server response:
    - Eager data: saved as image file (stdout not allowed)
    - Lazy data: protobuf (.pb) or pickle (.pkl) format

    Examples:
        biopb image process input.png --op mock_echo --output output.png
        biopb image process input.pb --op mock_echo -O output.pb
        biopb tensor get my-source -o - | biopb image process --op segment -O -
        biopb image process input.png --op segment --token mytoken123 -O output.pb
    """
    start_time = time.time()
    channel = _create_grpc_channel(server)
    fmt = _infer_format(output, format)
    metadata = [("authorization", f"Bearer {token}")] if token else None

    try:
        # Parse input
        is_file, data_or_path = _parse_input(input)
        image_data = _build_image_data(is_file, data_or_path)

        # Build request
        request = ProcessRequest(image_data=image_data)
        if op:
            request.op_name = op

        stderr_console.print(
            f"[green]Sending request to[/green] {server}"
            + (f" (op: {op})" if op else "")
        )

        # Call server
        stub = ProcessImageStub(channel)
        response: ProcessResponse = stub.Run(request, metadata=metadata, timeout=60)

        # Write output
        _write_output(response, output, fmt)
        _log_timing(start_time)

    except typer.Exit:
        raise
    except grpc.RpcError as exc:
        stderr_console.print(f"[red]gRPC error:[/red] {exc.code()} - {exc.details()}")
        raise typer.Exit(1)
    except Exception as exc:
        stderr_console.print(f"[red]Error processing image:[/red] {exc}")
        raise typer.Exit(1)
    finally:
        channel.close()