diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/inc/PyNvGopDecoder.hpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/inc/PyNvGopDecoder.hpp index c57d218..f8b9c22 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/inc/PyNvGopDecoder.hpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/inc/PyNvGopDecoder.hpp @@ -39,6 +39,8 @@ #include #include #include +#include +#include #define MAX_SIZE 2000 @@ -147,6 +149,24 @@ class PyNvGopDecoder { std::vector* out_if_no_color_conversion, std::vector* out_if_color_converted, bool skip_final_sync = false); + /** + * Decode multiple target frames from each serialized GOP bundle. + * + * Each element in datas represents one source/GOP decode task, while the matching + * element in frame_id_groups contains every target display frame in that GOP. + * + * @param datas Serialized GOP buffers, one per source/GOP group + * @param sizes Buffer sizes corresponding to datas + * @param source_names Stable source names, one per source/GOP group + * @param frame_id_groups Target display frame IDs for each GOP group + * @param as_bgr Whether RGB output should use BGR channel order + * @param output Flat RGB output in group order + */ + void decode_from_gop_groups(const std::vector& datas, const std::vector& sizes, + const std::vector& source_names, + const std::vector>& frame_id_groups, bool as_bgr, + std::vector& output); + /** * Load GOP data from multiple binary files in parallel * @@ -257,6 +277,25 @@ class PyNvGopDecoder { void ReleaseDecoder(); protected: + /** + * Native decode configuration for one GOP group. + * + * Keeping these fields together avoids four parallel vectors whose indices + * must stay aligned. The ordering operator also makes the configuration a + * direct key in the persistent decoder-slot lookup table. + */ + struct GroupedDecoderConfig { + int codec_id; + int width; + int height; + int frame_size; + + bool operator<(const GroupedDecoderConfig& other) const { + return std::tie(codec_id, width, height, frame_size) < + std::tie(other.codec_id, other.width, other.height, other.frame_size); + } + }; + int main_decode( const std::vector& color_ranges, const std::vector& codec_ids, std::vector& widths, std::vector& heights, std::vector& frame_sizes, const std::vector& filepaths, @@ -265,6 +304,25 @@ class PyNvGopDecoder { std::vector* out_if_no_color_conversion, std::vector* out_if_color_converted, bool skip_final_sync = false); + /** + * Match grouped decode requests to persistent decoder slots by decode + * configuration instead of by input-list position. A key can own several + * slots because groups with the same shape may decode concurrently. + * + * @param decoder_configs Codec and native shape for each input group + * @param decoder_slots Output mapping from group index to decoder slot + * @return 0 on success, non-zero error code on failure + */ + int AssignGroupedDecoderSlots(const std::vector& decoder_configs, + std::vector& decoder_slots); + + int main_decode_groups( + const std::vector& color_ranges, const std::vector& decoder_configs, + const std::vector& source_names, const std::vector>& frame_id_groups, + const std::vector& decoder_slots, bool as_bgr, + std::vector>>>& vpacket_queue, + std::vector& output); + /** * Perform GOP-based video demuxing and packet extraction for high-performance parallel decoding * @@ -660,7 +718,7 @@ class PyNvGopDecoder { // Lazy loading functions void ensureCudaContextInitialized(); - void ensureDemuxRunnersInitialized(); + void ensureDemuxRunnersInitialized(size_t required_count); void ensureDecodeRunnersInitialized(); void ensureMergeRunnersInitialized(); 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 bc6437b..66197a2 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 @@ -19,8 +19,10 @@ #include #include +#include #include #include +#include #include #include #include @@ -406,6 +408,13 @@ void PyNvGopDecoder::DecProc(AVColorRange color_range, NvDecoder* decoder, ConcurrentQueue>* packet_queue, const std::vector sorted_frame_ids, bool use_bgr_format, const std::string& filename, LastDecodedFrameInfo& last_decoded_frame_info) { + if (sorted_frame_ids.empty()) { + throw std::invalid_argument("[ERROR] DecProc requires at least one target frame"); + } + if (p_frames.size() < sorted_frame_ids.size()) { + throw std::invalid_argument("[ERROR] DecProc has fewer output buffers than target frames"); + } + std::stringstream ss; ss << "DecProc_Thread: fid[" << sorted_frame_ids[0] << "], " << std::this_thread::get_id(); nvtxRangePushA(ss.str().c_str()); @@ -438,6 +447,9 @@ void PyNvGopDecoder::DecProc(AVColorRange color_range, NvDecoder* decoder, int64_t timestamp = 0; pFrame = decoder->GetFrame(×tamp); // LOG(INFO) << " after get frame, frame_idx: " << frame_idx << " timestamp: " << timestamp; + if (frame_id_iter == sorted_frame_ids.end()) { + break; + } if (timestamp % 2 || timestamp / 2 == *frame_id_iter) { OutputFrame output_frame; if constexpr (std::is_same_v) { @@ -655,6 +667,123 @@ int PyNvGopDecoder::InitializeDecoders(const std::vector& codec_ids) { return 0; } +int PyNvGopDecoder::AssignGroupedDecoderSlots(const std::vector& decoder_configs, + std::vector& decoder_slots) { + nvtxRangePushA("Assign Grouped Decoder Slots"); + + const size_t num_groups = decoder_configs.size(); + if (num_groups > static_cast(max_num_files)) { + nvtxRangePop(); + LOG(ERROR) << "Grouped decoder configuration count exceeds max_num_files"; + return -1; + } + + for (const auto& config : decoder_configs) { + // DecodeFromGOPGroupsRGB accepts serialized bytes from Python, so callers + // can provide payloads that did not come from GetGOPGroups. Reject corrupt + // dimensions here before they reach NvDecoder or GPU-buffer allocation. + if (config.width <= 0 || config.height <= 0 || config.frame_size <= 0) { + nvtxRangePop(); + LOG(ERROR) << "Grouped decoder dimensions and frame sizes must be positive"; + return -1; + } + } + + ensureCudaContextInitialized(); + ensureDecodeRunnersInitialized(); + + // Build a transient dictionary over the persistent decoder vector. The + // dictionary is rebuilt each call so it cannot become stale when a slot is + // replaced. Each key maps to multiple slots to support concurrent GOPs + // with the same codec and shape. + std::map> available_slots; + for (size_t slot_idx = 0; slot_idx < vdec.size(); ++slot_idx) { + const int current_width = vdec[slot_idx]->GetCurrentWidth(); + const int current_height = vdec[slot_idx]->GetCurrentHeight(); + if (current_width <= 0 || current_height <= 0) { + continue; + } + const GroupedDecoderConfig config{static_cast(vdec[slot_idx]->GetCodec()), current_width, + current_height, vdec[slot_idx]->GetFrameSize()}; + available_slots[config].push_back(slot_idx); + } + + const size_t unassigned = std::numeric_limits::max(); + decoder_slots.assign(num_groups, unassigned); + std::vector slot_in_use(vdec.size(), false); + + // First reserve every exact match. Doing this before replacing any slot + // avoids an early unmatched group evicting a decoder needed by a later + // group in the same call. + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + auto candidates = available_slots.find(decoder_configs[group_idx]); + if (candidates == available_slots.end() || candidates->second.empty()) { + continue; + } + const size_t slot_idx = candidates->second.back(); + candidates->second.pop_back(); + decoder_slots[group_idx] = slot_idx; + slot_in_use[slot_idx] = true; + } + + const bool needs_new_decoder = + std::find(decoder_slots.begin(), decoder_slots.end(), unassigned) != decoder_slots.end(); + if (needs_new_decoder) { + ck(cuCtxPushCurrent(this->cu_context)); + } + + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + if (decoder_slots[group_idx] != unassigned) { + continue; + } + + size_t slot_idx = unassigned; + // Retain decoders for shapes that are not present in the current call, + // up to the configured max_num_files cache bound. Once the pool is + // full, replace an idle slot. + if (vdec.size() < static_cast(max_num_files)) { + slot_idx = vdec.size(); + } else { + const auto unused = std::find(slot_in_use.begin(), slot_in_use.end(), false); + if (unused != slot_in_use.end()) { + slot_idx = static_cast(std::distance(slot_in_use.begin(), unused)); + } + } + if (slot_idx == unassigned) { + if (needs_new_decoder) { + ck(cuCtxPopCurrent(NULL)); + } + nvtxRangePop(); + LOG(ERROR) << "No idle decoder slot is available for grouped decode"; + return -1; + } + + const auto& config = decoder_configs[group_idx]; + const auto codec = static_cast(config.codec_id); + nvtxRangePushA("Grouped Decoder Slot Creation"); + std::unique_ptr decoder(new NvDecoder(this->cu_stream, this->cu_context, true, codec, + false, true, false, false, nullptr, nullptr, false, + config.width, config.height)); + if (slot_idx == vdec.size()) { + vdec.push_back(std::move(decoder)); + slot_in_use.push_back(true); + } else { + decode_runners[slot_idx].join(); + vdec[slot_idx] = std::move(decoder); + reset_last_decoded_frame_info(last_decoded_frame_infos[slot_idx]); + slot_in_use[slot_idx] = true; + } + decoder_slots[group_idx] = slot_idx; + nvtxRangePop(); + } + + if (needs_new_decoder) { + ck(cuCtxPopCurrent(NULL)); + } + nvtxRangePop(); + return 0; +} + int PyNvGopDecoder::GetFileFrameBuffers(const std::vector* widths, const std::vector* heights, const std::vector* frame_sizes, bool convert_to_rgb, std::vector>& per_file_frame_buffers) { diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_constructors.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_constructors.cpp index f8c5da9..d957503 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_constructors.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_constructors.cpp @@ -139,13 +139,14 @@ void PyNvGopDecoder::ensureCudaContextInitialized() { } } -void PyNvGopDecoder::ensureDemuxRunnersInitialized() { - if (!demux_runners.empty()) { - return; // Already initialized +void PyNvGopDecoder::ensureDemuxRunnersInitialized(size_t required_count) { + if (required_count > static_cast(max_num_files)) { + throw std::invalid_argument("required demux runners exceed max_num_files"); } + // max_num_files is a request-capacity limit, not an eager thread count. demux_runners.reserve(max_num_files); - for (size_t i = 0; i < max_num_files; ++i) { + while (demux_runners.size() < required_count) { demux_runners.emplace_back(); } } @@ -603,6 +604,206 @@ void Init_PyNvGopDecoder(py::module& m) { ... print(f" First frame IDs: {first_ids}") ... print(f" GOP lengths: {gop_lens}") )pbdoc") + .def( + "GetGOPGroups", + [](std::shared_ptr& dec, const py::list& requests) { + struct SourceRequest { + std::string filepath; + // Sorted, unique decode targets. The original request order and + // duplicates are retained separately in original_positions. + std::vector frame_ids; + std::map> original_positions; + }; + struct GroupResult { + // Serialized demux output for one GOP: encoded packet bytes plus lookup metadata, + // not decoded pixels. It becomes the group's ``gop_data`` returned to Python. + SerializedPacketBundle bundle; + // Unique requested frame IDs inside this GOP, ordered for decoding. + std::vector frame_ids; + // frame_positions[i] contains every index where frame_ids[i] appeared in the + // original request. Example: [8, 6, 8] becomes IDs [6, 8] with [[1], [0, 2]]. + std::vector> frame_positions; + // Half-open display-frame range [first_frame_id, + // first_frame_id + gop_len) covered by bundle. + int first_frame_id; + int gop_len; + }; + + std::vector source_requests; + source_requests.reserve(requests.size()); + for (const auto& item : requests) { + const py::dict request = py::cast(item); + SourceRequest source_request{ + request["filepath"].cast(), + request["frame_ids"].cast>(), + {}, + }; + for (size_t frame_position = 0; frame_position < source_request.frame_ids.size(); + ++frame_position) { + const int frame_id = source_request.frame_ids[frame_position]; + if (frame_id < 0) { + throw std::invalid_argument("frame IDs must be non-negative"); + } + source_request.original_positions[frame_id].push_back( + static_cast(frame_position)); + } + + source_request.frame_ids.clear(); + source_request.frame_ids.reserve(source_request.original_positions.size()); + for (const auto& [frame_id, _] : source_request.original_positions) { + source_request.frame_ids.push_back(frame_id); + } + source_requests.push_back(std::move(source_request)); + } + + std::vector> results_by_source(source_requests.size()); + // Index of the next sorted frame ID not yet assigned to a GOP for + // each source request. + std::vector next_frame_indices(source_requests.size(), 0); + { + py::gil_scoped_release release; + while (true) { + // get_gop_list extracts one GOP per source in a call. A source + // stays pending while it still has target IDs beyond the GOPs + // extracted in previous rounds. + std::vector pending_source_indices; + for (size_t source_idx = 0; source_idx < source_requests.size(); ++source_idx) { + if (next_frame_indices[source_idx] < + source_requests[source_idx].frame_ids.size()) { + pending_source_indices.push_back(source_idx); + } + } + if (pending_source_indices.empty()) { + break; + } + + std::vector pending_filepaths; + std::vector representative_ids; + pending_filepaths.reserve(pending_source_indices.size()); + representative_ids.reserve(pending_source_indices.size()); + for (const size_t source_idx : pending_source_indices) { + const auto& source_request = source_requests[source_idx]; + pending_filepaths.push_back(source_request.filepath); + // The first unassigned target locates the next GOP for + // this source; all remaining targets in that GOP are + // consumed together below. + representative_ids.push_back( + source_request.frame_ids[next_frame_indices[source_idx]]); + } + + auto bundles = dec->get_gop_list(pending_filepaths, representative_ids); + // get_gop_list promises one result per input path in the same + // order. Check that contract before mapping round-local results + // back to their original request indices. + if (bundles.size() != pending_source_indices.size()) { + throw std::runtime_error( + "GetGOPList returned a different number of bundles than requested"); + } + + for (size_t pending_idx = 0; pending_idx < pending_source_indices.size(); + ++pending_idx) { + const size_t source_idx = pending_source_indices[pending_idx]; + auto& bundle = bundles[pending_idx]; + if (bundle.first_frame_ids.size() != 1 || bundle.gop_lens.size() != 1) { + throw std::runtime_error( + "GetGOPList returned invalid per-source GOP metadata"); + } + + const int first_frame_id = bundle.first_frame_ids[0]; + const int gop_len = bundle.gop_lens[0]; + const int64_t gop_end = + static_cast(first_frame_id) + static_cast(gop_len); + const auto& source_request = source_requests[source_idx]; + const int representative_id = + source_request.frame_ids[next_frame_indices[source_idx]]; + if (gop_len <= 0 || representative_id < first_frame_id || + representative_id >= gop_end) { + throw std::runtime_error( + "demuxed GOP range does not contain its representative frame"); + } + + std::vector grouped_ids; + std::vector> grouped_positions; + while (next_frame_indices[source_idx] < source_request.frame_ids.size()) { + const int frame_id = source_request.frame_ids[next_frame_indices[source_idx]]; + if (frame_id >= gop_end) { + break; + } + if (frame_id < first_frame_id) { + throw std::runtime_error("target frame precedes its demuxed GOP start"); + } + grouped_ids.push_back(frame_id); + grouped_positions.push_back(source_request.original_positions.at(frame_id)); + ++next_frame_indices[source_idx]; + } + + results_by_source[source_idx].push_back( + {std::move(bundle), std::move(grouped_ids), std::move(grouped_positions), + first_frame_id, gop_len}); + } + } + } + + py::list result; + for (size_t source_idx = 0; source_idx < results_by_source.size(); ++source_idx) { + for (auto& group : results_by_source[source_idx]) { + auto& bundle = group.bundle; + auto capsule = py::capsule(bundle.data.release(), + [](void* ptr) { delete[] static_cast(ptr); }); + py::array_t numpy_data({bundle.size}, {sizeof(uint8_t)}, + static_cast(capsule.get_pointer()), + capsule); + py::dict group_dict; + group_dict["gop_data"] = std::move(numpy_data); + // Zero-based index into the input requests list. All GOPs + // split from the same source request keep this index so the + // caller can scatter decoded frames back to that request. + group_dict["source_index"] = source_idx; + group_dict["source_name"] = source_requests[source_idx].filepath; + group_dict["frame_ids"] = group.frame_ids; + group_dict["frame_positions"] = group.frame_positions; + group_dict["first_frame_id"] = group.first_frame_id; + group_dict["gop_len"] = group.gop_len; + result.append(std::move(group_dict)); + } + } + return result; + }, + py::arg("requests"), + R"pbdoc( + Extract one serialized payload for each unique source/GOP. + + Args: + requests: Source request dictionaries. Each dictionary must contain + ``filepath`` and ``frame_ids``. Decode targets are sorted and + de-duplicated per request, while every original position is + retained in ``frame_positions``. + + Returns: + A source-major list of group dictionaries. Requests spanning GOP + boundaries are split automatically. Each dictionary contains: + + Group dictionaries are variable length: their ``frame_ids`` lists + need not be aligned across groups. The length of each list is the + number of unique requested frames contained in that source/GOP, + rather than a conventional batch dimension. + + - ``gop_data``: encoded packets and packet metadata for one GOP. + - ``source_index``: zero-based index of the originating item in + ``requests``; groups split from one request share this value. + - ``source_name``: the request's filepath. + - ``frame_ids``: sorted unique targets contained in this GOP. + - ``frame_positions``: original request positions for each target. + - ``first_frame_id`` and ``gop_len``: the GOP's half-open display + frame range ``[first_frame_id, first_frame_id + gop_len)``. + + Pass the returned list directly to :meth:`DecodeFromGOPGroupsRGB`. + + Example: + >>> groups = decoder.GetGOPGroups( + ... [{"filepath": camera_5_path, "frame_ids": [6, 7, 8, 9]}]) + >>> decoded_groups = decoder.DecodeFromGOPGroupsRGB(groups) + )pbdoc") .def( "DecodeFromGOPRGB", [](std::shared_ptr& dec, const py::array_t& numpy_data, @@ -793,6 +994,141 @@ void Init_PyNvGopDecoder(py::module& m) { >>> # Convert to PyTorch tensors on GPU (shape (height, width, 3), uint8) >>> rgb_tensors = [torch.as_tensor(frame).clone() for frame in rgb_frames] )pbdoc") + .def( + "DecodeFromGOPGroupsRGB", + [](std::shared_ptr& dec, const py::list& groups, bool as_bgr) { + try { + using ByteArray = py::array_t; + struct GroupLayout { + // Index of the originating GetGOPGroups request. Multiple + // GOPs split from one request have the same source_index. + int source_index; + std::string source_name; + std::vector frame_ids; + std::vector> frame_positions; + int first_frame_id; + int gop_len; + }; + + const size_t num_groups = groups.size(); + std::vector gop_datas(num_groups); + std::vector datas(num_groups); + std::vector sizes(num_groups); + std::vector source_names(num_groups); + std::vector> frame_id_groups(num_groups); + std::vector layouts(num_groups); + + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + py::dict group = py::cast(groups[group_idx]); + for (const char* required_key : + {"gop_data", "source_index", "source_name", "frame_ids", "frame_positions", + "first_frame_id", "gop_len"}) { + if (!group.contains(required_key)) { + throw std::invalid_argument(std::string("group is missing '") + required_key + + "'"); + } + } + + gop_datas[group_idx] = group["gop_data"].cast(); + const auto& data = gop_datas[group_idx]; + datas[group_idx] = static_cast(data.data()); + sizes[group_idx] = data.size(); + + GroupLayout layout{ + group["source_index"].cast(), + group["source_name"].cast(), + group["frame_ids"].cast>(), + group["frame_positions"].cast>>(), + group["first_frame_id"].cast(), + group["gop_len"].cast(), + }; + if (layout.source_index < 0) { + throw std::invalid_argument("group source_index must be non-negative"); + } + if (layout.frame_positions.size() != layout.frame_ids.size()) { + throw std::invalid_argument( + "group frame_positions must have one entry per frame_id"); + } + for (const auto& positions : layout.frame_positions) { + if (positions.empty() || std::any_of(positions.begin(), positions.end(), + [](int position) { return position < 0; })) { + throw std::invalid_argument( + "group frame_positions entries must be non-empty and non-negative"); + } + } + + source_names[group_idx] = layout.source_name; + frame_id_groups[group_idx] = layout.frame_ids; + layouts[group_idx] = std::move(layout); + } + + std::vector result; + { + py::gil_scoped_release release; + dec->decode_from_gop_groups(datas, sizes, source_names, frame_id_groups, as_bgr, + result); + } + + py::list decoded_groups; + size_t result_offset = 0; + for (const auto& layout : layouts) { + py::list frames; + for (size_t frame_idx = 0; frame_idx < layout.frame_ids.size(); ++frame_idx) { + if (result_offset >= result.size()) { + throw std::runtime_error( + "grouped decode returned fewer frames than requested"); + } + frames.append(py::cast(result[result_offset++])); + } + + py::dict decoded_group; + decoded_group["source_index"] = layout.source_index; + decoded_group["source_name"] = layout.source_name; + decoded_group["frame_ids"] = layout.frame_ids; + decoded_group["frame_positions"] = layout.frame_positions; + decoded_group["first_frame_id"] = layout.first_frame_id; + decoded_group["gop_len"] = layout.gop_len; + decoded_group["frames"] = std::move(frames); + decoded_groups.append(std::move(decoded_group)); + } + if (result_offset != result.size()) { + throw std::runtime_error("grouped decode returned more frames than requested"); + } + return decoded_groups; + } catch (const std::exception& e) { + throw std::runtime_error(e.what()); + } + }, + py::arg("groups"), py::arg("as_bgr") = false, + R"pbdoc( + Decode several target frames from each unique source/GOP bundle. + + Unlike :meth:`DecodeFromGOPListRGB`, which assigns one decoder task to + every output frame, this method assigns one decoder task to every GOP + group. All target frames in a group are produced during the same packet + traversal and NVDEC decode chain. + + Groups are variable length: each group's ``frame_ids`` and returned + ``frames`` lists contain the unique requested frames in that source/GOP, + so their lengths need not be aligned across groups. + + Args: + groups: Group dictionaries returned by :meth:`GetGOPGroups`. Each + dictionary carries one serialized GOP, its unique target frame + IDs, and every target's original positions. + as_bgr: Return BGR when true, RGB when false. + + Returns: + One dictionary per input group. Metadata and ``frame_positions`` are + preserved, and ``frames`` contains one RGBFrame per unique frame ID. + Callers can scatter each frame to all corresponding original + positions without decoding duplicates. + + Example: + >>> groups = demuxer.GetGOPGroups( + ... [{"filepath": camera_5_path, "frame_ids": [6, 7, 8, 9]}]) + >>> decoded_groups = decoder.DecodeFromGOPGroupsRGB(groups) + )pbdoc") .def( "DecodeFromGOPList", [](std::shared_ptr& dec, diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_random_decoder.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_random_decoder.cpp index d125dde..9248989 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_random_decoder.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_random_decoder.cpp @@ -71,7 +71,7 @@ void PyNvGopDecoder::decode_from_video(const std::vector& filepaths // lazy loading ensureCudaContextInitialized(); - ensureDemuxRunnersInitialized(); + ensureDemuxRunnersInitialized(total_frames); ensureDecodeRunnersInitialized(); // reset last decoded frame infos diff --git a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_separate_decoder.cpp b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_separate_decoder.cpp index 86700bb..b95a5c5 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_separate_decoder.cpp +++ b/packages/on_demand_video_decoder/ext_impl/src/PyNvOnDemandDecoder/src/PyNvGopDecoder_separate_decoder.cpp @@ -59,7 +59,7 @@ void PyNvGopDecoder::get_gop_internal( } // lazy loading - ensureDemuxRunnersInitialized(); + ensureDemuxRunnersInitialized(total_frames); // Initialize demuxers st = InitializeDemuxers(filepaths, demuxers, fastStreamInfos); @@ -580,6 +580,290 @@ void PyNvGopDecoder::decode_from_gop_list(const std::vector& dat nvtxRangePop(); } +void PyNvGopDecoder::decode_from_gop_groups(const std::vector& datas, + const std::vector& sizes, + const std::vector& source_names, + const std::vector>& frame_id_groups, bool as_bgr, + std::vector& output) { + nvtxRangePushA("DecodeFromGOPGroups"); + + const size_t num_groups = datas.size(); + if (sizes.size() != num_groups || source_names.size() != num_groups || + frame_id_groups.size() != num_groups) { + nvtxRangePop(); + throw std::invalid_argument( + "[ERROR] gop_datas, source_names, and frame_id_groups must have the same length"); + } + if (num_groups > static_cast(max_num_files)) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] number of GOP groups exceeds max_num_files"); + } + + if (num_groups == 0) { + output.clear(); + nvtxRangePop(); + return; + } + + std::vector color_ranges; + std::vector decoder_configs; + std::vector gop_lens; + std::vector first_frame_ids; + std::vector> packets_bytes; + std::vector> decode_idxs; + std::vector packet_binary_data_ptrs; + std::vector packet_binary_data_sizes; + + color_ranges.reserve(num_groups); + decoder_configs.reserve(num_groups); + gop_lens.reserve(num_groups); + first_frame_ids.reserve(num_groups); + packets_bytes.reserve(num_groups); + decode_idxs.reserve(num_groups); + packet_binary_data_ptrs.reserve(num_groups); + packet_binary_data_sizes.reserve(num_groups); + + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + const auto& target_ids = frame_id_groups[group_idx]; + if (target_ids.empty()) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] frame_id_groups must not contain an empty group"); + } + if (!std::is_sorted(target_ids.begin(), target_ids.end()) || + std::adjacent_find(target_ids.begin(), target_ids.end()) != target_ids.end()) { + nvtxRangePop(); + throw std::invalid_argument( + "[ERROR] frame IDs in each group must be strictly increasing and unique"); + } + if (target_ids.front() < 0) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] frame IDs must be non-negative"); + } + + std::vector bundle_color_ranges; + std::vector bundle_codec_ids; + std::vector bundle_widths; + std::vector bundle_heights; + std::vector bundle_frame_sizes; + std::vector bundle_gop_lens; + std::vector bundle_first_frame_ids; + std::vector> bundle_packets_bytes; + std::vector> bundle_decode_idxs; + std::vector bundle_packet_binary_data_ptrs; + std::vector bundle_packet_binary_data_sizes; + + const uint32_t frames_in_bundle = parseSerializedPacketData( + datas[group_idx], sizes[group_idx], bundle_color_ranges, bundle_codec_ids, bundle_widths, + bundle_heights, bundle_frame_sizes, bundle_gop_lens, bundle_first_frame_ids, bundle_packets_bytes, + bundle_decode_idxs, bundle_packet_binary_data_ptrs, bundle_packet_binary_data_sizes); + + if (frames_in_bundle != 1) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] grouped GOP payload must contain exactly one block"); + } + const int64_t first = bundle_first_frame_ids[0]; + const int64_t end = first + static_cast(bundle_gop_lens[0]); + if (target_ids.front() < first || target_ids.back() >= end) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] grouped GOP payload does not contain every target frame"); + } + + color_ranges.push_back(bundle_color_ranges[0]); + decoder_configs.push_back( + {bundle_codec_ids[0], bundle_widths[0], bundle_heights[0], bundle_frame_sizes[0]}); + gop_lens.push_back(bundle_gop_lens[0]); + first_frame_ids.push_back(bundle_first_frame_ids[0]); + packets_bytes.push_back(std::move(bundle_packets_bytes[0])); + decode_idxs.push_back(std::move(bundle_decode_idxs[0])); + packet_binary_data_ptrs.push_back(bundle_packet_binary_data_ptrs[0]); + packet_binary_data_sizes.push_back(bundle_packet_binary_data_sizes[0]); + } + + std::vector decoder_slots; + int status = AssignGroupedDecoderSlots(decoder_configs, decoder_slots); + if (status != 0) { + nvtxRangePop(); + throw std::runtime_error("[ERROR] AssignGroupedDecoderSlots failed"); + } + + std::vector>>> vpacket_queue(num_groups); + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + const size_t slot_idx = decoder_slots[group_idx]; + int skip_packets = 0; + const int last_frame_id = last_decoded_frame_infos[slot_idx].frame_id; + if (last_decoded_frame_infos[slot_idx].filename != source_names[group_idx] || + last_frame_id < first_frame_ids[group_idx] || + last_frame_id >= first_frame_ids[group_idx] + gop_lens[group_idx] || + last_frame_id >= frame_id_groups[group_idx].front()) { + skip_packets = 0; + } else { + skip_packets = last_decoded_frame_infos[slot_idx].packet_id; + } + if (skip_packets == 0) { + reset_last_decoded_frame_info(last_decoded_frame_infos[slot_idx]); + } + + vpacket_queue[group_idx] = std::make_unique>>(); + vpacket_queue[group_idx]->setSize(MAX_SIZE); + + size_t offset = 0; + for (size_t packet_idx = 0; packet_idx < packets_bytes[group_idx].size(); ++packet_idx) { + const int packet_bytes = packets_bytes[group_idx][packet_idx]; + int decode_idx = decode_idxs[group_idx][packet_idx]; + + if (skip_packets > 0) { + --skip_packets; + if (packet_bytes > 0) { + offset += static_cast(packet_bytes); + } + continue; + } + + if (packet_bytes == -1) { + vpacket_queue[group_idx]->push_back(std::make_tuple(nullptr, -1, 0)); + } else if (packet_bytes == 0) { + vpacket_queue[group_idx]->push_back(std::make_tuple(nullptr, 0, 0)); + } else if (packet_bytes > 0) { + if (offset + static_cast(packet_bytes) > packet_binary_data_sizes[group_idx]) { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] GOP packet data is truncated"); + } + uint8_t* video_data = const_cast(packet_binary_data_ptrs[group_idx] + offset); + offset += static_cast(packet_bytes); + vpacket_queue[group_idx]->push_back( + std::make_tuple(video_data, packet_bytes, decode_idx * 2)); + } else { + nvtxRangePop(); + throw std::invalid_argument("[ERROR] invalid negative GOP packet size"); + } + } + } + + status = main_decode_groups(color_ranges, decoder_configs, source_names, frame_id_groups, decoder_slots, + as_bgr, vpacket_queue, output); + if (status != 0) { + nvtxRangePop(); + throw std::runtime_error("[ERROR] main_decode_groups failed"); + } + + nvtxRangePop(); +} + +int PyNvGopDecoder::main_decode_groups( + const std::vector& color_ranges, const std::vector& decoder_configs, + const std::vector& source_names, const std::vector>& frame_id_groups, + const std::vector& decoder_slots, bool as_bgr, + std::vector>>>& vpacket_queue, + std::vector& output) { + ensureCudaContextInitialized(); + ensureDecodeRunnersInitialized(); + + const size_t num_groups = frame_id_groups.size(); + if (decoder_configs.size() != num_groups || decoder_slots.size() != num_groups) { + LOG(ERROR) << "Grouped decoder configuration or slot mapping size does not match group count"; + return -1; + } + std::unordered_set unique_slots; + for (const size_t slot_idx : decoder_slots) { + if (slot_idx >= vdec.size() || !unique_slots.insert(slot_idx).second) { + LOG(ERROR) << "Grouped decoder slot mapping contains an invalid or duplicate slot"; + return -1; + } + } + size_t total_output_frames = 0; + std::vector output_widths; + std::vector output_heights; + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + total_output_frames += frame_id_groups[group_idx].size(); + output_widths.insert(output_widths.end(), frame_id_groups[group_idx].size(), + decoder_configs[group_idx].width); + output_heights.insert(output_heights.end(), frame_id_groups[group_idx].size(), + decoder_configs[group_idx].height); + } + + const std::vector no_frame_sizes; + int status = InitGpuMemPool(output_heights, output_widths, no_frame_sizes, true); + if (status != 0) { + LOG(ERROR) << "InitGpuMemPool failed for grouped GOP decode"; + return status; + } + std::vector> flat_frame_buffers; + status = GetFileFrameBuffers(&output_widths, &output_heights, &no_frame_sizes, true, flat_frame_buffers); + if (status != 0) { + LOG(ERROR) << "GetFileFrameBuffers failed for grouped GOP decode"; + return status; + } + + std::vector> group_frame_buffers(num_groups); + size_t flat_buffer_idx = 0; + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + group_frame_buffers[group_idx].reserve(frame_id_groups[group_idx].size()); + for (size_t target_idx = 0; target_idx < frame_id_groups[group_idx].size(); ++target_idx) { + group_frame_buffers[group_idx].push_back(flat_frame_buffers[flat_buffer_idx++][0]); + } + } + + std::vector> rgb_frames(num_groups); + + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + try { + const size_t slot_idx = decoder_slots[group_idx]; + const AVColorRange color_range = static_cast(color_ranges[group_idx]); + rgb_frames[group_idx].reserve(frame_id_groups[group_idx].size()); +#ifdef PROCESS_SYNC + DecProc(color_range, vdec[slot_idx].get(), rgb_frames[group_idx], + group_frame_buffers[group_idx], vpacket_queue[group_idx].get(), + frame_id_groups[group_idx], as_bgr, source_names[group_idx], + last_decoded_frame_infos[slot_idx]); +#else + decode_runners[slot_idx].join(); + decode_runners[slot_idx].start(PyNvGopDecoder::DecProc, color_range, + vdec[slot_idx].get(), std::ref(rgb_frames[group_idx]), + group_frame_buffers[group_idx], vpacket_queue[group_idx].get(), + frame_id_groups[group_idx], as_bgr, source_names[group_idx], + std::ref(last_decoded_frame_infos[slot_idx])); +#endif + } catch (const std::exception& error) { + force_join_all(); + LOG(ERROR) << "Grouped DecProc failed: " << error.what(); + return -1; + } + } + +#ifndef PROCESS_SYNC + try { + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + decode_runners[decoder_slots[group_idx]].join(); + } + } catch (const std::exception& error) { + force_join_all(); + LOG(ERROR) << "Grouped decode thread join failed: " << error.what(); + return -1; + } +#endif + + output.clear(); + output.reserve(total_output_frames); + for (size_t group_idx = 0; group_idx < num_groups; ++group_idx) { + if (rgb_frames[group_idx].size() != frame_id_groups[group_idx].size()) { + force_join_all(); + LOG(ERROR) << "Grouped RGB decode produced " << rgb_frames[group_idx].size() + << " frames, expected " << frame_id_groups[group_idx].size() << " for group " + << group_idx; + return -1; + } + for (auto& frame : rgb_frames[group_idx]) { + output.push_back(std::move(frame)); + } + } + + // DecProc queues color conversion on cu_stream. Joining its CPU runner only + // guarantees that the work was submitted; the output frames are ready when + // this public synchronous API returns only after the stream is synchronized. + CUDA_DRVAPI_CALL(cuStreamSynchronize(this->cu_stream)); + return 0; +} + int PyNvGopDecoder::main_decode( const std::vector& color_ranges, const std::vector& codec_ids, std::vector& widths, std::vector& heights, std::vector& frame_sizes, const std::vector& filepaths, diff --git a/packages/on_demand_video_decoder/ext_impl/src/VideoCodecSDKUtils/helper_classes/NvCodec/NvDecoder/NvDecoder.h b/packages/on_demand_video_decoder/ext_impl/src/VideoCodecSDKUtils/helper_classes/NvCodec/NvDecoder/NvDecoder.h index 5b48f84..370e9d8 100644 --- a/packages/on_demand_video_decoder/ext_impl/src/VideoCodecSDKUtils/helper_classes/NvCodec/NvDecoder/NvDecoder.h +++ b/packages/on_demand_video_decoder/ext_impl/src/VideoCodecSDKUtils/helper_classes/NvCodec/NvDecoder/NvDecoder.h @@ -133,6 +133,13 @@ class NvDecoder { */ CUcontext GetContext() { return m_cuContext; } + /** + * @brief Return the codec and current decoded dimensions. + */ + cudaVideoCodec GetCodec() const { return m_eCodec; } + int GetCurrentWidth() const { return static_cast(m_nWidth); } + int GetCurrentHeight() const { return static_cast(m_nLumaHeight); } + /** * @brief This function is used to get the output frame width. * NV12/P016 output format width is 2 byte aligned because of U and V interleave diff --git a/packages/on_demand_video_decoder/tests/test_grouped_resolution_growth.py b/packages/on_demand_video_decoder/tests/test_grouped_resolution_growth.py new file mode 100644 index 0000000..a279a4d --- /dev/null +++ b/packages/on_demand_video_decoder/tests/test_grouped_resolution_growth.py @@ -0,0 +1,114 @@ +# 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. + +"""Regression coverage for grouped decoder resolution-aware slot reuse.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest +import torch + +av = pytest.importorskip("av") + +from accvlab import on_demand_video_decoder as nvc # noqa: E402 + + +def _write_h264_clip(path: Path, *, width: int, height: int) -> None: + try: + output = av.open(str(path), mode="w") + stream = output.add_stream("libx264", rate=10) + except (av.AVError, ValueError) as exc: + pytest.skip(f"PyAV libx264 encoder is unavailable: {exc}") + stream.width = width + stream.height = height + stream.pix_fmt = "yuv420p" + stream.gop_size = 4 + stream.codec_context.max_b_frames = 0 + for frame_id in range(8): + pixels = np.full((height, width, 3), frame_id * 17, dtype=np.uint8) + frame = av.VideoFrame.from_ndarray(pixels, format="rgb24") + for packet in stream.encode(frame): + output.mux(packet) + for packet in stream.encode(): + output.mux(packet) + output.close() + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="grouped decode requires CUDA") +def test_grouped_decoder_replaces_single_slot_when_resolution_changes(tmp_path: Path) -> None: + small_path = tmp_path / "small.mp4" + large_path = tmp_path / "large.mp4" + _write_h264_clip(small_path, width=64, height=64) + _write_h264_clip(large_path, width=192, height=128) + + demuxer = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) + decoder = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) + requests = [ + (small_path, (64, 64, 3)), + (large_path, (128, 192, 3)), + ] * 10 + for path, expected_shape in requests: + groups = demuxer.GetGOPGroups([{"filepath": str(path), "frame_ids": [0]}]) + decoded = decoder.DecodeFromGOPGroupsRGB(groups) + assert decoded[0]["frames"][0].shape == expected_shape + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="grouped decode requires CUDA") +def test_grouped_decoder_pool_keeps_inactive_shape_slots(tmp_path: Path) -> None: + small_path = tmp_path / "small.mp4" + large_path = tmp_path / "large.mp4" + _write_h264_clip(small_path, width=64, height=64) + _write_h264_clip(large_path, width=192, height=128) + + demuxer = nvc.CreateGopDecoder(maxfiles=2, iGpu=0) + decoder = nvc.CreateGopDecoder(maxfiles=2, iGpu=0) + requests = [ + (small_path, (64, 64, 3)), + (large_path, (128, 192, 3)), + ] * 10 + for path, expected_shape in requests: + groups = demuxer.GetGOPGroups([{"filepath": str(path), "frame_ids": [0]}]) + decoded = decoder.DecodeFromGOPGroupsRGB(groups) + assert decoded[0]["frames"][0].shape == expected_shape + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="grouped decode requires CUDA") +def test_grouped_decoder_pool_matches_shape_when_group_order_changes(tmp_path: Path) -> None: + small_a_path = tmp_path / "small_a.mp4" + small_b_path = tmp_path / "small_b.mp4" + large_path = tmp_path / "large.mp4" + _write_h264_clip(small_a_path, width=64, height=64) + _write_h264_clip(small_b_path, width=64, height=64) + _write_h264_clip(large_path, width=192, height=128) + + demuxer = nvc.CreateGopDecoder(maxfiles=3, iGpu=0) + decoder = nvc.CreateGopDecoder(maxfiles=3, iGpu=0) + requests = [ + ( + [small_a_path, large_path, small_b_path], + [(64, 64, 3), (128, 192, 3), (64, 64, 3)], + ), + ( + [large_path, small_b_path, small_a_path], + [(128, 192, 3), (64, 64, 3), (64, 64, 3)], + ), + ] * 10 + for paths, expected_shapes in requests: + groups = demuxer.GetGOPGroups([{"filepath": str(path), "frame_ids": [0]} for path in paths]) + decoded = decoder.DecodeFromGOPGroupsRGB(groups) + actual_shapes = [group["frames"][0].shape for group in decoded] + assert actual_shapes == expected_shapes diff --git a/packages/on_demand_video_decoder/tests/test_open_gop_overlap.py b/packages/on_demand_video_decoder/tests/test_open_gop_overlap.py index d8cc6df..257b4a9 100644 --- a/packages/on_demand_video_decoder/tests/test_open_gop_overlap.py +++ b/packages/on_demand_video_decoder/tests/test_open_gop_overlap.py @@ -138,6 +138,113 @@ def test_decodefromgoplist_rejects_cross_gop_frame(self, decoder): assert "GOP range" in str(exc_info.value) or "frame_id" in str(exc_info.value) +class TestGroupedGopApis: + """One source/GOP payload and one decode chain can produce many target frames.""" + + @staticmethod + def _scatter(decoded_groups, frame_counts): + frames_by_source = [[None] * frame_count for frame_count in frame_counts] + for group in decoded_groups: + assert len(group["frames"]) == len(group["frame_positions"]) + source_frames = frames_by_source[group["source_index"]] + for frame, positions in zip(group["frames"], group["frame_positions"]): + for position in positions: + assert source_frames[position] is None + source_frames[position] = frame + assert all(frame is not None for frames in frames_by_source for frame in frames) + return frames_by_source + + @pytest.mark.skipif(not os.path.isdir("/proc/self/task"), reason="requires Linux thread accounting") + def test_demux_runners_are_created_for_active_sources_only(self): + demuxer = nvc.CreateGopDecoder(maxfiles=64, iGpu=0) + threads_before = len(os.listdir("/proc/self/task")) + + groups = demuxer.GetGOPGroups([{"filepath": OPEN_GOP_SAMPLE, "frame_ids": [6]}]) + + threads_created = len(os.listdir("/proc/self/task")) - threads_before + assert len(groups) == 1 + assert threads_created < 8, ( + f"one active source created {threads_created} threads; " + "maxfiles must remain a capacity limit rather than an eager thread count" + ) + + def test_same_gop_is_serialized_and_decoded_once(self): + demuxer = nvc.CreateGopDecoder(maxfiles=4, iGpu=0) + grouped_decoder = nvc.CreateGopDecoder(maxfiles=4, iGpu=0) + baseline_decoder = nvc.CreateGopDecoder(maxfiles=5, iGpu=0) + requested_ids = [9, 6, 8, 7, 7] + expected_ids = [6, 7, 8, 9] + + groups = demuxer.GetGOPGroups([{"filepath": OPEN_GOP_SAMPLE, "frame_ids": requested_ids}]) + assert len(groups) == 1 + group = groups[0] + assert group["source_index"] == 0 + assert group["source_name"] == OPEN_GOP_SAMPLE + assert group["frame_ids"] == expected_ids + assert group["frame_positions"] == [[1], [3, 4], [2], [0]] + assert group["first_frame_id"] == 0 + assert group["gop_len"] == 20 + + native_demuxer = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) + native_groups = native_demuxer.GetGOPGroups( + [{"filepath": OPEN_GOP_SAMPLE, "frame_ids": requested_ids}] + ) + assert len(native_groups) == 1 + for key in ( + "source_index", + "source_name", + "frame_ids", + "frame_positions", + "first_frame_id", + "gop_len", + ): + assert native_groups[0][key] == group[key] + + legacy_groups = demuxer.GetGOPList([OPEN_GOP_SAMPLE] * len(expected_ids), expected_ids) + legacy_bytes = sum(gop_data.nbytes for gop_data, _, _ in legacy_groups) + assert group["gop_data"].nbytes * 3 < legacy_bytes + + baseline = baseline_decoder.DecodeN12ToRGB([OPEN_GOP_SAMPLE] * len(requested_ids), requested_ids) + baseline_tensors = [torch.as_tensor(frame).clone() for frame in baseline] + + decoded_groups = grouped_decoder.DecodeFromGOPGroupsRGB(groups) + assert len(decoded_groups) == 1 + assert len(decoded_groups[0]["frames"]) == len(expected_ids) + grouped = self._scatter(decoded_groups, [len(requested_ids)])[0] + for actual, expected in zip(grouped, baseline_tensors): + assert torch.equal(torch.as_tensor(actual), expected) + + def test_cross_gop_request_retains_order_and_duplicate_positions(self): + demuxer = nvc.CreateGopDecoder(maxfiles=2, iGpu=0) + grouped_decoder = nvc.CreateGopDecoder(maxfiles=3, iGpu=0) + baseline_decoder = nvc.CreateGopDecoder(maxfiles=4, iGpu=0) + requested_ids = [45, 6, 25, 6] + + groups = demuxer.GetGOPGroups([{"filepath": OPEN_GOP_SAMPLE, "frame_ids": requested_ids}]) + assert [group["frame_ids"] for group in groups] == [[6], [25], [45]] + assert [group["frame_positions"] for group in groups] == [[[1, 3]], [[2]], [[0]]] + assert [(group["first_frame_id"], group["gop_len"]) for group in groups] == [ + (0, 20), + (20, 20), + (40, 20), + ] + + decoded_groups = grouped_decoder.DecodeFromGOPGroupsRGB(groups) + assert sum(len(group["frames"]) for group in decoded_groups) == 3 + grouped = self._scatter(decoded_groups, [len(requested_ids)])[0] + baseline = baseline_decoder.DecodeN12ToRGB([OPEN_GOP_SAMPLE] * len(requested_ids), requested_ids) + for actual, expected in zip(grouped, baseline): + assert torch.equal(torch.as_tensor(actual), torch.as_tensor(expected)) + + def test_grouped_decode_rejects_targets_spanning_two_gops(self): + decoder = nvc.CreateGopDecoder(maxfiles=1, iGpu=0) + group = decoder.GetGOPGroups([{"filepath": OPEN_GOP_SAMPLE, "frame_ids": [6]}])[0] + group["frame_ids"] = [6, 25] + group["frame_positions"] = [[0], [1]] + with pytest.raises(Exception, match="does not contain every target frame"): + decoder.DecodeFromGOPGroupsRGB([group]) + + class TestSharedGopStoreOpenGop: """End-to-end check that ``GetGOPList`` + ``SharedGopStore.put`` produce independent entries for adjacent open-GOP GOPs."""