1.0.22

from io import BytesIO

import bentoml
from <http://bentoml.io|bentoml.io> import JSON, Image
from PIL.Image import Image as PILImage


class MyRunnable(bentoml.Runnable):
    SUPPORTED_RESOURCES = ("cpu",)
    SUPPORTS_CPU_MULTI_THREADING = False

    @bentoml.Runnable.method()
    def predict(self, image: PILImage):
        buffer = BytesIO()
        image.save(buffer, format=image.format)
        return {
            "size": image.size,
            "format": image.format,
            "length": len(buffer.getvalue()),
        }


runner = bentoml.Runner(MyRunnable, name="runner")


svc = bentoml.Service("test_runner", runners=[runner])


@svc.api(input=Image(), output=JSON())
async def classify(input: PILImage) -> dict:
    buffer = BytesIO()
    input.save(buffer, format=input.format)
    runner_result = await runner.predict.async_run(input)
    return {
        "input_size": input.size,
        "input_format": input.format,
        "input_length": len(buffer.getvalue()),
        "runner_result": runner_result,
    }