################################################################################
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
"""Decoded Image encode/decode conversion helpers for multimodal operators."""
import io
import logging
from pyflink.common.image import Image, ImageMode
from pyflink.model.dependencies import dependency_install_requirement
# Pillow warns or rejects very large images to protect users from decompression
# bombs. Keep an explicit upper bound for decode safety while allowing common
# high-resolution images; this is about 1.6 GB for RGB uint8 payloads.
_MAX_IMAGE_PIXELS = 178_956_970
_logger = logging.getLogger(__name__)
def _get_numpy():
import numpy as np
return np
_PILImage = None
_pil_open = None
_PIL_IMAGE_ERRORS = ()
_IMAGE_DATA_ERRORS = (OSError, ValueError, SyntaxError)
def _load_pillow():
global _PILImage, _pil_open, _PIL_IMAGE_ERRORS, _IMAGE_DATA_ERRORS
if _PILImage is not None:
return _PILImage, _pil_open
try:
from PIL import Image as PILImage
except ImportError as e:
raise ImportError(
"Pillow is required for image operations. "
f"Install it with: pip install {dependency_install_requirement('PIL')}"
) from e
PILImage.MAX_IMAGE_PIXELS = _MAX_IMAGE_PIXELS
_PILImage = PILImage
_pil_open = _original_pil_open(PILImage)
_PIL_IMAGE_ERRORS = (PILImage.DecompressionBombError,)
_IMAGE_DATA_ERRORS = (OSError, ValueError, SyntaxError) + _PIL_IMAGE_ERRORS
return _PILImage, _pil_open
def _original_pil_open(PILImage):
pil_open = PILImage.open
wrapper_globals = getattr(pil_open, "__globals__", {})
original_open = wrapper_globals.get("_image_open")
# Some libraries wrap PIL.Image.open but keep the original as _image_open.
# Prefer it so corrupt inputs do not trigger wrapper-specific side effects.
if (
original_open is not None
and getattr(original_open, "__module__", None) == "PIL.Image"):
return original_open
return pil_open
def get_pillow_image_module():
"""Return Pillow's Image module, importing Pillow lazily."""
return _load_pillow()[0]
def _get_pil_open():
"""Return the cached original PIL.Image.open implementation."""
return _load_pillow()[1]
_IMAGE_MODES_BY_NAME = {mode.canonical_name: mode for mode in ImageMode}
_IMAGE_MODE_NAMES = sorted(_IMAGE_MODES_BY_NAME)
_PIL_NATIVE_MODES = {"L", "LA", "RGB", "RGBA"}
_PIL_16BIT_MODES = {"I;16", "I;16L", "I;16B"}
_UINT8_UINT16_SCALE = 257
_PIL_MODES_BY_IMAGE_MODE = {
ImageMode.L: "L",
ImageMode.LA: "LA",
ImageMode.RGB: "RGB",
ImageMode.RGBA: "RGBA",
ImageMode.L16: "I;16",
}
_PIL_IMAGE_FILTER_MODES = {"L", "LA", "RGB", "RGBA"}
_PIL_IMAGE_FILTER_MODE_BY_CHANNELS = {1: "L", 2: "LA", 3: "RGB", 4: "RGBA"}
_UINT8_MODES_BY_CHANNELS = {
mode.channels: mode.canonical_name
for mode in ImageMode
if mode.byte_width == 1
}
_UINT16_MODES_BY_CHANNELS = {
mode.channels: mode.canonical_name
for mode in ImageMode
if mode.byte_width == 2
}
DEFAULT_IMAGE_ENCODE_QUALITY = 85
SUPPORTED_IMAGE_FORMATS = {"JPEG", "PNG", "TIFF", "WEBP", "BMP", "GIF"}
def _normalize_image_mode(mode):
if not isinstance(mode, str):
raise ValueError(f"mode must be a string, got {mode!r}")
normalized = mode.upper()
if normalized not in _IMAGE_MODES_BY_NAME:
raise ValueError(
f"Unsupported decoded image mode {mode!r}; expected one of "
f"{_IMAGE_MODE_NAMES}"
)
return normalized
def normalize_image_mode(mode):
"""Normalize and validate a decoded Image mode name."""
return _normalize_image_mode(mode)
def image_mode_dtype(mode):
"""Return the numpy dtype used by a decoded Image mode."""
np = _get_numpy()
return np.dtype(_normalize_image_mode_value(mode)._numpy_dtype)
[docs]def image_mode_channels(mode):
"""Return the channel count used by a decoded Image mode."""
return _normalize_image_mode_value(mode).channels
def is_image_data_error(error):
"""Return whether an exception represents corrupt or invalid image data."""
return isinstance(error, _IMAGE_DATA_ERRORS)
def _infer_image_mode(pixel_array):
np = _get_numpy()
dtype = np.dtype(pixel_array.dtype)
if pixel_array.ndim == 2:
channels = 1
elif pixel_array.ndim == 3:
channels = pixel_array.shape[2]
else:
raise ValueError(
"Expected a 2-D or 3-D image array, "
f"got {pixel_array.ndim}-D"
)
if dtype == np.dtype(np.uint8):
mode = _UINT8_MODES_BY_CHANNELS.get(channels)
elif dtype == np.dtype(np.uint16):
mode = _UINT16_MODES_BY_CHANNELS.get(channels)
else:
mode = None
if mode is None:
raise ValueError(
"Unsupported decoded Image dtype/channel combination: "
f"dtype={pixel_array.dtype}, channels={channels}. "
"Supported image modes are L, LA, RGB, RGBA, and L16."
)
return mode, channels
def _ensure_channel_axis(pixel_array):
if pixel_array.ndim == 2:
return pixel_array[:, :, None]
return pixel_array
def _normalize_image_mode_value(mode):
if isinstance(mode, ImageMode):
return mode
if isinstance(mode, str):
normalized = _normalize_image_mode(mode)
return ImageMode.from_canonical_name(normalized)
raise TypeError(
"Image mode must be an ImageMode or canonical mode string, got %s"
% type(mode).__name__
)
def _image_mode_name(mode):
if isinstance(mode, ImageMode):
return mode.canonical_name
return _normalize_image_mode(mode)
def _ensure_image_array(pixel_array):
np = _get_numpy()
pixel_array = _ensure_channel_axis(np.asarray(pixel_array))
if not pixel_array.flags["C_CONTIGUOUS"]:
pixel_array = np.ascontiguousarray(pixel_array)
return pixel_array
[docs]def image_to_ndarray(image, mode=None):
"""Return decoded :class:`Image` pixels as an ndarray for operator internals.
When ``mode`` is ``None``, this returns the stored Image data unchanged,
including its canonical 3-D ``(height, width, channels)`` shape. Callers may
rely on this 3-D HWC shape; single-channel decoded images must not be
collapsed to 2-D in the ``mode=None`` path. When a target mode is provided,
Pillow handles color/channel conversion and the codec applies linear
8-bit/16-bit scaling. Single-channel targets such as ``L`` and ``L16`` are
returned in Pillow's 2-D array shape.
"""
if image is None:
return None
if not isinstance(image, Image):
raise TypeError(f"Expected Image, got {type(image).__name__}")
if mode is None:
return image.data
target_mode = _normalize_image_mode_value(mode)
if image.mode is target_mode:
if target_mode.channels == 1:
return image.data[:, :, 0]
return image.data
return convert_image_array_to_mode(image.data, target_mode)
[docs]def ndarray_to_image(pixel_array, mode=None):
"""Convert an ndarray to a decoded :class:`Image`."""
if pixel_array is None:
return None
if mode is not None:
target_mode = _image_mode_name(mode)
pixel_array = convert_image_array_to_mode(pixel_array, target_mode)
image_mode = _normalize_image_mode_value(target_mode)
else:
inferred_mode, _ = _infer_image_mode(pixel_array)
image_mode = _normalize_image_mode_value(inferred_mode)
normalized = _ensure_image_array(pixel_array)
return Image(data=normalized, mode=image_mode)
[docs]def pil_to_image(pil_image, mode=None):
"""Convert a PIL image to a decoded :class:`Image`.
With ``mode=None``, directly representable Pillow modes are preserved.
Other Pillow modes are converted to RGB/RGBA using Pillow's own conversion
rules, which may perform color-space conversion. Explicit 8-bit/16-bit
target conversions use the codec's linear bit-depth scaling.
"""
image_mode = _normalize_image_mode_value(mode) if mode is not None else None
pixel_array = _canonicalize_decoded_pil_image(pil_image, image_mode)
if image_mode is None:
image_mode = _normalize_image_mode_value(_infer_image_mode(pixel_array)[0])
return Image(data=_ensure_image_array(pixel_array), mode=image_mode)
[docs]def convert_image_array_to_mode(pixel_array, mode):
"""Convert an image ndarray to one of the built-in image modes.
Pillow performs the color/channel conversion. Bit-depth conversion between
16-bit and 8-bit modes uses the same linear scaling as Rust's image crate:
``uint16 // 257`` when reducing to 8-bit and ``uint8 * 257`` when expanding
to ``L16``.
"""
np = _get_numpy()
target_mode = _normalize_image_mode_value(mode)
pixel_array = np.asarray(pixel_array)
source_mode_name, _ = _infer_image_mode(pixel_array)
source_mode = _normalize_image_mode_value(source_mode_name)
if source_mode is target_mode:
if target_mode.channels == 1 and pixel_array.ndim == 3:
pixel_array = pixel_array[:, :, 0]
return pixel_array.astype(np.dtype(target_mode._numpy_dtype), copy=False)
if source_mode is ImageMode.L16 and target_mode.byte_width == 1:
l_image = _l16_array_to_l_array(pixel_array)
if target_mode is ImageMode.L:
return l_image
pil_image = _image_array_to_pil_for_mode_conversion(l_image)
converted = pil_image.convert(_PIL_MODES_BY_IMAGE_MODE[target_mode])
return np.asarray(converted).astype(
np.dtype(target_mode._numpy_dtype), copy=False
)
if source_mode.byte_width == 1 and target_mode is ImageMode.L16:
pil_image = _image_array_to_pil_for_mode_conversion(pixel_array)
l_image = np.asarray(pil_image.convert(_PIL_MODES_BY_IMAGE_MODE[ImageMode.L]))
return _l_array_to_l16_array(l_image)
pil_image = _image_array_to_pil_for_mode_conversion(pixel_array)
converted = pil_image.convert(_PIL_MODES_BY_IMAGE_MODE[target_mode])
return np.asarray(converted).astype(np.dtype(target_mode._numpy_dtype), copy=False)
def _l16_array_to_l_array(pixel_array):
np = _get_numpy()
pixel_array = np.asarray(pixel_array)
if pixel_array.ndim == 3:
pixel_array = pixel_array[:, :, 0]
return (pixel_array // _UINT8_UINT16_SCALE).astype(np.uint8, copy=False)
def _l_array_to_l16_array(pixel_array):
np = _get_numpy()
return (np.asarray(pixel_array).astype(np.uint16) * _UINT8_UINT16_SCALE).astype(
np.uint16, copy=False
)
def _image_array_to_pil_for_mode_conversion(pixel_array):
np = _get_numpy()
pixel_array = np.asarray(pixel_array)
if pixel_array.dtype.byteorder in (">", "<"):
pixel_array = pixel_array.astype(pixel_array.dtype.newbyteorder("="), copy=False)
inferred_mode, _ = _infer_image_mode(pixel_array)
if pixel_array.ndim == 3 and pixel_array.shape[2] == 1:
pixel_array = pixel_array[:, :, 0]
pil_mode = _PIL_MODES_BY_IMAGE_MODE[_normalize_image_mode_value(inferred_mode)]
return _image_array_to_pil_with_raw_decoder(pixel_array, pil_mode)
def _image_array_to_pil_with_raw_decoder(pixel_array, pil_mode):
np = _get_numpy()
PILImage = get_pillow_image_module()
pixel_array = np.ascontiguousarray(pixel_array)
height, width = pixel_array.shape[:2]
return PILImage.frombytes(
pil_mode,
(width, height),
pixel_array.tobytes(),
"raw",
pil_mode,
0,
1,
)
def _pillow_default_output_mode(pil_image):
# The built-in Image type must choose a representable target mode. After
# that choice, Pillow owns the actual color conversion, precision reduction,
# and any conversion failure.
if (
"A" in pil_image.getbands()
or pil_image.info.get("transparency") is not None):
return "RGBA"
return "RGB"
def _canonicalize_decoded_pil_image(pil_image, mode):
np = _get_numpy()
if mode is not None:
if pil_image.mode in _PIL_NATIVE_MODES or pil_image.mode in _PIL_16BIT_MODES:
return convert_image_array_to_mode(_pil_image_to_array(pil_image), mode)
base = pil_image.convert(_pillow_default_output_mode(pil_image))
return convert_image_array_to_mode(
_pil_image_to_array(base),
mode,
)
if pil_image.mode in _PIL_NATIVE_MODES:
return _pil_image_to_array(pil_image)
if pil_image.mode in _PIL_16BIT_MODES:
return _pil_image_to_array(pil_image).astype(np.uint16, copy=False)
# For Pillow modes that the built-in Image type cannot represent directly,
# delegate to Pillow's standard RGB/RGBA conversion. This may perform
# color-space conversion or precision reduction according to Pillow rules.
return _pil_image_to_array(pil_image.convert(_pillow_default_output_mode(pil_image)))
def _pil_image_to_array(pil_image):
np = _get_numpy()
pixel_array = np.asarray(pil_image)
if pixel_array.dtype.byteorder in (">", "<"):
pixel_array = pixel_array.astype(pixel_array.dtype.newbyteorder("="), copy=False)
return pixel_array
[docs]def image_array_to_pil(pixel_array, mode=None):
"""Convert ndarray pixels to a PIL Image, converting mode only when requested."""
np = _get_numpy()
if mode is not None:
pixel_array = convert_image_array_to_mode(pixel_array, mode)
pixel_array = np.asarray(pixel_array)
if pixel_array.dtype.byteorder in (">", "<"):
pixel_array = pixel_array.astype(pixel_array.dtype.newbyteorder("="), copy=False)
inferred_mode, _ = _infer_image_mode(pixel_array)
if pixel_array.ndim == 3 and pixel_array.shape[2] == 1:
pixel_array = pixel_array[:, :, 0]
pil_mode = _PIL_MODES_BY_IMAGE_MODE[_normalize_image_mode_value(inferred_mode)]
return _image_array_to_pil_with_raw_decoder(pixel_array, pil_mode)
def _pil_fallback_mode_for_array(pixel_array):
channels = 1 if pixel_array.ndim == 2 else pixel_array.shape[2]
return {1: "L", 2: "LA", 3: "RGB", 4: "RGBA"}[channels]
[docs]def image_array_to_pil_compatible(pixel_array):
"""Convert ndarray pixels to a PIL Image, normalizing only when required.
Pillow cannot operate directly on high-precision multichannel arrays.
This helper preserves natively supported modes and falls back to the
uint8 mode with the same channel count for PIL-only transforms.
"""
try:
return image_array_to_pil(pixel_array)
except ValueError:
return image_array_to_pil(
pixel_array, mode=_pil_fallback_mode_for_array(pixel_array)
)
def image_array_to_filterable_pil(pixel_array):
"""Convert ndarray pixels to a PIL Image supported by Pillow ImageFilter.
Pillow can represent modes such as ``I;16`` for L16 input, but common
ImageFilter operations do not accept every representable mode. Normalize
unsupported filter modes to the 8-bit mode with the same channel count.
L16 input is reduced with the codec's linear ``uint16 // 257`` scaling.
"""
pil_image = image_array_to_pil_compatible(pixel_array)
if pil_image.mode in _PIL_IMAGE_FILTER_MODES:
return pil_image
channels = 1 if pixel_array.ndim == 2 else pixel_array.shape[2]
target_mode = _PIL_IMAGE_FILTER_MODE_BY_CHANNELS[channels]
return image_array_to_pil_compatible(
convert_image_array_to_mode(pixel_array, target_mode)
)
def normalize_image_format(image_format, param_name="image_format", allow_none=False):
"""Normalize and validate an encoded image format name."""
if image_format is None and allow_none:
return None
if not isinstance(image_format, str):
raise ValueError(f"{param_name} must be a string")
normalized = image_format.upper()
if normalized not in SUPPORTED_IMAGE_FORMATS:
raise ValueError(
f"{param_name} must be one of {sorted(SUPPORTED_IMAGE_FORMATS)}, "
f"got {image_format!r}"
)
return normalized
def validate_image_quality(quality):
"""Validate JPEG/WebP encode quality."""
if not isinstance(quality, int) or quality < 1 or quality > 100:
raise ValueError(f"quality must be an integer in [1, 100], got {quality!r}")
return quality
# Image decode helpers
# Data-level exceptions from PIL / cv2 that indicate corrupt input. Kept narrow
# to avoid masking real bugs. The tuple is extended with Pillow-specific errors
# when Pillow is first loaded.
def _read_image_dimensions(image_bytes):
"""Read image dimensions from file header without decoding.
Parses JPEG (SOF0/SOF2), PNG (IHDR), WebP (VP8/VP8L), BMP, and GIF
headers to extract width and height. Returns ``(width, height)`` on
success, or ``(None, None)`` if the format is unrecognized or the
header is too short to parse.
This is a zero-allocation, zero-decode operation - safe to call on
every image before PIL decoding as a memory-bomb guard.
"""
if image_bytes is None or len(image_bytes) < 2:
return None, None
# JPEG: scan for SOF0 (0xFFC0) or SOF2 (0xFFC2) marker
if image_bytes[:2] == b'\xff\xd8':
# Scan markers; each marker is 0xFF + non-zero type byte.
# SOF markers contain height (2B) and width (2B) at offset 5-8
# within the marker payload.
i = 2
while i + 9 <= len(image_bytes):
if image_bytes[i] != 0xFF:
break
marker = image_bytes[i + 1]
# Skip padding 0xFF bytes
if marker == 0xFF:
i += 1
continue
# Standalone markers (no payload): RST0-RST7, SOI, EOI, TEM
if marker == 0xD8 or marker == 0xD9 or marker == 0x01:
i += 2
continue
if 0xD0 <= marker <= 0xD7:
i += 2
continue
# SOS marker - image data follows, stop scanning
if marker == 0xDA:
break
# Read marker payload length (big-endian uint16, includes itself)
seg_len = (image_bytes[i + 2] << 8) | image_bytes[i + 3]
# SOF0, SOF1, SOF2 (progressive): height at offset +5, width at +7
if marker in (0xC0, 0xC1, 0xC2):
h = (image_bytes[i + 5] << 8) | image_bytes[i + 6]
w = (image_bytes[i + 7] << 8) | image_bytes[i + 8]
return w, h
i += 2 + seg_len
return None, None
# PNG: IHDR chunk at offset 16
if len(image_bytes) >= 24 and image_bytes[:4] == b'\x89PNG':
w = int.from_bytes(image_bytes[16:20], 'big')
h = int.from_bytes(image_bytes[20:24], 'big')
return w, h
# WebP
if len(image_bytes) >= 30 and image_bytes[:4] == b'RIFF' and image_bytes[8:12] == b'WEBP':
# Lossy (VP8): bits 0-3 of header are a signature, then
# width (14 bits) at byte 26, height (14 bits) at byte 28
if image_bytes[12:16] == b'VP8 ':
w = int.from_bytes(image_bytes[26:28], 'little') & 0x3FFF
h = int.from_bytes(image_bytes[28:30], 'little') & 0x3FFF
return w, h
# Lossless (VP8L): byte 20 is signature 0x2F, then
# width-1 (14 bits) and height-1 (14 bits) packed in 4 bytes
if image_bytes[12:16] == b'VP8L' and len(image_bytes) >= 25:
if image_bytes[20] == 0x2F:
bits = int.from_bytes(image_bytes[21:25], 'little')
w = (bits & 0x3FFF) + 1
h = ((bits >> 14) & 0x3FFF) + 1
return w, h
# Extended WebP (VP8X): canvas width/height minus 1 are stored
# as 24-bit little-endian integers at bytes 24..29.
if image_bytes[12:16] == b'VP8X':
w = int.from_bytes(image_bytes[24:27], 'little') + 1
h = int.from_bytes(image_bytes[27:30], 'little') + 1
return w, h
return None, None
# BMP: width at offset 18, height at offset 22 (LE int32)
if len(image_bytes) >= 26 and image_bytes[:2] == b'BM':
w = int.from_bytes(image_bytes[18:22], 'little')
h = abs(int.from_bytes(image_bytes[22:26], 'little'))
return w, h
# GIF: width at offset 6, height at offset 8 (LE uint16)
if len(image_bytes) >= 10 and image_bytes[:4] == b'GIF8':
w = int.from_bytes(image_bytes[6:8], 'little')
h = int.from_bytes(image_bytes[8:10], 'little')
return w, h
return None, None
def detect_image_format(image_bytes):
"""Detect image format from magic bytes. Returns a format string for PIL save().
Returns None if image_bytes is None or format is unrecognized.
"""
if image_bytes is None:
return None
if len(image_bytes) >= 2 and image_bytes[:2] == b'\xff\xd8':
return "JPEG"
if len(image_bytes) >= 4 and image_bytes[:4] == b'\x89PNG':
return "PNG"
if len(image_bytes) >= 12 and image_bytes[:4] == b'RIFF' and image_bytes[8:12] == b'WEBP':
return "WEBP"
if len(image_bytes) >= 2 and image_bytes[:2] == b'BM':
return "BMP"
if len(image_bytes) >= 4 and image_bytes[:4] == b'GIF8':
return "GIF"
if len(image_bytes) >= 4 and image_bytes[:4] in (b'II*\x00', b'MM\x00*'):
return "TIFF"
return None
def read_encoded_image_metadata(image_bytes):
"""Read encoded image metadata without materializing decoded pixels.
Returns ``None`` when the encoded payload cannot be parsed or verified.
"""
if image_bytes is None:
return None
if not isinstance(image_bytes, (bytes, bytearray)):
raise TypeError(f"image_bytes must be bytes, got {type(image_bytes).__name__}")
pil_open = _get_pil_open()
try:
with pil_open(io.BytesIO(image_bytes)) as pil_image:
metadata = (
pil_image.width,
pil_image.height,
len(pil_image.getbands()),
pil_image.mode,
pil_image.format or detect_image_format(image_bytes) or "UNKNOWN",
)
# verify() checks the encoded container without loading the pixel
# array, which keeps metadata/filter paths lightweight.
pil_image.verify()
return metadata
except _IMAGE_DATA_ERRORS as e:
_logger.debug("Failed to read image metadata: %s", e)
return None
def _convert_for_jpeg(pil_image):
if pil_image.mode in _PIL_16BIT_MODES:
return image_array_to_pil(
convert_image_array_to_mode(_pil_image_to_array(pil_image), ImageMode.RGB)
)
if pil_image.mode not in ("L", "RGB"):
# JPEG only supports 8-bit L/RGB in Pillow; convert explicitly at the
# output boundary instead of letting save() fail with a vague error.
return pil_image.convert("RGB")
return pil_image
def _image_array_to_jpeg_pil(pixel_array):
"""Convert ndarray pixels to a PIL Image compatible with JPEG output."""
try:
pil_image = image_array_to_pil(pixel_array)
except ValueError:
# JPEG cannot preserve high-precision multichannel pixels; normalize
# explicitly at the output boundary.
pil_image = image_array_to_pil(pixel_array, mode="RGB")
return _convert_for_jpeg(pil_image)
def _validate_max_pixels(max_pixels):
if max_pixels is not None and max_pixels <= 0:
raise ValueError(f"max_pixels must be positive, got {max_pixels!r}")
def _check_encoded_image_pixel_limit(image_bytes, max_pixels):
if max_pixels is None:
return
w, h = _read_image_dimensions(image_bytes)
if w is not None and h is not None and w * h > max_pixels:
raise ValueError(
f"Image dimensions {w}x{h} exceed max_pixels={max_pixels}"
)
[docs]def decode_image(image_bytes, mode=None, max_pixels=None):
"""Decode encoded image bytes into a decoded :class:`Image`.
When ``mode`` is ``None``, Pillow modes that cannot be represented directly
by the built-in Image type are converted to RGB/RGBA using Pillow's
conversion rules. Explicit 8-bit/16-bit target conversions use linear
scaling: ``uint16 // 257`` when reducing to 8-bit and ``uint8 * 257`` when
expanding to ``L16``.
"""
if image_bytes is None:
return None
target_mode = _normalize_image_mode(mode) if mode is not None else None
_validate_max_pixels(max_pixels)
if not isinstance(image_bytes, (bytes, bytearray)):
raise TypeError(f"image_bytes must be bytes, got {type(image_bytes).__name__}")
pil_open = _get_pil_open()
_check_encoded_image_pixel_limit(image_bytes, max_pixels)
with pil_open(io.BytesIO(image_bytes)) as pil_image:
return pil_to_image(pil_image, mode=target_mode)
def _verify_encoded_image_bytes(image_bytes, mode=None, max_pixels=None):
"""Verify encoded image headers/containers without materializing pixels."""
if image_bytes is None:
return False
if mode is not None:
_normalize_image_mode(mode)
_validate_max_pixels(max_pixels)
if not isinstance(image_bytes, (bytes, bytearray)):
raise TypeError(f"image_bytes must be bytes, got {type(image_bytes).__name__}")
pil_open = _get_pil_open()
_check_encoded_image_pixel_limit(image_bytes, max_pixels)
with pil_open(io.BytesIO(image_bytes)) as pil_image:
pil_image.verify()
return True
def is_decodable_image_bytes(image_bytes, mode=None, max_pixels=None):
"""Return whether encoded bytes pass lightweight header/container checks.
This uses Pillow ``verify()`` and does not guarantee that the full pixel
stream can be materialized by ``decode_image``.
"""
try:
return _verify_encoded_image_bytes(image_bytes, mode=mode, max_pixels=max_pixels)
except _IMAGE_DATA_ERRORS as e:
_logger.debug("Failed to verify image bytes: %s", e)
return False
def _encode_pixel_array(pixel_array, target_format, quality):
if target_format.upper() == "JPEG":
# JPEG cannot preserve alpha or high-precision multichannel pixels, so
# normalize only at the explicit JPEG output boundary.
pil_image = _image_array_to_jpeg_pil(pixel_array)
else:
try:
pil_image = image_array_to_pil(pixel_array)
except ValueError as e:
raise ValueError(
f"Cannot encode decoded Image input as {target_format}: {e}"
) from e
buf = io.BytesIO()
pil_image.save(buf, format=target_format, quality=quality)
return buf.getvalue()
def _encode_pil_image(pil_image, target_format, quality):
if target_format.upper() == "JPEG":
pil_image = _convert_for_jpeg(pil_image)
buf = io.BytesIO()
pil_image.save(buf, format=target_format, quality=quality)
return buf.getvalue()
def transcode_image_bytes(
image_bytes,
output_format=None,
quality=DEFAULT_IMAGE_ENCODE_QUALITY,
):
"""Transcode encoded image bytes to encoded image bytes."""
if image_bytes is None:
return None
if not isinstance(image_bytes, (bytes, bytearray)):
raise TypeError(
"image_bytes must be encoded image bytes, "
f"got {type(image_bytes).__name__}"
)
pil_open = _get_pil_open()
output_format = normalize_image_format(
output_format, param_name="output_format", allow_none=True
)
quality = validate_image_quality(quality)
with pil_open(io.BytesIO(image_bytes)) as pil_image:
target_format = output_format or pil_image.format or "JPEG"
return _encode_pil_image(pil_image, target_format, quality)
def image_batch_to_ndarrays(series, mode=None, max_pixels=None):
"""Convert a batch of decoded Images into ndarrays (fail-fast)."""
images = []
valid_idx = []
for i, item in enumerate(series):
if item is None:
continue
if not isinstance(item, Image):
raise TypeError(f"Expected Image, got {type(item).__name__}")
if max_pixels is not None and item.height * item.width > max_pixels:
raise ValueError(
f"Image dimensions {item.width}x{item.height} exceed "
f"max_pixels={max_pixels}"
)
pixel_array = image_to_ndarray(item, mode=mode)
if pixel_array is None:
continue
images.append(pixel_array)
valid_idx.append(i)
return images, valid_idx