diff --git a/packages/on_demand_video_decoder/data/temporal_variants/vfr_h264_yuv420p.mp4 b/packages/on_demand_video_decoder/data/temporal_variants/vfr_h264_yuv420p.mp4 new file mode 100644 index 00000000..a3d832f6 Binary files /dev/null and b/packages/on_demand_video_decoder/data/temporal_variants/vfr_h264_yuv420p.mp4 differ diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvBatchAsyncGopDecoder.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvBatchAsyncGopDecoder.cpp index 4b8e0264..a5a80a36 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvBatchAsyncGopDecoder.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvBatchAsyncGopDecoder.cpp @@ -266,9 +266,9 @@ void PyNvBatchAsyncGopDecoder::build_yuv_frame(Pixel_Format fmt, size_t H, size_ break; // TODO(P016): LoadDLPack rejects "|u2" typestr, so no DLPack tensor can be built. case Pixel_Format_P016: - out.views.push_back(CAIMemoryView{{H, W, 1}, {W, 1, 1}, "|u2", stream_id, dst_ptr, false}); + out.views.push_back(CAIMemoryView{{H, W, 1}, {W * 2, 2, 2}, "|u2", stream_id, dst_ptr, false}); out.views.push_back(CAIMemoryView{ - {H / 2, W / 2, 2}, {W / 2 * 2, 2, 1}, "|u2", stream_id, dst_ptr + 2 * H * W, false}); + {H / 2, W / 2, 2}, {W * 2, 4, 2}, "|u2", stream_id, dst_ptr + 2 * H * W, false}); break; // TODO(YUV444): needs a flat (H*3, W) DLPack view and extBuf support for 3-plane layouts. case Pixel_Format_YUV444: diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_common.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_common.cpp index 66197a24..f14e4444 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_common.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_common.cpp @@ -141,14 +141,14 @@ int PyNvGopDecoder::GetYUVFromFrame(NvDecoder* decoder, const uint8_t* pFrame, u } break; case Pixel_Format_P016: { decoded_frame.views.push_back(CAIMemoryView{{height, width, 1}, - {width, 1, 1}, + {width * 2, 2, 2}, "|u2", reinterpret_cast(decoder->GetStream()), (CUdeviceptr)(pFrame_buffer), false}); decoded_frame.views.push_back( CAIMemoryView{{height / 2, width / 2, 2}, - {width / 2 * 2, 2, 1}, + {width * 2, 4, 2}, "|u2", reinterpret_cast(decoder->GetStream()), (CUdeviceptr)(pFrame_buffer + 2 * (width * height)), diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvVideoReader.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvVideoReader.cpp index 9b4c0334..9961e651 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvVideoReader.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvVideoReader.cpp @@ -721,13 +721,13 @@ DecodedFrameExt PyNvVideoReader::returnYUVFrame(void* pFrame_buffer, void* pFram } break; case Pixel_Format_P016: { frame.views.push_back(CAIMemoryView{{height, width, 1}, - {width, 1, 1}, + {width * 2, 2, 2}, "|u2", reinterpret_cast(this->decoder->GetStream()), (CUdeviceptr)(pFrame_buffer), false}); frame.views.push_back(CAIMemoryView{{height / 2, width / 2, 2}, - {width / 2 * 2, 2, 1}, + {width * 2, 4, 2}, "|u2", reinterpret_cast(this->decoder->GetStream()), (CUdeviceptr)(pFrame_buffer + 2 * (width * height)), diff --git a/packages/on_demand_video_decoder/pyproject.toml b/packages/on_demand_video_decoder/pyproject.toml index 65482734..f229c1d0 100644 --- a/packages/on_demand_video_decoder/pyproject.toml +++ b/packages/on_demand_video_decoder/pyproject.toml @@ -14,7 +14,7 @@ dynamic = ["version"] description = "On-demand video decoder (part of the ACCV-Lab package)." requires-python = ">=3.8" dependencies = [ - "torch>=2.0.0", + "torch>=2.3.0", "numpy>=1.22.2", ] diff --git a/packages/on_demand_video_decoder/tests/common/__init__.py b/packages/on_demand_video_decoder/tests/common/__init__.py new file mode 100644 index 00000000..3a76188c --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/__init__.py @@ -0,0 +1,15 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Shared decoder conformance tests.""" diff --git a/packages/on_demand_video_decoder/tests/common/adapters.py b/packages/on_demand_video_decoder/tests/common/adapters.py new file mode 100644 index 00000000..d31b3092 --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/adapters.py @@ -0,0 +1,558 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Thin adapters for APIs that naturally decode video paths and frame indices.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Callable, List, Literal, Sequence, Tuple + +import torch + +from common.model import CanonicalFrame, OutputContract + +OutputFormat = Literal["rgb", "bgr", "yuv"] + + +_PIXEL_FORMAT_NAMES = { + 3: "NV12", + 4: "YUV444", + 5: "P016", + 6: "YUV444_16BIT", +} + + +class DecoderTestAdapter(ABC): + """Minimum interface required by the common decode tests.""" + + name: str + output_formats: Tuple[OutputFormat, ...] + + def __init__(self, decoder: object) -> None: + self.decoder = decoder + + @abstractmethod + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + """Decode one requested frame for every video entry.""" + + def normalize(self, frame: object, output_format: OutputFormat) -> CanonicalFrame: + """Take an owned snapshot suitable for exact comparison.""" + + if output_format in {"rgb", "bgr"}: + tensor = torch.as_tensor(frame, device="cuda").clone() + if tensor.ndim != 3 or tensor.shape[-1] != 3: + raise AssertionError(f"RGB/BGR output must have shape (H, W, 3), got {tuple(tensor.shape)}") + return CanonicalFrame( + format=output_format.upper(), + planes=(tensor,), + width=int(tensor.shape[1]), + height=int(tensor.shape[0]), + ) + + planes = [] + for view in frame.cuda(): + plane = torch.as_tensor(view, device="cuda").clone() + if plane.ndim == 3 and plane.shape[-1] == 1: + plane = plane.squeeze(-1) + planes.append(plane) + if not planes: + raise AssertionError("YUV output did not expose any planes") + + format_name = _PIXEL_FORMAT_NAMES.get(int(frame.format), f"UNKNOWN_{int(frame.format)}") + return CanonicalFrame( + format=format_name, + planes=tuple(planes), + width=int(planes[0].shape[1]), + height=int(planes[0].shape[0]), + ) + + def output_contract(self, output_format: OutputFormat) -> OutputContract: + # Native YUV output is a format family: its concrete layout and dtype + # depend on the source pixel format and bit depth. + kind = "YUV" if output_format == "yuv" else output_format.upper() + dtypes = (torch.uint8, torch.uint16) if output_format == "yuv" else (torch.uint8,) + return OutputContract(kind=kind, dtypes=dtypes) + + def close(self) -> None: + self.decoder = None + + +class RandomAdapter(DecoderTestAdapter): + name = "random" + output_formats = ("rgb", "bgr", "yuv") + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + if output_format == "yuv": + return self.decoder.Decode(list(videos), list(frame_indices)) + return self.decoder.DecodeN12ToRGB( + list(videos), + list(frame_indices), + output_format == "bgr", + ) + + +class RandomFastInitAdapter(DecoderTestAdapter): + name = "random_fast_init" + output_formats = ("rgb", "bgr", "yuv") + + def __init__( + self, + decoder: object, + get_fast_init_info: Callable[[Sequence[str]], object], + ) -> None: + super().__init__(decoder) + self.get_fast_init_info = get_fast_init_info + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + videos = list(videos) + frame_indices = list(frame_indices) + stream_infos = self.get_fast_init_info(videos) + if output_format == "yuv": + return self.decoder.Decode( + videos, + frame_indices, + fastStreamInfos=stream_infos, + ) + return self.decoder.DecodeN12ToRGB( + videos, + frame_indices, + output_format == "bgr", + fastStreamInfos=stream_infos, + ) + + +class StreamAdapter(DecoderTestAdapter): + name = "stream" + output_formats = ("rgb", "bgr", "yuv") + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + if output_format == "yuv": + return self.decoder.Decode(list(videos), list(frame_indices)) + return self.decoder.DecodeN12ToRGB( + list(videos), + list(frame_indices), + output_format == "bgr", + ) + + +class StreamAsyncAdapter(DecoderTestAdapter): + name = "stream_async" + output_formats = ("rgb", "bgr") + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + videos = list(videos) + frame_indices = list(frame_indices) + as_bgr = output_format == "bgr" + self.decoder.DecodeN12ToRGBAsync(videos, frame_indices, as_bgr) + return self.decoder.DecodeN12ToRGBAsyncGetBuffer( + videos, + frame_indices, + as_bgr, + ) + + +class BatchStreamAsyncAdapter(DecoderTestAdapter): + name = "batch_stream_async" + output_formats = ("rgb", "bgr") + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + videos = list(videos) + frame_indices_2d = [[frame_index] for frame_index in frame_indices] + as_bgr = output_format == "bgr" + self.decoder.Decode(videos, frame_indices_2d, as_bgr) + frames_2d = self.decoder.GetBuffer(videos, frame_indices_2d, as_bgr) + for video_index, frames in enumerate(frames_2d): + if len(frames) != 1: + raise AssertionError( + f"batch output {video_index} must contain exactly one frame, got {len(frames)}" + ) + return [frames[0] for frames in frames_2d] + + +def _decode_gop_data( + decoder: object, + gop_data: Sequence[object], + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, +) -> List[object]: + if output_format == "yuv": + return decoder.DecodeFromGOPList( + list(gop_data), + list(videos), + list(frame_indices), + ) + return decoder.DecodeFromGOPListRGB( + list(gop_data), + list(videos), + list(frame_indices), + output_format == "bgr", + ) + + +class BatchGopAsyncAdapter(DecoderTestAdapter): + name = "batch_gop_async" + output_formats = ("rgb", "bgr", "yuv") + + def __init__(self, demuxer: object, decoder: object) -> None: + super().__init__(decoder) + self.demuxer = demuxer + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + videos = list(videos) + frame_indices = list(frame_indices) + gop_list = self.demuxer.GetGOPList( + videos, + frame_indices, + useGOPCache=False, + ) + gop_data_2d = [[bundle[0]] for bundle in gop_list] + frame_indices_2d = [[frame_index] for frame_index in frame_indices] + + if output_format == "yuv": + self.decoder.DecodeFromGOPList( + gop_data_2d, + videos, + frame_indices_2d, + ) + frames_2d = self.decoder.DecodeFromGOPListGetBuffer( + videos, + frame_indices_2d, + ) + else: + as_bgr = output_format == "bgr" + self.decoder.DecodeFromGOPListRGB( + gop_data_2d, + videos, + frame_indices_2d, + as_bgr, + ) + frames_2d = self.decoder.DecodeFromGOPListRGBGetBuffer( + videos, + frame_indices_2d, + as_bgr, + ) + + for video_index, frames in enumerate(frames_2d): + if len(frames) != 1: + raise AssertionError( + f"batch output {video_index} must contain exactly one frame, got {len(frames)}" + ) + return [frames[0] for frames in frames_2d] + + def close(self) -> None: + self.demuxer = None + super().close() + + +class GopListAdapter(DecoderTestAdapter): + name = "gop_list" + output_formats = ("rgb", "bgr", "yuv") + + def __init__(self, demuxer: object, decoder: object) -> None: + super().__init__(decoder) + self.demuxer = demuxer + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + gop_list = self.demuxer.GetGOPList( + list(videos), + list(frame_indices), + useGOPCache=False, + ) + gop_data = [bundle[0] for bundle in gop_list] + return _decode_gop_data( + self.decoder, + gop_data, + videos, + frame_indices, + output_format, + ) + + def close(self) -> None: + self.demuxer = None + super().close() + + +class GopListFastInitAdapter(DecoderTestAdapter): + name = "gop_list_fast_init" + output_formats = ("rgb", "bgr", "yuv") + + def __init__( + self, + demuxer: object, + decoder: object, + get_fast_init_info: Callable[[Sequence[str]], object], + ) -> None: + super().__init__(decoder) + self.demuxer = demuxer + self.get_fast_init_info = get_fast_init_info + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + videos = list(videos) + frame_indices = list(frame_indices) + gop_list = self.demuxer.GetGOPList( + videos, + frame_indices, + fastStreamInfos=self.get_fast_init_info(videos), + useGOPCache=False, + ) + return _decode_gop_data( + self.decoder, + [bundle[0] for bundle in gop_list], + videos, + frame_indices, + output_format, + ) + + def close(self) -> None: + self.demuxer = None + super().close() + + +class GroupAdapter(DecoderTestAdapter): + name = "group" + output_formats = ("rgb", "bgr") + + def __init__(self, demuxer: object, decoder: object) -> None: + super().__init__(decoder) + self.demuxer = demuxer + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + if len(videos) != len(frame_indices): + raise ValueError("videos and frame_indices must have the same length") + requests = [ + {"filepath": video, "frame_ids": [frame_index]} + for video, frame_index in zip(videos, frame_indices) + ] + groups = self.demuxer.GetGOPGroups(requests) + decoded_groups = self.decoder.DecodeFromGOPGroupsRGB( + groups, + output_format == "bgr", + ) + + frames = [None] * len(videos) + for group in decoded_groups: + source_index = group["source_index"] + for frame, positions in zip(group["frames"], group["frame_positions"]): + for position in positions: + if position != 0: + raise AssertionError( + f"source {source_index} returned unexpected frame position {position}" + ) + if frames[source_index] is not None: + raise AssertionError(f"source {source_index} returned more than one frame") + frames[source_index] = frame + + if any(frame is None for frame in frames): + raise AssertionError("group API did not return one frame for every input video") + return frames + + def close(self) -> None: + self.demuxer = None + super().close() + + +class GopFileAdapter(DecoderTestAdapter): + name = "gop_file" + output_formats = ("rgb", "bgr", "yuv") + + def __init__( + self, + demuxer: object, + decoder: object, + save_gop_to_file: Callable[[object, str], object], + ) -> None: + super().__init__(decoder) + self.demuxer = demuxer + self.save_gop_to_file = save_gop_to_file + + def decode( + self, + videos: Sequence[str], + frame_indices: Sequence[int], + output_format: OutputFormat, + ) -> List[object]: + gop_list = self.demuxer.GetGOPList( + list(videos), + list(frame_indices), + useGOPCache=False, + ) + with TemporaryDirectory(prefix="decoder-common-gop-") as temp_dir: + gop_paths = [] + for index, (gop_data, _, _) in enumerate(gop_list): + gop_path = Path(temp_dir) / f"gop-{index}.bin" + self.save_gop_to_file(gop_data, str(gop_path)) + gop_paths.append(str(gop_path)) + loaded_gop_data = self.decoder.LoadGopsToList(gop_paths) + return _decode_gop_data( + self.decoder, + loaded_gop_data, + videos, + frame_indices, + output_format, + ) + + def close(self) -> None: + self.demuxer = None + super().close() + + +@dataclass(frozen=True) +class AdapterFactory: + name: str + output_formats: Tuple[OutputFormat, ...] + create: Callable[[], DecoderTestAdapter] + + +def adapter_factories(nvc: object) -> List[AdapterFactory]: + """Create fresh decoders for each test without adding a registry system.""" + + max_files = 8 + return [ + AdapterFactory( + RandomAdapter.name, + RandomAdapter.output_formats, + lambda: RandomAdapter(nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0)), + ), + AdapterFactory( + RandomFastInitAdapter.name, + RandomFastInitAdapter.output_formats, + lambda: RandomFastInitAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.GetFastInitInfo, + ), + ), + AdapterFactory( + StreamAdapter.name, + StreamAdapter.output_formats, + lambda: StreamAdapter(nvc.CreateSampleReader(num_of_set=1, num_of_file=max_files, iGpu=0)), + ), + AdapterFactory( + StreamAsyncAdapter.name, + StreamAsyncAdapter.output_formats, + lambda: StreamAsyncAdapter(nvc.CreateSampleReader(num_of_set=1, num_of_file=max_files, iGpu=0)), + ), + AdapterFactory( + BatchStreamAsyncAdapter.name, + BatchStreamAsyncAdapter.output_formats, + lambda: BatchStreamAsyncAdapter( + nvc.CreateBatchAsyncStreamReader( + num_of_set=1, + num_of_file=max_files, + max_frames_per_decode_call=1, + iGpu=0, + ) + ), + ), + AdapterFactory( + BatchGopAsyncAdapter.name, + BatchGopAsyncAdapter.output_formats, + lambda: BatchGopAsyncAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.CreateBatchAsyncGopDecoder( + maxfiles=max_files, + max_frames_per_decode_call=1, + iGpu=0, + ), + ), + ), + AdapterFactory( + GopListAdapter.name, + GopListAdapter.output_formats, + lambda: GopListAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + ), + ), + AdapterFactory( + GopListFastInitAdapter.name, + GopListFastInitAdapter.output_formats, + lambda: GopListFastInitAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.GetFastInitInfo, + ), + ), + AdapterFactory( + GroupAdapter.name, + GroupAdapter.output_formats, + lambda: GroupAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + ), + ), + AdapterFactory( + GopFileAdapter.name, + GopFileAdapter.output_formats, + lambda: GopFileAdapter( + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.CreateGopDecoder(maxfiles=max_files, iGpu=0), + nvc.SaveGopToFile, + ), + ), + ] diff --git a/packages/on_demand_video_decoder/tests/common/cases.py b/packages/on_demand_video_decoder/tests/common/cases.py new file mode 100644 index 00000000..20f6bdcc --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/cases.py @@ -0,0 +1,182 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Deterministic requests shared by all file-and-frame decoder APIs.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, List, Tuple + +import torch + +from common.model import DecodeCase, InvalidDecodeCase, ResourceDecodeCase + +DATA_DIR = Path(__file__).resolve().parents[2] / "data" +SAMPLE_CLIP_DIR = DATA_DIR / "sample_clip" +PIX_FMT_VARIANTS_DIR = DATA_DIR / "pix_fmt_variants" +OPEN_GOP_VARIANTS_DIR = DATA_DIR / "open_gop_variant" +TEMPORAL_VARIANTS_DIR = DATA_DIR / "temporal_variants" + + +def sample_videos() -> Tuple[str, ...]: + videos = tuple(str(path) for path in sorted(SAMPLE_CLIP_DIR.glob("*.mp4"))) + if len(videos) < 5: + raise RuntimeError(f"at least five common test videos are required in {SAMPLE_CLIP_DIR}") + return videos + + +def common_cases() -> List[DecodeCase]: + videos = sample_videos() + return [ + DecodeCase("single_first", (videos[0],), (0,)), + DecodeCase("multi_video_mixed", videos[:3], (0, 37, 200)), + DecodeCase( + "same_video_unsorted_duplicate", + (videos[0], videos[0], videos[0], videos[0]), + (60, 0, 60, 29), + ), + DecodeCase( + "boundaries", + (videos[0], videos[0], videos[0]), + (0, 100, 200), + ), + ] + + +def resource_cases() -> Tuple[ResourceDecodeCase, ...]: + """Committed codec fixtures that each need one decode smoke path.""" + + cases = ( + ResourceDecodeCase( + "h264_avc1", + str(PIX_FMT_VARIANTS_DIR / "h264_avc1_yuv420p.mp4"), + 33, + "rgb", + "RGB", + torch.uint8, + ((256, 256, 3),), + ), + ResourceDecodeCase( + "hevc_hev1_10bit", + str(PIX_FMT_VARIANTS_DIR / "hevc_hev1_yuv420p10le.mp4"), + 33, + "yuv", + "P016", + torch.uint16, + ((256, 256), (128, 128, 2)), + ), + ResourceDecodeCase( + "hevc_hvc1_10bit", + str(PIX_FMT_VARIANTS_DIR / "hevc_hvc1_yuv420p10le.mp4"), + 33, + "yuv", + "P016", + torch.uint16, + ((256, 256), (128, 128, 2)), + ), + ResourceDecodeCase( + "vfr_h264", + str(TEMPORAL_VARIANTS_DIR / "vfr_h264_yuv420p.mp4"), + 33, + "rgb", + "RGB", + torch.uint8, + ((256, 256, 3),), + ), + # Display frame 39 is a leading RASL picture associated with CRA 40. + ResourceDecodeCase( + "open_gop", + str(OPEN_GOP_VARIANTS_DIR / "moving_shape_open_gop_h265.mp4"), + 39, + "rgb", + "RGB", + torch.uint8, + ((256, 256, 3),), + ), + ) + for case in cases: + if not Path(case.video).is_file(): + raise RuntimeError(f"common decode resource is missing: {case.video}") + return cases + + +INVALID_CASE_NAMES = ( + "too_few_frame_indices", + "too_many_frame_indices", + "missing_file", + "empty_file", + "non_video_file", + "truncated_video", + "negative_frame_index", + "out_of_range_frame_index", + "none_frame_index", + "string_frame_index", + "float_frame_index", + "too_many_inputs", +) + + +def _insert_middle(values: Tuple[Any, ...], invalid_value: Any) -> Tuple[Any, ...]: + return values[:2] + (invalid_value,) + values[2:] + + +def invalid_case(name: str, temp_dir: Path) -> InvalidDecodeCase: + """Build one malformed request without requiring external media tools.""" + + videos = sample_videos() + valid_videos = videos[:4] + valid_frames = (0, 30, 60, 90, 120) + + if name == "too_few_frame_indices": + return InvalidDecodeCase(name, valid_videos, valid_frames[:3]) + if name == "too_many_frame_indices": + return InvalidDecodeCase(name, valid_videos, valid_frames) + + if name in {"missing_file", "empty_file", "non_video_file", "truncated_video"}: + invalid_path = temp_dir / f"{name}.mp4" + if name == "empty_file": + invalid_path.write_bytes(b"") + elif name == "non_video_file": + invalid_path.write_bytes(b"not a video\n" * 8) + elif name == "truncated_video": + invalid_path.write_bytes(Path(videos[0]).read_bytes()[:64]) + return InvalidDecodeCase( + name, + _insert_middle(valid_videos, str(invalid_path)), + valid_frames, + ) + + invalid_frame_indices = { + "negative_frame_index": -1, + "out_of_range_frame_index": 999999, + "none_frame_index": None, + "string_frame_index": "60", + "float_frame_index": 60.5, + } + if name in invalid_frame_indices: + return InvalidDecodeCase( + name, + videos[:5], + _insert_middle((0, 30, 90, 120), invalid_frame_indices[name]), + ) + + if name == "too_many_inputs": + return InvalidDecodeCase( + name, + tuple(videos[index % len(videos)] for index in range(9)), + tuple(index * 10 for index in range(9)), + ) + + raise ValueError(f"unknown invalid decode case: {name}") diff --git a/packages/on_demand_video_decoder/tests/common/checker.py b/packages/on_demand_video_decoder/tests/common/checker.py new file mode 100644 index 00000000..c70879b8 --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/checker.py @@ -0,0 +1,33 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Extension point for a future independent binary correctness checker.""" + +from __future__ import annotations + +from typing import Protocol, Sequence + +from common.adapters import OutputFormat +from common.model import CanonicalFrame, DecodeCase + + +class BinaryCorrectnessChecker(Protocol): + """Optional checker supplied by an environment with an approved reference.""" + + def validate( + self, + case: DecodeCase, + output_format: OutputFormat, + frames: Sequence[CanonicalFrame], + ) -> None: ... diff --git a/packages/on_demand_video_decoder/tests/common/conftest.py b/packages/on_demand_video_decoder/tests/common/conftest.py new file mode 100644 index 00000000..7ca257de --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/conftest.py @@ -0,0 +1,106 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Fixtures for the common decoder contract.""" + +from __future__ import annotations + +from typing import Iterator, Optional, Tuple + +import pytest + +import accvlab.on_demand_video_decoder as nvc +from common.adapters import DecoderTestAdapter, OutputFormat, adapter_factories +from common.cases import INVALID_CASE_NAMES, common_cases, invalid_case, resource_cases +from common.checker import BinaryCorrectnessChecker + +_ADAPTER_FACTORIES = adapter_factories(nvc) +_ADAPTER_INPUTS = tuple( + (factory, output_format) for factory in _ADAPTER_FACTORIES for output_format in factory.output_formats +) +_INVALID_ADAPTER_INPUTS = tuple( + (factory, output_format) + for factory in _ADAPTER_FACTORIES + for output_format in factory.output_formats + if output_format != "bgr" +) +_COMMON_CASES = common_cases() +_RESOURCE_CASES = resource_cases() +_RESOURCE_ADAPTER_INPUTS = tuple( + (factory, resource_case) + for resource_case in _RESOURCE_CASES + for factory in _ADAPTER_FACTORIES + if resource_case.preferred_output_format in factory.output_formats +) + + +@pytest.fixture( + params=_ADAPTER_INPUTS, + ids=lambda value: f"{value[0].name}-{value[1]}", +) +def decoder_input(request) -> Iterator[Tuple[DecoderTestAdapter, OutputFormat]]: + factory, output_format = request.param + adapter = factory.create() + try: + yield adapter, output_format + finally: + adapter.close() + + +@pytest.fixture(params=_COMMON_CASES, ids=lambda case: case.name) +def decode_case(request): + return request.param + + +@pytest.fixture( + params=_INVALID_ADAPTER_INPUTS, + ids=lambda value: f"{value[0].name}-{value[1]}", +) +def invalid_decoder_input(request) -> Iterator[Tuple[DecoderTestAdapter, OutputFormat]]: + factory, output_format = request.param + adapter = factory.create() + try: + yield adapter, output_format + finally: + adapter.close() + + +@pytest.fixture(params=INVALID_CASE_NAMES) +def invalid_decode_case(request, tmp_path): + return invalid_case(request.param, tmp_path) + + +@pytest.fixture( + params=_RESOURCE_ADAPTER_INPUTS, + ids=lambda value: (f"{value[0].name}-{value[1].name}-{value[1].preferred_output_format}"), +) +def resource_decoder_input(request): + factory, resource_case = request.param + adapter = factory.create() + try: + yield adapter, resource_case + finally: + adapter.close() + + +@pytest.fixture(params=_RESOURCE_CASES, ids=lambda case: case.name) +def resource_case(request): + return request.param + + +@pytest.fixture +def binary_correctness_checker() -> Optional[BinaryCorrectnessChecker]: + """Reserved for an approved binary reference checker; intentionally unset.""" + + return None diff --git a/packages/on_demand_video_decoder/tests/common/model.py b/packages/on_demand_video_decoder/tests/common/model.py new file mode 100644 index 00000000..0dfc4d25 --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/model.py @@ -0,0 +1,110 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Small data model used by the common decoder tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Literal, Tuple + +import torch + + +@dataclass(frozen=True) +class DecodeCase: + """A common decode request: one frame index for every video entry.""" + + name: str + videos: Tuple[str, ...] + frame_indices: Tuple[int, ...] + + def __post_init__(self) -> None: + if len(self.videos) != len(self.frame_indices): + raise ValueError("videos and frame_indices must have the same length") + if not self.videos: + raise ValueError("a common decode case must not be empty") + + +@dataclass(frozen=True) +class InvalidDecodeCase: + """A malformed request that must reach the adapter unchanged.""" + + name: str + videos: Tuple[Any, ...] + frame_indices: Tuple[Any, ...] + + +@dataclass(frozen=True) +class ResourceDecodeCase: + """One committed fixture and its exact normalized output contract.""" + + name: str + video: str + frame_index: int + preferred_output_format: Literal["rgb", "yuv"] + expected_format: str + expected_dtype: torch.dtype + expected_plane_shapes: Tuple[Tuple[int, ...], ...] + + +@dataclass(frozen=True) +class OutputContract: + """Properties shared by every frame returned through one adapter.""" + + kind: str + dtypes: Tuple[torch.dtype, ...] + device_type: str = "cuda" + + +@dataclass(frozen=True) +class CanonicalFrame: + """Owned tensors used to compare results from otherwise different APIs.""" + + format: str + planes: Tuple[torch.Tensor, ...] + width: int + height: int + + @property + def dtype(self) -> torch.dtype: + return self.planes[0].dtype + + @property + def device(self) -> torch.device: + if not self.planes: + raise AssertionError("a canonical frame must expose at least one plane") + devices = {plane.device for plane in self.planes} + if len(devices) != 1: + device_names = sorted(str(device) for device in devices) + raise AssertionError(f"all planes must be on the same device, got {device_names}") + return self.planes[0].device + + @property + def device_type(self) -> str: + return self.device.type + + +def assert_frames_equal(actual: CanonicalFrame, expected: CanonicalFrame) -> None: + """Assert exact equality and provide a useful plane-level failure.""" + + assert actual.format == expected.format + assert actual.width == expected.width + assert actual.height == expected.height + assert len(actual.planes) == len(expected.planes) + for plane_index, (actual_plane, expected_plane) in enumerate(zip(actual.planes, expected.planes)): + assert actual_plane.shape == expected_plane.shape, f"plane {plane_index} shape differs" + assert actual_plane.dtype == expected_plane.dtype, f"plane {plane_index} dtype differs" + assert actual_plane.device == expected_plane.device, f"plane {plane_index} device differs" + assert torch.equal(actual_plane, expected_plane), f"plane {plane_index} pixels differ" diff --git a/packages/on_demand_video_decoder/tests/common/test_decode_common.py b/packages/on_demand_video_decoder/tests/common/test_decode_common.py new file mode 100644 index 00000000..4bcf5e41 --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/test_decode_common.py @@ -0,0 +1,125 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Minimum decode behavior shared by file-and-frame decoder APIs.""" + +from __future__ import annotations + +from typing import List, Sequence, Tuple + +import torch + +from common.adapters import DecoderTestAdapter, OutputFormat +from common.cases import sample_videos +from common.checker import BinaryCorrectnessChecker +from common.model import CanonicalFrame, DecodeCase, assert_frames_equal + +DecoderInput = Tuple[DecoderTestAdapter, OutputFormat] + + +def _decode_and_normalize( + adapter: DecoderTestAdapter, + output_format: OutputFormat, + videos: Sequence[str], + frame_indices: Sequence[int], +) -> List[CanonicalFrame]: + frames = adapter.decode(videos, frame_indices, output_format) + assert frames is not None + assert len(frames) == len(videos) + assert all(frame is not None for frame in frames) + return [adapter.normalize(frame, output_format) for frame in frames] + + +def _assert_output_contract( + adapter: DecoderTestAdapter, + output_format: OutputFormat, + frame: CanonicalFrame, +) -> None: + contract = adapter.output_contract(output_format) + assert frame.width > 0 + assert frame.height > 0 + assert frame.planes + assert frame.dtype in contract.dtypes + assert frame.device_type == contract.device_type + + if contract.kind in {"RGB", "BGR"}: + assert frame.format == contract.kind + assert len(frame.planes) == 1 + assert frame.planes[0].shape == (frame.height, frame.width, 3) + else: + assert frame.format in {"NV12", "P016", "YUV444", "YUV444_16BIT"} + if frame.format in {"NV12", "P016"}: + assert len(frame.planes) == 2 + assert frame.planes[0].shape == (frame.height, frame.width) + assert frame.planes[1].shape == (frame.height // 2, frame.width // 2, 2) + else: + assert len(frame.planes) == 3 + assert all(plane.shape == (frame.height, frame.width) for plane in frame.planes) + + +def test_common_decode_contract( + decoder_input: DecoderInput, + decode_case: DecodeCase, + binary_correctness_checker: BinaryCorrectnessChecker, +) -> None: + adapter, output_format = decoder_input + frames = _decode_and_normalize( + adapter, + output_format, + decode_case.videos, + decode_case.frame_indices, + ) + for frame in frames: + _assert_output_contract(adapter, output_format, frame) + + if decode_case.name == "same_video_unsorted_duplicate": + assert_frames_equal(frames[0], frames[2]) + + if binary_correctness_checker is not None: + binary_correctness_checker.validate(decode_case, output_format, frames) + + +def test_decode_is_deterministic(decoder_input: DecoderInput) -> None: + adapter, output_format = decoder_input + videos = sample_videos()[:3] + frame_indices = (0, 37, 100) + first = _decode_and_normalize(adapter, output_format, videos, frame_indices) + second = _decode_and_normalize(adapter, output_format, videos, frame_indices) + for actual, expected in zip(second, first): + assert_frames_equal(actual, expected) + + +def test_request_order_is_preserved(decoder_input: DecoderInput) -> None: + adapter, output_format = decoder_input + video = sample_videos()[0] + videos = (video, video, video) + ordered = _decode_and_normalize(adapter, output_format, videos, (0, 30, 60)) + reordered = _decode_and_normalize(adapter, output_format, videos, (60, 0, 30)) + + assert_frames_equal(reordered[0], ordered[2]) + assert_frames_equal(reordered[1], ordered[0]) + assert_frames_equal(reordered[2], ordered[1]) + + +def test_same_frame_is_batch_independent(decoder_input: DecoderInput) -> None: + adapter, output_format = decoder_input + videos = sample_videos() + single = _decode_and_normalize(adapter, output_format, (videos[0],), (30,))[0] + mixed = _decode_and_normalize( + adapter, + output_format, + (videos[1], videos[0], videos[2]), + (10, 30, 50), + )[1] + assert_frames_equal(mixed, single) diff --git a/packages/on_demand_video_decoder/tests/common/test_decode_invalid.py b/packages/on_demand_video_decoder/tests/common/test_decode_invalid.py new file mode 100644 index 00000000..377519ae --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/test_decode_invalid.py @@ -0,0 +1,71 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Invalid requests shared by all file-and-frame decoder APIs.""" + +from __future__ import annotations + +from typing import Tuple + +import pytest + +from common.adapters import DecoderTestAdapter, OutputFormat +from common.cases import sample_videos +from common.model import InvalidDecodeCase + +DecoderInput = Tuple[DecoderTestAdapter, OutputFormat] +EXPECTED_EXCEPTIONS = (RuntimeError, ValueError, TypeError) + + +def _assert_recovery(adapter: DecoderTestAdapter, output_format: OutputFormat) -> None: + frames = adapter.decode((sample_videos()[0],), (0,), output_format) + assert frames is not None + assert len(frames) == 1 + assert frames[0] is not None + recovered = adapter.normalize(frames[0], output_format) + assert recovered.width > 0 + assert recovered.height > 0 + assert recovered.planes + + +def test_empty_request_is_safe_and_recovers( + invalid_decoder_input: DecoderInput, +) -> None: + adapter, output_format = invalid_decoder_input + + try: + frames = adapter.decode((), (), output_format) + except EXPECTED_EXCEPTIONS as error: + assert str(error).strip(), "decode error must contain an actionable message" + else: + assert frames == [] + + _assert_recovery(adapter, output_format) + + +def test_invalid_request_fails_atomically_and_recovers( + invalid_decoder_input: DecoderInput, + invalid_decode_case: InvalidDecodeCase, +) -> None: + adapter, output_format = invalid_decoder_input + + with pytest.raises(EXPECTED_EXCEPTIONS) as error: + adapter.decode( + invalid_decode_case.videos, + invalid_decode_case.frame_indices, + output_format, + ) + assert str(error.value).strip(), "decode error must contain an actionable message" + + _assert_recovery(adapter, output_format) diff --git a/packages/on_demand_video_decoder/tests/common/test_decode_resources.py b/packages/on_demand_video_decoder/tests/common/test_decode_resources.py new file mode 100644 index 00000000..5f1c668a --- /dev/null +++ b/packages/on_demand_video_decoder/tests/common/test_decode_resources.py @@ -0,0 +1,80 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +"""Format contracts and cross-API equality for committed video fixtures.""" + +from __future__ import annotations + +from fractions import Fraction +from typing import Tuple + +import accvlab.on_demand_video_decoder as nvc +from common.adapters import DecoderTestAdapter, adapter_factories +from common.cases import resource_cases +from common.model import CanonicalFrame, ResourceDecodeCase, assert_frames_equal + +ResourceDecoderInput = Tuple[DecoderTestAdapter, ResourceDecodeCase] + + +def _decode_one(adapter: DecoderTestAdapter, case: ResourceDecodeCase) -> CanonicalFrame: + frames = adapter.decode( + (case.video,), + (case.frame_index,), + case.preferred_output_format, + ) + + assert frames is not None + assert len(frames) == 1 + assert frames[0] is not None + return adapter.normalize(frames[0], case.preferred_output_format) + + +def test_decode_committed_resource(resource_decoder_input: ResourceDecoderInput) -> None: + adapter, case = resource_decoder_input + frame = _decode_one(adapter, case) + + assert frame.format == case.expected_format + assert frame.dtype == case.expected_dtype + assert tuple(tuple(plane.shape) for plane in frame.planes) == case.expected_plane_shapes + + +def test_resource_adapters_are_pixel_identical(resource_case: ResourceDecodeCase) -> None: + baseline = None + compared_adapters = 0 + for factory in adapter_factories(nvc): + if resource_case.preferred_output_format not in factory.output_formats: + continue + adapter = factory.create() + try: + actual = _decode_one(adapter, resource_case) + finally: + adapter.close() + if baseline is None: + baseline = actual + else: + assert_frames_equal(actual, baseline) + compared_adapters += 1 + + assert compared_adapters >= 2 + + +def test_vfr_fixture_reports_variable_timing() -> None: + case = next(case for case in resource_cases() if case.name == "vfr_h264") + info = nvc.GetFastInitInfo([case.video])[0] + average_rate = Fraction(info.avg_frame_rate_num, info.avg_frame_rate_den) + real_rate = Fraction(info.r_frame_rate_num, info.r_frame_rate_den) + + assert (info.width, info.height) == (256, 256) + assert info.duration > 0 + assert average_rate != real_rate diff --git a/packages/on_demand_video_decoder/tests/test_batch_async_gop_decoder.py b/packages/on_demand_video_decoder/tests/test_batch_async_gop_decoder.py index 6e78032e..db115333 100644 --- a/packages/on_demand_video_decoder/tests/test_batch_async_gop_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_batch_async_gop_decoder.py @@ -156,25 +156,6 @@ def test_construct_rejects_invalid_args(kwargs): # =========================================================================== -def test_validate_empty_filepaths(): - """Empty filepaths list is rejected before any decode work.""" - dec = _make_async_dec() - with pytest.raises(RuntimeError, match="filepaths must not be empty"): - dec.DecodeFromGOPListRGB([], [], [], False) - - -def test_validate_too_many_files(): - """More filepaths than num_of_file is rejected.""" - files = _sample_files() - dec = _make_async_dec(V=1) # only room for 1 video - if len(files) <= 1: - pytest.skip("need at least 2 sample videos") - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, [[0]] * len(files)) - with pytest.raises(RuntimeError, match="exceeds maxfiles"): - dec.DecodeFromGOPListRGB(numpy_datas, files, [[0]] * len(files), False) - - def test_validate_too_many_frames(): """More frames than max_frames_per_decode_call is rejected.""" files = _sample_files() @@ -369,27 +350,6 @@ def test_rgb_output_shape(): assert len(out[v]) == F, f"out[{v}] has {len(out[v])} frames, expected {F}" -def test_rgb_output_dtype_and_device(): - """Each RGBFrame converts to a uint8 (H, W, 3) CUDA tensor.""" - files = _sample_files() - V = len(files) - frame_ids_2d = [[0]] * V - - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, frame_ids_2d) - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPListRGB(numpy_datas, files, frame_ids_2d, False) - out = dec.DecodeFromGOPListRGBGetBuffer(files, frame_ids_2d, False) - - for v in range(V): - t = torch.as_tensor(out[v][0], device="cuda") - assert t.dtype == torch.uint8, f"v={v}: dtype={t.dtype}" - assert t.ndim == 3, f"v={v}: ndim={t.ndim}" - assert t.shape[-1] == 3, f"v={v}: shape={tuple(t.shape)}" - assert t.device.type == "cuda" - - def test_rgb_single_video_multi_frame(): """V=1, F=4 is supported.""" files = _sample_files() @@ -405,39 +365,6 @@ def test_rgb_single_video_multi_frame(): assert len(out) == 1 and len(out[0]) == len(_SAME_GOP_FRAMES) -def test_rgb_single_frame_per_video(): - """F=1 is supported (one frame per video).""" - files = _sample_files() - V = len(files) - frame_ids_2d = [[0]] * V - - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, frame_ids_2d) - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPListRGB(numpy_datas, files, frame_ids_2d, False) - out = dec.DecodeFromGOPListRGBGetBuffer(files, frame_ids_2d, False) - assert len(out) == V - for v in range(V): - assert len(out[v]) == 1 - - -@pytest.mark.parametrize("as_bgr", [False, True]) -def test_rgb_as_bgr_flag(as_bgr): - """as_bgr=True and as_bgr=False both complete without error.""" - files = _sample_files() - V = len(files) - frame_ids_2d = [[0]] * V - - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, frame_ids_2d) - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPListRGB(numpy_datas, files, frame_ids_2d, as_bgr) - out = dec.DecodeFromGOPListRGBGetBuffer(files, frame_ids_2d, as_bgr) - assert out[0][0] is not None - - # =========================================================================== # Section E — functional YUV decode (shape, views, format) # =========================================================================== @@ -462,58 +389,6 @@ def test_yuv_output_shape(): assert len(out[v]) == F -def test_yuv_frame_has_views(): - """Each DecodedFrameExt has the correct number of CAIMemoryView plane views.""" - files = _sample_files() - V = len(files) - frame_ids_2d = [[0]] * V - - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, frame_ids_2d) - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPList(numpy_datas, files, frame_ids_2d) - out = dec.DecodeFromGOPListGetBuffer(files, frame_ids_2d) - - # Raw Pixel_Format enum values (from PyCAIMemoryView.hpp): - # NV12=3 → 2 planes (Y, UV interleaved) - # YUV444=4 → 3 planes (Y, U, V separate) - # P016=5 → 2 planes (Y, UV interleaved 16-bit) - # YUV444_16Bit=6 → 3 planes (Y, U, V separate 16-bit) - EXPECTED_VIEWS = {3: 2, 4: 3, 5: 2, 6: 3} - - for v in range(V): - frame = out[v][0] - views = frame.cuda() - fmt = frame.format - expected = EXPECTED_VIEWS.get(fmt, 2) - assert ( - len(views) == expected - ), f"v={v}: format={fmt}, expected {expected} plane views, got {len(views)}" - # Y-plane shape: (H, W, 1) - assert len(views[0].shape) == 3 - assert views[0].shape[2] == 1 - - -def test_yuv_frame_format_set(): - """DecodedFrameExt.format is not UNDEFINED.""" - files = _sample_files() - V = len(files) - frame_ids_2d = [[0]] * V - - gop_dec = _make_gop_dec() - numpy_datas = _build_numpy_datas(gop_dec, files, frame_ids_2d) - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPList(numpy_datas, files, frame_ids_2d) - out = dec.DecodeFromGOPListGetBuffer(files, frame_ids_2d) - - for v in range(V): - assert ( - out[v][0].format != nvc.Pixel_Format_UNDEFINED if hasattr(nvc, "Pixel_Format_UNDEFINED") else True - ) - - # =========================================================================== # Section F — precision: 2D RGB output must bit-match 1D GOP reference # =========================================================================== @@ -804,28 +679,6 @@ def test_resubmit_only_latest_result_retrievable(): dec2.DecodeFromGOPListRGBGetBuffer(files, frame_ids_a, False) -def test_invalid_frame_id_propagates_exception(): - """Out-of-range frame_id (beyond video length) is rethrown at GetBuffer. - - The bundle for frame 0 does not cover frame 999999, so the worker throws - "no serialized GOP bundle covers frame 999999". This tests that async - worker errors are stored and re-raised at GetBuffer time. - """ - files = _sample_files() - V = len(files) - # Frame id 999999 is almost certainly out of range for any test video. - frame_ids_2d = [[999999]] * V - gop_dec = _make_gop_dec() - # Bundle obtained for a valid frame; the invalid frame_id causes a decode error. - gop_data = _get_numpy_data(gop_dec, files[0], 0) - numpy_datas = [[gop_data]] * V - - dec = _make_async_dec(V=V, F=1) - dec.DecodeFromGOPListRGB(numpy_datas, files, frame_ids_2d, False) - with pytest.raises(RuntimeError): - dec.DecodeFromGOPListRGBGetBuffer(files, frame_ids_2d, False) - - def test_wrong_gop_bundle_propagates_async_error(): """Valid frame_id but bundle from the wrong GOP is rethrown at GetBuffer. diff --git a/packages/on_demand_video_decoder/tests/test_batch_async_stream_decoder.py b/packages/on_demand_video_decoder/tests/test_batch_async_stream_decoder.py index 1d67a013..13cf5a85 100644 --- a/packages/on_demand_video_decoder/tests/test_batch_async_stream_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_batch_async_stream_decoder.py @@ -125,32 +125,6 @@ def test_rejects_direct_construction(): # =========================================================================== -def test_validate_size_mismatch(): - """filepaths.size() != frame_ids_2d.size() rejected at entry.""" - files = _select_sample_videos() - r = _make_reader(V=len(files)) - bad_frame_ids = [[0]] # length 1, but files has more entries - with pytest.raises(RuntimeError, match=r"filepaths\.size\(\).*frame_ids_2d\.size\(\)"): - r.Decode(files, bad_frame_ids, False) - - -def test_validate_empty_filepaths(): - """Empty filepaths list is rejected.""" - r = _make_reader() - with pytest.raises(RuntimeError, match=r"filepaths must not be empty"): - r.Decode([], [], False) - - -def test_validate_too_many_files(): - """Exceeding num_of_file is rejected at entry.""" - files = _select_sample_videos() - r = _make_reader(V=1) # num_of_file=1, but we'll pass len(files) > 1 - if len(files) <= 1: - pytest.skip("need at least 2 sample videos for this test") - with pytest.raises(RuntimeError, match=r"exceeds num_of_file"): - r.Decode(files, [[0]] * len(files), False) - - def test_validate_too_many_frames(): """Exceeding max_frames_per_decode_call is rejected at entry.""" files = _select_sample_videos() @@ -217,41 +191,6 @@ def test_decode_basic_2d_shape(): assert len(out[v]) == F, f"out[{v}] inner len should be F={F}, got {len(out[v])}" -def test_decode_basic_2d_dtype_and_device(): - """Each frame is a uint8, 3-channel tensor on CUDA.""" - files = _select_sample_videos() - V = len(files) - F = 2 - frame_ids_2d = [[0, 7]] * V - - r = _make_reader(V=V, F=F) - r.Decode(files, frame_ids_2d, as_bgr=False) - out = r.GetBuffer(files, frame_ids_2d, as_bgr=False) - - for v in range(V): - for f in range(F): - t = torch.as_tensor(out[v][f], device="cuda") - assert t.dtype == torch.uint8, f"out[{v}][{f}].dtype = {t.dtype}" - assert t.ndim == 3, f"out[{v}][{f}].ndim = {t.ndim}" - assert t.shape[-1] == 3, f"out[{v}][{f}].shape = {tuple(t.shape)}" - assert t.device.type == "cuda" - - -def test_decode_single_frame_per_video(): - """F=1 is supported (degenerates to behavior similar to 1D API).""" - files = _select_sample_videos() - V = len(files) - frame_ids_2d = [[0]] * V - - r = _make_reader(V=V, F=1) - r.Decode(files, frame_ids_2d, as_bgr=False) - out = r.GetBuffer(files, frame_ids_2d, as_bgr=False) - - assert len(out) == V - for v in range(V): - assert len(out[v]) == 1 - - def test_decode_single_video_multi_frame(): """V=1 with multiple frames is supported.""" files = _select_sample_videos() @@ -408,17 +347,3 @@ def test_resubmit_keeps_only_latest_result(): r2.Decode(files, frame_ids_b, False) with pytest.raises(RuntimeError, match=r"do not match buffered result"): r2.GetBuffer(files, frame_ids_a, False) - - -def test_invalid_file_propagates_exception(): - """A decode error in the worker (e.g. nonexistent file) is rethrown at Get.""" - files = _select_sample_videos() - V = len(files) - bad_files = list(files) - bad_files[0] = "/__definitely_not_a_real_file__.mp4" - frame_ids_2d = [[0]] * V - - r = _make_reader(V=V) - r.Decode(bad_files, frame_ids_2d, False) - with pytest.raises(RuntimeError): - r.GetBuffer(bad_files, frame_ids_2d, False) diff --git a/packages/on_demand_video_decoder/tests/test_decoder_with_fast_init.py b/packages/on_demand_video_decoder/tests/test_decoder_with_fast_init.py deleted file mode 100644 index fa14b086..00000000 --- a/packages/on_demand_video_decoder/tests/test_decoder_with_fast_init.py +++ /dev/null @@ -1,88 +0,0 @@ -# Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# -# 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 pytest -import sys - -import random - -import utils -import accvlab.on_demand_video_decoder as nvc - - -def test_random_access_fast_init_single(): - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - fast_stream_infos = nvc.GetFastInitInfo(files) - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - gop_decoded = utils.gop_decode_bgr_with_fast_init(nv_gop_dec, files, frames, fast_stream_infos) - assert gop_decoded is not None, f"gop_decoded is None for DecodeN12ToRGB, frames: {frames}" - - -def test_separate_access_fast_init_single(): - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec1 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - nv_gop_dec2 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - fast_stream_infos = nvc.GetFastInitInfo(files) - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - gop_decoded = utils.gop_decode_bgr_ddseparate_with_fast_init( - nv_gop_dec1, nv_gop_dec2, files, frames, fast_stream_infos - ) - assert gop_decoded is not None, f"gop_decoded is None for DecodeN12ToRGB, frames: {frames}" - - -if __name__ == "__main__": - sys.exit(pytest.main([__file__, "-v"])) diff --git a/packages/on_demand_video_decoder/tests/test_pix_fmt_detection.py b/packages/on_demand_video_decoder/tests/test_pix_fmt_detection.py index 7f226fa9..2c44a63e 100644 --- a/packages/on_demand_video_decoder/tests/test_pix_fmt_detection.py +++ b/packages/on_demand_video_decoder/tests/test_pix_fmt_detection.py @@ -21,7 +21,7 @@ 10-bit HEVC stream this mis-sized the GPU output buffer by a factor of 2 and crashed the GOP decode path with `CUDA_ERROR_INVALID_VALUE`. -These tests drive `GetGOPList` + `DecodeFromGOPList` over the matrix of stream +This parametrized test drives `GetGOPList` + `DecodeFromGOPList` over the matrix of stream shapes the package supports on NVDEC (HEVC at both 8-bit and 10-bit with both hev1 and hvc1 sample-entry tags, plus H.264 8-bit), and assert the decoded planes carry the dtype implied by the stream's real bit-depth (uint8 for @@ -44,11 +44,11 @@ # package targets — see the support matrix for codec/bit-depth coverage: # https://developer.nvidia.com/video-encode-decode-support-matrix VARIANTS = [ - ("hevc_hev1_yuv420p.mp4", "hev1", 8, "|u1"), - ("hevc_hev1_yuv420p10le.mp4", "hev1", 10, "|u2"), - ("hevc_hvc1_yuv420p.mp4", "hvc1", 8, "|u1"), - ("hevc_hvc1_yuv420p10le.mp4", "hvc1", 10, "|u2"), - ("h264_avc1_yuv420p.mp4", "avc1", 8, "|u1"), + ("hevc_hev1_yuv420p.mp4", "hev1", 8, 3, "|u1", ((256, 256, 1), (128, 128, 2))), + ("hevc_hev1_yuv420p10le.mp4", "hev1", 10, 5, "|u2", ((256, 256, 1), (128, 128, 2))), + ("hevc_hvc1_yuv420p.mp4", "hvc1", 8, 3, "|u1", ((256, 256, 1), (128, 128, 2))), + ("hevc_hvc1_yuv420p10le.mp4", "hvc1", 10, 5, "|u2", ((256, 256, 1), (128, 128, 2))), + ("h264_avc1_yuv420p.mp4", "avc1", 8, 3, "|u1", ((256, 256, 1), (128, 128, 2))), ] @@ -60,11 +60,18 @@ def _video_path(name): @pytest.mark.parametrize( - "filename, codec_tag, bit_depth, expected_dtype", + "filename, codec_tag, bit_depth, expected_format, expected_dtype, expected_shapes", VARIANTS, ids=[v[0] for v in VARIANTS], ) -def test_decode_from_gop_round_trip(filename, codec_tag, bit_depth, expected_dtype): +def test_decode_from_gop_round_trip( + filename, + codec_tag, + bit_depth, + expected_format, + expected_dtype, + expected_shapes, +): """End-to-end: GetGOPList -> DecodeFromGOPList must produce a plane of the correct dtype for the stream's actual bit-depth.""" path = _video_path(filename) @@ -72,49 +79,50 @@ def test_decode_from_gop_round_trip(filename, codec_tag, bit_depth, expected_dty demuxer = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) decoder = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) - gop_list = demuxer.GetGOPList([path], [0], useGOPCache=True) + frame_id = 33 + gop_list = demuxer.GetGOPList([path], [frame_id], useGOPCache=True) assert gop_list, f"GetGOPList returned empty for {filename}" gop_data, first_ids, gop_lens = gop_list[0] assert gop_data.size > 0, f"GOP data is empty for {filename}" - assert first_ids == [0], f"unexpected first_ids={first_ids} for {filename}" + assert first_ids[0] <= frame_id, f"unexpected first_ids={first_ids} for {filename}" assert gop_lens and gop_lens[0] > 0, f"unexpected gop_lens={gop_lens} for {filename}" + assert frame_id < first_ids[0] + gop_lens[0] - frames = decoder.DecodeFromGOPList([gop_data], [path], [0]) + frames = decoder.DecodeFromGOPList([gop_data], [path], [frame_id]) assert len(frames) == 1, f"expected 1 frame, got {len(frames)} for {filename}" + assert frames[0].format == expected_format planes = frames[0].cuda() - assert len(planes) >= 1, f"no planes returned for {filename}" - - cai = planes[0].__cuda_array_interface__ - assert cai["typestr"] == expected_dtype, ( - f"Y plane dtype mismatch for {filename}: got {cai['typestr']!r}, " - f"expected {expected_dtype!r} for {bit_depth}-bit" - ) + assert tuple(tuple(plane.shape) for plane in planes) == expected_shapes expected_bytes_per_sample = 2 if bit_depth >= 10 else 1 - actual_bytes_per_sample = int(cai["typestr"][-1]) + luma_width = expected_shapes[0][1] + expected_strides = ( + ( + luma_width * expected_bytes_per_sample, + expected_bytes_per_sample, + expected_bytes_per_sample, + ), + ( + luma_width * expected_bytes_per_sample, + 2 * expected_bytes_per_sample, + expected_bytes_per_sample, + ), + ) + actual_strides = tuple(tuple(plane.__cuda_array_interface__["strides"]) for plane in planes) + assert ( + actual_strides == expected_strides + ), f"plane strides mismatch for {filename}: got {actual_strides}, expected {expected_strides}" + + for plane_index, plane in enumerate(planes): + cai = plane.__cuda_array_interface__ + assert cai["typestr"] == expected_dtype, ( + f"plane {plane_index} dtype mismatch for {filename}: got {cai['typestr']!r}, " + f"expected {expected_dtype!r} for {bit_depth}-bit" + ) + + actual_bytes_per_sample = int(planes[0].__cuda_array_interface__["typestr"][-1]) assert actual_bytes_per_sample == expected_bytes_per_sample, ( f"Y plane element size mismatch for {filename}: got " f"{actual_bytes_per_sample}B, expected {expected_bytes_per_sample}B" ) - - -@pytest.mark.parametrize( - "filename, codec_tag, bit_depth, expected_dtype", - VARIANTS, - ids=[v[0] for v in VARIANTS], -) -def test_decode_does_not_raise_invalid_value(filename, codec_tag, bit_depth, expected_dtype): - """Focused regression: the specific failure mode we fixed was - `CUDA_ERROR_INVALID_VALUE` thrown from the GOP decode path because the GPU buffer - was half the size NVDEC writes. Reproduce the exact call sequence the - customer used and assert it does not raise that error.""" - path = _video_path(filename) - demuxer = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) - decoder = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) - - gop_data, _, _ = demuxer.GetGOPList([path], [0], useGOPCache=True)[0] - # No prior RGB call to "prime" the GPU pool — exercise the raw YUV path - # directly, which was the broken path before the SPS fallback. - frames = decoder.DecodeFromGOPList([gop_data], [path], [0]) - assert frames, f"DecodeFromGOPList returned no frames for {filename}" diff --git a/packages/on_demand_video_decoder/tests/test_random_decoder.py b/packages/on_demand_video_decoder/tests/test_random_decoder.py index 910a625b..fa287a16 100644 --- a/packages/on_demand_video_decoder/tests/test_random_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_random_decoder.py @@ -15,9 +15,6 @@ import pytest import sys -import random - -import utils import accvlab.on_demand_video_decoder as nvc @@ -29,29 +26,5 @@ def test_pynvgopdecoder_rejects_direct_construction(): nvc.PyNvGopDecoder(1, 0) -def test_random_access_single(): - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - gop_decoded = utils.gop_decode_bgr(nv_gop_dec, files, frames) - assert gop_decoded is not None, f"gop_decoded is None for DecodeN12ToRGB, frames: {frames}" - - if __name__ == "__main__": sys.exit(pytest.main([__file__, "-v"])) diff --git a/packages/on_demand_video_decoder/tests/test_separate_decoder.py b/packages/on_demand_video_decoder/tests/test_separate_decoder.py index d55fd547..b8842497 100644 --- a/packages/on_demand_video_decoder/tests/test_separate_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_separate_decoder.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -12,262 +12,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -import pytest -import sys - -import torch -import random -import os - -import utils -import accvlab.on_demand_video_decoder as nvc - - -def test_separate_access_from_gop_file_to_list_api_single(): - """ - Test LoadGopsToList + DecodeFromGOPListRGB API combination with file persistence. - - This test validates the workflow: - 1. Extract GOP data using GetGOPList and save to separate files - 2. Load GOP data from files as a list using LoadGopsToList - 3. Decode using DecodeFromGOPListRGB - 4. Compare with OpenCV baseline - """ - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec1 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - nv_gop_dec2 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - num_frames_to_use = len(frames) - - # Step 1: Extract and save GOP data for each video - packet_files = [] - for i in range(len(files)): - ((numpy_data, first_frame_ids, gop_lens),) = nv_gop_dec1.GetGOPList( - files[i : i + 1], frames[i : i + 1] - ) - packet_file = os.path.join("./", f"packets_list_{c:02d}_{i:02d}.bin") - nvc.SaveGopToFile(numpy_data, packet_file) - packet_files.append(packet_file) - - # Verify file was created - assert os.path.exists(packet_file), f"Packet file not created: {packet_file}" - assert os.path.getsize(packet_file) == numpy_data.size, f"File size mismatch for {packet_file}" - - # Step 2: Load GOP data as a list using LoadGopsToList - gop_data_list = nv_gop_dec2.LoadGopsToList(packet_files) - - # Validate the loaded list - assert isinstance( - gop_data_list, list - ), f"LoadGopsToList should return a list, got {type(gop_data_list)}" - assert len(gop_data_list) == len( - packet_files - ), f"LoadGopsToList returned {len(gop_data_list)} items, expected {len(packet_files)}" - - # Step 3: Decode using DecodeFromGOPListRGB - decoded_frames = nv_gop_dec2.DecodeFromGOPListRGB( - gop_data_list, # List of GOP data arrays - files, # List of file paths - frames, # List of frame IDs - as_bgr=True, - ) - - assert decoded_frames is not None and len(decoded_frames) == num_frames_to_use, ( - f"DecodeFromGOPListRGB returned {len(decoded_frames) if decoded_frames else 0} " - f"frames, expected {num_frames_to_use}" - ) - - # Convert to tensor format for comparison - gop_decoded = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - assert len(gop_decoded) == num_frames_to_use - - # Cleanup: Remove temporary packet files - for packet_file in packet_files: - os.remove(packet_file) - - -def test_separate_access_from_multi_packets_on_the_fly(): - """ - Test per-video GetGOPList + DecodeFromGOPListRGB with packets collected on the fly. - - Calls GetGOPList once per video (one file at a time), collects the per-video - numpy arrays, then batch-decodes with DecodeFromGOPListRGB in a single call. - This exercises the code path where demux is interleaved with other work before - a single decode call at the end. - """ - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec1 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - nv_gop_dec2 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - packets_list = [] - for i in range(len(files)): - ((numpy_data, first_frame_ids, gop_lens),) = nv_gop_dec1.GetGOPList( - files[i : i + 1], frames[i : i + 1] - ) - packets_list.append(numpy_data) - - gop_decoded = utils.gop_decode_bgr_ddseparate_from_multi_packets( - nv_gop_dec2, files, frames, packets_list - ) - - assert gop_decoded is not None, f"gop_decoded is None for DecodeN12ToRGB, frames: {frames}" - - -def test_separate_access_gop_list_api(): - """ - Test GetGOPList + DecodeFromGOPListRGB API combination. - - This test validates the GetGOPList workflow by: - 1. Extracting separate GOP bundles for each video using GetGOPList - 2. Extracting gop_data from each bundle (ignoring metadata) - 3. Decoding all videos at once using DecodeFromGOPListRGB - 4. Comparing results with OpenCV baseline - - This combination demonstrates the complete workflow for per-video GOP management. - """ - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - # Stage 1 decoder: Extract per-video GOP data - nv_gop_dec1 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - # Stage 2 decoder: Decode from individual GOP bundles - nv_gop_dec2 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - num_frames_to_use = len(frames) - - # Stage 1: Extract per-video GOP data using GetGOPList - # Returns: [(gop_data1, first_ids1, gop_lens1), (gop_data2, ...), ...] - gop_list = nv_gop_dec1.GetGOPList(files, frames) - - # Validate GetGOPList output - assert gop_list is not None and len(gop_list) == len(files), ( - f"GetGOPList returned invalid data. " - f"Expected {len(files)} bundles, got {len(gop_list) if gop_list else 0}" - ) - - # Validate each bundle structure - for i, bundle in enumerate(gop_list): - assert ( - isinstance(bundle, tuple) and len(bundle) == 3 - ), f"Bundle {i} has invalid structure. Expected tuple of 3 elements, got {type(bundle)}" - gop_data, first_frame_ids, gop_lens = bundle - assert gop_data is not None and len(gop_data) > 0, f"Bundle {i} has empty or None gop_data" - - # Stage 2: Decode all videos at once using DecodeFromGOPListRGB - gop_data_list = [gop_data for gop_data, _, _ in gop_list] - - decoded_frames = nv_gop_dec2.DecodeFromGOPListRGB( - gop_data_list, # List of per-video GOP data from GetGOPList - files, # List of file paths (one per video) - frames, # List of frame IDs (one per video) - as_bgr=True, # Output in BGR format for comparison - ) - - assert decoded_frames is not None and len(decoded_frames) == num_frames_to_use, ( - f"DecodeFromGOPListRGB returned {len(decoded_frames) if decoded_frames else 0} " - f"frames, expected {num_frames_to_use}" - ) - - # Convert to tensor format for comparison - gop_decoded_list = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - assert len(gop_decoded_list) == num_frames_to_use - - -def test_separate_access_gop_list_raw_api(): - """Test GetGOPList + DecodeFromGOPList raw API (YUV output).""" - clip_dir = os.path.join(utils.get_data_dir(), "sample_clip") - files = [os.path.join(clip_dir, name) for name in sorted(os.listdir(clip_dir))[:3]] - frames = [0, 7, 14] - max_num_files_to_use = len(files) - expected_frame_size = 256 * 256 * 3 // 2 - - nv_gop_dec1 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - nv_gop_dec2 = nvc.CreateGopDecoder( - maxfiles=max_num_files_to_use, - iGpu=0, - ) - - gop_list = nv_gop_dec1.GetGOPList(files, frames) - assert gop_list is not None and len(gop_list) == len(files), ( - f"GetGOPList returned invalid data. " - f"Expected {len(files)} bundles, got {len(gop_list) if gop_list else 0}" - ) - - gop_data_list = [gop_data for gop_data, _, _ in gop_list] - decoded_frames = nv_gop_dec2.DecodeFromGOPList(gop_data_list, files, frames) - - assert decoded_frames is not None and len(decoded_frames) == len(files), ( - f"DecodeFromGOPList returned {len(decoded_frames) if decoded_frames else 0} " - f"frames, expected {len(files)}" - ) - - for file_name, frame_id, decoded_frame in zip(files, frames, decoded_frames): - assert decoded_frame.framesize() == expected_frame_size, ( - f"frame size mismatch for {file_name} frame {frame_id}: " - f"{decoded_frame.framesize()} != {expected_frame_size}" - ) - - -if __name__ == "__main__": - sys.exit(pytest.main([__file__, "-v"])) +"""Separate demux/decode workflows are covered by the common decoder tests.""" diff --git a/packages/on_demand_video_decoder/tests/test_stream_asnyc_decoder.py b/packages/on_demand_video_decoder/tests/test_stream_asnyc_decoder.py index 95257983..d75eb3f6 100644 --- a/packages/on_demand_video_decoder/tests/test_stream_asnyc_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_stream_asnyc_decoder.py @@ -15,49 +15,11 @@ import pytest import sys import torch -import random -import time import utils import accvlab.on_demand_video_decoder as nvc -def test_async_decode_basic_flow(): - """ - Test 1.1: Basic async decode flow - Test that async decode works: start async -> wait -> get result - """ - max_num_files_to_use = 6 - path_base = utils.get_data_dir() - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frame_id_list = [0] * len(files) - - # Start async decode - nv_stream_dec.DecodeN12ToRGBAsync(files, frame_id_list, False) - - # Get result (will wait for async decode to complete) - decoded_frames = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_id_list, False) - - # Verify results - assert decoded_frames is not None, "decoded_frames is None" - assert len(decoded_frames) == len(files), f"Expected {len(files)} frames, got {len(decoded_frames)}" - - # Verify each frame is valid RGBFrame - for i, frame in enumerate(decoded_frames): - assert frame is not None, f"Frame {i} is None" - assert hasattr(frame, 'shape'), f"Frame {i} has no shape attribute" - assert hasattr(frame, 'data'), f"Frame {i} has no data attribute" - - def test_async_decode_prefetch_flow(): """ Test 1.2: Prefetch mechanism @@ -250,31 +212,6 @@ def test_async_decode_getbuffer_on_empty_buffer_throws_error(): nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_0, False) -def test_async_decode_error_handling_invalid_file(): - """ - Test 4.1: Error handling - invalid file path - Test that decoding failure propagates correctly - """ - max_num_files_to_use = 6 - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - # Use invalid file path - invalid_files = ["/nonexistent/path/to/video.mp4"] - frame_id_list = [0] - - # Start async decode with invalid file (should fail in background) - nv_stream_dec.DecodeN12ToRGBAsync(invalid_files, frame_id_list, False) - - # GetBuffer should rethrow the exception - with pytest.raises(RuntimeError): - nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(invalid_files, frame_id_list, False) - - def test_async_decode_error_message_is_preserved(): """ Test 4.1b: Error handling - exception message is preserved from async thread @@ -330,34 +267,6 @@ def test_async_decode_error_message_is_preserved(): # This could be improved in the future by wrapping exceptions with file context. -def test_async_decode_error_handling_invalid_frame_id(): - """ - Test 4.2: Error handling - invalid frame_id - Test that invalid frame_id causes exception - """ - max_num_files_to_use = 6 - path_base = utils.get_data_dir() - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - files = utils.select_random_clip(path_base) - assert files is not None - - # Use very large frame_id that doesn't exist - invalid_frame_ids = [999999] * len(files) - - # Start async decode with invalid frame_id - nv_stream_dec.DecodeN12ToRGBAsync(files, invalid_frame_ids, False) - - # GetBuffer should rethrow the exception - with pytest.raises(RuntimeError): - nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, invalid_frame_ids, False) - - def test_async_decode_vs_sync_decode_result_comparison(): """ Test 9.2: Compare async decode results with sync decode results @@ -416,60 +325,6 @@ def test_async_decode_vs_sync_decode_result_comparison(): assert max_diff < 1.0, f"Frame {i} pixel values differ: max_diff={max_diff}" -def test_async_decode_multiple_frames_sequential(): - """ - Test: Sequential processing of multiple frames with prefetching - Test the complete workflow: decode frame 0, then prefetch and process frame 3, then frame 6 - """ - max_num_files_to_use = 6 - path_base = utils.get_data_dir() - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - files = utils.select_random_clip(path_base) - assert files is not None - - # Process frame 0 - frame_0 = [0] * len(files) - nv_stream_dec.DecodeN12ToRGBAsync(files, frame_0, False) - frames_0 = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_0, False) - assert frames_0 is not None - - # Deep copy frame 0 (as required by documentation) - tensor_0 = [torch.as_tensor(frame, device='cuda').clone() for frame in frames_0] - - # Prefetch frame 3 - frame_3 = [3] * len(files) - nv_stream_dec.DecodeN12ToRGBAsync(files, frame_3, False) - - # Process frame 3 - frames_3 = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_3, False) - assert frames_3 is not None - - # Deep copy frame 3 - tensor_3 = [torch.as_tensor(frame, device='cuda').clone() for frame in frames_3] - - # Prefetch frame 6 - frame_6 = [6] * len(files) - nv_stream_dec.DecodeN12ToRGBAsync(files, frame_6, False) - - # Process frame 6 - frames_6 = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_6, False) - assert frames_6 is not None - - # Deep copy frame 6 - tensor_6 = [torch.as_tensor(frame, device='cuda').clone() for frame in frames_6] - - # Verify all frames are valid - assert len(tensor_0) == len(files) - assert len(tensor_3) == len(files) - assert len(tensor_6) == len(files) - - @pytest.mark.timeout(30) def test_destructor_with_pending_task(): """ @@ -524,38 +379,6 @@ def test_getbuffer_without_async_call(): nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_id_list, False) -def test_recovery_after_exception(): - """ - Test: Reader should be usable after exception - Test that reader can continue working after an exception occurred - """ - max_num_files_to_use = 6 - path_base = utils.get_data_dir() - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - # Trigger an exception with invalid file - nv_stream_dec.DecodeN12ToRGBAsync(["/invalid/nonexistent/path.mp4"], [0], False) - with pytest.raises(RuntimeError): - nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(["/invalid/nonexistent/path.mp4"], [0], False) - - # Verify reader can still work after exception - files = utils.select_random_clip(path_base) - assert files is not None - - frame_id_list = [0] * len(files) - - nv_stream_dec.DecodeN12ToRGBAsync(files, frame_id_list, False) - frames = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_id_list, False) - - assert frames is not None - assert len(frames) == len(files) - - def test_multiple_async_without_getbuffer(): """ Test: Multiple consecutive Async calls without GetBuffer @@ -658,26 +481,6 @@ def test_double_getbuffer_same_params(): nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer(files, frame_id_list, False) -def test_empty_list_params(): - """ - Test: Empty list parameters should be handled gracefully - Test that empty filepaths/frame_ids don't cause crashes - """ - max_num_files_to_use = 6 - - nv_stream_dec = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - # Empty lists - return empty - nv_stream_dec.DecodeN12ToRGBAsync([], [], False) - frames = nv_stream_dec.DecodeN12ToRGBAsyncGetBuffer([], [], False) - # If it succeeds, result should be empty - assert len(frames) == 0 - - @pytest.mark.timeout(30) def test_sync_api_waits_for_async(): """ diff --git a/packages/on_demand_video_decoder/tests/test_stream_decoder.py b/packages/on_demand_video_decoder/tests/test_stream_decoder.py index 393142a8..702913a9 100644 --- a/packages/on_demand_video_decoder/tests/test_stream_decoder.py +++ b/packages/on_demand_video_decoder/tests/test_stream_decoder.py @@ -16,7 +16,6 @@ import sys import ctypes -import random import threading import time @@ -32,31 +31,6 @@ def test_pynvsamplereader_rejects_direct_construction(): nvc.PyNvSampleReader(1, 1) -def test_stream_access_single(): - max_num_files_to_use = 6 - iter_num = 10 - path_base = utils.get_data_dir() - - nv_gop_dec = nvc.CreateSampleReader( - num_of_set=10, - num_of_file=max_num_files_to_use, - iGpu=0, - ) - - frame_min = 0 - frame_max = 200 - - for c in range(iter_num): - files = utils.select_random_clip(path_base) - assert files is not None, f"files is None for select_random_clip, path_base: {path_base}" - - frames = [random.randint(frame_min, frame_max) for _ in range(len(files))] - print(f"Comparison: {c}, frames: {frames}") - - gop_decoded = utils.gop_decode_bgr(nv_gop_dec, files, frames) - assert gop_decoded is not None, f"gop_decoded is None for DecodeN12ToRGB, frames: {frames}" - - def _heartbeats_during(call, margin=0.02): """ Run `call` while a background thread records timestamps in a tight pure-Python loop. diff --git a/packages/on_demand_video_decoder/tests/test_stream_resource_free.py b/packages/on_demand_video_decoder/tests/test_stream_resource_free.py deleted file mode 100644 index 638cefe7..00000000 --- a/packages/on_demand_video_decoder/tests/test_stream_resource_free.py +++ /dev/null @@ -1,982 +0,0 @@ -# Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# -# 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. - -""" -Resource Management Tests for Stream Decoder (CreateSampleReader) - -This module tests CPU and GPU memory resource release mechanisms for: -- Synchronous decoding: DecodeN12ToRGB() -- Asynchronous decoding: DecodeN12ToRGBAsync() + DecodeN12ToRGBAsyncGetBuffer() - -Key Points: -- NvDecoder uses CUDA Driver API (cuMemAlloc), NOT PyTorch allocator -- torch.cuda.memory_allocated() CANNOT detect NvDecoder memory -- We MUST use pynvml to get accurate GPU memory usage -""" - -import gc -import sys -import time - -import pytest -import torch - -import utils -import accvlab.on_demand_video_decoder as nvc -from utils import CPUMemoryMonitor, GPUMemoryMonitor, force_cleanup, measure_memory_delta - -# ============================================================================ -# Test Fixtures -# ============================================================================ - - -@pytest.fixture(scope="session", autouse=True) -def nvdec_warmup(): - """Pre-initialize NVDEC before any leak tests. - - The first NVDEC decoder creation on a cold GPU allocates ~30-40 MB of - runtime overhead (JIT, internal state) that is never returned to the OS. - Without warmup, the first test to create a decoder sees this overhead as - a 'leak' because the baseline was measured before NVDEC was initialized. - Running a throwaway decode here ensures all subsequent baseline measurements - are taken on a warmed-up NVDEC, so only real leaks show as deltas. - """ - path_base = utils.get_data_dir() - files = utils.select_random_clip(path_base) - if files: - decoder = nvc.CreateSampleReader(num_of_set=1, num_of_file=len(files), iGpu=0) - try: - decoder.DecodeN12ToRGB(files, [0] * len(files), False) - except Exception: - pass - try: - decoder.DecodeN12ToRGBAsync(files, [0] * len(files), False) - decoder.DecodeN12ToRGBAsyncGetBuffer(files, [0] * len(files), False) - except Exception: - pass - del decoder - force_cleanup() - - -@pytest.fixture -def gpu_monitor(): - """Fixture to provide GPU memory monitor.""" - with GPUMemoryMonitor(gpu_id=0) as monitor: - yield monitor - - -@pytest.fixture -def cpu_monitor(): - """Fixture to provide CPU memory monitor.""" - yield CPUMemoryMonitor() - - -@pytest.fixture -def video_files(): - """Fixture to provide test video files.""" - path_base = utils.get_data_dir() - files = utils.select_random_clip(path_base) - assert files is not None, f"No video files found in {path_base}" - return files - - -# ============================================================================ -# P0 Tests - Core Functionality (Must Pass) -# ============================================================================ - - -class TestP0CoreResourceRelease: - """ - P0 Priority Tests - Core resource release functionality. - These tests must pass for the decoder to be considered stable. - """ - - # Memory tolerance in MB - based on actual test results: - # - Set to 10 MB to allow for CUDA runtime overhead and measurement noise - GPU_TOLERANCE_MB = 10.0 - CPU_TOLERANCE_MB = 10.0 - - def test_01_basic_del_releases_gpu_memory(self, gpu_monitor, video_files): - """ - Test 1: Normal usage followed by `del` releases GPU resources. - - Scenario: Create decoder → decode several frames → del decoder - Verify: GPU memory returns to initial level (within tolerance) - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - # Create decoder and decode - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - frame_ids = [0] * len(video_files) - _ = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - - # Record memory after decoding - after_decode_gpu = gpu_monitor.get_used_memory_mb() - print( - f"\nGPU memory after decode: {after_decode_gpu:.1f} MB " - f"(baseline: {baseline_gpu:.1f} MB, delta: {after_decode_gpu - baseline_gpu:.1f} MB)" - ) - - # Delete decoder and force cleanup - del decoder - force_cleanup() - - # Check memory released - final_gpu = gpu_monitor.get_used_memory_mb() - is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - print( - f"GPU memory after cleanup: {final_gpu:.1f} MB " - f"(baseline: {baseline_gpu:.1f} MB, delta: {delta:.1f} MB)" - ) - - assert is_ok, ( - f"GPU memory leak detected! " - f"Baseline: {baseline_gpu:.1f} MB, Final: {final_gpu:.1f} MB, " - f"Delta: {delta:.1f} MB (tolerance: {self.GPU_TOLERANCE_MB} MB)" - ) - - # def test_02_sync_decode_loop_memory_stable(self, gpu_monitor, video_files): - # """ - # Test 2: Synchronous decode loop - memory should remain stable. - - # Scenario: Create single decoder → decode N times in loop → del decoder - # Verify: Memory doesn't continuously grow during decoding - # """ - # force_cleanup() - # baseline_gpu = gpu_monitor.get_used_memory_mb() - - # decoder = nvc.CreateSampleReader( - # num_of_set=1, - # num_of_file=6, - # iGpu=0, - # ) - - # num_iterations = 50 - # memory_samples = [] - - # for i in range(num_iterations): - # frame_ids = [i % 100] * len(video_files) # Vary frame IDs - # frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - - # # Deep copy to tensor and release reference - # tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - # del frames, tensors - - # # Sample memory every 10 iterations - # if (i + 1) % 10 == 0: - # mem = gpu_monitor.get_used_memory_mb() - # memory_samples.append(mem) - # print(f"Iteration {i+1}: GPU memory = {mem:.1f} MB") - - # # Check memory stability (no continuous growth) - # # Compare first half average with second half average - # mid = len(memory_samples) // 2 - # first_half_avg = sum(memory_samples[:mid]) / mid if mid > 0 else 0 - # second_half_avg = sum(memory_samples[mid:]) / (len(memory_samples) - mid) - - # growth = second_half_avg - first_half_avg - # print( - # f"\nMemory growth: {growth:.1f} MB " - # f"(first half avg: {first_half_avg:.1f} MB, second half avg: {second_half_avg:.1f} MB)" - # ) - - # # Cleanup - # del decoder - # force_cleanup() - - # final_gpu = gpu_monitor.get_used_memory_mb() - # is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - # print(f"Final GPU memory: {final_gpu:.1f} MB (delta from baseline: {delta:.1f} MB)") - - # # Memory should not grow significantly during iteration - # assert growth < self.GPU_TOLERANCE_MB, f"Memory growing during decode loop! Growth: {growth:.1f} MB" - - # # Memory should be released after del - # assert is_ok, f"GPU memory leak after del! Delta: {delta:.1f} MB" - - # TODO: Re-enable this test once a more scientific measurement methodology is in place. - # - # Why this test is currently disabled: - # On some CI environments (e.g. different GPU driver / CUDA version), the first - # DecodeN12ToRGBAsync call leaves a variable amount (~32 MB observed, but - # potentially more on other hardware) of CUDA stream-ordered pool allocations - # (USED_MEM_CURRENT) that persist after `del decoder`. The root cause is unclear - # (likely NVDEC driver-internal async context state or a thread/stream ordering - # effect specific to the background-thread async path), and the residual size is - # not predictable across environments, so no fixed tolerance can be justified. - # Setting a threshold like 64 MB would pass today but could silently let a real - # leak through on a GPU where the one-time overhead is larger. - # The same applies to the pynvml GPU-delta assertion: it reflects device-global - # used memory that also captures driver/context retention unrelated to our code. - # def test_03_async_decode_loop_memory_stable(self, gpu_monitor, video_files): - # """ - # Test 3: Asynchronous decode loop - memory should remain stable. - - # Scenario: Create decoder → loop (Async + GetBuffer) N times → del decoder - # Verify: Memory doesn't continuously grow during async decoding - # """ - # force_cleanup() - # baseline_gpu = gpu_monitor.get_used_memory_mb() - # baseline_pool_used = gpu_monitor.get_pool_used_mb() - - # decoder = nvc.CreateSampleReader( - # num_of_set=1, - # num_of_file=6, - # iGpu=0, - # ) - - # num_iterations = 50 - # memory_samples = [] - - # for i in range(num_iterations): - # frame_ids = [i % 100] * len(video_files) - - # # Async decode - # decoder.DecodeN12ToRGBAsync(video_files, frame_ids, False) - # frames = decoder.DecodeN12ToRGBAsyncGetBuffer(video_files, frame_ids, False) - - # # Deep copy and release - # tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - # del frames, tensors - - # if (i + 1) % 10 == 0: - # mem = gpu_monitor.get_used_memory_mb() - # memory_samples.append(mem) - # print(f"Iteration {i+1}: GPU memory = {mem:.1f} MB") - - # # Check memory stability - # mid = len(memory_samples) // 2 - # first_half_avg = sum(memory_samples[:mid]) / mid if mid > 0 else 0 - # second_half_avg = sum(memory_samples[mid:]) / (len(memory_samples) - mid) - # growth = second_half_avg - first_half_avg - - # print(f"\nMemory growth: {growth:.1f} MB") - - # del decoder - # force_cleanup() - - # pool_used_delta = gpu_monitor.get_pool_used_mb() - baseline_pool_used - # final_gpu = gpu_monitor.get_used_memory_mb() - # is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - # print( - # f"Final GPU memory: {final_gpu:.1f} MB (delta: {delta:.1f} MB), pool used delta: {pool_used_delta:.1f} MB" - # ) - - # assert growth < self.GPU_TOLERANCE_MB, f"Memory growing during async loop! Growth: {growth:.1f} MB" - # assert pool_used_delta <= 0, ( - # f"CUDA pool leak: {pool_used_delta:.1f} MB more actively allocated after del than before" - # ) - # assert is_ok, f"GPU memory leak after del! Delta: {delta:.1f} MB" - - # TODO: Re-enable this test once a more scientific measurement methodology is in place. - # - # Why this test is currently disabled: - # The current approach evaluates GPU memory leaks by sampling the device's - # used-memory counter (via pynvml) before and after running the scenario, and - # asserting on the delta. That signal mixes together several things that are - # not "a leak in our decoder": - # 1. pynvml/NVML reports device-global used memory -- it includes anything - # the CUDA primary context and driver retain, not just allocations owned - # by our decoder object. - # 2. The CUDA primary context's lifetime is governed by driver-level - # reference counting (cuDevicePrimaryCtxRetain/Release), so context-level - # memory is not guaranteed to be returned on a single destroy cycle. - # 3. PyTorch's caching allocator does not promptly return freed blocks to - # the OS/driver -- see "CUDA semantics: Memory management" in the - # PyTorch docs; torch.cuda.empty_cache() is best-effort, not a guarantee. - # 4. NVDEC has its own driver-level caches that can persist across - # create/destroy cycles in our code. - # The combined effect is that this test can both false-positive (driver/ - # allocator retention is reported as a leak) and false-negative (a real leak - # gets masked by allocator reuse). - # - # A more rigorous replacement should isolate decoder-owned allocations from - # context/driver/allocator retention -- e.g. running the scenario in a fresh - # subprocess and comparing post-teardown memory, instrumenting the C++ - # allocator directly, or using CUDA memory tracking tooling (compute-sanitizer - # leak-check). - # def test_07_repeated_create_destroy_memory_stable(self, gpu_monitor, video_files): - # """ - # Test 7: Repeated create/destroy cycle - no cumulative memory leak. - - # Scenario: Loop M times (create decoder → decode → del decoder) - # Verify: Memory returns to baseline after each cycle, no accumulation - # """ - - # MAX_LEAK_PER_CYCLE_MB = 1.0 - - # force_cleanup() - # baseline_gpu = gpu_monitor.get_used_memory_mb() - - # num_cycles = 50 # Increased from 10 to better detect memory leaks - # memory_after_cycles = [] - - # for cycle in range(num_cycles): - # # Create decoder - # decoder = nvc.CreateSampleReader( - # num_of_set=1, - # num_of_file=6, - # iGpu=0, - # ) - - # # Decode several frames - # for _ in range(5): - # frame_ids = [0] * len(video_files) - # frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - # tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - # del frames, tensors - - # # Delete decoder - # del decoder - # force_cleanup() - - # mem = gpu_monitor.get_used_memory_mb() - # memory_after_cycles.append(mem) - - # # Print every 10 cycles to reduce output noise - # if (cycle + 1) % 10 == 0 or cycle == 0: - # print( - # f"Cycle {cycle + 1}: GPU memory after cleanup = {mem:.1f} MB " - # f"(delta from baseline: {mem - baseline_gpu:.1f} MB)" - # ) - - # # Check no cumulative leak - # final_gpu = memory_after_cycles[-1] - # is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - # # Also check that memory isn't growing across cycles - # growth = memory_after_cycles[-1] - memory_after_cycles[0] - - # # Calculate average growth per cycle to detect linear leak - # avg_growth_per_cycle = growth / (num_cycles - 1) if num_cycles > 1 else 0 - - # print(f"\n=== Memory Leak Analysis ===") - # print(f"Total cycles: {num_cycles}") - # print(f"First cycle memory: {memory_after_cycles[0]:.1f} MB") - # print(f"Final cycle memory: {memory_after_cycles[-1]:.1f} MB") - # print(f"Cumulative growth: {growth:.1f} MB") - # print(f"Average growth per cycle: {avg_growth_per_cycle:.2f} MB") - # print(f"Max allowed leak per cycle: {MAX_LEAK_PER_CYCLE_MB:.2f} MB") - # print(f"Final delta from baseline: {delta:.1f} MB") - - # # For repeated cycle tests, only check per-cycle leak rate - # # (cumulative growth is just: num_cycles × per_cycle_rate, so checking both is redundant) - # # - # # Note: A small per-cycle leak (e.g., 0.78 MB) may be due to: - # # - CUDA/NVDEC driver-level caching - # # - Primary context reference counting overhead - # # This is acceptable as long as the rate is bounded. - # assert avg_growth_per_cycle <= MAX_LEAK_PER_CYCLE_MB, ( - # f"Memory leak detected! Average {avg_growth_per_cycle:.2f} MB per cycle " - # f"(threshold: {MAX_LEAK_PER_CYCLE_MB:.2f} MB). " - # f"Total leak: {growth:.1f} MB over {num_cycles} cycles." - # ) - - def test_09_del_with_pending_async_task_no_deadlock(self, gpu_monitor, video_files): - """ - Test 9: Delete decoder while async task is pending - should not deadlock. - - Scenario: Call DecodeN12ToRGBAsync() then immediately del decoder - Verify: No deadlock, resources properly released - """ - force_cleanup() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - frame_ids = [0] * len(video_files) - - # Start async decode but don't get result - decoder.DecodeN12ToRGBAsync(video_files, frame_ids, False) - - # Immediately delete - this should not deadlock - # (destructor should wait for async task to complete) - start_time = time.time() - del decoder - elapsed = time.time() - start_time - - print(f"\nDestructor completed in {elapsed:.2f} seconds") - - # Should complete within reasonable time (not deadlock) - assert elapsed < 30.0, f"Destructor took too long ({elapsed:.2f}s), possible deadlock!" - - force_cleanup() - - # NOTE: post-del GPU/pool memory assertions are omitted here for the same - # reason test_03 is disabled: the first async decode on some environments - # leaves a variable amount of CUDA pool allocations (USED_MEM_CURRENT) that - # persist after destruction, and no environment-independent threshold can be - # justified. The primary goal of this test is the deadlock check above. - - -# ============================================================================ -# P1 Tests - Important Functionality -# ============================================================================ - - -class TestP1ExplicitResourceRelease: - """ - P1 Priority Tests - Explicit resource release APIs and exception handling. - """ - - GPU_TOLERANCE_MB = 50.0 - - def test_04_release_device_memory_effective(self, gpu_monitor, video_files): - """ - Test 4: release_device_memory() effectively releases GPU memory. - - Scenario: Create decoder → decode → call release_device_memory() - Verify: GPU memory significantly decreases after call - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - # Decode to allocate GPU memory - frame_ids = [0] * len(video_files) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - after_decode_gpu = gpu_monitor.get_used_memory_mb() - gpu_used_by_decoder = after_decode_gpu - baseline_gpu - print( - f"\nGPU memory after decode: {after_decode_gpu:.1f} MB " - f"(decoder using: {gpu_used_by_decoder:.1f} MB)" - ) - - # Release device memory - decoder.release_device_memory() - torch.cuda.synchronize() - time.sleep(0.2) - - after_release_gpu = gpu_monitor.get_used_memory_mb() - released = after_decode_gpu - after_release_gpu - - print( - f"GPU memory after release_device_memory(): {after_release_gpu:.1f} MB " - f"(released: {released:.1f} MB)" - ) - - # Memory should decrease (at least some should be released) - # Note: Not all memory may be released due to internal state - assert released > 0 or gpu_used_by_decoder < 50, ( - f"release_device_memory() didn't release any memory! " - f"Before: {after_decode_gpu:.1f} MB, After: {after_release_gpu:.1f} MB" - ) - - del decoder - force_cleanup() - - def test_05_clear_all_readers_effective(self, gpu_monitor, video_files): - """ - Test 5: clearAllReaders() effectively releases resources. - - Scenario: Create decoder → decode multiple different videos → call clearAllReaders() - Verify: GPU memory decreases after clearing readers - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=3, # Multiple sets to hold more reader state - num_of_file=6, - iGpu=0, - ) - - # Decode multiple times with different frame IDs to create reader state - for frame_offset in range(3): - frame_ids = [frame_offset * 10] * len(video_files) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - after_decode_gpu = gpu_monitor.get_used_memory_mb() - print( - f"\nGPU memory after multiple decodes: {after_decode_gpu:.1f} MB " - f"(delta: {after_decode_gpu - baseline_gpu:.1f} MB)" - ) - - # Clear all readers - decoder.clearAllReaders() - torch.cuda.synchronize() - time.sleep(0.2) - - after_clear_gpu = gpu_monitor.get_used_memory_mb() - released = after_decode_gpu - after_clear_gpu - - print( - f"GPU memory after clearAllReaders(): {after_clear_gpu:.1f} MB " f"(released: {released:.1f} MB)" - ) - - del decoder - force_cleanup() - - final_gpu = gpu_monitor.get_used_memory_mb() - is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - print(f"Final GPU memory: {final_gpu:.1f} MB (delta: {delta:.1f} MB)") - assert is_ok, f"GPU memory leak after clearAllReaders + del! Delta: {delta:.1f} MB" - - def test_06_release_memory_then_continue_decode(self, gpu_monitor, video_files): - """ - Test 6: Decoder remains usable after release_device_memory(). - - Scenario: Create decoder → decode → release_device_memory() → decode again - Verify: Decoder works correctly after memory release - - Note: This test validates that after calling release_device_memory(), - the decoder can still decode frames correctly. The GPU memory pool - should be re-allocated automatically on the next decode operation. - """ - force_cleanup() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - frame_ids = [0] * len(video_files) - - # First decode - frames_1 = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensor_1 = torch.as_tensor(frames_1[0], device='cuda').clone() - del frames_1 - - print(f"\nFirst decode successful, frame shape: {tensor_1.shape}") - - # Release memory - decoder.release_device_memory() - torch.cuda.synchronize() - - # Second decode - should still work (but currently returns empty frames!) - frames_2 = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensor_2 = torch.as_tensor(frames_2[0], device='cuda').clone() - del frames_2 - - print(f"Second decode successful, frame shape: {tensor_2.shape}") - - # Verify frames are identical (same content) - assert tensor_2.shape[0] > 0, f"Frame has zero height! Shape: {tensor_2.shape}" - assert ( - tensor_1.shape == tensor_2.shape - ), f"Frame shapes don't match! {tensor_1.shape} vs {tensor_2.shape}" - max_diff = (tensor_1.float() - tensor_2.float()).abs().max().item() - print(f"Max pixel difference between frames: {max_diff}") - assert max_diff < 1.0, f"Frames differ too much! Max diff: {max_diff}" - - del decoder, tensor_1, tensor_2 - force_cleanup() - - def test_10_multiple_async_without_getbuffer_no_leak(self, gpu_monitor, video_files): - """ - Test 10: Multiple Async calls without GetBuffer - no memory leak. - - Scenario: Call Async multiple times (overwriting previous) → del decoder - Verify: Overwritten results don't cause memory leak - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - # Call Async multiple times without GetBuffer - # Each call should discard previous pending result - for i in range(10): - frame_ids = [i * 5] * len(video_files) - decoder.DecodeN12ToRGBAsync(video_files, frame_ids, False) - # Don't call GetBuffer - let it be overwritten - - # Only get the last one - last_frame_ids = [45] * len(video_files) - frames = decoder.DecodeN12ToRGBAsyncGetBuffer(video_files, last_frame_ids, False) - assert frames is not None and len(frames) == len(video_files) - - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - del decoder - force_cleanup() - - final_gpu = gpu_monitor.get_used_memory_mb() - is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - print(f"\nGPU memory delta: {delta:.1f} MB (tolerance: {self.GPU_TOLERANCE_MB} MB)") - - assert is_ok, f"GPU memory leak with multiple Async without GetBuffer! " f"Delta: {delta:.1f} MB" - - def test_08_release_decoder_effective(self, gpu_monitor, video_files): - """ - Test 8: release_decoder() effectively releases GPU memory. - - Scenario: Create decoder → decode → call release_decoder() - Verify: GPU memory significantly decreases after call - - Note: release_decoder() is more thorough than release_device_memory(): - - release_device_memory(): only releases GPU memory pool - - release_decoder(): deletes all readers (NvDecoder, FFmpegDemuxer, etc.) - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - # Decode multiple frames to allocate more GPU memory - for frame_offset in range(5): - frame_ids = [frame_offset * 10] * len(video_files) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - after_decode_gpu = gpu_monitor.get_used_memory_mb() - gpu_used_by_decoder = after_decode_gpu - baseline_gpu - print( - f"\nGPU memory after decode: {after_decode_gpu:.1f} MB " - f"(decoder using: {gpu_used_by_decoder:.1f} MB)" - ) - - # Release decoder (more thorough than release_device_memory) - decoder.release_decoder() - torch.cuda.synchronize() - time.sleep(0.2) - - after_release_gpu = gpu_monitor.get_used_memory_mb() - released = after_decode_gpu - after_release_gpu - - print( - f"GPU memory after release_decoder(): {after_release_gpu:.1f} MB " - f"(released: {released:.1f} MB)" - ) - - # Memory should decrease significantly - assert released > 0 or gpu_used_by_decoder < 50, ( - f"release_decoder() didn't release any memory! " - f"Before: {after_decode_gpu:.1f} MB, After: {after_release_gpu:.1f} MB" - ) - - del decoder - force_cleanup() - - def test_09_release_decoder_then_continue_decode(self, gpu_monitor, video_files): - """ - Test 9: Decoder remains usable after release_decoder(). - - Scenario: Create decoder → decode → release_decoder() → decode again - Verify: Decoder works correctly after releasing all readers - - Note: After release_decoder(), new readers will be created automatically - on the next decode operation. - """ - force_cleanup() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - frame_ids = [0] * len(video_files) - - # First decode - frames_1 = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensor_1 = torch.as_tensor(frames_1[0], device='cuda').clone() - shape_1 = tensor_1.shape - del frames_1 - - print(f"\nFirst decode successful, frame shape: {shape_1}") - - # Release decoder (delete all readers) - decoder.release_decoder() - torch.cuda.synchronize() - - # Second decode - should work (new readers will be created) - frames_2 = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensor_2 = torch.as_tensor(frames_2[0], device='cuda').clone() - shape_2 = tensor_2.shape - del frames_2 - - print(f"Second decode after release_decoder(), frame shape: {shape_2}") - - # Verify frame shapes match - assert shape_1 == shape_2, ( - f"Frame shapes don't match after release_decoder()! " f"Before: {shape_1}, After: {shape_2}" - ) - - del decoder - force_cleanup() - - # TODO: Re-enable this test once a more scientific measurement methodology is in place. - # - # Why this test is currently disabled: - # Same root cause as test_07_repeated_create_destroy_memory_stable above: - # we are diffing pynvml's device-global used-memory across cycles, but that - # counter also reflects CUDA primary context retention, PyTorch caching- - # allocator behavior (torch.cuda.empty_cache() is best-effort -- see PyTorch - # "CUDA semantics: Memory management"), and NVDEC driver-level caching. - # The per-cycle "growth" produced by this signal is therefore not a - # trustworthy leak metric for the decoder specifically. - # - # A proper test for repeated release_decoder() cycles needs measurement that - # isolates decoder-owned allocations from context/driver/allocator retention -- - # e.g. subprocess isolation, instrumenting the C++ allocator, or CUDA - # leak-check tooling (compute-sanitizer). - # def test_10_repeated_release_decoder_no_leak(self, gpu_monitor, video_files): - # """ - # Test 10: Repeated release_decoder() cycles don't cause memory leak. - - # Scenario: Repeatedly (decode → release_decoder()) in a loop - # Verify: No cumulative memory growth - # """ - # MAX_LEAK_PER_CYCLE_MB = 1.0 - - # force_cleanup() - # baseline_gpu = gpu_monitor.get_used_memory_mb() - - # decoder = nvc.CreateSampleReader( - # num_of_set=1, - # num_of_file=6, - # iGpu=0, - # ) - - # num_cycles = 20 - # memory_after_cycles = [] - - # print(f"\n=== Running {num_cycles} decode-release_decoder cycles ===") - - # for cycle in range(num_cycles): - # # Decode - # frame_ids = [cycle % 50] * len(video_files) - # frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - # tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - # del frames, tensors - - # # Release decoder - # decoder.release_decoder() - # torch.cuda.synchronize() - - # current_gpu = gpu_monitor.get_used_memory_mb() - # memory_after_cycles.append(current_gpu) - - # if (cycle + 1) % 5 == 0: - # print( - # f"Cycle {cycle + 1}: {current_gpu:.1f} MB " - # f"(delta from baseline: {current_gpu - baseline_gpu:.1f} MB)" - # ) - - # # Analyze memory growth - # growth = memory_after_cycles[-1] - memory_after_cycles[0] - # avg_growth_per_cycle = growth / (num_cycles - 1) if num_cycles > 1 else 0 - - # print(f"\n=== Memory Analysis ===") - # print(f"First cycle: {memory_after_cycles[0]:.1f} MB") - # print(f"Last cycle: {memory_after_cycles[-1]:.1f} MB") - # print(f"Growth: {growth:.1f} MB") - # print(f"Avg growth per cycle: {avg_growth_per_cycle:.2f} MB") - - # # Should not have significant per-cycle leak - # assert avg_growth_per_cycle < MAX_LEAK_PER_CYCLE_MB, ( - # f"Memory leak in release_decoder() cycles! " f"Avg growth: {avg_growth_per_cycle:.2f} MB/cycle" - # ) - - # del decoder - # force_cleanup() - - def test_11_release_decoder_with_pending_async(self, gpu_monitor, video_files): - """ - Test 11: release_decoder() properly handles pending async task. - - Scenario: Start async decode → immediately call release_decoder() - Verify: No crash, proper cleanup - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - # Start async decode - frame_ids = [0] * len(video_files) - decoder.DecodeN12ToRGBAsync(video_files, frame_ids, False) - - # Immediately release decoder (should wait for async to complete) - decoder.release_decoder() - torch.cuda.synchronize() - - # Try to decode again (should create new readers) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - assert frames is not None and len(frames) == len(video_files) - - tensor = torch.as_tensor(frames[0], device='cuda') - assert tensor.numel() > 0, "Got empty frame after release_decoder with pending async!" - - del frames - del decoder - force_cleanup() - - final_gpu = gpu_monitor.get_used_memory_mb() - is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - print(f"\nGPU memory delta: {delta:.1f} MB") - assert is_ok, f"Memory leak with release_decoder + pending async! Delta: {delta:.1f} MB" - - def test_12_exception_then_resource_release(self, gpu_monitor, video_files): - """ - Test 12: Resources properly released after exception. - - Scenario: Trigger exception with invalid file → catch exception → del decoder - Verify: Exception doesn't prevent resource release - """ - force_cleanup() - baseline_gpu = gpu_monitor.get_used_memory_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - # First, do some successful decoding - frame_ids = [0] * len(video_files) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - after_decode_gpu = gpu_monitor.get_used_memory_mb() - print(f"\nGPU memory after successful decode: {after_decode_gpu:.1f} MB") - - # Now trigger an exception - invalid_files = ["/nonexistent/invalid/path.mp4"] - invalid_frame_ids = [0] - - with pytest.raises(RuntimeError): - decoder.DecodeN12ToRGB(invalid_files, invalid_frame_ids, False) - - print("Exception caught as expected") - - # Delete decoder - should still release resources - del decoder - force_cleanup() - - final_gpu = gpu_monitor.get_used_memory_mb() - is_ok, delta = measure_memory_delta(baseline_gpu, final_gpu, self.GPU_TOLERANCE_MB) - - print(f"GPU memory after exception + del: {final_gpu:.1f} MB (delta: {delta:.1f} MB)") - - assert is_ok, f"GPU memory leak after exception! Delta: {delta:.1f} MB" - - -# ============================================================================ -# CPU Memory Tests -# ============================================================================ - - -class TestCPUMemoryRelease: - """Tests for CPU memory resource release.""" - - CPU_TOLERANCE_MB = 20.0 - - def test_cpu_memory_stable_sync_loop(self, cpu_monitor, video_files): - """Test CPU memory stability during sync decode loop.""" - force_cleanup() - baseline_cpu = cpu_monitor.get_rss_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - memory_samples = [] - for i in range(30): - frame_ids = [i % 100] * len(video_files) - frames = decoder.DecodeN12ToRGB(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - if (i + 1) % 10 == 0: - gc.collect() - mem = cpu_monitor.get_rss_mb() - memory_samples.append(mem) - print(f"Iteration {i+1}: CPU RSS = {mem:.1f} MB") - - del decoder - force_cleanup() - - final_cpu = cpu_monitor.get_rss_mb() - delta = final_cpu - baseline_cpu - - print(f"\nCPU RSS delta: {delta:.1f} MB (tolerance: {self.CPU_TOLERANCE_MB} MB)") - - assert delta < self.CPU_TOLERANCE_MB, f"CPU memory leak! Delta: {delta:.1f} MB" - - def test_cpu_memory_stable_async_loop(self, cpu_monitor, video_files): - """Test CPU memory stability during async decode loop.""" - force_cleanup() - baseline_cpu = cpu_monitor.get_rss_mb() - - decoder = nvc.CreateSampleReader( - num_of_set=1, - num_of_file=6, - iGpu=0, - ) - - for i in range(30): - frame_ids = [i % 100] * len(video_files) - decoder.DecodeN12ToRGBAsync(video_files, frame_ids, False) - frames = decoder.DecodeN12ToRGBAsyncGetBuffer(video_files, frame_ids, False) - tensors = [torch.as_tensor(f, device='cuda').clone() for f in frames] - del frames, tensors - - del decoder - force_cleanup() - - final_cpu = cpu_monitor.get_rss_mb() - delta = final_cpu - baseline_cpu - - print(f"\nCPU RSS delta: {delta:.1f} MB (tolerance: {self.CPU_TOLERANCE_MB} MB)") - - assert delta < self.CPU_TOLERANCE_MB, f"CPU memory leak! Delta: {delta:.1f} MB" - - -if __name__ == "__main__": - sys.exit(pytest.main([__file__, "-v"])) diff --git a/packages/on_demand_video_decoder/tests/utils.py b/packages/on_demand_video_decoder/tests/utils.py index 4a3223c9..dc2115c4 100644 --- a/packages/on_demand_video_decoder/tests/utils.py +++ b/packages/on_demand_video_decoder/tests/utils.py @@ -18,103 +18,11 @@ import random import time -import numpy as np import psutil import pynvml import torch -def is_diff_in_range(to_comp_1, to_comp_2, tolerance): - # support input as YUV tuple or single channel - if isinstance(to_comp_1, tuple) and isinstance(to_comp_2, tuple): - results = [is_diff_in_range(c1, c2, tolerance) for c1, c2 in zip(to_comp_1, to_comp_2)] - in_range = all(r[0] for r in results) - max_diff = max(r[1] for r in results) - count_in_range = sum(r[2] for r in results) - return in_range, max_diff, count_in_range - - # support single channel - if torch.is_tensor(to_comp_1): - to_comp_1 = to_comp_1.cpu() - if torch.is_tensor(to_comp_2): - to_comp_2 = to_comp_2.cpu() - to_comp_1 = np.array(to_comp_1) - to_comp_2 = np.array(to_comp_2) - use_int = issubclass(to_comp_1.dtype.type, np.integer) and issubclass(to_comp_2.dtype.type, np.integer) - type_to_use = np.int64 if use_int else np.float64 - diffs = np.abs(np.array(to_comp_1).astype(type_to_use) - np.array(to_comp_2).astype(type_to_use)) - max_diff = diffs.max() - in_range = max_diff <= tolerance - count_in_range = np.sum(diffs > tolerance) - return in_range, max_diff, count_in_range - - -def diff(gop_decoded, opencv_decoded, file_names, frames, num_files, diff_tolerance=21): - for file_name, frame, i in zip(file_names, frames, range(num_files)): - same_with_opencv_dec, max_diff, count_in_range = is_diff_in_range( - gop_decoded[i], opencv_decoded[i], diff_tolerance - ) - print(file_name, " frame_id: ", frame, " pass: ", same_with_opencv_dec, " max_diff: ", max_diff) - if not same_with_opencv_dec: - print(f"Error: {file_name} {frame} is not same with opencv_decoded") - return max_diff - return 0 - - -def gop_decode_bgr(nv_gop_dec, file_path_list, frame_id_list): - try: - decoded_frames = nv_gop_dec.DecodeN12ToRGB(file_path_list, frame_id_list, True) - torch.cuda.nvtx.range_push("unsqueeze_tensor_list") - res = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - torch.cuda.nvtx.range_pop() - return res - except Exception as e: - print(f"Error: {e}") - return None - - -def gop_decode_bgr_with_fast_init(nv_gop_dec, file_path_list, frame_id_list, fast_stream_infos): - try: - decoded_frames = nv_gop_dec.DecodeN12ToRGB( - file_path_list, frame_id_list, as_bgr=True, fastStreamInfos=fast_stream_infos - ) - torch.cuda.nvtx.range_push("unsqueeze_tensor_list") - res = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - torch.cuda.nvtx.range_pop() - return res - except Exception as e: - print(f"Error: {e}") - return None - - -def gop_decode_bgr_ddseparate_with_fast_init( - nv_gop_dec1, nv_gop_dec2, file_path_list, frame_id_list, fast_stream_infos -): - try: - gop_list = nv_gop_dec1.GetGOPList(file_path_list, frame_id_list, fastStreamInfos=fast_stream_infos) - gop_data_list = [data for data, _, _ in gop_list] - decoded_frames = nv_gop_dec2.DecodeFromGOPListRGB( - gop_data_list, file_path_list, frame_id_list, as_bgr=True - ) - res = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - return res - except Exception as e: - print(f"Error: {e}") - return None - - -def gop_decode_bgr_ddseparate_from_multi_packets(nv_gop_dec, file_path_list, frame_id_list, packets_list): - try: - decoded_frames = nv_gop_dec.DecodeFromGOPListRGB( - packets_list, file_path_list, frame_id_list, as_bgr=True - ) - res = [torch.unsqueeze(torch.as_tensor(df), 0) for df in decoded_frames] - return res - except Exception as e: - print(f"Error: {e}") - return None - - def get_data_dir(): """ Return absolute path to the test video data directory.