| | from typing import Any, Dict, List |
| | import base64 |
| | import io |
| | import tempfile |
| | from PIL import Image |
| | import logging |
| |
|
| | def process_image(img): |
| | crash_img = img.crop((0,0, 200, 200)) |
| |
|
| | if crash_img: |
| | img_io = io.BytesIO() |
| | crash_img.save(img_io, "PNG") |
| | img_io.seek(0) |
| | |
| | return {"data": base64.b64encode(img_io.read()).decode("utf-8"), "mime_type": "image/png"} |
| | else: |
| | return {"error": "No crash diagram detected"} |
| |
|
| |
|
| | class EndpointHandler: |
| | def __init__(self, path: str = ""): |
| | """Initialize the endpoint handler. |
| | |
| | Args: |
| | path: Path to the model artifacts |
| | """ |
| | logging.warning("initialized") |
| | pass |
| | |
| | def __call__(self, data: Any) -> List[List[Dict[str, str]]]: |
| | logging.warning("inside __call__") |
| | logging.warning(f"data keys {data.keys()}") |
| | inputs = data.pop("inputs", data) |
| | logging.warning(f"inputs keys {inputs.keys()}") |
| | imagedata = inputs.pop("imagedata", inputs) |
| | if isinstance(imagedata, str): |
| | logging.warning("decoding pdfdata") |
| | image_bytes = base64.b64decode(imagedata) |
| | img = Image.open(io.BytesIO(image_bytes)) |
| | logging.warning(f"image {str(img)}") |
| | return process_image(img) |
