wehub-resource-sync
567 lines
21 KiB
Python
Executable File
567 lines
21 KiB
Python
Executable File
import os
|
|
import gc
|
|
import imageio
|
|
import inspect
|
|
import numpy as np
|
|
import torch
|
|
import torchvision
|
|
import cv2
|
|
from einops import rearrange
|
|
from PIL import Image
|
|
import time
|
|
import torch.nn.functional as F
|
|
import torchvision.transforms.functional as TF
|
|
|
|
|
|
def filter_kwargs(cls, kwargs):
|
|
sig = inspect.signature(cls.__init__)
|
|
valid_params = set(sig.parameters.keys()) - {"self", "cls"}
|
|
filtered_kwargs = {k: v for k, v in kwargs.items() if k in valid_params}
|
|
return filtered_kwargs
|
|
|
|
|
|
def get_width_and_height_from_image_and_base_resolution(image, base_resolution):
|
|
target_pixels = int(base_resolution) * int(base_resolution)
|
|
original_width, original_height = Image.open(image).size
|
|
ratio = (target_pixels / (original_width * original_height)) ** 0.5
|
|
width_slider = round(original_width * ratio)
|
|
height_slider = round(original_height * ratio)
|
|
return height_slider, width_slider
|
|
|
|
|
|
def color_transfer(sc, dc):
|
|
"""
|
|
Transfer color distribution from of sc, referred to dc.
|
|
|
|
Args:
|
|
sc (numpy.ndarray): input image to be transfered.
|
|
dc (numpy.ndarray): reference image
|
|
|
|
Returns:
|
|
numpy.ndarray: Transferred color distribution on the sc.
|
|
"""
|
|
|
|
def get_mean_and_std(img):
|
|
x_mean, x_std = cv2.meanStdDev(img)
|
|
x_mean = np.hstack(np.around(x_mean, 2))
|
|
x_std = np.hstack(np.around(x_std, 2))
|
|
return x_mean, x_std
|
|
|
|
sc = cv2.cvtColor(sc, cv2.COLOR_RGB2LAB)
|
|
s_mean, s_std = get_mean_and_std(sc)
|
|
dc = cv2.cvtColor(dc, cv2.COLOR_RGB2LAB)
|
|
t_mean, t_std = get_mean_and_std(dc)
|
|
img_n = ((sc - s_mean) * (t_std / s_std)) + t_mean
|
|
np.putmask(img_n, img_n > 255, 255)
|
|
np.putmask(img_n, img_n < 0, 0)
|
|
dst = cv2.cvtColor(cv2.convertScaleAbs(img_n), cv2.COLOR_LAB2RGB)
|
|
return dst
|
|
|
|
|
|
def save_videos_grid(
|
|
videos: torch.Tensor,
|
|
path: str,
|
|
rescale=False,
|
|
n_rows=6,
|
|
fps=12,
|
|
imageio_backend=True,
|
|
color_transfer_post_process=False,
|
|
):
|
|
videos = rearrange(videos, "b c t h w -> t b c h w")
|
|
outputs = []
|
|
for x in videos:
|
|
x = torchvision.utils.make_grid(x, nrow=n_rows)
|
|
x = x.transpose(0, 1).transpose(1, 2).squeeze(-1)
|
|
if rescale:
|
|
x = (x + 1.0) / 2.0 # -1,1 -> 0,1
|
|
x = (x * 255).numpy().astype(np.uint8)
|
|
outputs.append(Image.fromarray(x))
|
|
|
|
if color_transfer_post_process:
|
|
for i in range(1, len(outputs)):
|
|
outputs[i] = Image.fromarray(color_transfer(np.uint8(outputs[i]), np.uint8(outputs[0])))
|
|
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
if imageio_backend:
|
|
if path.endswith("mp4"):
|
|
imageio.mimsave(path, outputs, fps=fps)
|
|
else:
|
|
imageio.mimsave(path, outputs, duration=(1000 * 1 / fps))
|
|
else:
|
|
if path.endswith("mp4"):
|
|
path = path.replace(".mp4", ".gif")
|
|
outputs[0].save(path, format="GIF", append_images=outputs, save_all=True, duration=100, loop=0)
|
|
|
|
|
|
def get_image_to_video_latent(validation_image_start, validation_image_end, video_length, sample_size):
|
|
if validation_image_start is not None and validation_image_end is not None:
|
|
if type(validation_image_start) is str and os.path.isfile(validation_image_start):
|
|
image_start = clip_image = Image.open(validation_image_start).convert("RGB")
|
|
image_start = image_start.resize([sample_size[1], sample_size[0]])
|
|
clip_image = clip_image.resize([sample_size[1], sample_size[0]])
|
|
else:
|
|
image_start = clip_image = validation_image_start
|
|
image_start = [_image_start.resize([sample_size[1], sample_size[0]]) for _image_start in image_start]
|
|
clip_image = [_clip_image.resize([sample_size[1], sample_size[0]]) for _clip_image in clip_image]
|
|
|
|
if type(validation_image_end) is str and os.path.isfile(validation_image_end):
|
|
image_end = Image.open(validation_image_end).convert("RGB")
|
|
image_end = image_end.resize([sample_size[1], sample_size[0]])
|
|
else:
|
|
image_end = validation_image_end
|
|
image_end = [_image_end.resize([sample_size[1], sample_size[0]]) for _image_end in image_end]
|
|
|
|
if type(image_start) is list:
|
|
clip_image = clip_image[0]
|
|
start_video = torch.cat(
|
|
[
|
|
torch.from_numpy(np.array(_image_start)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0)
|
|
for _image_start in image_start
|
|
],
|
|
dim=2,
|
|
)
|
|
input_video = torch.tile(start_video[:, :, :1], [1, 1, video_length, 1, 1])
|
|
input_video[:, :, : len(image_start)] = start_video
|
|
|
|
input_video_mask = torch.zeros_like(input_video[:, :1])
|
|
input_video_mask[:, :, len(image_start) :] = 255
|
|
else:
|
|
input_video = torch.tile(
|
|
torch.from_numpy(np.array(image_start)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0),
|
|
[1, 1, video_length, 1, 1],
|
|
)
|
|
input_video_mask = torch.zeros_like(input_video[:, :1])
|
|
input_video_mask[:, :, 1:] = 255
|
|
|
|
if type(image_end) is list:
|
|
image_end = [
|
|
_image_end.resize(image_start[0].size if type(image_start) is list else image_start.size)
|
|
for _image_end in image_end
|
|
]
|
|
end_video = torch.cat(
|
|
[
|
|
torch.from_numpy(np.array(_image_end)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0)
|
|
for _image_end in image_end
|
|
],
|
|
dim=2,
|
|
)
|
|
input_video[:, :, -len(end_video) :] = end_video
|
|
|
|
input_video_mask[:, :, -len(image_end) :] = 0
|
|
else:
|
|
image_end = image_end.resize(image_start[0].size if type(image_start) is list else image_start.size)
|
|
input_video[:, :, -1:] = torch.from_numpy(np.array(image_end)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0)
|
|
input_video_mask[:, :, -1:] = 0
|
|
|
|
input_video = input_video / 255
|
|
|
|
elif validation_image_start is not None:
|
|
if type(validation_image_start) is str and os.path.isfile(validation_image_start):
|
|
image_start = clip_image = Image.open(validation_image_start).convert("RGB")
|
|
image_start = image_start.resize([sample_size[1], sample_size[0]])
|
|
clip_image = clip_image.resize([sample_size[1], sample_size[0]])
|
|
else:
|
|
image_start = clip_image = validation_image_start
|
|
image_start = [_image_start.resize([sample_size[1], sample_size[0]]) for _image_start in image_start]
|
|
clip_image = [_clip_image.resize([sample_size[1], sample_size[0]]) for _clip_image in clip_image]
|
|
image_end = None
|
|
|
|
if type(image_start) is list:
|
|
clip_image = clip_image[0]
|
|
start_video = torch.cat(
|
|
[
|
|
torch.from_numpy(np.array(_image_start)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0)
|
|
for _image_start in image_start
|
|
],
|
|
dim=2,
|
|
)
|
|
input_video = torch.tile(start_video[:, :, :1], [1, 1, video_length, 1, 1])
|
|
input_video[:, :, : len(image_start)] = start_video
|
|
input_video = input_video / 255
|
|
|
|
input_video_mask = torch.zeros_like(input_video[:, :1])
|
|
input_video_mask[:, :, len(image_start) :] = 255
|
|
else:
|
|
input_video = (
|
|
torch.tile(
|
|
torch.from_numpy(np.array(image_start)).permute(2, 0, 1).unsqueeze(1).unsqueeze(0),
|
|
[1, 1, video_length, 1, 1],
|
|
)
|
|
/ 255
|
|
)
|
|
input_video_mask = torch.zeros_like(input_video[:, :1])
|
|
input_video_mask[
|
|
:,
|
|
:,
|
|
1:,
|
|
] = 255
|
|
else:
|
|
image_start = None
|
|
image_end = None
|
|
input_video = torch.zeros([1, 3, video_length, sample_size[0], sample_size[1]])
|
|
input_video_mask = torch.ones([1, 1, video_length, sample_size[0], sample_size[1]]) * 255
|
|
clip_image = None
|
|
|
|
del image_start
|
|
del image_end
|
|
gc.collect()
|
|
|
|
return input_video, input_video_mask, clip_image
|
|
|
|
|
|
def get_video_to_video_latent(
|
|
input_video_path, video_length, sample_size, fps=None, validation_video_mask=None, ref_image=None
|
|
):
|
|
if input_video_path is not None:
|
|
if isinstance(input_video_path, str):
|
|
cap = cv2.VideoCapture(input_video_path)
|
|
input_video = []
|
|
|
|
original_fps = cap.get(cv2.CAP_PROP_FPS)
|
|
frame_skip = 1 if fps is None else int(original_fps // fps)
|
|
|
|
frame_count = 0
|
|
|
|
while True:
|
|
ret, frame = cap.read()
|
|
if not ret:
|
|
break
|
|
|
|
if frame_count % frame_skip == 0:
|
|
frame = cv2.resize(frame, (sample_size[1], sample_size[0]))
|
|
input_video.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
|
|
|
frame_count += 1
|
|
|
|
cap.release()
|
|
else:
|
|
input_video = input_video_path
|
|
|
|
input_video = torch.from_numpy(np.array(input_video))[:video_length]
|
|
input_video = input_video.permute([3, 0, 1, 2]).unsqueeze(0) / 255
|
|
|
|
if validation_video_mask is not None:
|
|
validation_video_mask = (
|
|
Image.open(validation_video_mask).convert("L").resize((sample_size[1], sample_size[0]))
|
|
)
|
|
input_video_mask = np.where(np.array(validation_video_mask) < 240, 0, 255)
|
|
|
|
input_video_mask = (
|
|
torch.from_numpy(np.array(input_video_mask))
|
|
.unsqueeze(0)
|
|
.unsqueeze(-1)
|
|
.permute([3, 0, 1, 2])
|
|
.unsqueeze(0)
|
|
)
|
|
input_video_mask = torch.tile(input_video_mask, [1, 1, input_video.size()[2], 1, 1])
|
|
input_video_mask = input_video_mask.to(input_video.device, input_video.dtype)
|
|
else:
|
|
input_video_mask = torch.zeros_like(input_video[:, :1])
|
|
input_video_mask[:, :, :] = 255
|
|
else:
|
|
input_video, input_video_mask = None, None
|
|
|
|
if ref_image is not None:
|
|
if isinstance(ref_image, str):
|
|
clip_image = Image.open(ref_image).convert("RGB")
|
|
else:
|
|
clip_image = Image.fromarray(np.array(ref_image, np.uint8))
|
|
else:
|
|
clip_image = None
|
|
|
|
if ref_image is not None:
|
|
if isinstance(ref_image, str):
|
|
ref_image = Image.open(ref_image).convert("RGB")
|
|
ref_image = ref_image.resize((sample_size[1], sample_size[0]))
|
|
ref_image = torch.from_numpy(np.array(ref_image))
|
|
ref_image = ref_image.unsqueeze(0).permute([3, 0, 1, 2]).unsqueeze(0) / 255
|
|
else:
|
|
ref_image = torch.from_numpy(np.array(ref_image))
|
|
ref_image = ref_image.unsqueeze(0).permute([3, 0, 1, 2]).unsqueeze(0) / 255
|
|
return input_video, input_video_mask, ref_image, clip_image
|
|
|
|
|
|
def process_video(input_video_path, input_mask_video_path, ref_images, video_length, sample_size):
|
|
"""Process input video and mask for editing"""
|
|
if input_video_path is not None:
|
|
cap = cv2.VideoCapture(input_video_path)
|
|
frames = []
|
|
while cap.isOpened():
|
|
ret, frame = cap.read()
|
|
if not ret:
|
|
break
|
|
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
|
frames.append(Image.fromarray(frame))
|
|
cap.release()
|
|
|
|
frames = frames[:video_length]
|
|
if len(frames) < video_length:
|
|
frames += [frames[-1]] * (video_length - len(frames))
|
|
|
|
resized_frames = [frame.resize([sample_size[1], sample_size[0]]) for frame in frames]
|
|
|
|
input_video = (
|
|
torch.stack([torch.from_numpy(np.array(frame)).permute(2, 0, 1) for frame in resized_frames])
|
|
.permute(1, 0, 2, 3)
|
|
.unsqueeze(0)
|
|
) # [1, C, T, H, W]
|
|
|
|
else:
|
|
# 直接生成全零张量，形状为 [1, 3, T, H, W]，归一化到 [-1, 1]
|
|
input_video = torch.zeros((1, 3, video_length, sample_size[0], sample_size[1])).float()
|
|
# 生成 input_video_mask 张量
|
|
if input_mask_video_path is not None:
|
|
mask_cap = cv2.VideoCapture(input_mask_video_path)
|
|
mask_frames = []
|
|
while mask_cap.isOpened():
|
|
ret, frame = mask_cap.read()
|
|
if not ret:
|
|
break
|
|
if len(frame.shape) == 3:
|
|
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
|
_, mask = cv2.threshold(frame, 127, 255, cv2.THRESH_BINARY)
|
|
mask_frames.append(mask)
|
|
mask_cap.release()
|
|
|
|
mask_frames = mask_frames[:video_length]
|
|
if len(mask_frames) < video_length:
|
|
mask_frames += [mask_frames[-1]] * (video_length - len(mask_frames))
|
|
|
|
resized_masks = [Image.fromarray(mask).resize([sample_size[1], sample_size[0]]) for mask in mask_frames]
|
|
|
|
input_video_mask = (
|
|
torch.stack([torch.from_numpy(np.array(mask)) for mask in resized_masks]).unsqueeze(0).unsqueeze(0) / 255.0
|
|
) # [1, 1, T, H, W]
|
|
|
|
else:
|
|
# 直接生成全1张量，形状为 [1, 1, T, H, W]，表示所有区域有效
|
|
input_video_mask = torch.ones((1, 1, video_length, sample_size[0], sample_size[1])).float()
|
|
|
|
if input_video_path is not None and input_video is not None:
|
|
input_video = input_video * (torch.tile(input_video_mask, [1, 3, 1, 1, 1]) < 0.5) + (128.0) * (
|
|
torch.tile(input_video_mask, [1, 3, 1, 1, 1]) >= 0.5
|
|
)
|
|
input_video = input_video.div_(127.5).sub_(1.0)
|
|
|
|
if ref_images is not None:
|
|
for i, ref_img in enumerate(ref_images):
|
|
if ref_img is not None:
|
|
ref_img = Image.open(ref_img).convert("RGB")
|
|
ref_img = TF.to_tensor(ref_img).sub_(0.5).div_(0.5).unsqueeze(1)
|
|
if ref_img.shape[-2:] != sample_size:
|
|
canvas_height, canvas_width = sample_size
|
|
ref_height, ref_width = ref_img.shape[-2:]
|
|
white_canvas = torch.ones((3, 1, canvas_height, canvas_width)) # [-1, 1]
|
|
scale = min(canvas_height / ref_height, canvas_width / ref_width)
|
|
new_height = int(ref_height * scale)
|
|
new_width = int(ref_width * scale)
|
|
resized_image = (
|
|
F.interpolate(
|
|
ref_img.squeeze(1).unsqueeze(0),
|
|
size=(new_height, new_width),
|
|
mode="bilinear",
|
|
align_corners=False,
|
|
)
|
|
.squeeze(0)
|
|
.unsqueeze(1)
|
|
)
|
|
top = (canvas_height - new_height) // 2
|
|
left = (canvas_width - new_width) // 2
|
|
white_canvas[:, :, top : top + new_height, left : left + new_width] = resized_image
|
|
ref_img = white_canvas
|
|
ref_images[i] = ref_img
|
|
ref_images = torch.cat(ref_images, dim=1).unsqueeze(0)
|
|
|
|
return input_video, input_video_mask, ref_images
|
|
|
|
|
|
class Time_Logger:
|
|
total = 0
|
|
count = 0
|
|
name = None
|
|
|
|
def log(self, t):
|
|
self.total += t
|
|
self.count += 1
|
|
|
|
def get_avg(self):
|
|
# 转换为秒
|
|
return self.total / self.count
|
|
|
|
|
|
def get_time_stat(logger: Time_Logger):
|
|
def time_stat(func):
|
|
logger.name = func.__name__
|
|
|
|
def wrapper(*args, **kwargs):
|
|
start = time.time()
|
|
result = func(*args, **kwargs)
|
|
end = time.time()
|
|
logger.log(end - start)
|
|
return result
|
|
|
|
return wrapper
|
|
|
|
return time_stat
|
|
|
|
|
|
def interpolate_MUSE(masks, new_depth):
|
|
bs, c, depth, height, width = masks.shape
|
|
if depth > 1:
|
|
first_frame = masks[:, :, 0:1, :, :]
|
|
remaining_frames = masks[:, :, 1:, :, :]
|
|
|
|
if (depth - 1) % 4 == 0:
|
|
# 重新排列为分组形式以便进行最大池化
|
|
grouped_frames = remaining_frames.view(
|
|
bs,
|
|
c,
|
|
(depth - 1) // 4, # 组数
|
|
4, # 每组4帧
|
|
height,
|
|
width,
|
|
)
|
|
# 对每组4帧进行最大池化
|
|
pooled_frames = torch.max(grouped_frames, dim=3)[0] # 在第4个维度(每组帧)上取最大值
|
|
# 合并首帧和池化后的帧
|
|
masks = torch.cat([first_frame, pooled_frames], dim=2)
|
|
else:
|
|
# 如果不能被4整除，使用原来的插值方法作为后备
|
|
masks = F.interpolate(masks, size=(new_depth, height, width), mode="nearest-exact")
|
|
else:
|
|
masks = masks
|
|
return masks
|
|
|
|
|
|
def check_noise_predictions(noise_pred, step_index, timestep, name="NOISE"):
|
|
"""
|
|
检查预测噪声是否有问题
|
|
"""
|
|
print(f"\n[{name} CHECK] Step {step_index}, Timestep: {timestep}")
|
|
|
|
# 基本形状信息
|
|
print(f" Noise shape: {noise_pred.shape}")
|
|
|
|
# 检查NaN值
|
|
nan_count = torch.isnan(noise_pred).sum().item()
|
|
if nan_count > 0:
|
|
print(f" *** ERROR: Found {nan_count} NaN values in noise prediction! ***")
|
|
return False
|
|
|
|
# 检查无穷大值
|
|
inf_count = torch.isinf(noise_pred).sum().item()
|
|
if inf_count > 0:
|
|
print(f" *** ERROR: Found {inf_count} Inf values in noise prediction! ***")
|
|
return False
|
|
|
|
# 统计信息
|
|
with torch.no_grad():
|
|
mean_val = noise_pred.mean().item()
|
|
std_val = noise_pred.std().item()
|
|
min_val = noise_pred.min().item()
|
|
max_val = noise_pred.max().item()
|
|
abs_max = noise_pred.abs().max().item()
|
|
|
|
print(f" Mean: {mean_val:.6f}, Std: {std_val:.6f}")
|
|
print(f" Min: {min_val:.6f}, Max: {max_val:.6f}")
|
|
print(f" Abs Max: {abs_max:.6f}")
|
|
|
|
# 检查异常大的值
|
|
if abs_max > 1e6:
|
|
print(f" *** WARNING: Very large values detected (>{1e6})! ***")
|
|
return False
|
|
|
|
# 检查零值过多（可能表示数值下溢）
|
|
zero_ratio = (noise_pred.abs() < 1e-10).sum().item() / noise_pred.numel()
|
|
if zero_ratio > 0.5: # 超过50%的值接近零
|
|
print(f" *** WARNING: High zero ratio ({zero_ratio * 100:.2f}%) detected! ***")
|
|
|
|
# 检查梯度爆炸迹象
|
|
if abs_max > 1e3:
|
|
print(f" *** WARNING: Large values ({abs_max:.2f}) detected, potential gradient explosion! ***")
|
|
|
|
return True
|
|
|
|
|
|
def check_weight_overflow(model, name_prefix=""):
|
|
"""检查模型权重是否存在溢出"""
|
|
overflow_detected = False
|
|
|
|
for name, param in model.named_parameters():
|
|
full_name = f"{name_prefix}.{name}" if name_prefix else name
|
|
|
|
# 计算统计信息
|
|
mean_val = param.mean().item()
|
|
std_val = param.std().item()
|
|
min_val = param.min().item()
|
|
max_val = param.max().item()
|
|
|
|
print(f"[STATS] {full_name}: mean={mean_val:.6f}, std={std_val:.6f}, min={min_val:.6f}, max={max_val:.6f}")
|
|
|
|
# 检查NaN
|
|
if torch.isnan(param).any():
|
|
print(f"[ERROR] NaN detected in {full_name}")
|
|
overflow_detected = True
|
|
|
|
# 检查无穷大值
|
|
if torch.isinf(param).any():
|
|
print(f"[ERROR] Inf detected in {full_name}")
|
|
overflow_detected = True
|
|
|
|
# 检查异常大的值
|
|
max_val = param.abs().max().item()
|
|
if max_val > 1e6:
|
|
print(f"[WARN] Large values detected in {full_name}: max={max_val}")
|
|
|
|
# 检查零值过多（可能表示数值下溢）
|
|
zero_ratio = (param.abs() < 1e-10).sum().item() / param.numel()
|
|
if zero_ratio > 0.9:
|
|
print(f"[WARN] High zero ratio in {full_name}: {zero_ratio * 100:.2f}%")
|
|
|
|
return overflow_detected
|
|
|
|
|
|
def postprocess_videoframe(videos: torch.Tensor, rescale=False, n_rows=6):
|
|
|
|
videos = rearrange(videos, "b c t h w -> t b c h w")
|
|
outputs = []
|
|
for x in videos:
|
|
x = torchvision.utils.make_grid(x, nrow=n_rows)
|
|
x = x.transpose(0, 1).transpose(1, 2).squeeze(-1)
|
|
if rescale:
|
|
x = (x + 1.0) / 2.0 # -1,1 -> 0,1
|
|
x = (x * 255).numpy().astype(np.uint8)
|
|
outputs.append(x)
|
|
|
|
current_h, current_w = outputs[0].shape[:2]
|
|
|
|
pil_outputs = []
|
|
for i, img in enumerate(outputs):
|
|
try:
|
|
# 1. 如果是 PyTorch Tensor，先转到 CPU Numpy
|
|
if hasattr(img, "cpu"):
|
|
img = img.detach().cpu().numpy()
|
|
|
|
# 2. 如果是 Numpy 数组
|
|
if isinstance(img, np.ndarray):
|
|
# 检查维度: 如果是 (C, H, W) -> 转为 (H, W, C)
|
|
# 判据: 第一个维度是 3，且最后两个维度比较大
|
|
if img.ndim == 3 and img.shape[0] == 3 and img.shape[2] > 3:
|
|
img = img.transpose(1, 2, 0)
|
|
|
|
# 检查数值范围: 如果是 0.0-1.0 的浮点数 -> 转为 0-255 uint8
|
|
if img.dtype != np.uint8:
|
|
if img.max() <= 1.05: # 稍微放宽一点防止 1.0001
|
|
img = (img * 255).clip(0, 255)
|
|
img = img.astype(np.uint8)
|
|
|
|
# 3. 转换为 PIL
|
|
pil_outputs.append(Image.fromarray(img))
|
|
|
|
except Exception as e:
|
|
print(f"[Error] Frame {i} convert failed: {e}. type={type(img)}, shape={getattr(img, 'shape', 'N/A')}")
|
|
# 出错时放入一张黑图防止整个接口崩溃
|
|
pil_outputs.append(Image.new("RGB", (current_w, current_h), (0, 0, 0)))
|
|
|
|
return pil_outputs
|