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:
        client = 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 to track the request results. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

- nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

- mm_crop_zone: how much to crop from the center of the image.

- mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- slice_thickness: slice thickness parameter of the exam.

- pixel_spacing_column: pixel spacing column parameter of the exam.

- type_of_scan: type of scan, should be macula, widescan, optic_disk

- zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

- scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

- num_slices: number of slices in the exam

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 to track the request results.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys

            - nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

            - nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

            - mm_crop_zone: how much to crop from the center of the image.

            - mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - slice_thickness: slice thickness parameter of the exam.

            - pixel_spacing_column: pixel spacing column parameter of the exam.

            - type_of_scan: type of scan, should be macula, widescan, optic_disk

            - zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

            - scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

            - num_slices: number of slices in the exam


    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, logger)
    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
    )

    start = time.time()
    images = asyncio.run(
        _read_from_gcp_bucket_async(input_bucket_to_use, input_location, examination_id, config, 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:
        request_uuids = []
        result_paths = []
        filenames = []
        statuses = []

        for filename, image in images:
            client = grpcclient.InferenceServerClient(
                url=config.grcp_model_server_address,
                verbose=False,
                channel_args=(("grpc.lb_policy_name", "round_robin"),),
            )
            model_config = client.get_model_config(
                model_name=model_name, model_version=model_version, as_json=True
            )["config"]
            img = image
            img = img[np.newaxis, ..., np.newaxis].astype(np.float32)

            inputs = [
                grpcclient.InferInput("IMAGE", img.shape, np_to_triton_dtype(img.dtype)),
                grpcclient.InferInput("INPUT_JSON", (1, 1), "BYTES"),
            ]

            inputs[0].set_data_from_numpy(img)

            slice_idx = int(filename.split(".")[1])
            inference_params = inference_parameters.replace("'", '"')
            dict_inference_parameters = json.loads(inference_params)
            dict_inference_parameters["slice_idx"] = slice_idx

            zone_to_run = define_zone_to_run(dict_inference_parameters, logger)
            request_uuid = str(uuid.uuid4())
            if zone_to_run is not None:
                dict_inference_parameters["zone_of_interest"] = zone_to_run
                inference_params = json.dumps(dict_inference_parameters)
                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_uuids.append(request_uuid)
                result_paths.append(
                    f"{output_bucket_to_use}/{config.output_folder_name}/{request_uuid}.json"
                )
                filenames.append(filename)

                statuses.append("sent")

                response = client.async_infer(
                    model_name=model_name,
                    model_version=model_version,
                    inputs=inputs,
                    outputs=outputs,
                    callback=partial(
                        result_image_bucket_callback,
                        model_config=model_config,
                        filename=filename,
                        request_uuid=request_uuid,
                        client=client,
                        webhook_url=webhook_url,
                        write_to_gcs=write_to_gcs,
                        output_bucket_name=output_bucket_to_use,
                        examination_id=examination_id,
                        logger=logger
                    ),
                )
            else:
                filenames.append(filename)
                request_uuids.append(request_uuid)
                result_paths.append(None)
                statuses.append("no_zone_to_run")

        return JSONResponse(
            content={
                "filename": filenames if len(filenames) > 1 else filenames[0],
                "request_uuid": request_uuids if len(request_uuids) > 1 else request_uuids[0],
                "result_path": result_paths if len(result_paths) > 1 else result_paths[0],
                "status": statuses if len(statuses) > 1 else statuses[0],
                "examination_id": examination_id,
            },
            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), 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 Azure Blob 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 to track the request results. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- scan_width: width of the scan window.

- mm_crop_zone: how much to crop from the center of the image.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- pixel_spacing_column: pixel spacing column parameter of the exam.

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),
    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 Azure Blob 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 to track the request results.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys

            - scan_width: width of the scan window.

            - mm_crop_zone: how much to crop from the center of the image.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - pixel_spacing_column: pixel spacing column parameter of the exam.

    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, logger)
    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
    )

    images = _read_from_azure_uae_bucket(input_bucket_to_use, input_location)

    try:
        # Process the image contents here (e.g., save it, analyze it, etc.)
        response = _process_images(
            images,
            inference_parameters,
            output_bucket_to_use,
            webhook_url,
            config,
            write_to_gcs,
            examination_id,
            logger
        )
        return JSONResponse(
            content={
                "filename": response["filenames"],
                "request_uuid": response["request_uuids"],
                "result_path": response["result_paths"],
                "status": "sent",
                "examination_id": examination_id,
            },
            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:

- slice_idx: index of the slice to process.

- nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

- nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

- mm_crop_zone: how much to crop from the center of the image.

- mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- slice_thickness: slice thickness parameter of the exam.

- pixel_spacing_column: pixel spacing column parameter of the exam.

- zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

- num_slices: number of slices in the exam

- type_of_scan: type of scan, should be macula, widescan, optic_disk

- scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

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 to track the request results.

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:

            - slice_idx: index of the slice to process.

            - nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

            - nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

            - mm_crop_zone: how much to crop from the center of the image.

            - mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - slice_thickness: slice thickness parameter of the exam.

            - pixel_spacing_column: pixel spacing column parameter of the exam.

            - zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

            - num_slices: number of slices in the exam

            - type_of_scan: type of scan, should be macula, widescan, optic_disk

            - scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

        *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 to track the request results.

    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, logger)
    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:
        inference_params = inference_parameters.replace("'", '"')

        inference_params_dict = json.loads(inference_params)
        zone_to_run = define_zone_to_run(inference_params_dict, logger)
        request_uuid = str(uuid.uuid4())
        if zone_to_run is not None:
            contents = image.file.read()

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

            img = cv2.imdecode(image_bytes, cv2.IMREAD_GRAYSCALE)
            img = img[np.newaxis, ..., np.newaxis].astype(np.float32)

            inputs = [
                grpcclient.InferInput("IMAGE", img.shape, np_to_triton_dtype(img.dtype)),
                grpcclient.InferInput("INPUT_JSON", (1, 1), "BYTES"),
            ]
            inputs[0].set_data_from_numpy(img)

            inference_params_dict["zone_of_interest"] = zone_to_run
            inference_params = json.dumps(inference_params_dict)

            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"]))
            ]

            response = 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,
                    examination_id=examination_id,
                    logger=logger
                ),
            )

            return JSONResponse(
                content={
                    "filename": image.filename,
                    "status": "sent",
                    "request_uuid": request_uuid,
                },
                status_code=200,
            )
        else:
            return JSONResponse(
                content={
                    "filename": image.filename,
                    "message": "No zone to run",
                    "status": "no_zone_to_run",
                    "request_uuid": request_uuid,
                },
                status_code=200,
            )
    except Exception as e:
        return JSONResponse(
            content={
                "filename": image.filename,
                "message": str(e),
                "status": "error",
                "request_uuid": request_uuid,
            },
            status_code=400,
        )