deepstream-generate-pipeline

DeepStream Pipeline Builder Generate ready-to-run pipelines for NVIDIA DeepStream SDK by collecting pipeline requirements through an interactive questionnaire, then assembling the pipeline using a standalone BM25 retrieval backend with structural metadata boosting (similarity search over 270+ verified pipelines, zero external dependencies). Prerequisites - Python: 3.8+ (stdlib only — no pip packages required) - DeepStream SDK: Installed at (for validation and element verification) - GStreamer: and on (installed with DeepStream) - Platform: x86 dGPU (T4, A100, L40, RTX, etc.) or aarch64 — Jets…

)\n\n# ---------------------------------------------------------------------------\n# Known DeepStream/GStreamer element properties\n# ---------------------------------------------------------------------------\nKNOWN_ELEMENT_PROPERTIES = {\n \"filesrc\": {\n \"location\", \"num-buffers\", \"typefind\", \"blocksize\", \"do-timestamp\",\n },\n \"nvinfer\": {\n \"config-file-path\", \"batch-size\", \"unique-id\", \"infer-on-gie-id\",\n \"input-tensor-meta\", \"interval\", \"model-engine-file\", \"gpu-id\",\n \"process-mode\", \"network-type\", \"cluster-mode\", \"operate-on-gie-id\",\n \"operate-on-class-ids\", \"filter-out-class-ids\", \"output-tensor-meta\",\n \"clip-object-outside-roi\", \"infer-on-class-ids\", \"raw-output-file-write\",\n },\n \"nvstreammux\": {\n \"name\", \"batch-size\", \"width\", \"height\", \"batched-push-timeout\",\n \"live-source\", \"enable-padding\", \"num-surfaces-per-frame\",\n \"gpu-id\", \"nvbuf-memory-type\", \"buffer-pool-size\",\n \"compute-hw\", \"interpolation-method\", \"adaptive-batching\",\n \"sync-inputs\",\n },\n \"nvvideoconvert\": {\n \"flip-method\", \"src-crop\", \"dest-crop\", \"gpu-id\",\n \"nvbuf-memory-type\", \"interpolation-method\", \"compute-hw\",\n },\n \"nvtracker\": {\n \"ll-lib-file\", \"ll-config-file\", \"gpu-id\", \"tracker-width\",\n \"tracker-height\", \"display-tracking-id\", \"compute-hw\",\n },\n \"nvdsosd\": {\n \"process-mode\", \"display-text\", \"display-bbox\", \"display-mask\",\n \"display-clock\", \"clock-font\", \"clock-font-size\", \"gpu-id\",\n },\n \"nvmultistreamtiler\": {\n \"rows\", \"columns\", \"width\", \"height\", \"gpu-id\",\n \"nvbuf-memory-type\", \"compute-hw\", \"show-source\",\n },\n \"nvdspreprocess\": {\n \"config-file\", \"gpu-id\", \"operate-on-gie-id\",\n },\n \"nvv4l2decoder\": {\n \"gpu-id\", \"num-extra-surfaces\", \"cudadec-memtype\",\n \"drop-frame-interval\", \"disable-dpb\",\n },\n \"nvv4l2h264enc\": {\n \"bitrate\", \"maxperf-enable\", \"preset-level\", \"profile\",\n \"control-rate\", \"gpu-id\", \"iframeinterval\",\n \"qp-range\", \"constqp\", \"initqp\",\n },\n \"nvv4l2h265enc\": {\n \"bitrate\", \"maxperf-enable\", \"preset-level\", \"profile\",\n \"control-rate\", \"gpu-id\", \"iframeinterval\",\n \"qp-range\", \"constqp\", \"initqp\",\n },\n \"nveglglessink\": {\"sync\", \"qos\", \"max-lateness\", \"window-x\", \"window-y\"},\n \"nv3dsink\": {\"sync\", \"qos\"},\n \"fakesink\": {\"sync\", \"async\", \"dump\", \"signal-handoffs\", \"num-buffers\"},\n \"filesink\": {\"location\", \"sync\", \"async\", \"append\", \"buffer-mode\", \"buffer-size\"},\n \"jpegenc\": {\"quality\", \"idct-method\", \"snapshot\"},\n \"jpegparse\": set(),\n \"nvjpegdec\": {\"gpu-id\"},\n \"pngenc\": {\"compression-level\", \"snapshot\"},\n \"pngdec\": set(),\n \"h264parse\": {\"config-interval\", \"disable-passthrough\"},\n \"h265parse\": {\"config-interval\", \"disable-passthrough\"},\n \"qtdemux\": {\"name\"},\n \"qtmux\": {\"fragment-duration\", \"faststart\"},\n \"rtph264pay\": {\"config-interval\", \"pt\", \"mtu\"},\n \"rtph265pay\": {\"config-interval\", \"pt\", \"mtu\"},\n \"udpsink\": {\"host\", \"port\", \"sync\", \"async\"},\n \"uridecodebin\": {\"uri\", \"caps\", \"buffer-size\", \"use-buffering\"},\n \"v4l2src\": {\"device\", \"num-buffers\", \"io-mode\"},\n \"videotestsrc\": {\"pattern\", \"num-buffers\", \"is-live\"},\n \"videoconvert\": set(),\n \"videoparse\": {\"width\", \"height\", \"format\", \"framerate\"},\n \"xvimagesink\": {\"sync\"},\n \"decodebin\": {\"name\"},\n \"dsexample\": {\n \"full-frame\", \"processing-width\", \"processing-height\",\n \"gpu-id\", \"blur-objects\", \"unique-id\",\n },\n \"nvdewarper\": {\n \"config-file\", \"source-id\", \"nvbuf-memory-type\", \"gpu-id\",\n \"num-output-buffers\",\n },\n \"nvsegvisual\": {\"width\", \"height\", \"gpu-id\"},\n \"nvdspostprocess\": {\"postprocesslib-config-file\", \"postprocesslib-name\", \"gpu-id\"},\n \"queue\": {\"max-size-buffers\", \"max-size-bytes\", \"max-size-time\", \"leaky\"},\n \"tee\": {\"name\"},\n \"nvstreamdemux\": {\"name\"},\n \"nvvideotestsrc\": {\n \"num-buffers\", \"is-live\", \"location\", \"file-loop\",\n \"max-jitter\", \"fixed-jitter\",\n },\n \"nvurisrcbin\": {\"uri\"},\n \"nvegltransform\": set(),\n \"nvdslogger\": {\"fps-measurement-interval-sec\"},\n \"fpsdisplaysink\": {\"video-sink\", \"sync\", \"text-overlay\"},\n \"identity\": {\"silent\", \"dump\", \"single-segment\"},\n \"rtspsrc\": {\"location\", \"latency\", \"protocols\", \"user-id\", \"user-pw\"},\n \"rtph264depay\": set(),\n \"avdec_h264\": set(),\n \"videoscale\": set(),\n \"audioconvert\": set(),\n \"audioresample\": set(),\n \"aacparse\": set(),\n \"avdec_aac\": set(),\n \"nvdsaudiotemplate\": {\"customlib-name\", \"customlib-props\"},\n \"tcpserversink\": {\"host\", \"port\"},\n \"tcpclientsrc\": {\"host\", \"port\"},\n \"rtpgstpay\": {\"config-interval\"},\n \"rtpstreampay\": set(),\n \"rtpstreamdepay\": set(),\n \"rtpgstdepay\": set(),\n \"nvinferserver\": {\n \"config-file-path\", \"unique-id\", \"infer-on-gie-id\", \"batch-size\",\n \"interval\", \"gpu-id\", \"process-mode\", \"input-tensor-meta\",\n \"operate-on-gie-id\", \"operate-on-class-ids\", \"infer-on-class-ids\",\n \"output-tensor-meta\",\n },\n \"nvdsanalytics\": {\"config-file\", \"unique-id\"},\n \"nvmsgconv\": {\n \"config\", \"payload-type\", \"msg2p-lib\", \"comp-id\", \"debug-payload-dir\",\n \"msg2p-newapi\",\n },\n \"nvmsgbroker\": {\n \"proto-lib\", \"conn-str\", \"topic\", \"config\", \"comp-id\", \"sync\",\n \"new-api\",\n },\n \"nvdsmetamux\": set(),\n}\n\nPASSTHROUGH_ELEMENTS = {\n \"capsfilter\", \"identity\", \"multiqueue\",\n \"input-selector\", \"output-selector\",\n}\n\n# Elements that output video/x-raw in system (CPU) memory — NOT NVMM.\n# Feeding these directly into NVMM-requiring elements causes linking errors.\nSYSTEM_MEMORY_OUTPUT = {\"nvjpegdec\", \"pngdec\", \"jpegdec\", \"videoconvert\", \"videoscale\"}\n\n# Elements that require video/x-raw(memory:NVMM) on their sink pads.\nNVMM_INPUT_REQUIRED = {\"nvstreammux\", \"nvv4l2h264enc\", \"nvv4l2h265enc\"}\n\n# Platform-specific sink elements — using the wrong one causes runtime failures.\nDGPU_ONLY_SINKS = {\"nveglglessink\", \"nvegltransform\"}\nJETSON_ONLY_SINKS = {\"nv3dsink\"}\n\n# Element ordering: elements that should appear BEFORE others in the pipeline.\n# Key = element that must come first, Value = set of elements it must precede.\nEXPECTED_ORDER = {\n \"filesrc\": {\"nvv4l2decoder\", \"decodebin\", \"jpegparse\", \"h264parse\", \"h265parse\", \"qtdemux\"},\n \"nvv4l2decoder\": {\"nvstreammux\", \"nvinfer\", \"nvinferserver\", \"nvvideoconvert\"},\n \"nvstreammux\": {\"nvinfer\", \"nvinferserver\", \"nvtracker\", \"nvdsosd\", \"nvmultistreamtiler\"},\n \"nvinfer\": {\"nvtracker\", \"nvdsosd\", \"nveglglessink\", \"nv3dsink\", \"filesink\", \"fakesink\"},\n \"nvinferserver\": {\"nvtracker\", \"nvdsosd\", \"nveglglessink\", \"nv3dsink\", \"filesink\", \"fakesink\"},\n \"nvtracker\": {\"nvdsosd\", \"nveglglessink\", \"nv3dsink\", \"filesink\"},\n}\n\n\n# ---------------------------------------------------------------------------\n# Element existence check via gst-inspect-1.0\n# ---------------------------------------------------------------------------\ndef check_element_exists(element_name):\n \"\"\"Check if a GStreamer element is registered using gst-inspect-1.0.\"\"\"\n if not _SAFE_ELEMENT_NAME.match(element_name):\n return None\n try:\n result = subprocess.run(\n [\"gst-inspect-1.0\", element_name],\n capture_output=True, text=True, timeout=5\n )\n return result.returncode == 0\n except (subprocess.TimeoutExpired, FileNotFoundError):\n return None\n\n\n# ---------------------------------------------------------------------------\n# Pipeline parser\n# ---------------------------------------------------------------------------\ndef extract_elements_and_properties(pipeline_str):\n \"\"\"\n Parse a gst-launch-1.0 pipeline string and extract elements with properties.\n Returns list of dicts: [{\"name\": str, \"properties\": dict, \"raw\": str}]\n \"\"\"\n pipeline_str = re.sub(r\"^gst-launch-1\\.0\\s+\", \"\", pipeline_str.strip())\n pipeline_str = re.sub(r\"^(?:-[a-zA-Z]+\\s+)+\", \"\", pipeline_str)\n pipeline_str = re.sub(r\"\\\\\\s*\\n\\s*\", \" \", pipeline_str)\n\n segments = re.split(r\"\\s+!\\s+\", pipeline_str)\n\n elements = []\n pad_refs = []\n\n for segment in segments:\n segment = segment.strip()\n if not segment:\n continue\n\n if segment.startswith((\"'\", '\"', \"video/\", \"audio/\", \"image/\", \"application/\")):\n continue\n\n parts = segment.split()\n if not parts:\n continue\n\n idx = 0\n # A segment may start with pad refs (e.g. \"m.sink_0 nvstreammux name=m ...\")\n while idx \u003c len(parts):\n tok = parts[idx]\n if \".\" in tok and not tok.startswith(\"/\") and \"=\" not in tok:\n pad_refs.append(tok)\n idx += 1\n else:\n break\n\n if idx >= len(parts):\n continue\n\n element_name = parts[idx]\n\n if element_name.startswith((\"'\", '\"', \"video/\", \"audio/\", \"image/\")):\n continue\n\n properties = {}\n for part in parts[idx + 1:]:\n if part.startswith((\"'\", '\"')):\n break\n if \"=\" in part:\n key, _, val = part.partition(\"=\")\n properties[key] = val\n\n elements.append({\n \"name\": element_name,\n \"properties\": properties,\n \"raw\": segment,\n })\n\n return elements, pad_refs\n\n\n# ---------------------------------------------------------------------------\n# Validation checks\n# ---------------------------------------------------------------------------\ndef validate_syntax(pipeline_str):\n \"\"\"Check for common syntax errors.\"\"\"\n errors = []\n warnings = []\n\n single_quotes = pipeline_str.count(\"'\")\n double_quotes = pipeline_str.count('\"')\n if single_quotes % 2 != 0:\n errors.append(\"Unbalanced single quotes in pipeline\")\n if double_quotes % 2 != 0:\n errors.append(\"Unbalanced double quotes in pipeline\")\n\n if re.search(r\"!\\s*!\", pipeline_str):\n errors.append(\"Empty pipe segment detected (consecutive '!' without element)\")\n\n cleaned = re.sub(r\"^gst-launch-1\\.0\\s+\", \"\", pipeline_str.strip())\n cleaned = re.sub(r\"^(?:-[a-zA-Z]+\\s+)+\", \"\", cleaned)\n cleaned = re.sub(r\"\\\\\\s*\\n\\s*\", \" \", cleaned).strip()\n if cleaned.startswith(\"!\"):\n errors.append(\"Pipeline starts with '!' — missing source element\")\n if cleaned.endswith(\"!\"):\n errors.append(\"Pipeline ends with '!' — missing sink element\")\n\n # Underscore-typo check: scope to the segment that owns nvinfer/nvinferserver\n # so a hyphenated config-file-path on a different element (or in a comment)\n # does not mask an underscore typo on the actual nvinfer.\n if re.search(r\"\\bnvinfer(?:server)?\\b[^!]*\\bconfig_file_path=\", pipeline_str):\n errors.append(\"nvinfer uses 'config-file-path' (hyphens), not 'config_file_path' (underscores)\")\n\n # Note: this only inspects the segment immediately following the\n # 'nvstreammux' token (non-greedy match up to the next '!' or end of\n # string). Properties placed after a '!' separator — rare, but possible\n # via pipeline-continuation patterns — will not be considered. Adequate\n # for the single-line pipelines this skill produces today.\n mux_match = re.search(r\"nvstreammux\\b(.*?)(?:!|$)\", pipeline_str)\n if mux_match and \"name=\" not in mux_match.group(1):\n warnings.append(\"nvstreammux without 'name=' — multi-source linking may fail\")\n\n return errors, warnings\n\n\ndef validate_elements(pipeline_str):\n \"\"\"Validate all elements exist and properties are recognized.\"\"\"\n elements, pad_refs = extract_elements_and_properties(pipeline_str)\n errors = []\n warnings = []\n checked_elements = set()\n\n has_gst_inspect = which(\"gst-inspect-1.0\") is not None\n\n for elem in elements:\n name = elem[\"name\"]\n\n if name not in checked_elements:\n checked_elements.add(name)\n if has_gst_inspect and name not in KNOWN_ELEMENT_PROPERTIES and name not in PASSTHROUGH_ELEMENTS:\n exists = check_element_exists(name)\n if exists is False:\n errors.append(f\"Unknown element '{name}' — not found by gst-inspect-1.0\")\n continue\n elif exists is None:\n warnings.append(f\"Could not verify element '{name}' — gst-inspect-1.0 timed out\")\n\n if name in KNOWN_ELEMENT_PROPERTIES:\n known_props = KNOWN_ELEMENT_PROPERTIES[name]\n for prop in elem[\"properties\"]:\n if known_props and prop not in known_props and prop != \"name\":\n warnings.append(\n f\"Unrecognized property '{prop}' on element '{name}'\"\n f\" — verify with gst-inspect-1.0 {name}\"\n )\n\n return elements, pad_refs, errors, warnings\n\n\ndef validate_memory_format(pipeline_str, elements, pad_refs):\n \"\"\"\n Detect memory format mismatches — e.g. nvjpegdec (system memory) feeding\n directly into nvstreammux (requires NVMM) without nvvideoconvert in between.\n\n Handles two patterns:\n 1. Consecutive elements: ... ! nvjpegdec ! nvstreammux ...\n 2. Pad-ref branches: ... ! nvjpegdec ! mux.sink_N (branch end)\n \"\"\"\n errors = []\n flagged_pairs = set()\n\n # Build mux alias set from named nvstreammux elements\n mux_aliases = set()\n for e in elements:\n if e[\"name\"] == \"nvstreammux\" and \"name\" in e[\"properties\"]:\n mux_aliases.add(e[\"properties\"][\"name\"])\n\n # --- Check 1: consecutive elements in the parsed list ---\n for i in range(len(elements) - 1):\n curr = elements[i][\"name\"]\n nxt = elements[i + 1][\"name\"]\n if curr in SYSTEM_MEMORY_OUTPUT and nxt in NVMM_INPUT_REQUIRED:\n pair = (curr, nxt)\n if pair not in flagged_pairs:\n flagged_pairs.add(pair)\n errors.append(\n f\"'{curr}' outputs system memory (video/x-raw) but \"\n f\"'{nxt}' requires NVMM — \"\n f\"insert 'nvvideoconvert' between them\"\n )\n\n if not mux_aliases:\n return errors\n\n # --- Check 2: segment before a mux pad ref (catches multi-branch) ---\n clean = re.sub(r\"^gst-launch-1\\.0\\s+\", \"\", pipeline_str.strip())\n clean = re.sub(r\"^(?:-[a-zA-Z]+\\s+)+\", \"\", clean)\n clean = re.sub(r\"\\\\\\s*\\n\\s*\", \" \", clean)\n segments = re.split(r\"\\s+!\\s+\", clean)\n\n for i, seg in enumerate(segments):\n seg_stripped = seg.strip()\n has_mux_pad = any(\n seg_stripped.startswith(f\"{a}.sink_\") or f\" {a}.sink_\" in seg_stripped\n for a in mux_aliases\n )\n if not has_mux_pad or i == 0:\n continue\n prev_seg = segments[i - 1].strip()\n prev_parts = prev_seg.split()\n if not prev_parts:\n continue\n prev_elem = prev_parts[0]\n if prev_elem in SYSTEM_MEMORY_OUTPUT:\n pair = (prev_elem, \"nvstreammux\")\n if pair not in flagged_pairs:\n flagged_pairs.add(pair)\n errors.append(\n f\"'{prev_elem}' outputs system memory (video/x-raw) but \"\n f\"feeds into 'nvstreammux' via pad ref — \"\n f\"insert 'nvvideoconvert' between them\"\n )\n\n return errors\n\n\ndef validate_pipeline_structure(elements, pad_refs):\n \"\"\"Check source/sink presence, required props, and named-pad consistency.\"\"\"\n warnings = []\n\n if not elements:\n return [\"Pipeline is empty — no elements found\"], warnings\n\n errors = []\n element_names = [e[\"name\"] for e in elements]\n named_elements = {}\n for e in elements:\n if \"name\" in e[\"properties\"]:\n named_elements[e[\"properties\"][\"name\"]] = e[\"name\"]\n\n source_elements = {\n \"filesrc\", \"uridecodebin\", \"v4l2src\", \"videotestsrc\", \"rtspsrc\",\n \"multifilesrc\", \"nvvideotestsrc\", \"nvurisrcbin\", \"tcpclientsrc\",\n }\n has_source = any(e in source_elements for e in element_names)\n if not has_source:\n warnings.append(\n \"No recognized source element (filesrc, uridecodebin, v4l2src, etc.)\"\n \" — may be intentional if using a named pad\"\n )\n\n sink_elements = {\n \"nveglglessink\", \"nv3dsink\", \"filesink\", \"fakesink\", \"udpsink\",\n \"xvimagesink\", \"autovideosink\", \"fpsdisplaysink\", \"tcpserversink\",\n }\n has_sink = any(e in sink_elements for e in element_names)\n if not has_sink:\n warnings.append(\"No recognized sink element — pipeline needs a sink to terminate\")\n\n for elem in elements:\n if elem[\"name\"] == \"nvinfer\" and \"config-file-path\" not in elem[\"properties\"]:\n errors.append(\"nvinfer missing required 'config-file-path' property\")\n if elem[\"name\"] == \"nvinferserver\" and \"config-file-path\" not in elem[\"properties\"]:\n errors.append(\"nvinferserver missing required 'config-file-path' property\")\n if elem[\"name\"] == \"filesrc\" and \"location\" not in elem[\"properties\"]:\n errors.append(\"filesrc missing required 'location' property\")\n\n # Named-pad cross-reference validation\n for ref in pad_refs:\n alias = ref.split(\".\")[0]\n if alias not in named_elements:\n if not any(alias == e[\"properties\"].get(\"name\", \"\") for e in elements):\n warnings.append(\n f\"Pad reference '{ref}' uses alias '{alias}'\"\n f\" but no element has name='{alias}'\"\n )\n\n # batch-size / sink-pad count consistency\n for elem in elements:\n if elem[\"name\"] == \"nvstreammux\" and \"batch-size\" in elem[\"properties\"]:\n try:\n batch = int(elem[\"properties\"][\"batch-size\"])\n except ValueError:\n continue\n alias = elem[\"properties\"].get(\"name\", \"\")\n if alias:\n pad_count = sum(\n 1 for r in pad_refs\n if r.startswith(f\"{alias}.sink_\")\n )\n if pad_count > 0 and pad_count != batch and batch > 1:\n warnings.append(\n f\"nvstreammux '{alias}' has batch-size={batch}\"\n f\" but {pad_count} sink pad(s) connected\"\n )\n\n return errors, warnings\n\n\ndef validate_platform_sink(elements):\n \"\"\"\n Detect platform-sink mismatches:\n - nveglglessink used alongside nv3dsink hints (Jetson pipeline with dGPU sink)\n - Both Jetson-only and dGPU-only sinks in the same pipeline\n \"\"\"\n warnings = []\n element_names = {e[\"name\"] for e in elements}\n\n has_dgpu_sink = bool(element_names & DGPU_ONLY_SINKS)\n has_jetson_sink = bool(element_names & JETSON_ONLY_SINKS)\n\n if has_dgpu_sink and has_jetson_sink:\n warnings.append(\n \"Pipeline mixes dGPU-only sinks (nveglglessink) and Jetson-only \"\n \"sinks (nv3dsink) — pick one for your target platform\"\n )\n\n return warnings\n\n\ndef validate_element_ordering(elements):\n \"\"\"\n Detect nonsensical element ordering — e.g. an encoder appearing before a\n decoder, or a sink element appearing before inference elements.\n Only flags clear-cut ordering violations for elements in EXPECTED_ORDER.\n \"\"\"\n warnings = []\n # Build position map: element_name -> first occurrence index\n first_pos = {}\n for i, e in enumerate(elements):\n if e[\"name\"] not in first_pos:\n first_pos[e[\"name\"]] = i\n\n for before_elem, after_set in EXPECTED_ORDER.items():\n if before_elem not in first_pos:\n continue\n before_idx = first_pos[before_elem]\n for after_elem in after_set:\n if after_elem in first_pos and first_pos[after_elem] \u003c before_idx:\n warnings.append(\n f\"Element '{after_elem}' appears before '{before_elem}' \"\n f\"— expected '{before_elem}' first\"\n )\n\n return warnings\n\n\ndef validate_with_gst_launch(pipeline_str, pad_refs=None):\n \"\"\"\n Use gst-launch-1.0 for a dry-run parse check by substituting\n fakesrc/fakesink and stripping elements that need real data.\n\n Multi-stream pipelines (those with named pad refs like m.sink_0) are\n skipped — fakesrc cannot negotiate caps through named pads on\n nvstreammux, causing false-positive linking errors.\n \"\"\"\n errors = []\n warnings = []\n\n if pad_refs and len(pad_refs) > 0:\n return errors, warnings\n\n if not which(\"gst-launch-1.0\"):\n warnings.append(\"gst-launch-1.0 not found — skipping live pipeline parse check\")\n return errors, warnings\n\n test_pipeline = re.sub(r\"^gst-launch-1\\.0\\s+\", \"\", pipeline_str.strip())\n test_pipeline = re.sub(r\"^(?:-[a-zA-Z]+\\s+)+\", \"\", test_pipeline)\n test_pipeline = re.sub(r\"\\\\\\s*\\n\\s*\", \" \", test_pipeline)\n\n source_replacements = {\n \"filesrc\": \"fakesrc\",\n \"v4l2src\": \"fakesrc\",\n \"videotestsrc\": \"fakesrc\",\n \"rtspsrc\": \"fakesrc\",\n \"uridecodebin\": \"fakesrc\",\n \"multifilesrc\": \"fakesrc\",\n \"nvvideotestsrc\": \"fakesrc\",\n \"nvurisrcbin\": \"fakesrc\",\n \"tcpclientsrc\": \"fakesrc\",\n }\n sink_replacements = {\n \"nveglglessink\": \"fakesink\",\n \"nv3dsink\": \"fakesink\",\n \"filesink\": \"fakesink\",\n \"udpsink\": \"fakesink\",\n \"xvimagesink\": \"fakesink\",\n \"autovideosink\": \"fakesink\",\n \"fpsdisplaysink\": \"fakesink\",\n \"tcpserversink\": \"fakesink\",\n }\n\n test_str = test_pipeline\n\n # Anchor replacements to segment-leading positions ((^|!\\s+)) so that a\n # property value containing the element name (e.g. config-file-path=\n # /path/with/filesrc.txt) is not mistakenly rewritten — that would corrupt\n # the dry-run input and produce false-positive parse errors.\n for real, fake in source_replacements.items():\n test_str = re.sub(rf\"(^|!\\s+){real}\\b[^!]*\", rf\"\\g\u003c1>{fake} num-buffers=1 \", test_str)\n\n for real, fake in sink_replacements.items():\n test_str = re.sub(rf\"(^|!\\s+){real}\\b[^!]*\", rf\"\\g\u003c1>{fake} \", test_str)\n\n test_str = re.sub(\n r\"\\b(config-file-path|config-file|ll-config-file|ll-lib-file\"\n r\"|postprocesslib-config-file|postprocesslib-name)=[^\\s!]+\",\n \"\", test_str,\n )\n\n strip_elements = [\n \"qtdemux\", \"h264parse\", \"h265parse\", \"nvv4l2decoder\", \"nvjpegdec\",\n \"jpegparse\", \"decodebin\", \"pngdec\", \"jpegdec\", \"videoparse\",\n \"qtmux\", \"matroskamux\", \"mp4mux\", \"flvmux\", \"mpegtsmux\",\n \"rtph264pay\", \"rtph265pay\", \"rtph264depay\", \"avdec_h264\",\n \"rtpgstpay\", \"rtpstreampay\", \"rtpstreamdepay\", \"rtpgstdepay\",\n \"aacparse\", \"avdec_aac\", \"audioconvert\", \"audioresample\",\n ]\n for elem in strip_elements:\n test_str = re.sub(rf\"\\b{elem}\\b[^!]*!\\s*\", \"\", test_str)\n test_str = re.sub(rf\"\\b{elem}\\b[^!]*$\", \"\", test_str)\n\n test_str = re.sub(r\"\\s+\", \" \", test_str).strip()\n test_str = re.sub(r\"!\\s*!\", \"!\", test_str)\n test_str = re.sub(r\"^\\s*!\\s*\", \"\", test_str)\n test_str = re.sub(r\"\\s*!\\s*$\", \"\", test_str)\n\n if not test_str.strip():\n warnings.append(\"Pipeline too simple to perform live parse validation\")\n return errors, warnings\n\n try:\n try:\n gst_args = shlex.split(test_str)\n except ValueError:\n warnings.append(\"Live parse check skipped — pipeline string could not be tokenized safely\")\n return errors, warnings\n\n result = subprocess.run(\n [\"gst-launch-1.0\", \"--gst-debug-level=0\"] + gst_args,\n capture_output=True, text=True, timeout=5,\n )\n\n stderr = result.stderr.strip()\n if \"erroneous pipeline\" in stderr.lower():\n for line in stderr.split(\"\\n\"):\n if \"erroneous pipeline\" in line.lower():\n line = line.strip()[:300]\n errors.append(f\"GStreamer parse error: {line}\")\n elif result.returncode != 0 and \"error\" in stderr.lower():\n noise_patterns = [\n \"no such file\", \"configuration file not provided\",\n \"doesn't want to preroll\", \"not negotiated\",\n \"internal data stream\", \"gstnvinfer\",\n \"no element\", \"could not set property\",\n ]\n for line in stderr.split(\"\\n\"):\n line = line.strip()\n if not line or \"error\" not in line.lower():\n continue\n line_lower = line.lower()\n if any(p in line_lower for p in noise_patterns):\n continue\n warnings.append(f\"GStreamer runtime warning: {line[:300]}\")\n\n except subprocess.TimeoutExpired:\n warnings.append(\"Live parse check timed out — pipeline may hang or require hardware not present\")\n except FileNotFoundError:\n warnings.append(\"gst-launch-1.0 not found — skipping live parse check\")\n\n return errors, warnings\n\n\n# ---------------------------------------------------------------------------\n# Main\n# ---------------------------------------------------------------------------\ndef main():\n parser = argparse.ArgumentParser(description=\"Validate a GStreamer/DeepStream pipeline\")\n parser.add_argument(\"pipeline\", nargs=\"?\", help=\"Pipeline string to validate\")\n parser.add_argument(\"--pipeline\", dest=\"pipeline_flag\", help=\"Pipeline string (alternative flag)\")\n parser.add_argument(\n \"--format\", dest=\"output_format\",\n choices=(\"summary\", \"json\"), default=\"json\",\n help=(\n \"Output format. 'json' = full verbose output (default — preserves \"\n \"existing behavior). 'summary' = one-line human status, with errors \"\n \"or warnings printed below if present.\"\n ),\n )\n\n args = parser.parse_args()\n pipeline_str = args.pipeline or args.pipeline_flag\n fmt = args.output_format\n\n if not pipeline_str:\n print(\"ERROR: No pipeline provided\", file=sys.stderr)\n print('Usage: python3 validate_pipeline.py \"gst-launch-1.0 ...\"', file=sys.stderr)\n sys.exit(1)\n\n if len(pipeline_str) > 16384:\n if fmt == \"summary\":\n print(\"invalid: Pipeline string exceeds maximum allowed length (16384 chars)\")\n else:\n result = {\"valid\": False, \"errors\": [\"Pipeline string exceeds maximum allowed length (16384 chars)\"], \"warnings\": []}\n print(json.dumps(result, indent=2))\n sys.exit(1)\n\n all_errors = []\n all_warnings = []\n\n syntax_errors, syntax_warnings = validate_syntax(pipeline_str)\n all_errors.extend(syntax_errors)\n all_warnings.extend(syntax_warnings)\n\n elements, pad_refs, elem_errors, elem_warnings = validate_elements(pipeline_str)\n all_errors.extend(elem_errors)\n all_warnings.extend(elem_warnings)\n\n mem_errors = validate_memory_format(pipeline_str, elements, pad_refs)\n all_errors.extend(mem_errors)\n\n struct_errors, struct_warnings = validate_pipeline_structure(elements, pad_refs)\n all_errors.extend(struct_errors)\n all_warnings.extend(struct_warnings)\n\n platform_warnings = validate_platform_sink(elements)\n all_warnings.extend(platform_warnings)\n\n ordering_warnings = validate_element_ordering(elements)\n all_warnings.extend(ordering_warnings)\n\n gst_errors, gst_warnings = validate_with_gst_launch(pipeline_str, pad_refs)\n all_errors.extend(gst_errors)\n all_warnings.extend(gst_warnings)\n\n result = {\n \"valid\": len(all_errors) == 0,\n \"elements_found\": [e[\"name\"] for e in elements],\n \"num_elements\": len(elements),\n \"pad_refs\": pad_refs,\n \"errors\": all_errors,\n \"warnings\": all_warnings,\n }\n\n if fmt == \"summary\":\n live_parse_skipped = bool(pad_refs)\n if result[\"valid\"]:\n note = \"\"\n if live_parse_skipped:\n note = \" · live-parse skipped (multi-stream)\"\n print(\n \"valid · {n} elements · {w} warnings{note}\".format(\n n=len(elements), w=len(all_warnings), note=note,\n )\n )\n else:\n print(\n \"invalid · {n} elements · {e} errors · {w} warnings\".format(\n n=len(elements), e=len(all_errors), w=len(all_warnings),\n )\n )\n for err in all_errors:\n print(\" error: {}\".format(err))\n for warn in all_warnings:\n print(\" warning: {}\".format(warn))\n else:\n print(json.dumps(result, indent=2))\n\n sys.exit(0 if result[\"valid\"] else 1)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":30313,"content_sha256":"632ce394c09dc530e5c91a80a3c6711689dbb594504cb239d2f2bf0e5c04cfa3"},{"filename":"skill-card.md","content":"## Description: \u003cbr>\nBuild DeepStream GStreamer pipelines interactively by collecting pipeline requirements through an interactive questionnaire, then assembling the pipeline using a standalone BM25 retrieval backend with structural metadata boosting over 270+ verified pipelines. \u003cbr>\n\nThis skill is ready for commercial/non-commercial use. \u003cbr>\n\n## Owner\nNVIDIA \u003cbr>\n\n### License/Terms of Use: \u003cbr>\nCC-BY-4.0 AND Apache-2.0 \u003cbr>\n## Use Case: \u003cbr>\nDevelopers and engineers use this skill to rapidly generate ready-to-run gst-launch-1.0 pipelines for NVIDIA DeepStream SDK video analytics workflows including object detection, tracking, and streaming on dGPU, Jetson, and SBSA platforms. \u003cbr>\n\n### Deployment Geography for Use: \u003cbr>\nGlobal \u003cbr>\n\n## Known Risks and Mitigations: \u003cbr>\nRisk: Review before execution as proposals could introduce incorrect or misleading guidance into skills. \u003cbr>\nMitigation: Review and scan skill before deployment. \u003cbr>\n\n## Reference(s): \u003cbr>\n- [Assembly Rules](references/assembly-rules.md) \u003cbr>\n- [Output Format](references/output-format.md) \u003cbr>\n- [Requirement Extraction](references/requirement-extraction.md) \u003cbr>\n- [Security and Limitations](references/security-and-limitations.md) \u003cbr>\n- [NVIDIA DeepStream SDK](https://developer.nvidia.com/deepstream-sdk) \u003cbr>\n- [DeepStream NGC Container](https://catalog.ngc.nvidia.com/orgs/nvidia/containers/deepstream) \u003cbr>\n\n\n## Skill Output: \u003cbr>\n**Output Type(s):** [Shell commands, Configuration instructions] \u003cbr>\n**Output Format:** [Markdown with inline bash code blocks] \u003cbr>\n**Output Parameters:** [1D] \u003cbr>\n**Other Properties Related to Output:** [None] \u003cbr>\n\n## Evaluation Agents Used: \u003cbr>\n- Tier 3 agent details not available in this report \u003cbr>\n\n\n\n## Evaluation Tasks: \u003cbr>\nEvaluated via NVSkills-Eval 3-Tier Evaluation with external profile. Tier 3 live agent evaluation not available in this report. \u003cbr>\n\n## Evaluation Metrics Used: \u003cbr>\nReported benchmark dimensions: \u003cbr>\n- Security: Checks whether skill-assisted execution avoids unsafe behavior such as secret leakage, destructive commands, or unauthorized access. \u003cbr>\n- Correctness: Checks whether the agent follows the expected workflow and produces the correct final output. \u003cbr>\n- Discoverability: Checks whether the agent loads the skill when relevant and avoids using it when irrelevant. \u003cbr>\n- Effectiveness: Checks whether the agent performs measurably better with the skill than without it. \u003cbr>\n- Efficiency: Checks whether the agent uses fewer tokens and avoids redundant work. \u003cbr>\n\n\n\n## Skill Version(s): \u003cbr>\n1.0.0 (source: frontmatter) \u003cbr>\n\n## Ethical Considerations: \u003cbr>\nNVIDIA believes Trustworthy AI is a shared responsibility and we have established policies and practices to enable development for a wide array of AI applications. When downloaded or used in accordance with our terms of service, developers should work with their internal team to ensure this skill meets requirements for the relevant industry and use case and addresses unforeseen product misuse. \u003cbr>\n\n(For Release on NVIDIA Platforms Only) \u003cbr>\nPlease report quality, risk, security vulnerabilities or NVIDIA AI Concerns [here](https://app.intigriti.com/programs/nvidia/nvidiavdp/detail). \u003cbr>\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":3231,"content_sha256":"d9cc368ffee96ccdba58d56f01e7cd7bd11762853a339834b09591e8367848f3"},{"filename":"tests/__init__.py","content":"# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n# SPDX-License-Identifier: Apache-2.0\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n","content_type":"text/x-python; charset=utf-8","language":"python","size":679,"content_sha256":"c4dc0cf54a15db963e684ed996a7ceb38a3737390327929f05930ac903f33514"},{"filename":"tests/test_cli_format.py","content":"#!/usr/bin/env python3\n\n# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n# SPDX-License-Identifier: Apache-2.0\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nEnd-to-end CLI tests for the --format flag on generate_pipeline.py and\nvalidate_pipeline.py. Run the scripts as subprocesses so the full surface\n(argparse, formatting branches, exit codes) is exercised.\n\"\"\"\n\nimport json\nimport os\nimport subprocess\nimport sys\nimport unittest\n\nSCRIPTS_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), \"..\", \"scripts\"))\nGENERATE = os.path.join(SCRIPTS_DIR, \"generate_pipeline.py\")\nVALIDATE = os.path.join(SCRIPTS_DIR, \"validate_pipeline.py\")\n\nGENERATE_BASE_ARGS = [\n \"--query\", \"infer on 3 videos and display\",\n \"--source-type\", \"Local video file\",\n \"--num-sources\", \"3\",\n \"--inference\", \"primary\",\n \"--tracker\", \"none\",\n \"--sink\", \"display\",\n \"--platform\", \"dGPU\",\n \"--extras\", \"none\",\n]\n\nVALID_PIPELINE = (\n \"gst-launch-1.0 videotestsrc num-buffers=1 ! \"\n \"video/x-raw,format=NV12,width=320,height=240 ! fakesink\"\n)\nINVALID_PIPELINE = \"gst-launch-1.0 filesrc ! ! fakesink\"\n\n\ndef run_generate(extra_args=None):\n args = [sys.executable, GENERATE] + GENERATE_BASE_ARGS + (extra_args or [])\n return subprocess.run(args, capture_output=True, text=True, timeout=30)\n\n\ndef run_validate(pipeline, extra_args=None):\n args = [sys.executable, VALIDATE, pipeline] + (extra_args or [])\n return subprocess.run(args, capture_output=True, text=True, timeout=30)\n\n\nclass TestGenerateFormat(unittest.TestCase):\n \"\"\"generate_pipeline.py --format {json,compact,summary}\"\"\"\n\n def test_default_is_full_json(self):\n \"\"\"Default (no --format) must remain full JSON for backward compat.\"\"\"\n result = run_generate()\n self.assertEqual(result.returncode, 0)\n payload = json.loads(result.stdout)\n # Full JSON includes the retrieved_pipelines list (~10 entries)\n self.assertIn(\"retrieved_pipelines\", payload)\n self.assertIsInstance(payload[\"retrieved_pipelines\"], list)\n self.assertIn(\"confidence\", payload)\n self.assertIn(\"context\", payload)\n\n def test_format_json_matches_default(self):\n \"\"\"--format=json explicitly should match the default behavior.\"\"\"\n default = run_generate()\n explicit = run_generate([\"--format\", \"json\"])\n self.assertEqual(default.stdout, explicit.stdout)\n\n def test_format_compact_is_smaller_json(self):\n \"\"\"--format=compact returns JSON with top_pipeline only, not retrieved_pipelines.\"\"\"\n result = run_generate([\"--format\", \"compact\"])\n self.assertEqual(result.returncode, 0)\n payload = json.loads(result.stdout)\n self.assertIn(\"top_pipeline\", payload)\n self.assertNotIn(\"retrieved_pipelines\", payload)\n self.assertIn(\"confidence\", payload)\n self.assertIn(\"context\", payload)\n # top_pipeline is either a result dict or null\n if payload[\"top_pipeline\"] is not None:\n self.assertIn(\"pipeline\", payload[\"top_pipeline\"])\n self.assertIn(\"score\", payload[\"top_pipeline\"])\n\n def test_format_summary_is_single_line(self):\n \"\"\"--format=summary prints one line: confidence=… retrieved=… top_match=…\"\"\"\n result = run_generate([\"--format\", \"summary\"])\n self.assertEqual(result.returncode, 0)\n lines = result.stdout.strip().splitlines()\n self.assertEqual(len(lines), 1)\n self.assertIn(\"confidence=\", lines[0])\n self.assertIn(\"retrieved=\", lines[0])\n self.assertIn(\"top_match=\", lines[0])\n\n def test_compact_smaller_than_json(self):\n \"\"\"compact JSON output should be substantially smaller than full json.\"\"\"\n full = run_generate([\"--format\", \"json\"])\n compact = run_generate([\"--format\", \"compact\"])\n self.assertLess(len(compact.stdout), len(full.stdout))\n # compact should be at most 1/3 the size for typical queries\n self.assertLess(len(compact.stdout) * 3, len(full.stdout) + 200)\n\n def test_invalid_format_rejected(self):\n \"\"\"argparse should reject an unknown --format value.\"\"\"\n result = run_generate([\"--format\", \"bogus\"])\n self.assertNotEqual(result.returncode, 0)\n self.assertIn(\"invalid choice\", result.stderr.lower())\n\n\nclass TestValidateFormat(unittest.TestCase):\n \"\"\"validate_pipeline.py --format {json,summary}\"\"\"\n\n def test_default_is_full_json(self):\n \"\"\"Default (no --format) must remain full JSON for backward compat.\"\"\"\n result = run_validate(VALID_PIPELINE)\n # videotestsrc → fakesink may emit gst-launch dry-run warnings; valid still True\n payload = json.loads(result.stdout)\n self.assertIn(\"valid\", payload)\n self.assertIn(\"elements_found\", payload)\n self.assertIn(\"errors\", payload)\n self.assertIn(\"warnings\", payload)\n\n def test_format_json_matches_default(self):\n default = run_validate(VALID_PIPELINE)\n explicit = run_validate(VALID_PIPELINE, [\"--format\", \"json\"])\n self.assertEqual(default.stdout, explicit.stdout)\n\n def test_format_summary_valid_is_one_line(self):\n result = run_validate(VALID_PIPELINE, [\"--format\", \"summary\"])\n lines = result.stdout.strip().splitlines()\n # Summary header is exactly one line; warnings would be indented after.\n self.assertGreaterEqual(len(lines), 1)\n self.assertTrue(lines[0].startswith(\"valid · \"))\n self.assertIn(\"elements\", lines[0])\n self.assertIn(\"warnings\", lines[0])\n\n def test_format_summary_invalid_lists_errors(self):\n result = run_validate(INVALID_PIPELINE, [\"--format\", \"summary\"])\n self.assertNotEqual(result.returncode, 0)\n out = result.stdout\n self.assertTrue(out.startswith(\"invalid · \"))\n # Errors are indented under the header\n self.assertIn(\"\\n error: \", out)\n\n def test_format_summary_multistream_notes_skipped_live_parse(self):\n \"\"\"Multi-stream pipelines should mention 'live-parse skipped' in summary.\"\"\"\n multistream = (\n \"gst-launch-1.0 -e \"\n \"filesrc location=/dev/null ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_0 \"\n \"filesrc location=/dev/null ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_1 \"\n \"nvstreammux name=m batch-size=2 width=1920 height=1080 ! \"\n \"nvinfer config-file-path=/dev/null batch-size=2 ! \"\n \"nvvideoconvert ! nvdsosd ! nveglglessink\"\n )\n result = run_validate(multistream, [\"--format\", \"summary\"])\n # May be valid or invalid depending on env; either way the summary line\n # should mention live-parse skip when pad refs are present.\n self.assertIn(\"live-parse skipped\", result.stdout)\n\n def test_invalid_format_rejected(self):\n result = run_validate(VALID_PIPELINE, [\"--format\", \"bogus\"])\n self.assertNotEqual(result.returncode, 0)\n self.assertIn(\"invalid choice\", result.stderr.lower())\n\n def test_dash_e_flag_accepted(self):\n \"\"\"The Step 5 generated script uses 'gst-launch-1.0 -e' — validator must accept.\"\"\"\n with_e = \"gst-launch-1.0 -e videotestsrc num-buffers=1 ! fakesink\"\n result = run_validate(with_e, [\"--format\", \"summary\"])\n # The -e flag must be stripped cleanly — the validator should report\n # success (returncode == 0). Surface stdout/stderr in the failure\n # message so a regression is debuggable from the test output alone.\n self.assertEqual(\n result.returncode, 0,\n msg=(\n f\"validator rejected 'gst-launch-1.0 -e' pipeline \"\n f\"(returncode={result.returncode}).\\n\"\n f\"stdout: {result.stdout!r}\\nstderr: {result.stderr!r}\"\n ),\n )\n # First line of summary must not be the 'invalid · …' header.\n self.assertNotIn(\"invalid · \", result.stdout.split(\"\\n\")[0])\n # The summary should report a non-zero element count.\n self.assertRegex(result.stdout, r\"\\d+ elements\")\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":8657,"content_sha256":"ca7c13158d326a13e7e2291d16dd4fd5ae168ff7311202e2f6260159ea7cc837"},{"filename":"tests/test_golden.py","content":"#!/usr/bin/env python3\n\n# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n# SPDX-License-Identifier: Apache-2.0\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nGolden test set — regression tests for pipeline retrieval quality.\n\nEach test case specifies a query + user parameters and asserts that the top\nresults contain expected elements/patterns (or don't contain unwanted ones).\nRun this after any change to the retriever or data to ensure quality doesn't\nregress.\n\"\"\"\n\nimport os\nimport sys\nimport unittest\n\nSCRIPT_DIR = os.path.join(os.path.dirname(__file__), \"..\", \"scripts\")\nsys.path.insert(0, SCRIPT_DIR)\n\nfrom generate_pipeline import BM25Retriever\n\nDATA_CSV = os.path.join(os.path.dirname(__file__), \"..\", \"data\", \"data.csv\")\n\n\ndef _meta(platform=\"unknown\", source=\"unknown\", sink=\"unknown\",\n inference=\"unknown\", tracker=\"none\", num_sources=1):\n return {\n \"platform\": platform,\n \"source_type\": source,\n \"sink_type\": sink,\n \"inference\": inference,\n \"tracker\": tracker,\n \"num_sources\": num_sources,\n }\n\n\nclass GoldenTestBase(unittest.TestCase):\n @classmethod\n def setUpClass(cls):\n cls.retriever = BM25Retriever()\n cls.retriever.load_csv(DATA_CSV)\n\n def assert_top_n_contain(self, query, user_meta, expected_in, n=3,\n not_expected=None):\n \"\"\"Assert at least one of the top-N results contains all expected_in strings.\"\"\"\n results, _ = self.retriever.retrieve(query, user_meta=user_meta)\n top = results[:n]\n self.assertGreater(len(top), 0, f\"No results for query: {query}\")\n\n for expected in expected_in:\n found = any(expected in r[\"pipeline\"].lower() for r in top)\n self.assertTrue(found, f\"Expected '{expected}' in top-{n} results for: {query}\")\n\n if not_expected:\n for bad in not_expected:\n top_pipelines = \" \".join(r[\"pipeline\"].lower() for r in top[:1])\n self.assertNotIn(bad, top_pipelines,\n f\"Did NOT expect '{bad}' in #1 result for: {query}\")\n\n def assert_platform_match(self, query, user_meta, expected_platform, n=3):\n \"\"\"Assert top-N results prefer the expected platform.\"\"\"\n results, _ = self.retriever.retrieve(query, user_meta=user_meta)\n top = results[:n]\n platform_matches = sum(\n 1 for r in top if r[\"metadata\"][\"platform\"] == expected_platform\n )\n self.assertGreater(platform_matches, 0,\n f\"No {expected_platform} results in top-{n} for: {query}\")\n\n\nclass TestFlipRotate(GoldenTestBase):\n def test_flip_90_ccw(self):\n self.assert_top_n_contain(\n \"flip video 90 degrees counter-clockwise\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"flip-method=1\"],\n )\n\n def test_rotate_180(self):\n self.assert_top_n_contain(\n \"rotate incoming video by 180 degrees\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"flip-method=2\"],\n )\n\n def test_flip_horizontal(self):\n self.assert_top_n_contain(\n \"flip video horizontally\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"flip-method=4\"],\n )\n\n\nclass TestColorConversion(GoldenTestBase):\n def test_bgr10a2_to_i420(self):\n self.assert_top_n_contain(\n \"convert BGR10A2_LE to I420 and display\",\n _meta(platform=\"dgpu\", source=\"test\", sink=\"display\"),\n [\"bgr10a2_le\", \"i420\"],\n )\n\n def test_bgr10a2_to_rgba(self):\n self.assert_top_n_contain(\n \"convert BGR10A2_LE to RGBA and display\",\n _meta(platform=\"dgpu\", source=\"test\", sink=\"display\"),\n [\"bgr10a2_le\", \"rgba\"],\n )\n\n\nclass TestPrimaryInference(GoldenTestBase):\n def test_single_stream_display_dgpu(self):\n self.assert_top_n_contain(\n \"primary inference on a single mp4 video and display the output\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary\", num_sources=1),\n [\"nvinfer\", \"config-file-path\", \"nveglglessink\"],\n not_expected=[\"nv3dsink\"],\n )\n\n def test_single_stream_display_jetson(self):\n self.assert_top_n_contain(\n \"primary inference on a single mp4 video and display the output on Jetson\",\n _meta(platform=\"jetson\", source=\"video\", sink=\"display\",\n inference=\"primary\", num_sources=1),\n [\"nvinfer\", \"nv3dsink\"],\n )\n self.assert_platform_match(\n \"primary inference on Jetson\",\n _meta(platform=\"jetson\"), \"jetson\",\n )\n\n\nclass TestSecondaryInference(GoldenTestBase):\n def test_primary_secondary_display(self):\n self.assert_top_n_contain(\n \"primary and secondary inference on a single stream and display\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary+secondary\"),\n [\"nvinfer\", \"infer-on-gie-id\"],\n )\n\n\nclass TestTracker(GoldenTestBase):\n def test_inference_with_tracker(self):\n self.assert_top_n_contain(\n \"primary and secondary inference with tracker on a single stream\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary+secondary\", tracker=\"present\"),\n [\"nvtracker\"],\n )\n\n\nclass TestMultiStream(GoldenTestBase):\n def test_4_stream_inference(self):\n self.assert_top_n_contain(\n \"primary inference on 4 video streams and display\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary\", num_sources=4),\n [\"m.sink_0\", \"batch-size=4\"],\n n=5,\n )\n\n\nclass TestSaveOutput(GoldenTestBase):\n def test_save_to_mp4(self):\n self.assert_top_n_contain(\n \"encode video to h264 and mux into mp4 file\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"save-mp4\"),\n [\"filesink\", \"qtmux\"],\n n=5,\n )\n\n def test_save_h264(self):\n self.assert_top_n_contain(\n \"encode test video to h264 and save to file\",\n _meta(platform=\"dgpu\", source=\"test\", sink=\"save-file\"),\n [\"nvv4l2h264enc\", \"filesink\"],\n )\n\n\nclass TestPreprocessor(GoldenTestBase):\n def test_preprocess_before_primary(self):\n self.assert_top_n_contain(\n \"preprocess before primary inference on a single stream\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary+preprocess\"),\n [\"nvdspreprocess\", \"input-tensor-meta=1\"],\n )\n\n\nclass TestResize(GoldenTestBase):\n def test_resize_video(self):\n self.assert_top_n_contain(\n \"resize incoming video to specific height and width\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"nvvideoconvert\"],\n )\n\n\nclass TestCrop(GoldenTestBase):\n def test_crop_video(self):\n self.assert_top_n_contain(\n \"crop the incoming video frame\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"src-crop\"],\n )\n\n\nclass TestDewarp(GoldenTestBase):\n def test_dewarp_dgpu(self):\n self.assert_top_n_contain(\n \"dewarp a given mp4 video on dGPU\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\"),\n [\"nvdewarper\"],\n )\n\n\nclass TestPlatformFiltering(GoldenTestBase):\n def test_dgpu_query_prefers_dgpu(self):\n self.assert_platform_match(\n \"primary inference on video and display output\",\n _meta(platform=\"dgpu\", source=\"video\", sink=\"display\",\n inference=\"primary\"),\n \"dgpu\",\n )\n\n def test_jetson_query_prefers_jetson(self):\n self.assert_platform_match(\n \"primary inference on video and display output on Jetson\",\n _meta(platform=\"jetson\", source=\"video\", sink=\"display\",\n inference=\"primary\"),\n \"jetson\",\n )\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":8799,"content_sha256":"b8b347685a185b86b9208858be46bf58a940449f4811855fb303f25582465c75"},{"filename":"tests/test_retriever.py","content":"#!/usr/bin/env python3\n\n# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n# SPDX-License-Identifier: Apache-2.0\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"Unit tests for the BM25 pipeline retriever.\"\"\"\n\nimport os\nimport sys\nimport unittest\n\nSCRIPT_DIR = os.path.join(os.path.dirname(__file__), \"..\", \"scripts\")\nsys.path.insert(0, SCRIPT_DIR)\n\nfrom generate_pipeline import (\n BM25Retriever,\n expand_with_synonyms,\n extract_pipeline_metadata,\n normalize_user_params,\n tokenize,\n)\n\nDATA_CSV = os.path.join(os.path.dirname(__file__), \"..\", \"data\", \"data.csv\")\n\n\nclass TestTokenizer(unittest.TestCase):\n def test_lowercase(self):\n self.assertEqual(tokenize(\"Hello World\"), [\"hello\", \"world\"])\n\n def test_strips_punctuation(self):\n tokens = tokenize(\"nvinfer config-file-path=/foo/bar.txt\")\n self.assertIn(\"nvinfer\", tokens)\n self.assertIn(\"config-file-path\", tokens)\n\n def test_preserves_compound_tokens_splits_paths(self):\n \"\"\"\n The tokenizer keeps compound hyphenated\n tokens (config-file-path) intact but splits on '/' so a path like\n /opt/nvidia/deepstream/config.txt yields useful per-component tokens\n ('opt', 'nvidia', 'deepstream', 'config.txt') for IDF rather than\n collapsing into one giant unique token.\n \"\"\"\n tokens = tokenize(\"config-file-path=/opt/nvidia/deepstream/config.txt\")\n self.assertIn(\"config-file-path\", tokens)\n # Path components must be split out as separate tokens.\n self.assertIn(\"opt\", tokens)\n self.assertIn(\"nvidia\", tokens)\n self.assertIn(\"deepstream\", tokens)\n self.assertIn(\"config.txt\", tokens)\n # The whole path must NOT appear as a single token.\n self.assertNotIn(\"/opt/nvidia/deepstream/config.txt\", tokens)\n\n\nclass TestSynonymExpansion(unittest.TestCase):\n def test_infer_expands(self):\n expanded = expand_with_synonyms([\"infer\"])\n self.assertIn(\"inference\", expanded)\n self.assertIn(\"nvinfer\", expanded)\n self.assertIn(\"detect\", expanded)\n\n def test_no_duplicates(self):\n expanded = expand_with_synonyms([\"infer\", \"inference\"])\n counts = {}\n for tok in expanded:\n counts[tok] = counts.get(tok, 0) + 1\n for tok, count in counts.items():\n self.assertEqual(count, 1, f\"Duplicate token: {tok}\")\n\n def test_unknown_token_unchanged(self):\n expanded = expand_with_synonyms([\"xyzzy123\"])\n self.assertEqual(expanded, [\"xyzzy123\"])\n\n\nclass TestPipelineMetadata(unittest.TestCase):\n def test_dgpu_display(self):\n meta = extract_pipeline_metadata(\n \"gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! \"\n \"nvv4l2decoder ! nvinfer config-file-path=c.txt ! nveglglessink\"\n )\n self.assertEqual(meta[\"platform\"], \"dgpu\")\n self.assertEqual(meta[\"sink_type\"], \"display\")\n self.assertEqual(meta[\"source_type\"], \"video\")\n self.assertEqual(meta[\"inference\"], \"primary\")\n self.assertEqual(meta[\"tracker\"], \"none\")\n\n def test_jetson_display(self):\n meta = extract_pipeline_metadata(\n \"gst-launch-1.0 filesrc location=a.mp4 ! nvinfer config-file-path=c.txt ! nv3dsink\"\n )\n self.assertEqual(meta[\"platform\"], \"jetson\")\n self.assertEqual(meta[\"sink_type\"], \"display\")\n\n def test_primary_secondary_with_tracker(self):\n meta = extract_pipeline_metadata(\n \"nvinfer config-file-path=p.txt ! nvtracker ll-lib-file=t.so ! \"\n \"nvinfer config-file-path=s.txt infer-on-gie-id=1 ! nveglglessink\"\n )\n self.assertEqual(meta[\"inference\"], \"primary+secondary\")\n self.assertEqual(meta[\"tracker\"], \"present\")\n\n def test_multi_source_detection(self):\n meta = extract_pipeline_metadata(\n \"filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=4 ! \"\n \"nveglglessink filesrc location=b.mp4 ! m.sink_1 \"\n \"filesrc location=c.mp4 ! m.sink_2 \"\n \"filesrc location=d.mp4 ! m.sink_3\"\n )\n self.assertEqual(meta[\"num_sources\"], 4)\n\n def test_rtsp_source(self):\n meta = extract_pipeline_metadata(\n \"uridecodebin uri=rtsp://\u003crtsp-server>:\u003cport>/stream ! nveglglessink\"\n )\n self.assertEqual(meta[\"source_type\"], \"rtsp\")\n\n def test_save_mp4_sink(self):\n meta = extract_pipeline_metadata(\n \"filesrc location=a.mp4 ! nvv4l2h264enc ! h264parse ! qtmux ! filesink location=out.mp4\"\n )\n self.assertEqual(meta[\"sink_type\"], \"save-mp4\")\n\n def test_image_source(self):\n meta = extract_pipeline_metadata(\n \"filesrc location=img.jpg ! jpegparse ! nvjpegdec ! nveglglessink\"\n )\n self.assertEqual(meta[\"source_type\"], \"image\")\n\n def test_preprocess_inference(self):\n meta = extract_pipeline_metadata(\n \"nvdspreprocess config-file=c.txt ! nvinfer config-file-path=p.txt input-tensor-meta=1 ! nveglglessink\"\n )\n self.assertEqual(meta[\"inference\"], \"primary+preprocess\")\n\n def test_fakesink(self):\n meta = extract_pipeline_metadata(\"videotestsrc ! fakesink\")\n self.assertEqual(meta[\"source_type\"], \"test\")\n self.assertEqual(meta[\"sink_type\"], \"fakesink\")\n\n\nclass TestBM25Retriever(unittest.TestCase):\n @classmethod\n def setUpClass(cls):\n cls.retriever = BM25Retriever()\n cls.retriever.load_csv(DATA_CSV)\n\n def test_loads_documents(self):\n self.assertGreater(len(self.retriever.documents), 200)\n\n def test_idf_computed(self):\n self.assertGreater(len(self.retriever.idf), 0)\n self.assertIn(\"nvinfer\", self.retriever.idf)\n\n def test_avgdl_positive(self):\n self.assertGreater(self.retriever.avgdl, 0)\n\n def test_retrieve_returns_results(self):\n results, conf = self.retriever.retrieve(\"flip video 90 degrees\")\n self.assertGreater(len(results), 0)\n\n def test_retrieve_scores_descending(self):\n results, _ = self.retriever.retrieve(\"primary inference on mp4 video display\")\n scores = [r[\"score\"] for r in results]\n self.assertEqual(scores, sorted(scores, reverse=True))\n\n def test_retrieve_contains_metadata(self):\n results, _ = self.retriever.retrieve(\"primary inference\")\n for r in results:\n self.assertIn(\"metadata\", r)\n self.assertIn(\"platform\", r[\"metadata\"])\n\n def test_confidence_levels(self):\n _, conf = self.retriever.retrieve(\n \"primary inference on a single mp4 video and display the output\"\n )\n self.assertIn(conf, (\"high\", \"medium\", \"low\"))\n\n def test_structural_boost_prefers_matching_platform(self):\n dgpu_meta = {\n \"platform\": \"dgpu\", \"source_type\": \"video\", \"sink_type\": \"display\",\n \"inference\": \"primary\", \"tracker\": \"none\", \"num_sources\": 1,\n }\n jetson_meta = {\n \"platform\": \"jetson\", \"source_type\": \"video\", \"sink_type\": \"display\",\n \"inference\": \"primary\", \"tracker\": \"none\", \"num_sources\": 1,\n }\n query = \"primary inference on video and display output\"\n dgpu_results, _ = self.retriever.retrieve(query, user_meta=dgpu_meta)\n jetson_results, _ = self.retriever.retrieve(query, user_meta=jetson_meta)\n\n if dgpu_results and jetson_results:\n dgpu_top_platform = dgpu_results[0][\"metadata\"][\"platform\"]\n jetson_top_platform = jetson_results[0][\"metadata\"][\"platform\"]\n self.assertEqual(dgpu_top_platform, \"dgpu\")\n self.assertEqual(jetson_top_platform, \"jetson\")\n\n def test_empty_query_returns_empty(self):\n results, conf = self.retriever.retrieve(\"\")\n self.assertEqual(conf, \"low\")\n\n\nclass TestIndexCache(unittest.TestCase):\n \"\"\"\n Regression: the BM25 index is cached to a sibling\n JSON file keyed on (CSV mtime, internal schema version). A cached\n retriever must (a) reproduce the same indexed state as a fresh one and\n (b) be invalidated when the CSV is touched.\n \"\"\"\n\n def setUp(self):\n import shutil\n import tempfile\n self.tmpdir = tempfile.mkdtemp(prefix=\"ds-pg-cache-test-\")\n self.csv_copy = os.path.join(self.tmpdir, \"data.csv\")\n shutil.copyfile(DATA_CSV, self.csv_copy)\n self.cache_path = self.csv_copy + \".cache.json\"\n\n def tearDown(self):\n import shutil\n shutil.rmtree(self.tmpdir, ignore_errors=True)\n\n def test_cache_written_on_first_load(self):\n self.assertFalse(os.path.exists(self.cache_path))\n retriever = BM25Retriever()\n retriever.load_csv(self.csv_copy)\n self.assertTrue(os.path.exists(self.cache_path))\n\n def test_second_load_reproduces_state_from_cache(self):\n first = BM25Retriever()\n first.load_csv(self.csv_copy)\n second = BM25Retriever()\n second.load_csv(self.csv_copy)\n # Same documents, same IDF, same average doc length — proves the\n # cached state round-trips through JSON correctly.\n self.assertEqual(len(first.documents), len(second.documents))\n self.assertEqual(first.idf, second.idf)\n self.assertEqual(first.avgdl, second.avgdl)\n self.assertEqual(len(first.doc_tokens), len(second.doc_tokens))\n\n def test_cache_invalidated_when_csv_mtime_changes(self):\n retriever = BM25Retriever()\n retriever.load_csv(self.csv_copy)\n cache_mtime_before = os.path.getmtime(self.cache_path)\n\n # Touch the CSV so its mtime advances (sleep ensures the new mtime\n # exceeds filesystem resolution).\n import time\n time.sleep(1.1)\n os.utime(self.csv_copy, None)\n\n retriever2 = BM25Retriever()\n retriever2.load_csv(self.csv_copy)\n cache_mtime_after = os.path.getmtime(self.cache_path)\n # Cache was rewritten, not reused.\n self.assertGreater(cache_mtime_after, cache_mtime_before)\n\n def test_cache_invalidated_on_schema_version_bump(self):\n import json\n retriever = BM25Retriever()\n retriever.load_csv(self.csv_copy)\n # Rewrite the cache with an obviously-stale version stamp; the next\n # load must reject it and rebuild from CSV.\n with open(self.cache_path, \"r\", encoding=\"utf-8\") as f:\n payload = json.load(f)\n payload[\"version\"] = -999\n with open(self.cache_path, \"w\", encoding=\"utf-8\") as f:\n json.dump(payload, f)\n\n retriever2 = BM25Retriever()\n retriever2.load_csv(self.csv_copy)\n # Rebuilt cache must have the current schema version, not the bogus one.\n with open(self.cache_path, \"r\", encoding=\"utf-8\") as f:\n new_payload = json.load(f)\n self.assertNotEqual(new_payload[\"version\"], -999)\n\n\nclass TestNormalizeUserParams(unittest.TestCase):\n def test_basic_mapping(self):\n class Args:\n source_type = \"Local video file\"\n num_sources = 1\n inference = \"primary\"\n tracker = \"none\"\n sink = \"display\"\n platform = \"dGPU\"\n\n meta = normalize_user_params(Args())\n self.assertEqual(meta[\"platform\"], \"dgpu\")\n self.assertEqual(meta[\"source_type\"], \"video\")\n self.assertEqual(meta[\"sink_type\"], \"display\")\n self.assertEqual(meta[\"inference\"], \"primary\")\n self.assertEqual(meta[\"tracker\"], \"none\")\n\n def test_jetson_mapping(self):\n class Args:\n source_type = \"RTSP stream\"\n num_sources = 4\n inference = \"primary+secondary\"\n tracker = \"NvDCF\"\n sink = \"save-mp4\"\n platform = \"Jetson\"\n\n meta = normalize_user_params(Args())\n self.assertEqual(meta[\"platform\"], \"jetson\")\n self.assertEqual(meta[\"source_type\"], \"rtsp\")\n self.assertEqual(meta[\"tracker\"], \"present\")\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":12485,"content_sha256":"5f76092cb158fab87d864eeec8ae50ed53f4cf7b68598ed32882e44dc764aae5"},{"filename":"tests/test_validator.py","content":"#!/usr/bin/env python3\n\n# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n# SPDX-License-Identifier: Apache-2.0\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"Unit tests for the pipeline validator.\"\"\"\n\nimport os\nimport re\nimport sys\nimport unittest\n\nSCRIPT_DIR = os.path.join(os.path.dirname(__file__), \"..\", \"scripts\")\nsys.path.insert(0, SCRIPT_DIR)\n\nfrom validate_pipeline import (\n extract_elements_and_properties,\n validate_element_ordering,\n validate_memory_format,\n validate_pipeline_structure,\n validate_platform_sink,\n validate_syntax,\n validate_with_gst_launch,\n)\n\n\nclass TestSyntaxValidation(unittest.TestCase):\n def test_valid_pipeline(self):\n errors, warnings = validate_syntax(\n \"gst-launch-1.0 filesrc location=test.mp4 ! nveglglessink\"\n )\n self.assertEqual(errors, [])\n\n def test_unbalanced_single_quotes(self):\n errors, _ = validate_syntax(\"gst-launch-1.0 filesrc location='test.mp4 ! nveglglessink\")\n self.assertTrue(any(\"single quotes\" in e for e in errors))\n\n def test_unbalanced_double_quotes(self):\n errors, _ = validate_syntax('gst-launch-1.0 filesrc location=\"test.mp4 ! nveglglessink')\n self.assertTrue(any(\"double quotes\" in e for e in errors))\n\n def test_empty_pipe_segment(self):\n errors, _ = validate_syntax(\"gst-launch-1.0 filesrc location=t.mp4 ! ! nveglglessink\")\n self.assertTrue(any(\"Empty pipe\" in e for e in errors))\n\n def test_leading_pipe(self):\n errors, _ = validate_syntax(\"! filesrc location=t.mp4 ! nveglglessink\")\n self.assertTrue(any(\"starts with\" in e for e in errors))\n\n def test_trailing_pipe(self):\n errors, _ = validate_syntax(\"gst-launch-1.0 filesrc location=t.mp4 !\")\n self.assertTrue(any(\"ends with\" in e for e in errors))\n\n def test_underscore_property_typo(self):\n errors, _ = validate_syntax(\n \"gst-launch-1.0 nvinfer config_file_path=c.txt ! nveglglessink\"\n )\n self.assertTrue(any(\"hyphens\" in e for e in errors))\n\n def test_underscore_typo_not_masked_by_hyphen_form_elsewhere(self):\n \"\"\"\n Regression: the underscore-typo check must\n scope to the segment that owns nvinfer. A hyphenated config-file-path\n on a different element (here nvdspreprocess) must not silently mask\n an underscore typo on the actual nvinfer.\n \"\"\"\n errors, _ = validate_syntax(\n \"gst-launch-1.0 nvdspreprocess config-file=p.txt ! \"\n \"nvinfer config_file_path=c.txt ! nveglglessink\"\n )\n self.assertTrue(any(\"hyphens\" in e for e in errors))\n\n def test_streammux_without_name(self):\n _, warnings = validate_syntax(\n \"gst-launch-1.0 filesrc location=t.mp4 ! nvstreammux batch-size=1 ! nveglglessink\"\n )\n self.assertTrue(any(\"name=\" in w for w in warnings))\n\n\nclass TestElementParser(unittest.TestCase):\n def test_simple_pipeline(self):\n elements, pad_refs = extract_elements_and_properties(\n \"gst-launch-1.0 filesrc location=test.mp4 ! nveglglessink\"\n )\n names = [e[\"name\"] for e in elements]\n self.assertIn(\"filesrc\", names)\n self.assertIn(\"nveglglessink\", names)\n self.assertEqual(pad_refs, [])\n\n def test_properties_extracted(self):\n elements, _ = extract_elements_and_properties(\n \"nvinfer config-file-path=c.txt batch-size=4 unique-id=1 ! nveglglessink\"\n )\n nvinfer = [e for e in elements if e[\"name\"] == \"nvinfer\"][0]\n self.assertEqual(nvinfer[\"properties\"][\"config-file-path\"], \"c.txt\")\n self.assertEqual(nvinfer[\"properties\"][\"batch-size\"], \"4\")\n\n def test_pad_refs_extracted(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! \"\n \"m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink\"\n )\n self.assertIn(\"m.sink_0\", pad_refs)\n names = [e[\"name\"] for e in elements]\n self.assertIn(\"nvstreammux\", names)\n\n def test_multiple_pad_refs(self):\n # Both branches end with a pad ref followed by a ' ! ' separator so\n # the parser sees each pad ref as the leading token of its own\n # segment — making the assertion on m.sink_0 AND m.sink_1\n # deterministic.\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=a.mp4 ! m.sink_0 \"\n \"filesrc location=b.mp4 ! m.sink_1 \"\n \"nvstreammux name=m batch-size=2 width=1920 height=1080 ! nveglglessink\"\n )\n self.assertIn(\"m.sink_0\", pad_refs)\n self.assertIn(\"m.sink_1\", pad_refs)\n\n def test_caps_filter_skipped(self):\n elements, _ = extract_elements_and_properties(\n \"videotestsrc ! 'video/x-raw,format=NV12' ! nveglglessink\"\n )\n names = [e[\"name\"] for e in elements]\n self.assertNotIn(\"video/x-raw,format=NV12\", names)\n\n def test_gst_launch_prefix_stripped(self):\n elements, _ = extract_elements_and_properties(\n \"gst-launch-1.0 filesrc location=t.mp4 ! nveglglessink\"\n )\n self.assertNotIn(\"gst-launch-1.0\", [e[\"name\"] for e in elements])\n\n def test_flags_stripped(self):\n elements, _ = extract_elements_and_properties(\n \"gst-launch-1.0 -e filesrc location=t.mp4 ! nveglglessink\"\n )\n self.assertNotIn(\"-e\", [e[\"name\"] for e in elements])\n\n def test_multiple_flags_stripped(self):\n \"\"\"\n Regression: a run of leading flags such as\n '-e -v' must be stripped together. Previously only the first flag was\n removed and the second was parsed as a phantom element.\n \"\"\"\n elements, _ = extract_elements_and_properties(\n \"gst-launch-1.0 -e -v filesrc location=t.mp4 ! nveglglessink\"\n )\n names = [e[\"name\"] for e in elements]\n self.assertNotIn(\"-e\", names)\n self.assertNotIn(\"-v\", names)\n self.assertIn(\"filesrc\", names)\n\n def test_live_parse_substitution_anchors_to_segment_start(self):\n \"\"\"\n Regression: the live-parse source/sink\n replacement must not match the element name when it appears inside a\n property value (e.g. a config-file-path that contains 'filesrc' as\n part of a directory or filename). Previously \\\\b{real}\\\\b matched the\n substring inside the path and corrupted the dry-run input. The fix\n anchors to segment-leading positions ((^|!\\\\s+)).\n \"\"\"\n original = (\n \"nvinfer config-file-path=/foo/filesrc.txt ! \"\n \"filesrc location=t.mp4 ! fakesink\"\n )\n substituted = re.sub(\n r\"(^|!\\s+)filesrc\\b[^!]*\",\n r\"\\g\u003c1>fakesrc num-buffers=1 \",\n original,\n )\n # The 'filesrc' inside the config-file-path must remain untouched.\n self.assertIn(\"config-file-path=/foo/filesrc.txt\", substituted)\n # The actual filesrc element (after !) must be replaced with fakesrc.\n self.assertIn(\"fakesrc num-buffers=1\", substituted)\n # The original filesrc element's location property should be gone.\n self.assertNotIn(\"location=t.mp4\", substituted)\n\n\nclass TestStructureValidation(unittest.TestCase):\n def test_valid_pipeline(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvinfer config-file-path=c.txt ! nveglglessink\"\n )\n errors, warnings = validate_pipeline_structure(elements, pad_refs)\n self.assertEqual(errors, [])\n\n def test_missing_source_warning(self):\n elements, pad_refs = extract_elements_and_properties(\n \"nvinfer config-file-path=c.txt ! nveglglessink\"\n )\n _, warnings = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"source\" in w.lower() for w in warnings))\n\n def test_missing_sink_warning(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvinfer config-file-path=c.txt\"\n )\n _, warnings = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"sink\" in w.lower() for w in warnings))\n\n def test_nvinfer_missing_config(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvinfer batch-size=1 ! nveglglessink\"\n )\n errors, _ = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"config-file-path\" in e for e in errors))\n\n def test_filesrc_missing_location(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc ! nveglglessink\"\n )\n errors, _ = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"location\" in e for e in errors))\n\n def test_empty_pipeline(self):\n errors, _ = validate_pipeline_structure([], [])\n self.assertTrue(any(\"empty\" in e.lower() for e in errors))\n\n def test_bad_pad_ref_warns(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=a.mp4 ! x.sink_0 nveglglessink\"\n )\n _, warnings = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"x.sink_0\" in w for w in warnings))\n\n def test_valid_pad_ref_no_warning(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=1 ! nveglglessink\"\n )\n _, warnings = validate_pipeline_structure(elements, pad_refs)\n pad_warnings = [w for w in warnings if \"pad reference\" in w.lower()]\n self.assertEqual(pad_warnings, [])\n\n def test_batch_size_pad_mismatch(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=4 width=1920 height=1080 ! nveglglessink\"\n )\n _, warnings = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"batch-size=4\" in w and \"1 sink\" in w for w in warnings))\n\n\nclass TestMemoryFormatValidation(unittest.TestCase):\n \"\"\"Test memory format mismatch detection between elements.\"\"\"\n\n def test_nvjpegdec_direct_to_mux_detected(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! \"\n \"m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink\"\n )\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertTrue(len(errors) > 0)\n self.assertTrue(any(\"nvjpegdec\" in e and \"NVMM\" in e for e in errors))\n\n def test_nvjpegdec_with_videoconvert_passes(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! \"\n \"nvvideoconvert ! m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink\"\n )\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertEqual(errors, [])\n\n def test_nvv4l2decoder_to_mux_passes(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! \"\n \"m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink\"\n )\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertEqual(errors, [])\n\n def test_nvjpegdec_to_display_no_error(self):\n pipeline = \"gst-launch-1.0 filesrc location=a.jpg ! nvjpegdec ! nveglglessink\"\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertEqual(errors, [])\n\n def test_pngdec_direct_to_mux_detected(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.png ! pngdec ! \"\n \"m.sink_0 nvstreammux name=m batch-size=1 ! nveglglessink\"\n )\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertTrue(len(errors) > 0)\n self.assertTrue(any(\"pngdec\" in e for e in errors))\n\n def test_no_duplicate_errors(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! \"\n \"m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink\"\n )\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertEqual(len(errors), 1, f\"Expected 1 error but got {len(errors)}: {errors}\")\n\n\nclass TestKnownProperties(unittest.TestCase):\n \"\"\"Validate that the property dict covers all properties used in data.csv.\"\"\"\n\n def test_nvinfer_has_clip_object_outside_roi(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"clip-object-outside-roi\", KNOWN_ELEMENT_PROPERTIES[\"nvinfer\"])\n\n def test_nvinfer_has_output_tensor_meta(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"output-tensor-meta\", KNOWN_ELEMENT_PROPERTIES[\"nvinfer\"])\n\n def test_nvdspreprocess_has_operate_on_gie_id(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"operate-on-gie-id\", KNOWN_ELEMENT_PROPERTIES[\"nvdspreprocess\"])\n\n def test_nvv4l2h264enc_has_qp_range(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"qp-range\", KNOWN_ELEMENT_PROPERTIES[\"nvv4l2h264enc\"])\n\n def test_nvdewarper_has_num_output_buffers(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"num-output-buffers\", KNOWN_ELEMENT_PROPERTIES[\"nvdewarper\"])\n\n\nclass TestPlatformSinkValidation(unittest.TestCase):\n \"\"\"Test platform-sink mismatch detection.\"\"\"\n\n def test_mixed_dgpu_jetson_sinks_warns(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nveglglessink ! nv3dsink\"\n )\n warnings = validate_platform_sink(elements)\n self.assertTrue(any(\"mixes\" in w.lower() for w in warnings))\n\n def test_dgpu_only_no_warning(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nveglglessink\"\n )\n warnings = validate_platform_sink(elements)\n self.assertEqual(warnings, [])\n\n def test_jetson_only_no_warning(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nv3dsink\"\n )\n warnings = validate_platform_sink(elements)\n self.assertEqual(warnings, [])\n\n def test_fakesink_no_warning(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! fakesink\"\n )\n warnings = validate_platform_sink(elements)\n self.assertEqual(warnings, [])\n\n\nclass TestElementOrderingValidation(unittest.TestCase):\n \"\"\"Test element ordering checks.\"\"\"\n\n def test_correct_order_no_warning(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvv4l2decoder ! nvinfer config-file-path=c.txt ! nveglglessink\"\n )\n warnings = validate_element_ordering(elements)\n self.assertEqual(warnings, [])\n\n def test_nvinfer_before_decoder_warns(self):\n elements, _ = extract_elements_and_properties(\n \"nvinfer config-file-path=c.txt ! nvv4l2decoder ! filesrc location=t.mp4 ! nveglglessink\"\n )\n warnings = validate_element_ordering(elements)\n self.assertTrue(len(warnings) > 0)\n\n def test_sink_before_source_warns(self):\n elements, _ = extract_elements_and_properties(\n \"nveglglessink ! filesrc location=t.mp4 ! nvinfer config-file-path=c.txt\"\n )\n warnings = validate_element_ordering(elements)\n # nvinfer after nveglglessink is detected via filesrc -> nvinfer ordering check\n self.assertTrue(len(warnings) > 0)\n\n def test_tracker_before_infer_warns(self):\n elements, _ = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvtracker ll-lib-file=t.so ! nvinfer config-file-path=c.txt ! nveglglessink\"\n )\n warnings = validate_element_ordering(elements)\n self.assertTrue(any(\"nvtracker\" in w and \"nvinfer\" in w for w in warnings))\n\n\nclass TestExpandedNVMMChecks(unittest.TestCase):\n \"\"\"Test that NVMM checks cover encoders too.\"\"\"\n\n def test_system_mem_to_h264enc_detected(self):\n pipeline = \"gst-launch-1.0 filesrc location=a.jpg ! nvjpegdec ! nvv4l2h264enc ! filesink location=out.h264\"\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertTrue(any(\"nvjpegdec\" in e and \"nvv4l2h264enc\" in e for e in errors))\n\n def test_videoconvert_to_h264enc_detected(self):\n pipeline = \"gst-launch-1.0 filesrc location=a.mp4 ! videoconvert ! nvv4l2h264enc ! filesink location=out.h264\"\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertTrue(any(\"videoconvert\" in e and \"nvv4l2h264enc\" in e for e in errors))\n\n def test_nvvideoconvert_to_h264enc_passes(self):\n pipeline = \"gst-launch-1.0 filesrc location=a.mp4 ! nvvideoconvert ! nvv4l2h264enc ! filesink location=out.h264\"\n elements, pad_refs = extract_elements_and_properties(pipeline)\n errors = validate_memory_format(pipeline, elements, pad_refs)\n self.assertEqual(errors, [])\n\n\nclass TestNewElements(unittest.TestCase):\n \"\"\"Test that new elements are in the known properties dict.\"\"\"\n\n def test_nvinferserver_in_known(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"nvinferserver\", KNOWN_ELEMENT_PROPERTIES)\n self.assertIn(\"config-file-path\", KNOWN_ELEMENT_PROPERTIES[\"nvinferserver\"])\n\n def test_nvdsanalytics_in_known(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"nvdsanalytics\", KNOWN_ELEMENT_PROPERTIES)\n\n def test_nvmsgconv_in_known(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"nvmsgconv\", KNOWN_ELEMENT_PROPERTIES)\n\n def test_nvmsgbroker_in_known(self):\n from validate_pipeline import KNOWN_ELEMENT_PROPERTIES\n self.assertIn(\"nvmsgbroker\", KNOWN_ELEMENT_PROPERTIES)\n\n def test_nvinferserver_missing_config_flagged(self):\n elements, pad_refs = extract_elements_and_properties(\n \"filesrc location=t.mp4 ! nvinferserver batch-size=1 ! nveglglessink\"\n )\n errors, _ = validate_pipeline_structure(elements, pad_refs)\n self.assertTrue(any(\"nvinferserver\" in e and \"config-file-path\" in e for e in errors))\n\n\nclass TestDryRunMultiStream(unittest.TestCase):\n \"\"\"Dry-run is skipped for multi-stream pipelines with named pad refs.\"\"\"\n\n def test_skipped_when_pad_refs_present(self):\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! \"\n \"m.sink_0 nvstreammux name=m batch-size=2 width=1920 height=1080 ! \"\n \"nvinfer config-file-path=c.txt batch-size=2 ! nvvideoconvert ! nvdsosd ! \"\n \"nveglglessink filesrc location=b.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_1\"\n )\n pad_refs = [\"m.sink_0\", \"m.sink_1\"]\n errors, warnings = validate_with_gst_launch(pipeline, pad_refs)\n self.assertEqual(errors, [])\n self.assertEqual(warnings, [])\n\n def test_not_skipped_when_no_pad_refs(self):\n \"\"\"Dry-run should execute (not be skipped) for single-stream pipelines.\"\"\"\n pipeline = (\n \"gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! \"\n \"nvv4l2decoder ! nveglglessink\"\n )\n errors, warnings = validate_with_gst_launch(pipeline, pad_refs=[])\n self.assertIsInstance(errors, list)\n self.assertIsInstance(warnings, list)\n\n def test_not_skipped_when_pad_refs_none(self):\n pipeline = \"gst-launch-1.0 videotestsrc ! fakesink\"\n errors, warnings = validate_with_gst_launch(pipeline, pad_refs=None)\n self.assertIsNotNone(errors)\n self.assertIsNotNone(warnings)\n self.assertIsInstance(errors, list)\n self.assertIsInstance(warnings, list)\n\n\nif __name__ == \"__main__\":\n unittest.main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":21347,"content_sha256":"80dd19e4b57886bd1887a5cea1c665b29facd948052cf7843b498d60bbdcc499"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"DeepStream Pipeline Builder","type":"text"}]},{"type":"paragraph","content":[{"text":"Generate ready-to-run ","type":"text"},{"text":"gst-launch-1.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" pipelines for NVIDIA DeepStream SDK by collecting pipeline requirements through an interactive questionnaire, then assembling the pipeline using a standalone BM25 retrieval backend with structural metadata boosting (similarity search over 270+ verified pipelines, zero external dependencies).","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Prerequisites","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Python:","type":"text","marks":[{"type":"strong"}]},{"text":" 3.8+ (stdlib only — no pip packages required)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"DeepStream SDK:","type":"text","marks":[{"type":"strong"}]},{"text":" Installed at ","type":"text"},{"text":"/opt/nvidia/deepstream/deepstream/","type":"text","marks":[{"type":"code_inline"}]},{"text":" (for ","type":"text"},{"text":"gst-inspect-1.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" validation and element verification)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"GStreamer:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"gst-launch-1.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"gst-inspect-1.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" on ","type":"text"},{"text":"PATH","type":"text","marks":[{"type":"code_inline"}]},{"text":" (installed with DeepStream)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Platform:","type":"text","marks":[{"type":"strong"}]},{"text":" x86 dGPU (T4, A100, L40, RTX, etc.) or aarch64 — Jetson (Orin, Xavier, Nano) / SBSA (Grace, GH200)","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Usage Examples","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"text"},"content":[{"text":"# Fully specified — skips most questions\ndetect and track on 4 rtsp streams and display on jetson\n\n# Partially specified — asks remaining questions\ngive me a pipeline to infer on an image\n\n# Minimal — asks all 7 questions\nbuild a pipeline","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Supported Configurations","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Parameter","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Options","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Input","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Local video (.mp4/.h264/.h265), local image (.jpg/.png), RTSP stream, USB camera, test pattern","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Inference","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"None, primary (nvinfer), primary+secondary, with preprocessor, Triton (nvinferserver)","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Tracker","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"None, NvDCF, IOU, NvSORT, DeepSORT","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Sink","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Display (dGPU/Jetson), save (JPG/PNG/MP4/H264), RTSP out, fakesink","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Platform","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"x86 dGPU (T4, A100, L40, RTX, etc.) or aarch64 — Jetson (Orin, Xavier, Nano) / SBSA (Grace, GH200)","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Extras","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Resize, rotate/flip, crop, color format conversion","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Scripts","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Script","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Purpose","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"scripts/generate_pipeline.py","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"BM25 retrieval engine — scores and ranks pipelines from ","type":"text"},{"text":"data/data.csv","type":"text","marks":[{"type":"code_inline"}]},{"text":". Supports ","type":"text"},{"text":"--format {json,compact,summary}","type":"text","marks":[{"type":"code_inline"}]},{"text":" (default ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"scripts/validate_pipeline.py","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"4-stage validator: syntax, elements, properties, live parse. Supports ","type":"text"},{"text":"--format {json,summary}","type":"text","marks":[{"type":"code_inline"}]},{"text":" (default ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"scripts/lint_data.py","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Data quality linter for the pipeline CSV (","type":"text"},{"text":"--fix","type":"text","marks":[{"type":"code_inline"}]},{"text":" to auto-repair)","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Workflow","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 1 — Collect Pipeline Requirements","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"You MUST ","type":"text","marks":[{"type":"strong"}]},{"text":"Read references/requirement-extraction.md","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" before doing this step.","type":"text","marks":[{"type":"strong"}]},{"text":" It contains the query-inference table, compound-extraction examples, the full ","type":"text"},{"text":"AskUserQuestion","type":"text","marks":[{"type":"code_inline"}]},{"text":" question bank (with the default-first ordering contract), the automatic-OSD and extras/flip-method rules, and the dynamic question-reduction examples that this step depends on. Apply them exactly.","type":"text"}]}]},{"type":"paragraph","content":[{"text":"Order of operations:","type":"text","marks":[{"type":"strong"}]}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Infer everything you can from the query","type":"text","marks":[{"type":"strong"}]},{"text":" using the inference table in ","type":"text"},{"text":"references/requirement-extraction.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". The goal is to identify which of the 7 parameters (input source, num sources, inference, tracker, sink, platform, extras) the user has already specified.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Ask the user about the unknowns via ","type":"text","marks":[{"type":"strong"}]},{"text":"AskUserQuestion","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" in a single call.","type":"text","marks":[{"type":"strong"}]},{"text":" Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" silently default tracker/sink/platform/extras — these are real choices the user should make explicitly (display vs save, no tracker vs NvDCF, x86 dGPU vs aarch64 Jetson/SBSA, etc.). Skip only the questions whose answer is already clear from the query.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Quote the inferred parameters back to the user","type":"text","marks":[{"type":"strong"}]},{"text":" in the lead-in to the question call so they can see what you already extracted. Example: ","type":"text"},{"text":"\"From your query I have: 3 mp4 videos, primary inference. Just need a few more details:\"","type":"text","marks":[{"type":"em"}]}]}]}]},{"type":"paragraph","content":[{"text":"Follow the inference table, question bank, and OSD/extras rules in ","type":"text"},{"text":"references/requirement-extraction.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" to decide which questions to ask and how to place transform elements, then proceed to Step 2.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 2 — Build the Natural Language Query","type":"text"}]},{"type":"paragraph","content":[{"text":"From the user's answers, construct a single descriptive query string. Follow this pattern:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"text"},"content":[{"text":"Please provide a GStreamer pipeline that [operation] on [num_sources] [input_type] [input_detail] [tracker_detail] and [output_action] [platform_detail]","type":"text"}]},{"type":"paragraph","content":[{"text":"Examples of constructed queries:","type":"text","marks":[{"type":"strong"}]}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"User Selections","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Constructed Query","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Local video, 1 source, Primary detector, No tracker, Display, dGPU","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"Please provide a GStreamer pipeline that performs primary inference on a single mp4 video and displays the output\"","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"RTSP, 4 sources, Primary+Secondary, NvDCF, Save MP4, dGPU","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"Please provide a GStreamer pipeline that performs primary and secondary inference with NvDCF tracker on 4 RTSP streams and saves output to MP4 file\"","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Local video, 2 sources, Primary with preprocessor, IOU, Display, Jetson","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"Please provide a GStreamer pipeline that performs preprocessing before primary inference with IOU tracker on 2 mp4 streams and displays the output on Jetson\"","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Local image, 1 source, None, No tracker, Save file, dGPU, Rotate 90° cw","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"Please provide a GStreamer pipeline that rotates a single jpg image 90° clockwise before processing and saves it to a file\"","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Local video, 3 sources, Primary detector, NvDCF, Save MP4, dGPU, Rotate 180°","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"Please provide a GStreamer pipeline that rotates 3 mp4 videos 180° before primary inference with NvDCF tracker and saves output to MP4 file\"","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 3 — Run the Pipeline Generator Script","type":"text"}]},{"type":"paragraph","content":[{"text":"Execute the backend script with the constructed query and user parameters:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python3 \u003cskill-path>/scripts/generate_pipeline.py \\\n --query \"\u003cconstructed_query>\" \\\n --source-type \"\u003cLocal video file|Local image file|RTSP stream|USB camera|Test pattern>\" \\\n --num-sources \u003cN> \\\n --inference \"\u003cNone|primary|primary+secondary|primary+preprocess|primary+secondary+preprocess|primary-triton|primary+secondary-triton>\" \\\n --tracker \"\u003cnone|NvDCF|IOU|NvSORT|DeepSORT>\" \\\n --sink \"\u003cdisplay|display-jetson|save-jpg|save-png|save-mp4|save-h264|rtsp-out|fakesink>\" \\\n --platform \"\u003cdGPU|Jetson|SBSA>\" \\\n --extras \"\u003cnone|resize|rotate|crop|color-convert|osd>\" \\\n --format compact","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Always pass ","type":"text","marks":[{"type":"strong"}]},{"text":"--format compact","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":".","type":"text","marks":[{"type":"strong"}]},{"text":" The ","type":"text"},{"text":"compact","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode returns only confidence + the top retrieved pipeline (~25 lines), instead of dumping all 10 retrievals as ~150 lines of JSON in the chat. The ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode (default for backward compat) is only useful when debugging the retriever directly. A ","type":"text"},{"text":"summary","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode (single human-readable line) also exists for non-Claude callers.","type":"text"}]}]},{"type":"paragraph","content":[{"text":"The script will (zero external dependencies — pure Python stdlib):","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Load the pipeline dataset (270+ verified DeepStream pipelines)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Extract structural metadata from each pipeline (platform, source type, sink type, inference mode, tracker, stream count)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Score with BM25 (document-length-normalized) + domain-specific synonym expansion on both queries and documents","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Apply structural boosting — results matching the user's platform/source/sink/inference get boosted, mismatches get penalized","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Return the top-K results as JSON with a ","type":"text"},{"text":"confidence","type":"text","marks":[{"type":"code_inline"}]},{"text":" field (","type":"text"},{"text":"high","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"medium","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"low","type":"text","marks":[{"type":"code_inline"}]},{"text":") based on the top score","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Claude uses these retrieved examples + the assembly rules below to construct the final pipeline","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"When ","type":"text"},{"text":"confidence","type":"text","marks":[{"type":"code_inline"}]},{"text":" is ","type":"text"},{"text":"low","type":"text","marks":[{"type":"code_inline"}]},{"text":", rely more heavily on the assembly rules below rather than the retrieved examples.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 4 — Validate the Pipeline","type":"text"}]},{"type":"paragraph","content":[{"text":"Before presenting, run the validation script to catch syntax errors, unknown elements, and linking issues:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python3 \u003cskill-path>/scripts/validate_pipeline.py \"\u003cassembled_pipeline>\" --format summary","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Always pass ","type":"text","marks":[{"type":"strong"}]},{"text":"--format summary","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":".","type":"text","marks":[{"type":"strong"}]},{"text":" Summary prints a single status line (e.g. ","type":"text"},{"text":"valid · 11 elements · 0 warnings · live-parse skipped (multi-stream)","type":"text","marks":[{"type":"code_inline"}]},{"text":"), with errors/warnings indented underneath only if present. The default ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode emits ~40 lines of structured output and is only useful for programmatic callers.","type":"text"}]}]},{"type":"paragraph","content":[{"text":"The validator performs 4 checks:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Syntax check","type":"text","marks":[{"type":"strong"}]},{"text":" — unbalanced quotes, empty pipe segments, missing source/sink","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Element check","type":"text","marks":[{"type":"strong"}]},{"text":" — verifies each element exists via ","type":"text"},{"text":"gst-inspect-1.0","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Property check","type":"text","marks":[{"type":"strong"}]},{"text":" — validates known properties for DeepStream elements","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Live parse check","type":"text","marks":[{"type":"strong"}]},{"text":" — uses ","type":"text"},{"text":"gst-launch-1.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" itself to construct the pipeline graph (with fakesrc/fakesink substituted), catching linking errors and pad mismatches. ","type":"text"},{"text":"Automatically skipped for multi-stream pipelines","type":"text","marks":[{"type":"strong"}]},{"text":" (those with named pad refs like ","type":"text"},{"text":"m.sink_0","type":"text","marks":[{"type":"code_inline"}]},{"text":") since fakesrc cannot negotiate caps through named pads.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"If validation fails (","type":"text"},{"text":"\"valid\": false","type":"text","marks":[{"type":"code_inline"}]},{"text":"), fix the errors and re-validate before presenting. ","type":"text"},{"text":"Limit validation retries to a maximum of 2 attempts","type":"text","marks":[{"type":"strong"}]},{"text":" — if the pipeline still fails after 2 fixes, present it as-is (the remaining checks already cover syntax, element, property, and structural correctness). If there are only warnings, present the pipeline but mention the warnings to the user.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 5 — Present the Pipeline","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"5.1 — Output format (THE ONLY ACCEPTED FORM)","type":"text"}]},{"type":"paragraph","content":[{"text":"Your response ","type":"text"},{"text":"must","type":"text","marks":[{"type":"strong"}]},{"text":" be exactly five blocks, in this order:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"One-line ","type":"text"},{"text":"status badge","type":"text","marks":[{"type":"strong"}]},{"text":" (validation + confidence)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Single bash code block","type":"text","marks":[{"type":"strong"}]},{"text":" containing the full ","type":"text"},{"text":"gst-launch-1.0 -e …","type":"text","marks":[{"type":"code_inline"}]},{"text":" command with concrete absolute paths, on ","type":"text"},{"text":"one line","type":"text","marks":[{"type":"strong"}]},{"text":" (no ","type":"text"},{"text":"\\","type":"text","marks":[{"type":"code_inline"}]},{"text":" continuations, no shell variables, no shell wrapper)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Breakdown table","type":"text","marks":[{"type":"strong"}]},{"text":" grouped by stage","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Suggestions","type":"text","marks":[{"type":"strong"}]},{"text":" bullet list","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"(only if pre-flight failed) a ","type":"text"},{"text":"⚠","type":"text","marks":[{"type":"code_inline"}]},{"text":" line above the status badge stating which default path is missing","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"That is the ONLY accepted output shape for this step. The Section 5.3 template in ","type":"text"},{"text":"references/output-format.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" is the literal template — match it.","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"5.2 — Pre-flight check (run before composing the response)","type":"text"}]},{"type":"paragraph","content":[{"text":"Run one ","type":"text"},{"text":"Bash","type":"text","marks":[{"type":"code_inline"}]},{"text":" ","type":"text"},{"text":"ls","type":"text","marks":[{"type":"code_inline"}]},{"text":" over the default paths the pipeline will reference (sample video, PGIE config, tracker lib/config). The result tells you whether to mark the badge with ","type":"text"},{"text":"⚠ default path not found: \u003cpath>","type":"text","marks":[{"type":"code_inline"}]},{"text":" and bump the matching \"Use your own …\" suggestion to the top.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"ls /opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 \\\n /opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_infer_primary.txt \\\n 2>&1","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"5.3 / 5.4 — Worked example & forbidden anti-patterns","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"You MUST ","type":"text","marks":[{"type":"strong"}]},{"text":"Read references/output-format.md","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" before composing this response.","type":"text","marks":[{"type":"strong"}]},{"text":" It contains the literal Section 5.3 template your output must match exactly, and the Section 5.4 gallery of forbidden output shapes (heredoc wrappers, shell-var indirection, ","type":"text"},{"text":"\\","type":"text","marks":[{"type":"code_inline"}]},{"text":" line-continuations, stray \"Run it\" lines, ","type":"text"},{"text":"Write","type":"text","marks":[{"type":"code_inline"}]},{"text":"-to-script). Mirror Section 5.3; never emit any Section 5.4 form.","type":"text"}]}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"5.5 — Self-check before sending the response","type":"text"}]},{"type":"paragraph","content":[{"text":"Before you emit your reply, mentally tick each box. If any check fails, rewrite the response.","type":"text"}]},{"type":"checkbox_list","attrs":{"id":null},"content":[{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The pipeline is on ","type":"text"},{"text":"exactly one line","type":"text","marks":[{"type":"strong"}]},{"text":" inside a single ","type":"text"},{"text":"```bash","type":"text","marks":[{"type":"code_inline"}]},{"text":" code block.","type":"text"}]}]},{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The pipeline begins with ","type":"text"},{"text":"gst-launch-1.0 -e","type":"text","marks":[{"type":"code_inline"}]},{"text":" and contains only literal absolute paths (e.g. ","type":"text"},{"text":"/opt/nvidia/deepstream/...","type":"text","marks":[{"type":"code_inline"}]},{"text":") — no ","type":"text"},{"text":"$VAR","type":"text","marks":[{"type":"code_inline"}]},{"text":", no ","type":"text"},{"text":"${VAR:-default}","type":"text","marks":[{"type":"code_inline"}]},{"text":", no ","type":"text"},{"text":"cat >","type":"text","marks":[{"type":"code_inline"}]},{"text":", no ","type":"text"},{"text":"EOF","type":"text","marks":[{"type":"code_inline"}]},{"text":", no ","type":"text"},{"text":"\\","type":"text","marks":[{"type":"code_inline"}]},{"text":" line continuations.","type":"text"}]}]},{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The response does ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" contain any of: ","type":"text"},{"text":"cat > /tmp/pipeline.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"bash /tmp/pipeline.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"\u003c\u003c'EOF'","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"${VAR:-","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The response does ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" call the ","type":"text"},{"text":"Write","type":"text","marks":[{"type":"code_inline"}]},{"text":" tool. (Save-to-file is offered as a ","type":"text"},{"text":"suggestion bullet","type":"text","marks":[{"type":"em"}]},{"text":", not an action.)","type":"text"}]}]},{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The breakdown table is grouped by stage (Source / Mux / Inference / Tracking / Composition / Render — adapt names to the pipeline's actual stages, e.g. add an ","type":"text"},{"text":"Encode/Mux","type":"text","marks":[{"type":"code_inline"}]},{"text":" row for file sinks).","type":"text"}]}]},{"type":"checkbox_item","attrs":{"checked":false},"content":[{"type":"paragraph","content":[{"text":"The \"Save it to a script?\" line appears in the Suggestions list — never as a primary action.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"5.6 — Pre-flight failure variant","type":"text"}]},{"type":"paragraph","content":[{"text":"If the Section 5.2 ","type":"text"},{"text":"ls","type":"text","marks":[{"type":"code_inline"}]},{"text":" reported one or more missing default paths, prepend a ","type":"text"},{"text":"⚠","type":"text","marks":[{"type":"code_inline"}]},{"text":" line above the status badge and bump the matching \"Use your own …\" suggestion to the top:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"markdown"},"content":[{"text":"⚠ default path not found: `/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4` — substitute your own video path before running\n✓ Validated · 11 elements · 0 warnings · confidence: HIGH\n\n```bash\ngst-launch-1.0 -e filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! …\n```\n\n[breakdown + suggestions as in Section 5.3, with the \"Use your own video\" suggestion bumped to the top]","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"On length:","type":"text","marks":[{"type":"strong"}]},{"text":" 5–8 stream pipelines run long when on a single line. That is correct and intended — chat clients render bash code blocks faithfully and copy reproduces them correctly. Long ≠ split.","type":"text"}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 6 — Offer Refinement","type":"text"}]},{"type":"paragraph","content":[{"text":"After presenting the pipeline, ask the user if they want to adjust anything:","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Want me to modify anything? For example:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Change the number of streams","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Add/remove tracker or secondary inference","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Switch between display and file output","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Change the platform (x86 dGPU / aarch64 Jetson / SBSA)","type":"text"}]}]}]}]},{"type":"paragraph","content":[{"text":"If the user requests changes, go back to ","type":"text"},{"text":"Step 2","type":"text","marks":[{"type":"strong"}]},{"text":" with updated parameters — do NOT re-ask all 7 questions. Only ask about the specific parameter that changed, or just apply the change directly if it's clear.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 6.5 — Optional: Save Pipeline to a Script","type":"text"}]},{"type":"paragraph","content":[{"text":"Only do this step when the user explicitly asks (e.g. ","type":"text"},{"text":"\"save it\"","type":"text","marks":[{"type":"em"}]},{"text":", ","type":"text"},{"text":"\"save to pipeline.sh\"","type":"text","marks":[{"type":"em"}]},{"text":", ","type":"text"},{"text":"\"write it to a file\"","type":"text","marks":[{"type":"em"}]},{"text":", ","type":"text"},{"text":"\"put it in ~/run.sh\"","type":"text","marks":[{"type":"em"}]},{"text":"). Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" create the file proactively — Step 5 always shows the concrete pipeline in chat for direct copy-paste; saving is a follow-up convenience.","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Filename:","type":"text","marks":[{"type":"strong"}]},{"text":" Default to ","type":"text"},{"text":"/tmp/pipeline.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":" if the user just says ","type":"text"},{"text":"\"save it\"","type":"text","marks":[{"type":"em"}]},{"text":". Use the exact path the user named otherwise (e.g. ","type":"text"},{"text":"~/run.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"scripts/demo.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"File contents:","type":"text","marks":[{"type":"strong"}]},{"text":" Two lines — shebang + the same single-line pipeline shown in chat (concrete absolute paths, no shell vars). Keep them in sync — what the user runs from the file is bit-for-bit identical to what they could have copy-pasted.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"#!/usr/bin/env bash\ngst-launch-1.0 -e filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_0 … ! nvdsosd ! nveglglessink","type":"text"}]},{"type":"paragraph","content":[{"text":"Use the ","type":"text"},{"text":"Write","type":"text","marks":[{"type":"code_inline"}]},{"text":" tool to create the file.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Confirm to user","type":"text","marks":[{"type":"strong"}]},{"text":" with the run command:","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Saved to ","type":"text"},{"text":"\u003cpath>","type":"text","marks":[{"type":"code_inline"}]},{"text":". Run it with:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"bash \u003cpath>","type":"text"}]}]}]}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Pipeline Assembly Rules","type":"text"}]},{"type":"paragraph","content":[{"text":"When the script is not available or fails, assemble the pipeline using the rules in ","type":"text"},{"text":"references/assembly-rules.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/assembly-rules.md","title":null}}]},{"text":". These rules cover source elements, multi-stream patterns, inference chains, tracker configs, sink elements, and extra operations. They also serve as validation for script output.","type":"text"}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Error Handling","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Failure","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Cause","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Recovery","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"generate_pipeline.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" returns ","type":"text"},{"text":"confidence: low","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Query doesn't match any pipeline in the dataset closely","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Rely on the assembly rules in this skill instead of retrieved examples","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"validate_pipeline.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" reports unknown element","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"GStreamer/DeepStream not installed or not on ","type":"text"},{"text":"PATH","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Install DeepStream SDK; confirm ","type":"text"},{"text":"gst-inspect-1.0 nvinfer","type":"text","marks":[{"type":"code_inline"}]},{"text":" works","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Validation fails after 2 retries","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Unusual element combination or linking issue","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Present the pipeline as-is with a warning — syntax/element/property checks still passed","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Script not found at ","type":"text"},{"text":"\u003cskill-path>/scripts/","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Skill not installed correctly or path misconfigured","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Verify the skill directory is symlinked into ","type":"text"},{"text":".claude/skills/","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":".cursor/skills/","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Testing","type":"text"}]},{"type":"paragraph","content":[{"text":"Run the test suite to verify retrieval quality and validator correctness:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python3 -m unittest discover -s \u003cskill-path>/tests -v","type":"text"}]},{"type":"paragraph","content":[{"text":"The suite includes:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Unit tests","type":"text","marks":[{"type":"strong"}]},{"text":" for the BM25 retriever (tokenizer, synonym expansion, metadata extraction, scoring)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Unit tests","type":"text","marks":[{"type":"strong"}]},{"text":" for the validator (syntax, structure, property, named-pad checks)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Golden regression tests","type":"text","marks":[{"type":"strong"}]},{"text":" — 20+ query→expected-result pairs ensuring retrieval quality doesn't regress","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data quality linter","type":"text","marks":[{"type":"strong"}]},{"text":" — checks the CSV for duplicates, syntax issues, and structural bugs:","type":"text"}]}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"python3 \u003cskill-path>/scripts/lint_data.py # report issues\npython3 \u003cskill-path>/scripts/lint_data.py --fix # auto-fix and overwrite","type":"text"}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Security, Limitations & Notes","type":"text"}]},{"type":"paragraph","content":[{"text":"Security posture, known limitations, and operational notes are documented in ","type":"text"},{"text":"references/security-and-limitations.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". Read that file when you need details on subprocess safety, input validation, platform/SDK requirements, the multi-stream dry-run caveat, or sample-path/config-file reminders.","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"deepstream-generate-pipeline","owner":"NVIDIA CORPORATION","author":"@skillopedia","source":{"stars":3825,"repo_name":"skills","origin_url":"https://github.com/NVIDIA/skills/blob/HEAD/skills/deepstream-generate-pipeline/SKILL.md","repo_owner":"NVIDIA","body_sha256":"ad1c26f75e186253c49b02655dfd013034c90d9d3b8ce3e019cc253dd348594a","cluster_key":"16d7138de64472e28a31aaa50fd8f0306f10d45a50c968dc51e20232828b0355","clean_bundle":{"format":"clean-skill-bundle-v1","source":"NVIDIA/skills/skills/deepstream-generate-pipeline/SKILL.md","text_bytes":273859,"source_commit":"25340619336e89bcc40365c28bfe9e790430468e","skills_sh_rank":15054,"attachment_count":16,"text_attachments":16,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/deepstream-generate-pipeline/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"1.0.0","category":"media-content","category_label":"Media"},"exact_dupes_collapsed_into_this":0},"license":"CC-BY-4.0 AND Apache-2.0","service":"deepstream","version":"1.0.0","category":"media-content","reviewed":"2026-04-27T00:00:00.000Z","import_tag":"skills-sh-top20k-2026-11","description":"Build DeepStream GStreamer pipelines interactively. Use when the user asks about pipelines for video/image inference, detection, tracking, or streaming — including natural phrases like 'pipeline to infer on image', 'run inference on video', 'detect objects in stream', 'save inference output', 'deepstream pipeline', 'gst-launch pipeline', 'process video with detection', 'build a pipeline', or any request involving GStreamer/DeepStream elements (nvinfer, nvstreammux, nvtracker, etc.)."}},"renderedAt":1793666249978}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.