# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # 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 os from functools import wraps import math import torch from PIL import Image import torchvision from chronoedit._ext.imaginaire.utils import log _IMAGE_EXTENSIONS = [".png", ".jpg", ".jpeg", "webp"] def resize_input(video: torch.Tensor, resolution: list[int]): r""" Resizes and crops the input video tensor while preserving aspect ratio. The video is first resized so that the smaller dimension matches the target resolution, preserving the aspect ratio. Then, it's center-cropped to the target resolution. Args: video (torch.Tensor): Input video tensor of shape (T, C, H, W). resolution (list[int]): Target resolution [H, W]. Returns: torch.Tensor: Resized and cropped video tensor of shape (T, C, target_H, target_W). """ orig_h, orig_w = video.shape[2], video.shape[3] target_h, target_w = resolution scaling_ratio = max((target_w / orig_w), (target_h / orig_h)) resizing_shape = (int(math.ceil(scaling_ratio * orig_h)), int(math.ceil(scaling_ratio * orig_w))) video_resized = torchvision.transforms.functional.resize(video, resizing_shape) video_cropped = torchvision.transforms.functional.center_crop(video_resized, resolution) return video_cropped def read_and_process_image(img_path: str, resolution: list[int], num_video_frames: int, resize: bool = True): """ Reads an image, converts it to a video tensor, and processes it for model input. The image is loaded, converted to a tensor, and replicated to match the `num_video_frames`. It's then optionally resized and permuted to the standard video format (B, C, T, H, W). Args: img_path (str): Path to the input image file. resolution (list[int]): Target resolution [H, W] for resizing. num_video_frames (int): The number of frames the output video tensor should have. resize (bool, optional): Whether to resize the image to the target resolution. Defaults to True. Returns: torch.Tensor: Processed video tensor of shape (1, C, T, H, W). Raises: ValueError: If the image extension is not one of the supported types. """ ext = os.path.splitext(img_path)[1] if ext.lower() not in _IMAGE_EXTENSIONS: raise ValueError(f"Invalid image extension: {ext}") # Read the image img = Image.open(img_path) # Convert to tensor img = torchvision.transforms.functional.to_tensor(img) # Create a video tensor by repeating the first frame vid_input = img.unsqueeze(0) # Add temporal dimension T=1 # Repeat the first frame to match the desired number of video frames # Note: The actual content for frames > 0 will be generated by the model. vid_input = torch.cat([vid_input, torch.zeros_like(vid_input).repeat(num_video_frames - 1, 1, 1, 1)], dim=0) vid_input = (vid_input * 255.0).to(torch.uint8) # Convert to uint8 range if needed (might depend on model) if resize: # Resize and crop to the target resolution vid_input = resize_input(vid_input, resolution) # Convert to {B, C, T, H, W} format expected by the model vid_input = vid_input.unsqueeze(0).permute(0, 2, 1, 3, 4) # Add batch dim B=1 and permute return vid_input class sync_timer: """ Synchronized timer to count the inference time of `nn.Module.forward` or else. set env var SYNC_TIMER=1 to enable logging! Example as context manager: ```python with timer('name'): run() ``` Example as decorator: ```python @timer('name') def run(): pass ``` """ def __init__(self, name=None, flag_env="SYNC_TIMER"): self.name = name self.flag_env = flag_env def __enter__(self): if os.environ.get(self.flag_env, "0") == "1": self.start = torch.cuda.Event(enable_timing=True) self.end = torch.cuda.Event(enable_timing=True) self.start.record() return lambda: self.time def __exit__(self, exc_type, exc_value, exc_tb): if os.environ.get(self.flag_env, "0") == "1": self.end.record() torch.cuda.synchronize() self.time = self.start.elapsed_time(self.end) if self.name is not None: log.info(f"{self.name} takes {self.time/1000:.4f}s", rank0_only=False) def __call__(self, func): @wraps(func) def wrapper(*args, **kwargs): with self: result = func(*args, **kwargs) return result return wrapper