Files
2026-07-13 13:39:38 +08:00

180 lines
6.4 KiB
Python

# Copyright 2024 LiveKit, Inc.
#
# Licensed 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.
import io
from dataclasses import dataclass
from importlib import import_module
from typing import TYPE_CHECKING, Literal, Optional
from livekit import rtc
if TYPE_CHECKING:
from PIL import Image
@dataclass
class EncodeOptions:
"""Options for encoding rtc.VideoFrame to portable image formats."""
format: Literal["JPEG", "PNG"] = "JPEG"
"""The format to encode the image."""
resize_options: Optional["ResizeOptions"] = None
"""Options for resizing the image."""
quality: int | None = 75
"""Image compression quality, 0-100. Only applies to JPEG."""
@dataclass
class ResizeOptions:
"""Options for resizing rtc.VideoFrame as part of encoding to a portable image format."""
width: int
"""The desired resize width (in)"""
height: int
"""The desired height to resize the image to."""
strategy: Literal[
"center_aspect_fit",
"center_aspect_cover",
"scale_aspect_fit",
"scale_aspect_cover",
"skew",
]
"""The strategy to use when resizing the image:
- center_aspect_fit: Fit the image into the provided dimensions, with letterboxing
- center_aspect_cover: Fill the provided dimensions, with cropping
- scale_aspect_fit: Fit the image into the provided dimensions, preserving its original aspect ratio
- scale_aspect_cover: Fill the provided dimensions, preserving its original aspect ratio (image will be larger than the provided dimensions)
- skew: Precisely resize the image to the provided dimensions
""" # noqa: E501
def import_pil() -> None:
try:
if "Image" not in globals():
globals()["Image"] = import_module("PIL.Image")
except ImportError:
raise ImportError(
"You haven't included the 'images' optional dependencies. Please install the 'codecs' extra by running `pip install livekit-agents[images]`" # noqa: E501
) from None
def encode(frame: rtc.VideoFrame, options: EncodeOptions) -> bytes:
"""Encode a rtc.VideoFrame to a portable image format (JPEG or PNG).
See EncodeOptions for more details.
"""
import_pil()
img = _image_from_frame(frame)
resized = _resize_image(img, options)
buffer = io.BytesIO()
kwargs = {}
if options.format == "JPEG" and options.quality is not None:
kwargs["quality"] = options.quality
resized.save(buffer, options.format, **kwargs)
buffer.seek(0)
return buffer.read()
def _image_from_frame(frame: rtc.VideoFrame) -> "Image.Image":
converted = frame
if frame.type != rtc.VideoBufferType.RGBA:
converted = frame.convert(rtc.VideoBufferType.RGBA)
rgb_image = Image.frombytes("RGBA", (frame.width, frame.height), bytes(converted.data)).convert(
"RGB"
)
return rgb_image
def _resize_image(image: "Image.Image", options: EncodeOptions) -> "Image.Image":
if options.resize_options is None:
return image
resize_opts = options.resize_options
if resize_opts.strategy == "skew":
return image.resize((resize_opts.width, resize_opts.height))
elif resize_opts.strategy == "center_aspect_fit":
result = Image.new("RGB", (resize_opts.width, resize_opts.height)) # noqa
# Start with assuming the new image is narrower than the original
new_width = resize_opts.width
new_height = int(image.height * (resize_opts.width / image.width))
# If the new image is wider than the original
if resize_opts.width / resize_opts.height > image.width / image.height:
new_height = resize_opts.height
new_width = int(image.width * (resize_opts.height / image.height))
resized = image.resize((new_width, new_height))
Image.Image.paste(
result,
resized,
(
(resize_opts.width - new_width) // 2,
(resize_opts.height - new_height) // 2,
),
)
return result
elif resize_opts.strategy == "center_aspect_cover":
result = Image.new("RGB", (resize_opts.width, resize_opts.height)) # noqa
# Start with assuming the new image is shorter than the original
new_height = int(image.height * (resize_opts.width / image.width))
new_width = resize_opts.width
# If the new image is taller than the original
if resize_opts.height / resize_opts.width > image.height / image.width:
new_width = int(image.width * (resize_opts.height / image.height))
new_height = resize_opts.height
resized = image.resize((new_width, new_height))
Image.Image.paste( # noqa
result,
resized,
(
(resize_opts.width - new_width) // 2,
(resize_opts.height - new_height) // 2,
),
)
return result
elif resize_opts.strategy == "scale_aspect_cover":
# Start with assuming width is the limiting dimension
new_width = resize_opts.width
new_height = int(image.height * (resize_opts.width / image.width))
# If height is under the limit, scale based on height instead
if new_height < resize_opts.height:
new_height = resize_opts.height
new_width = int(image.width * (resize_opts.height / image.height))
return image.resize((new_width, new_height))
elif resize_opts.strategy == "scale_aspect_fit":
# Start with assuming width is the limiting dimension
new_width = resize_opts.width
new_height = int(image.height * (resize_opts.width / image.width))
# If height would exceed the limit, scale based on height instead
if new_height > resize_opts.height:
new_height = resize_opts.height
new_width = int(image.width * (resize_opts.height / image.height))
return image.resize((new_width, new_height))
raise ValueError(f"Unknown resize strategy: {resize_opts.strategy}")