Home

ping()

GET: /ping

Endpoint to check if the server is running.

Returns:

Name	Type	Description
Response		Response with status 200 if the server is running.

Source code in app.py

@app.get("/ping")
def ping():
    """
    # GET: /ping

    Endpoint to check if the server is running.

    Returns:
        Response: Response with status 200 if the server is running.
    """
    try:
        _ = grpcclient.InferenceServerClient(url=config.grcp_model_server_address, verbose=False)
        return Response(status_code=200)
    except Exception:
        return Response(status_code=400)

predict_bucket(input_location=Header(None), inference_parameters=Header(None), webhook_url=Header(None), write_to_gcs=Header(False), input_bucket_name=Header(None), output_bucket_name=Header(None), examination_id=Header(None))

POST: /bucket_invocations

Endpoint to process an image and send it to the inference server.

Headers

Input-Location: Location of the image in the GCS bucket. Webhook-Url: URL to send the results of the inference. Write-To-GCS: Bool flag to write the results to a GCS bucket. False by default. Input-Bucket-Name: Name of the input bucket. Output-Bucket-Name: Name of the output bucket. Examination-ID: ID of the examination, used for logging and tracking. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- threshold: Threshold to apply to the output of the model. Default is 0.25.

- slice_idx: list of slice indices to run, e.g [45, 46, 47, ...]

Returns:

Name	Type	Description
		JSON with the results of the inference:
