mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-06-06 13:30:41 +02:00
Add load_ options for Image component
This commit is contained in:
parent
3a560e3364
commit
245db9aa8f
@ -4,10 +4,18 @@ from pathlib import Path
|
|||||||
|
|
||||||
import PIL.Image
|
import PIL.Image
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
|
from urllib.request import urlopen
|
||||||
|
|
||||||
from pydase.data_service.data_service import DataService
|
from pydase.data_service.data_service import DataService
|
||||||
|
|
||||||
|
|
||||||
|
class Figure:
|
||||||
|
"""Mock class for matplotlib.Figure"""
|
||||||
|
|
||||||
|
def savefig(self, format="png"):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class Image(DataService):
|
class Image(DataService):
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@ -26,28 +34,42 @@ class Image(DataService):
|
|||||||
|
|
||||||
def load_from_path(self, path: Path | str) -> None:
|
def load_from_path(self, path: Path | str) -> None:
|
||||||
with PIL.Image.open(path) as image:
|
with PIL.Image.open(path) as image:
|
||||||
self._load_from_PIL_Image(image)
|
self._load_from_PIL(image)
|
||||||
|
|
||||||
def load_from_base64(self, value: bytes) -> None:
|
def load_from_matplotlib_figure(self, fig: Figure, format_: str = "png") -> None:
|
||||||
if isinstance(value, bytes):
|
buffer = io.BytesIO()
|
||||||
# Decode the base64 string
|
fig.savefig(buffer, format=format_)
|
||||||
image_data = base64.b64decode(value)
|
value_ = base64.b64encode(buffer.getvalue())
|
||||||
|
self._load_from_base64(value_, format_)
|
||||||
|
|
||||||
# Create a writable memory buffer for the image
|
def load_from_url(self, url: str):
|
||||||
image_buffer = io.BytesIO(image_data)
|
image = PIL.Image.open(urlopen(url))
|
||||||
|
self._load_from_PIL(image)
|
||||||
|
|
||||||
# Read the image from the buffer
|
def load_from_base64(self, value_: bytes, format_: str | None = None) -> None:
|
||||||
image = PIL.Image.open(image_buffer)
|
if format_ is None:
|
||||||
self._load_from_PIL_Image(image)
|
format_ = self._get_image_format_from_bytes(value_)
|
||||||
|
self._load_from_base64(value_, format_)
|
||||||
|
|
||||||
def _load_from_PIL_Image(self, image: PIL.Image.Image) -> None:
|
def _load_from_base64(self, value_: bytes, format_: str) -> None:
|
||||||
if isinstance(image, PIL.Image.Image):
|
value = value_.decode("utf-8") if isinstance(value_, bytes) else value_
|
||||||
if image.format is not None:
|
self._value = value
|
||||||
self._format = image.format
|
self._format = format_
|
||||||
buffered = io.BytesIO()
|
|
||||||
image.save(buffered, format=self._format)
|
|
||||||
img_base64 = base64.b64encode(buffered.getvalue())
|
|
||||||
|
|
||||||
self._value = img_base64.decode()
|
def _load_from_PIL(self, image: PIL.Image.Image) -> None:
|
||||||
else:
|
if image.format is not None:
|
||||||
logger.error("Image format is 'None'. Skipping...")
|
format_ = image.format
|
||||||
|
buffer = io.BytesIO()
|
||||||
|
image.save(buffer, format=format_)
|
||||||
|
value_ = base64.b64encode(buffer.getvalue())
|
||||||
|
self._load_from_base64(value_, format_)
|
||||||
|
else:
|
||||||
|
logger.error("Image format is 'None'. Skipping...")
|
||||||
|
|
||||||
|
def _get_image_format_from_bytes(self, value_: bytes):
|
||||||
|
image_data = base64.b64decode(value_)
|
||||||
|
# Create a writable memory buffer for the image
|
||||||
|
image_buffer = io.BytesIO(image_data)
|
||||||
|
# Read the image from the buffer
|
||||||
|
image = PIL.Image.open(image_buffer)
|
||||||
|
return image.format
|
||||||
|
Loading…
x
Reference in New Issue
Block a user