From e567f78876d314a0d87d8e492edcd4dafc05cd2e Mon Sep 17 00:00:00 2001 From: Terry Jia Date: Sun, 26 Jul 2026 07:48:30 -0400 Subject: [PATCH] feat: VIDEO_EDIT input type for video trim/crop rich widgets --- comfy_api/latest/_input/video_types.py | 41 +++++- comfy_api/latest/_input_impl/video_types.py | 120 +++++++++++++++-- comfy_api/latest/_io.py | 36 +++++ comfy_api/latest/_util/__init__.py | 3 +- comfy_api/latest/_util/video_types.py | 20 +++ comfy_extras/nodes_video.py | 135 ++++++++++++++++++- server.py | 137 ++++++++++++++------ 7 files changed, 438 insertions(+), 54 deletions(-) diff --git a/comfy_api/latest/_input/video_types.py b/comfy_api/latest/_input/video_types.py index b700d44f5..c9c153e06 100644 --- a/comfy_api/latest/_input/video_types.py +++ b/comfy_api/latest/_input/video_types.py @@ -4,7 +4,7 @@ from fractions import Fraction from typing import Optional, Union, IO import io import av -from .._util import VideoContainer, VideoCodec, VideoComponents +from .._util import VideoContainer, VideoCodec, VideoComponents, normalize_crop_rect class VideoInput(ABC): """ @@ -54,6 +54,45 @@ class VideoInput(ABC): """ pass + def as_cropped( + self, + x: int = 0, + y: int = 0, + width: int = 0, + height: int = 0, + ) -> VideoInput: + """ + Create a new VideoInput spatially cropped to the given pixel rectangle. + + The rectangle is clamped to the frame and even-aligned for encoder + compatibility. An empty or full-frame rectangle returns the input + unchanged. + + Default implementation materializes the video via get_components(); + subclasses should override with lazier strategies when possible. + """ + components = self.get_components() + rect = normalize_crop_rect( + x, y, width, height, components.images.shape[2], components.images.shape[1] + ) + if rect is None: + return self + from .._input_impl.video_types import VideoFromComponents + + cx, cy, cw, ch = rect + return VideoFromComponents( + VideoComponents( + images=components.images[:, cy:cy + ch, cx:cx + cw, :], + audio=components.audio, + frame_rate=components.frame_rate, + metadata=components.metadata, + alpha=components.alpha[:, cy:cy + ch, cx:cx + cw] + if components.alpha is not None + else None, + ), + bit_depth=self.get_bit_depth(), + ) + def get_stream_source(self) -> Union[str, io.BytesIO]: """ Get a streamable source for the video. This allows processing without diff --git a/comfy_api/latest/_input_impl/video_types.py b/comfy_api/latest/_input_impl/video_types.py index cf4119250..5f99e61fb 100644 --- a/comfy_api/latest/_input_impl/video_types.py +++ b/comfy_api/latest/_input_impl/video_types.py @@ -12,7 +12,7 @@ import numpy as np import math import os import torch -from .._util import VideoContainer, VideoCodec, VideoComponents +from .._util import VideoContainer, VideoCodec, VideoComponents, normalize_crop_rect import logging @@ -115,12 +115,17 @@ def mp4_output_open_kwargs(path: str | io.BytesIO, format: VideoContainer, codec return open_kwargs +def _rotation_quadrant(frame: av.VideoFrame) -> int: + return int(round(frame.rotation // 90)) % 4 if frame.rotation else 0 + + class VideoFromFile(VideoInput): """ Class representing video input from a file. """ - def __init__(self, file: str | io.BytesIO, *, start_time: float=0, duration: float=0): + def __init__(self, file: str | io.BytesIO, *, start_time: float=0, duration: float=0, + crop: tuple[int, int, int, int] | None = None): """ Initialize the VideoFromFile object based off of either a path on disk or a BytesIO object containing the file contents. @@ -128,6 +133,7 @@ class VideoFromFile(VideoInput): self.__file = file self.__start_time = start_time self.__duration = duration + self.__crop = crop def get_stream_source(self) -> str | io.BytesIO: """ @@ -157,7 +163,31 @@ class VideoFromFile(VideoInput): for stream in container.streams: if stream.type == 'video': assert isinstance(stream, av.VideoStream) - return stream.width, stream.height + if self.__crop is None: + return stream.width, stream.height + + display_width, display_height = self._get_display_dimensions() + rect = normalize_crop_rect(*self.__crop, display_width, display_height) + if rect is not None: + return rect[2], rect[3] + return display_width, display_height + raise ValueError(f"No video stream found in file '{self.__file}'") + + def _get_display_dimensions(self) -> tuple[int, int]: + if isinstance(self.__file, io.BytesIO): + self.__file.seek(0) + with av.open(self.__file, mode='r') as container: + for stream in container.streams: + if stream.type == 'video': + assert isinstance(stream, av.VideoStream) + width, height = stream.width, stream.height + try: + frame = next(container.decode(stream), None) + except av.error.FFmpegError: + frame = None + if frame is not None and _rotation_quadrant(frame) % 2: + width, height = height, width + return width, height raise ValueError(f"No video stream found in file '{self.__file}'") def get_bit_depth(self) -> int: @@ -327,6 +357,8 @@ class VideoFromFile(VideoInput): streams = [video_stream] has_first_audio_frame = False checked_alpha = False + crop_rect = None + crop_resolved = False # Default to False so we decode until EOF if duration is 0 video_done = False @@ -397,9 +429,16 @@ class VideoFromFile(VideoInput): img = np.ascontiguousarray(align_graph[2].pull().to_ndarray(format=image_format)[:frame.height, :frame.width]) else: img = frame.to_ndarray(format=image_format) - if frame.rotation != 0: - k = int(round(frame.rotation // 90)) - img = np.rot90(img, k=k, axes=(0, 1)).copy() + rotation_quadrant = _rotation_quadrant(frame) + if rotation_quadrant: + img = np.rot90(img, k=rotation_quadrant, axes=(0, 1)).copy() + if self.__crop is not None: + if not crop_resolved: + crop_rect = normalize_crop_rect(*self.__crop, img.shape[1], img.shape[0]) + crop_resolved = True + if crop_rect is not None: + cx, cy, cw, ch = crop_rect + img = np.ascontiguousarray(img[cy:cy + ch, cx:cx + cw]) if alphas is None: frames.append(torch.from_numpy(img)) else: @@ -484,6 +523,8 @@ class VideoFromFile(VideoInput): reuse_streams = False if self.__start_time or self.__duration: reuse_streams = False + if self.__crop is not None: + reuse_streams = False if not reuse_streams: if bit_depth is None: @@ -564,6 +605,12 @@ class VideoFromFile(VideoInput): if duration: duration_cap = math.ceil(duration * sample_rate) + import comfy.utils + raw_duration = self._get_raw_duration() + window_seconds = duration if duration else max(raw_duration - start_time, 0.0) + progress_total = max(1, int(round(window_seconds * float(rate)))) + pbar = comfy.utils.ProgressBar(progress_total) + streams = [video_stream] if audio_stream is None else [video_stream, audio_stream] pts_step = max(1, int(round((1 / rate) / video_stream.time_base))) video_done = False @@ -576,6 +623,8 @@ class VideoFromFile(VideoInput): source_size = None rotation_k = 0 rotation_filter = None + crop_rect = None + crop_filter = None audio_started = False samples_written = 0 pending_audio = [] @@ -649,11 +698,15 @@ class VideoFromFile(VideoInput): if end_pts is not None and frame.pts is not None: frame_duration = min(frame_duration, end_pts - frame.pts) if output is None: - rotation_k = int(round(frame.rotation // 90)) % 4 if frame.rotation else 0 + rotation_k = _rotation_quadrant(frame) if rotation_k % 2: out_width, out_height = frame.height, frame.width else: out_width, out_height = frame.width, frame.height + if self.__crop is not None: + crop_rect = normalize_crop_rect(*self.__crop, out_width, out_height) + if crop_rect is not None: + out_width, out_height = crop_rect[2], crop_rect[3] if out_width % 2 or out_height % 2: raise ValueError(f"H.264 output requires even dimensions, got {out_width}x{out_height}") source_size = (frame.width, frame.height) @@ -694,9 +747,22 @@ class VideoFromFile(VideoInput): g_sink = g.add("buffersink") tail.link_to(g_sink) g.configure() - rotation_filter = (g_src, g_sink) - rotation_filter[0].push(frame) - frame = rotation_filter[1].pull() + rotation_filter = (g, g_src, g_sink) + rotation_filter[1].push(frame) + frame = rotation_filter[2].pull() + if crop_rect is not None: + if crop_filter is None: + g = av.filter.Graph() + g_src = g.add_buffer(width=frame.width, height=frame.height, + format=frame.format.name, time_base=video_stream.time_base) + g_crop = g.add("crop", f"{crop_rect[2]}:{crop_rect[3]}:{crop_rect[0]}:{crop_rect[1]}") + g_sink = g.add("buffersink") + g_src.link_to(g_crop) + g_crop.link_to(g_sink) + g.configure() + crop_filter = (g, g_src, g_sink) + crop_filter[1].push(frame) + frame = crop_filter[2].pull() if frame.color_range == ColorRange.JPEG: # compress full-range sources (yuvj/MJPEG) to limited range frame = frame.reformat(format=pix_fmt, src_color_range="JPEG", dst_color_range="MPEG") @@ -735,6 +801,7 @@ class VideoFromFile(VideoInput): out_packet.duration = video_frame_durations.pop(out_packet.pts, 0) output.mux(out_packet) drain_audio() + pbar.update(1) elif packet.stream == audio_stream and not audio_done: for resampled in itertools.chain.from_iterable(map(resampler.resample, packet.decode())): @@ -804,11 +871,42 @@ class VideoFromFile(VideoInput): self.get_stream_source(), start_time=start_time + self.__start_time, duration=duration, + crop=self.__crop, ) if trimmed.get_duration() < duration and strict_duration: return None return trimmed + def as_cropped( + self, x: int = 0, y: int = 0, width: int = 0, height: int = 0 + ) -> VideoInput: + if int(width) <= 0 or int(height) <= 0: + return self + + display_width, display_height = self._get_display_dimensions() + outer = ( + normalize_crop_rect(*self.__crop, display_width, display_height) + if self.__crop is not None + else None + ) + if outer is None: + rect = normalize_crop_rect(x, y, width, height, display_width, display_height) + else: + inner = normalize_crop_rect(x, y, width, height, outer[2], outer[3]) + rect = ( + (outer[0] + inner[0], outer[1] + inner[1], inner[2], inner[3]) + if inner is not None + else None + ) + if rect is None: + return self + return VideoFromFile( + self.get_stream_source(), + start_time=self.__start_time, + duration=self.__duration, + crop=rect, + ) + class VideoFromComponents(VideoInput): """ @@ -825,6 +923,8 @@ class VideoFromComponents(VideoInput): images=self.__components.images, audio=self.__components.audio, frame_rate=self.__components.frame_rate, + metadata=self.__components.metadata, + alpha=self.__components.alpha, ) def get_bit_depth(self) -> int: diff --git a/comfy_api/latest/_io.py b/comfy_api/latest/_io.py index 6d9e08f47..6c293b22d 100644 --- a/comfy_api/latest/_io.py +++ b/comfy_api/latest/_io.py @@ -1416,6 +1416,41 @@ class BoundingBoxes(ComfyTypeIO): self.default = [] +@comfytype(io_type="VIDEO_EDIT") +class VideoEdit(ComfyTypeIO): + class VideoTrimSection(TypedDict): + start_time: float + duration: float + + class VideoCropSection(TypedDict): + x: int + y: int + width: int + height: int + + class VideoEditDict(TypedDict, total=False): + trim: 'VideoEdit.VideoTrimSection' + crop: 'VideoEdit.VideoCropSection' + Type = VideoEditDict + + class Input(WidgetInput): + def __init__(self, id: str, display_name: str=None, optional=False, tooltip: str=None, + socketless: bool=True, default: dict=None, features: list[str]=None, advanced: bool=None): + super().__init__(id, display_name, optional, tooltip, None, default, socketless, None, None, None, None, advanced) + self.features = features if features is not None else ["trim", "crop"] + if default is None: + self.default = {} + if "trim" in self.features: + self.default["trim"] = {"start_time": 0.0, "duration": 0.0} + if "crop" in self.features: + self.default["crop"] = {"x": 0, "y": 0, "width": 0, "height": 0} + + def as_dict(self): + return super().as_dict() | prune_dict({ + "features": self.features, + }) + + @comfytype(io_type="HISTOGRAM") class Histogram(ComfyTypeIO): """A histogram represented as a list of bin counts.""" @@ -2493,5 +2528,6 @@ __all__ = [ "Curve", "Histogram", "Range", + "VideoEdit", "NodeReplace", ] diff --git a/comfy_api/latest/_util/__init__.py b/comfy_api/latest/_util/__init__.py index b27f5a97e..c3c9f0183 100644 --- a/comfy_api/latest/_util/__init__.py +++ b/comfy_api/latest/_util/__init__.py @@ -1,4 +1,4 @@ -from .video_types import VideoContainer, VideoCodec, VideoComponents +from .video_types import VideoContainer, VideoCodec, VideoComponents, normalize_crop_rect from .geometry_types import VOXEL, MESH, SPLAT, File3D from .image_types import SVG @@ -7,6 +7,7 @@ __all__ = [ "VideoContainer", "VideoCodec", "VideoComponents", + "normalize_crop_rect", "VOXEL", "MESH", "SPLAT", diff --git a/comfy_api/latest/_util/video_types.py b/comfy_api/latest/_util/video_types.py index 6c9d6a526..2f8ae887d 100644 --- a/comfy_api/latest/_util/video_types.py +++ b/comfy_api/latest/_util/video_types.py @@ -48,3 +48,23 @@ class VideoComponents: audio: Optional[AudioInput] = None metadata: Optional[dict] = None alpha: Optional[MaskInput] = None + + +def normalize_crop_rect( + x: int, y: int, width: int, height: int, source_width: int, source_height: int +) -> Optional[tuple[int, int, int, int]]: + width = int(width) + height = int(height) + if width <= 0 or height <= 0: + return None + x = max(0, min(int(x), source_width - 1)) + y = max(0, min(int(y), source_height - 1)) + width = min(width, source_width - x) + height = min(height, source_height - y) + if x == 0 and y == 0 and width == source_width and height == source_height: + return None + width -= width % 2 + height -= height % 2 + if width <= 0 or height <= 0: + return None + return x, y, width, height diff --git a/comfy_extras/nodes_video.py b/comfy_extras/nodes_video.py index 45394ce4d..a20b29999 100644 --- a/comfy_extras/nodes_video.py +++ b/comfy_extras/nodes_video.py @@ -229,8 +229,14 @@ class LoadVideo(io.ComfyNode): display_name="Load Video", category="video", essentials_category="Basics", + has_intermediate_output=True, inputs=[ io.Combo.Input("file", options=sorted(files), upload=io.UploadType.video), + io.VideoEdit.Input( + "edit", + optional=True, + tooltip="Trim (seconds) and crop (pixels) applied on load. Zero values leave the video unchanged.", + ), ], outputs=[ io.Video.Output(), @@ -238,12 +244,17 @@ class LoadVideo(io.ComfyNode): ) @classmethod - def execute(cls, file) -> io.NodeOutput: + def execute(cls, file, edit=None) -> io.NodeOutput: video_path = folder_paths.get_annotated_filepath(file) - return io.NodeOutput(InputImpl.VideoFromFile(video_path)) + source = InputImpl.VideoFromFile(video_path) + video = apply_video_trim(source, (edit or {}).get("trim")) + video = apply_video_crop(video, (edit or {}).get("crop")) + if video is source: + return io.NodeOutput(video, ui=preview_input_video(file)) + return io.NodeOutput(video, ui=save_video_preview(video)) @classmethod - def fingerprint_inputs(s, file): + def fingerprint_inputs(s, file, edit=None): video_path = folder_paths.get_annotated_filepath(file) mod_time = os.path.getmtime(video_path) # Instead of hashing the file, we can just use the modification time to avoid @@ -251,12 +262,60 @@ class LoadVideo(io.ComfyNode): return mod_time @classmethod - def validate_inputs(s, file): + def validate_inputs(s, file, edit=None): if not folder_paths.exists_annotated_filepath(file): return "Invalid video file: {}".format(file) return True +def preview_input_video(file: str) -> ui.PreviewVideo: + name, _ = folder_paths.annotated_filepath(file) + subfolder, _, filename = name.replace("\\", "/").rpartition("/") + return ui.PreviewVideo([ui.SavedResult(filename, subfolder, io.FolderType.input)]) + + +def save_video_preview(video: Input.Video) -> ui.PreviewVideo: + width, height = video.get_dimensions() + full_output_folder, filename, counter, subfolder, _ = folder_paths.get_save_image_path( + "ComfyUI_temp_video", folder_paths.get_temp_directory(), width, height + ) + preview_format = Types.VideoContainer.MP4 + file = f"{filename}_{counter:05}_.{Types.VideoContainer.get_extension(preview_format)}" + video.save_to( + os.path.join(full_output_folder, file), + format=preview_format, + codec="auto", + ) + return ui.PreviewVideo([ui.SavedResult(file, subfolder, io.FolderType.temp)]) + + +def apply_video_trim(video: Input.Video, trim, strict_duration: bool = False) -> Input.Video: + trim = trim or {} + start_time = float(trim.get("start_time", 0.0)) + duration = float(trim.get("duration", 0.0)) + if duration < 0: + raise ValueError(f"Trim duration must be >= 0, got {duration}") + if start_time == 0.0 and duration == 0.0: + return video + + trimmed = video.as_trimmed(start_time, duration, strict_duration=strict_duration) + if trimmed is None: + raise ValueError( + f"Failed to trim video:\nSource duration: {video.get_duration()}\nStart time: {start_time}\nTarget duration: {duration}" + ) + return trimmed + + +def apply_video_crop(video: Input.Video, crop) -> Input.Video: + crop = crop or {} + return video.as_cropped( + int(crop.get("x", 0)), + int(crop.get("y", 0)), + int(crop.get("width", 0)), + int(crop.get("height", 0)), + ) + + class VideoSlice(io.ComfyNode): @classmethod def define_schema(cls): @@ -304,6 +363,72 @@ class VideoSlice(io.ComfyNode): ) +class VideoTrim(io.ComfyNode): + @classmethod + def define_schema(cls): + return io.Schema( + node_id="VideoTrim", + display_name="Trim Video", + search_aliases=["trim video duration", "skip first frames", "cut video", "start time"], + category="video", + is_experimental=True, + essentials_category="Video Tools", + has_intermediate_output=True, + inputs=[ + io.Video.Input("video"), + io.VideoEdit.Input( + "trim", + features=["trim"], + tooltip="Trim window in seconds. Duration 0 keeps the video until the end.", + ), + io.Boolean.Input( + "strict_duration", + default=False, + advanced=True, + tooltip="If True, when the specified duration is not possible, an error will be raised.", + ), + ], + outputs=[ + io.Video.Output(), + ], + ) + + @classmethod + def execute(cls, video: io.Video.Type, trim: io.VideoEdit.Type, strict_duration: bool) -> io.NodeOutput: + trimmed = apply_video_trim(video, (trim or {}).get("trim"), strict_duration=strict_duration) + return io.NodeOutput(trimmed, ui=save_video_preview(trimmed)) + + +class VideoCrop(io.ComfyNode): + @classmethod + def define_schema(cls): + return io.Schema( + node_id="VideoCrop", + display_name="Crop Video", + search_aliases=["crop video", "cut region", "spatial crop"], + category="video", + is_experimental=True, + essentials_category="Video Tools", + has_intermediate_output=True, + inputs=[ + io.Video.Input("video"), + io.VideoEdit.Input( + "crop", + features=["crop"], + tooltip="Crop region in pixels. Zero width/height keeps the full frame.", + ), + ], + outputs=[ + io.Video.Output(), + ], + ) + + @classmethod + def execute(cls, video: io.Video.Type, crop: io.VideoEdit.Type) -> io.NodeOutput: + cropped = apply_video_crop(video, (crop or {}).get("crop")) + return io.NodeOutput(cropped, ui=save_video_preview(cropped)) + + class VideoExtension(ComfyExtension): @override async def get_node_list(self) -> list[type[io.ComfyNode]]: @@ -314,6 +439,8 @@ class VideoExtension(ComfyExtension): GetVideoComponents, LoadVideo, VideoSlice, + VideoTrim, + VideoCrop, ] async def comfy_entrypoint() -> VideoExtension: diff --git a/server.py b/server.py index c9ffcaa0d..ad81c8be0 100644 --- a/server.py +++ b/server.py @@ -31,6 +31,7 @@ from io import BytesIO import aiohttp from aiohttp import web +import av import logging import mimetypes @@ -212,6 +213,52 @@ def create_block_external_middleware(): return block_external_middleware +def resolve_view_media_path(request, user_manager): + if "filename" not in request.rel_url.query: + return web.Response(status=400) + filename = request.rel_url.query["filename"] + + # The frontend's LoadImage combo widget uses asset_hash values + # (e.g. "blake3:...") as widget values. When litegraph renders the + # node preview, it constructs /view?filename=, so this + # endpoint must resolve blake3 hashes to their on-disk file paths. + if filename.startswith("blake3:"): + owner_id = user_manager.get_request_user_id(request) + result = resolve_hash_to_path(filename, owner_id=owner_id) + if result is None: + return web.Response(status=404) + return result.abs_path, result.download_name, result.content_type + + filename, output_dir = folder_paths.annotated_filepath(filename) + + if not filename: + return web.Response(status=400) + + # validation for security: prevent accessing arbitrary path + if filename[0] == '/' or '..' in filename: + return web.Response(status=400) + + if output_dir is None: + type = request.rel_url.query.get("type", "output") + output_dir = folder_paths.get_directory_by_type(type) + + if output_dir is None: + return web.Response(status=400) + + if "subfolder" in request.rel_url.query: + subfolder = request.rel_url.query["subfolder"] + if os.path.isabs(subfolder) or os.path.splitdrive(subfolder)[0]: + return web.Response(status=403) + base_dir = os.path.abspath(output_dir) + full_output_dir = os.path.join(base_dir, subfolder) + if os.path.commonpath((os.path.abspath(full_output_dir), base_dir)) != base_dir: + return web.Response(status=403) + output_dir = full_output_dir + + filename = os.path.basename(filename) + return os.path.join(output_dir, filename), filename, None + + class PromptServer(): def __init__(self, loop): PromptServer.instance = self @@ -516,44 +563,10 @@ class PromptServer(): @routes.get("/view") async def view_image(request): if "filename" in request.rel_url.query: - filename = request.rel_url.query["filename"] - - # The frontend's LoadImage combo widget uses asset_hash values - # (e.g. "blake3:...") as widget values. When litegraph renders the - # node preview, it constructs /view?filename=, so this - # endpoint must resolve blake3 hashes to their on-disk file paths. - if filename.startswith("blake3:"): - owner_id = self.user_manager.get_request_user_id(request) - result = resolve_hash_to_path(filename, owner_id=owner_id) - if result is None: - return web.Response(status=404) - file, filename, resolved_content_type = result.abs_path, result.download_name, result.content_type - else: - resolved_content_type = None - filename, output_dir = folder_paths.annotated_filepath(filename) - - if not filename: - return web.Response(status=400) - - # validation for security: prevent accessing arbitrary path - if filename[0] == '/' or '..' in filename: - return web.Response(status=400) - - if output_dir is None: - type = request.rel_url.query.get("type", "output") - output_dir = folder_paths.get_directory_by_type(type) - - if output_dir is None: - return web.Response(status=400) - - if "subfolder" in request.rel_url.query: - full_output_dir = os.path.join(output_dir, request.rel_url.query["subfolder"]) - if os.path.commonpath((os.path.abspath(full_output_dir), output_dir)) != output_dir: - return web.Response(status=403) - output_dir = full_output_dir - - filename = os.path.basename(filename) - file = os.path.join(output_dir, filename) + resolved = resolve_view_media_path(request, self.user_manager) + if isinstance(resolved, web.Response): + return resolved + file, filename, resolved_content_type = resolved if os.path.isfile(file): if 'preview' in request.rel_url.query: @@ -660,6 +673,54 @@ class PromptServer(): return web.Response(status=404) + @routes.get("/video_metadata") + async def get_video_metadata(request): + resolved = resolve_view_media_path(request, self.user_manager) + if isinstance(resolved, web.Response): + return resolved + file = resolved[0] + + if not os.path.isfile(file): + return web.Response(status=404) + + def probe_video_metadata(): + with av.open(file) as container: + stream = next((s for s in container.streams if s.type == "video"), None) + if stream is None: + return None + + fps = float(stream.average_rate) if stream.average_rate else None + duration = None + if stream.duration is not None and stream.time_base is not None: + duration = float(stream.duration * stream.time_base) + elif container.duration is not None: + duration = float(container.duration * av.time_base) + frame_count = stream.frames or None + if frame_count is None and duration is not None and fps is not None: + frame_count = round(duration * fps) + + return { + "fps": fps, + "duration": duration, + "frame_count": frame_count, + "width": stream.codec_context.width, + "height": stream.codec_context.height, + "size": os.path.getsize(file), + } + + try: + metadata = await asyncio.to_thread(probe_video_metadata) + except FileNotFoundError: + return web.Response(status=404) + except PermissionError: + return web.Response(status=403) + except av.error.FFmpegError: + return web.Response(status=415) + if metadata is None: + return web.Response(status=415) + + return web.json_response(metadata) + @routes.get("/view_metadata/{folder_name}") async def view_metadata(request): folder_name = request.match_info.get("folder_name", None)