filename		Name of the file that was processed. >1 if multiple files.
status		Status of the request. Can be "sent" or "error".
result_path		Path to the result in the GCS bucket. >1 if multiple files.
request_uuid		UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/bucket_invocations")
def predict_bucket(
    input_location: str = Header(None),
    inference_parameters: str = Header(None),
    webhook_url: str = Header(None),
    write_to_gcs: bool = Header(False),
    input_bucket_name: str = Header(None),
    output_bucket_name: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /bucket_invocations

    Endpoint to process an image and send it to the inference server.

    Headers:
        *Input-Location*: Location of the image in the GCS bucket.
        *Webhook-Url*: URL to send the results of the inference.
        *Write-To-GCS*: Bool flag to write the results to a GCS bucket. False by default.
        *Input-Bucket-Name*: Name of the input bucket.
        *Output-Bucket-Name*: Name of the output bucket.
        *Examination-ID*: ID of the examination, used for logging and tracking.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys

            - threshold: Threshold to apply to the output of the model. Default is 0.25.

            - slice_idx: list of slice indices to run, e.g [45, 46, 47, ...]

    Returns:
        JSON with the results of the inference:
        filename: Name of the file that was processed. >1 if multiple files.
        status: Status of the request. Can be "sent" or "error".
        result_path: Path to the result in the GCS bucket. >1 if multiple files.
        request_uuid: UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

    Raises:
        Response: Error response if the content type is not supported.
    """

    webhook_response = _check_webhook(webhook_url, examination_id=examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    # get input and output locations
    input_bucket_to_use = (
        input_bucket_name if input_bucket_name is not None else config.input_bucket_name
    )
    output_bucket_to_use = (
        output_bucket_name if output_bucket_name is not None else config.output_bucket_name
    )
    if inference_parameters is None:
        inference_parameters = "{'threshold': 0.25}"
    inference_parameters = inference_parameters.replace("'", '"')
    inference_params = json.loads(inference_parameters)

    slice_idx = None
    if "slice_idx" in inference_params.keys():
        slice_idx = inference_params["slice_idx"]
        # add a lead 0
        slice_idx = [f"{i:03d}" for i in slice_idx]
        del inference_params["slice_idx"]

    start = time.time()
    images = asyncio.run(
        _read_from_gcp_bucket_async(input_bucket_to_use, input_location, examination_id, slice_idx=slice_idx, logger=logger)
    )
    elapsed = time.time() - start

    logger.info(json.dumps({
        "status": "INFO",
        "message": f"Read {len(images)} images from GCP bucket {input_bucket_to_use}/{input_location}. Took {elapsed} seconds.",
        "examination_id": examination_id
    }))


    try:
        response = _process_images(
            images,
            inference_parameters,
            output_bucket_to_use,
            webhook_url,
            write_to_gcs,
            examination_id,
        )

        return JSONResponse(
            content={
                "filename": response["filename"],
                "request_uuid": response["request_uuid"],
                "result_path": response["result_path"],
                "examination_id": examination_id,
                "status": "sent",
            },
            status_code=200,
        )

    except Exception as e:
        return JSONResponse(
            content={"error": str(e), "examination_id": examination_id}, status_code=400
        )

predict_bucket_azure_uae(input_location=Header(None), webhook_url=Header(None), inference_parameters=Header(None), write_to_gcs=Header(False), input_bucket_name=Header(None), output_bucket_name=Header(None), examination_id=Header(None))

POST: /bucket_invocations_azure_uae

Endpoint to process images from Microsoft Azure Blob Storage (UAE).

Headers

Input-Location: Location of the image in the bucket. Examination-ID: ID of the examination, used for logging and tracking. Webhook-Url: URL to send the results of the inference. Input-Bucket-Name: Name of the input GCS bucket. Default is the bucket name in the config file. Output-Bucket-Name: Name of the output GCS bucket. Default is the bucket name in the config file. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- threshold: Threshold to apply to the output of the model. Default is 0.25.

Returns:

Name	Type	Description
		JSON with the results of the inference:
filename		Name of the file that was processed. >1 if multiple files.
status		Status of the request. Can be "sent" or "error".
result_path		Path to the result in the GCS bucket. >1 if multiple files.
request_uuid		UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/bucket_invocations_azure_uae")
def predict_bucket_azure_uae(
    input_location: str = Header(None),
    webhook_url: str = Header(None),
    inference_parameters: str = Header(None),
    write_to_gcs: bool = Header(False),
    input_bucket_name: str = Header(None),
    output_bucket_name: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /bucket_invocations_azure_uae

    Endpoint to process images from Microsoft Azure Blob Storage (UAE).

    Headers:
        *Input-Location*: Location of the image in the bucket.
        *Examination-ID*: ID of the examination, used for logging and tracking.
        *Webhook-Url*: URL to send the results of the inference.
        *Input-Bucket-Name*: Name of the input GCS bucket. Default is the bucket name in the config file.
        *Output-Bucket-Name*: Name of the output GCS bucket. Default is the bucket name in the config file.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys


            - threshold: Threshold to apply to the output of the model. Default is 0.25.

    Returns:
        JSON with the results of the inference:
        filename: Name of the file that was processed. >1 if multiple files.
        status: Status of the request. Can be "sent" or "error".
        result_path: Path to the result in the GCS bucket. >1 if multiple files.
        request_uuid: UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

    Raises:
        Response: Error response if the content type is not supported.
    """
    webhook_response = _check_webhook(webhook_url, examination_id=examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    input_bucket_to_use = (
        input_bucket_name if input_bucket_name is not None else config.azure_uae_input_bucket_name
    )
    output_bucket_to_use = (
        output_bucket_name
        if output_bucket_name is not None
        else config.azure_uae_output_bucket_name
    )

    images = _read_from_azure_blob(
        input_bucket_to_use,
        input_location,
    )

    if not images:
        return JSONResponse(content={"error": "No images found"}, status_code=400)

    try:
        response = _process_images(
            images,
            inference_parameters,
            output_bucket_to_use,
            webhook_url,
            write_to_gcs,
            examination_id,
        )

        return JSONResponse(
            content={
                "filename": response["filename"],
                "request_uuid": response["request_uuid"],
                "result_path": response["result_path"],
                "examination_id": examination_id,
                "status": "sent",
            },
            status_code=200,
        )

    except Exception as e:
        return JSONResponse(
            content={"error": str(e), "examination_id": examination_id}, status_code=400
        )

predict_image(image=File(...), inference_parameters=Header(None), webhook_url=Header(None), examination_id=Header(None))

POST: /invocations

Endpoint to process an image and send it to the inference server.

Parameters:

Name	Type	Description	Default
image	UploadFile	Image file to process (in the request body).	File(...)

Headers

Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys:

- threshold: Threshold to apply to the output of the model. Default is 0.25.

Content-Type: Type of the image. Can be "image/jpeg", "image/png", "image/tiff", "image/bmp", "image/jpg". Webhook-URL: URL to send the results of the inference. Examination-ID: ID of the examination, used for logging and tracking.

Returns:

Type	Description
	JSON with the results of the inference:
	filename: Name of the file that was processed.
	status: Status of the request. Can be "sent" or "error".
	request_uuid: UUID of the request, generated by the server. Used to track the request results.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/invocations")
def predict_image(
    image: UploadFile = File(...),
    inference_parameters: str = Header(None),
    webhook_url: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /invocations

    Endpoint to process an image and send it to the inference server.

    Args:
        image (UploadFile): Image file to process (in the request body).

    Headers:
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys:

            - threshold: Threshold to apply to the output of the model. Default is 0.25.

        *Content-Type*: Type of the image. Can be "image/jpeg", "image/png", "image/tiff", "image/bmp", "image/jpg".
        *Webhook-URL*: URL to send the results of the inference.
        *Examination-ID*: ID of the examination, used for logging and tracking.

    Returns:
        JSON with the results of the inference:
        - filename: Name of the file that was processed.
        - status: Status of the request. Can be "sent" or "error".
        - request_uuid: UUID of the request, generated by the server. Used to track the request results.

    Raises:
        Response: Error response if the content type is not supported.
    """

    client = grpcclient.InferenceServerClient(
        url=config.grcp_model_server_address,
        verbose=False,
        channel_args=(("grpc.lb_policy_name", "round_robin"),),
    )  # , concurrency=1, connection_timeout=10)
    model_config = client.get_model_config(
        model_name=config.model_name, model_version=config.model_version, as_json=True
    )["config"]

    content_type = image.content_type

    webhook_response = _check_webhook(webhook_url, examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    if content_type not in config.available_content_types:
        return Response(
            status=415,
            content="Cannot decode image data. Is content_type correct?",
            media_type="text/plain",
        )

    try:
        contents = image.file.read()

        image_bytes = np.frombuffer(contents, dtype=np.uint8)

        img = cv2.imdecode(image_bytes, cv2.IMREAD_GRAYSCALE)

        img = img.astype(np.uint8)
        img = img[np.newaxis, ...]

        inputs = [
            grpcclient.InferInput("IMAGE", img.shape, np_to_triton_dtype(img.dtype)),
            grpcclient.InferInput("INPUT_JSON_PARAMS", (1, 1), "BYTES"),
        ]
        inputs[0].set_data_from_numpy(img)
        inference_params = inference_parameters.replace("'", '"')
        inputs[1].set_data_from_numpy(np.array([[inference_params]] * 1, dtype=np.object_))

        outputs = [
            grpcclient.InferRequestedOutput(model_config["output"][i]["name"])
            for i in range(len(model_config["output"]))
        ]

        request_uuid = str(uuid.uuid4())

        _ = client.async_infer(
            model_name=config.model_name,
            model_version=config.model_version,
            inputs=inputs,
            outputs=outputs,
            callback=partial(
                result_callback,
                model_config=model_config,
                filename=image.filename,
                request_uuid=request_uuid,
                client=client,
                webhook_url=webhook_url,
                image_resolution=(img.shape[1], img.shape[2]),
                examination_id=examination_id,
            ),
        )

        return JSONResponse(
            content={
                "filename": image.filename,
                "status": "sent",
                "request_uuid": request_uuid,
                "examination_id": examination_id,
            },
            status_code=200,
        )
    except Exception as e:
        return JSONResponse(
            content={"message": str(e), "status": "error", "examination_id": examination_id},
            status_code=400,
        )

read_from_gcp_bucket(input_bucket, prefix, slice_idx=None)

Function to read images from a GCP bucket.

Parameters:

Name	Type	Description	Default
prefix	str	Prefix to search for images in the bucket.	required

Returns:

Name	Type	Description
List	Tuple[str, ndarray]	List of images read from the bucket.

Source code in app.py

def read_from_gcp_bucket(
    input_bucket: str, prefix: str, slice_idx: List[int] = None
) -> Tuple[str, np.ndarray]:
    """
    Function to read images from a GCP bucket.

    Args:
        prefix (str): Prefix to search for images in the bucket.

    Returns:
        List: List of images read from the bucket.
    """

    # make client
    client = storage.Client()

    # get bucket
    bucket = client.bucket(input_bucket)

    # get blobs
    blobs = []

    if slice_idx is not None:
        for idx in slice_idx:
            name = f"{prefix}/file.{idx}"
            blob = bucket.list_blobs(prefix=name)
            blobs.extend(blob)
    else:
        blobs = bucket.list_blobs(prefix=prefix)

    images = []
    for blob in blobs:
        # read image
        try:
            image = cv2.imdecode(
                np.frombuffer(blob.download_as_string(), dtype=np.uint8), cv2.IMREAD_GRAYSCALE
            )
            images.append((blob.name, image))
        except Exception as e:
            log_message = json.dumps(
                {
                    "status": "ERROR",
                    "message": f"Failed to read image {blob.name} from GCP bucket {input_bucket}: {str(e)}",
                }
            )
            logger.error(log_message)
            continue

    return images

result_callback(model_config, filename, request_uuid, result, error, client, webhook_url, image_resolution, examination_id)

Callback function to process the result of the inference request. Sends a webhook action to the callback service.

Parameters:

Name	Type	Description	Default
model_config	dict	Model configuration dictionary.	required
filename	str	Name of the file that was processed.	required
result	list	List of output tensors.	required
error	Exception	Error that occurred during the request.	required
request_uuid	str	UUID of the request.	required
client	object	Triton client object.	required
webhook_url	str	URL of the webhook service.	required
image_resolution	Tuple[int, int]	Resolution of the input image.	required
examination_id	str	ID of the examination, used for logging and tracking.	required

Source code in app.py

def result_callback(
    model_config: dict,
    filename: str,
    request_uuid: str,
    result: Optional[list],
    error: Optional[Exception],
    client: object,
    webhook_url: str,
    image_resolution: Tuple[int, int],
    examination_id: str,
) -> None:
    """
    Callback function to process the result of the inference request.
    Sends a webhook action to the callback service.

    Args:
        model_config (dict): Model configuration dictionary.
        filename (str): Name of the file that was processed.
        result (list): List of output tensors.
        error (Exception): Error that occurred during the request.
        request_uuid (str): UUID of the request.
        client (object): Triton client object.
        webhook_url (str): URL of the webhook service.
        image_resolution (Tuple[int, int]): Resolution of the input image.
        examination_id (str): ID of the examination, used for logging and tracking.
    """

    if error is None:
        output_data = get_output_data_from_response(result, model_config)

        formed_output = {
            "metadata": {
                "height": image_resolution[0],
                "width": image_resolution[1],
            },
            "instances": [
                {
                    "type": "polyline",
                    "points": output_data,
                    "className": "ilm",
                    "classId": 14,
                    "probability": 100,
                    "attributes": [],
                }
            ],
        }

        status_message = {
            "id": request_uuid,
            "status": "COMPLETED",
            "output": formed_output,
            "filename": filename,
        }

        response = requests.post(webhook_url, json=status_message)

        if response.status_code == 200:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "COMPLETED",
                    "filename": filename,
                    "request_uuid": request_uuid,
                }
            )

            logger.info(log_message)
        else:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "ERROR",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": f"Error sending webhook action for {filename, request_uuid}",
                }
            )
            logger.error(log_message)
    else:
        status_message = {
            "id": request_uuid,
            "status": "FAILED",
            "error": str(error),
            "filename": filename,
        }
        response = requests.post(webhook_url, json=status_message)

        if response.status_code == 200:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "FAILED",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": str(error),
                }
            )
            logger.error(log_message)
        else:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "ERROR",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": f"Error sending webhook action for {filename, request_uuid}",
                }
            )
            logger.error(log_message)

    client.close()

result_image_bucket_callback(model_config, filename, result, error, client, request_uuid, webhook_url, output_bucket_name, examination_id, write_to_gcs=False, image_resolution=None)

Callback function to process the result of the inference request. Writes the result to a GCS bucket.

Parameters:

Name	Type	Description	Default
model_config	dict	Model configuration dictionary.	required
filename	str	Name of the file that was processed.	required
result	list	List of output tensors.	required
error	Exception	Error that occurred during the request.	required
request_uuid	str	UUID of the request.	required
client	object	Triton client object.	required
webhook_url	str	URL of the webhook service.	required
write_to_gcs	bool	Flag to write the result to a GCS bucket.	False
output_bucket_name	str	Name of the output bucket.	required
image_resolution	Tuple[int, int]	Resolution of the input image.	None
examination_id	str	ID of the examination, used for logging and tracking.	required

Source code in app.py

def result_image_bucket_callback(
    model_config: dict,
    filename: str,
    result: Optional[list],
    error: Optional[Exception],
    client: object,
    request_uuid: str,
    webhook_url: str,
    output_bucket_name: str,
    examination_id: str,
    write_to_gcs: bool = False,
    image_resolution: Tuple[int, int] = None,
) -> None:
    """
    Callback function to process the result of the inference request.
    Writes the result to a GCS bucket.

    Args:
        model_config (dict): Model configuration dictionary.
        filename (str): Name of the file that was processed.
        result (list): List of output tensors.
        error (Exception): Error that occurred during the request.
        request_uuid (str): UUID of the request.
        client (object): Triton client object.
        webhook_url (str): URL of the webhook service.
        write_to_gcs (bool): Flag to write the result to a GCS bucket.
        output_bucket_name (str): Name of the output bucket.
        image_resolution (Tuple[int, int]): Resolution of the input image.
        examination_id (str): ID of the examination, used for logging and tracking.
    """
    if error is None:
        output_data = get_output_data_from_response(result, model_config)

        formed_output = {
            "metadata": {
                "height": image_resolution[0],
                "width": image_resolution[1],
            },
            "instances": [
                {
                    "type": "polyline",
                    "points": output_data,
                    "className": "ilm",
                    "classId": 14,
                    "probability": 100,
                    "attributes": [],
                }
            ],
        }

        status_message = {
            "id": request_uuid,
            "status": "COMPLETED",
            "output": formed_output,
            "filename": filename,
        }
    else:
        output_data = {"error": str(error)}
        status_message = {
            "id": request_uuid,
            "status": "FAILED",
            "error": str(error),
            "filename": filename,
        }

    if write_to_gcs:
        write_json_to_gcs(
            output_bucket_name, output_data, f"{config.output_folder_name}/{request_uuid}.json"
        )

    client.close()

    response = requests.post(webhook_url, json=status_message)

    if response.status_code == 200:
        log_message = json.dumps(
            {
                "model": model_config["name"],
                "examination_id": examination_id,
                "status": "COMPLETED" if error is None else "FAILED",
                "filename": filename,
                "request_uuid": request_uuid,
            }
        )
        logger.info(log_message)
    else:
        log_message = json.dumps(
            {
                "model": model_config["name"],
                "examination_id": examination_id,
                "status": "ERROR",
                "filename": filename,
                "request_uuid": request_uuid,
                "error": f"Error sending webhook action for {filename, request_uuid}",
            }
        )
        logger.error(log_message)

write_json_to_gcs(bucket_name, json_data, output_path)

Write JSON data to a specific folder in a GCS bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the GCS bucket.	required
json_data	dict	The JSON data to be written.	required
output_path	str	The GCS path where the JSON file will be stored (e.g., 'folder/output_file.json').	required

Source code in app.py

def write_json_to_gcs(bucket_name: str, json_data: dict, output_path: str):
    """
    Write JSON data to a specific folder in a GCS bucket.

    Args:
        bucket_name (str): The name of the GCS bucket.
        json_data (dict): The JSON data to be written.
        output_path (str): The GCS path where the JSON file will be stored (e.g., 'folder/output_file.json').
    """
    # Convert the JSON data to a string
    json_str = json.dumps(json_data, indent=4)

    # Run the blocking GCS code in a thread pool
    _upload_to_gcs(json_str, bucket_name, output_path)
    log_message = json.dumps(
        {
            "model": config.model_name,
            "status": "COMPLETED",
            "filename": output_path,
        }
    )
    logger.info(log_message)