180 lines
6.4 KiB
Python
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}")
|