{
  "name": "deepstream-generate-pipeline",
  "description": "Build DeepStream GStreamer pipelines interactively. Collects input source, inference, tracker, and output preferences, then assembles a ready-to-run gst-launch-1.0 pipeline using BM25 retrieval over 270+ verified pipelines (zero external dependencies).",
  "skills": "./"
}

{
  "skill_name": "deepstream-generate-pipeline",
  "evals": [
    {
      "id": 1,
      "name": "happy-path-video-inference-display",
      "prompt": "Give me a pipeline to run inference on an mp4 video and display the output",
      "expected_output": "A gst-launch-1.0 pipeline with filesrc, nvv4l2decoder, nvstreammux, nvinfer, nvdsosd, and nveglglessink",
      "files": [],
      "assertions": [
        {
          "text": "Pipeline starts with gst-launch-1.0",
          "type": "contains_phrase",
          "phrase": "gst-launch-1.0"
        },
        {
          "text": "Pipeline includes nvinfer for primary inference",
          "type": "contains_phrase",
          "phrase": "nvinfer"
        },
        {
          "text": "Pipeline includes nvstreammux for batching",
          "type": "contains_phrase",
          "phrase": "nvstreammux"
        },
        {
          "text": "Pipeline includes nvdsosd for on-screen display",
          "type": "contains_phrase",
          "phrase": "nvdsosd"
        },
        {
          "text": "Response does not contain API tokens or credentials",
          "type": "not_contains_pattern",
          "pattern": "(Bearer |sk-|token=)[A-Za-z0-9+/=]{10,}"
        }
      ]
    },
    {
      "id": 2,
      "name": "happy-path-image-inference-save-jpg",
      "prompt": "Build a pipeline to infer on a jpg image and save the output as a jpg file",
      "expected_output": "A gst-launch-1.0 pipeline with filesrc, jpegparse, nvjpegdec or nvv4l2decoder, nvstreammux, nvinfer, nvdsosd, jpegenc, and filesink",
      "files": [],
      "assertions": [
        {
          "text": "Pipeline starts with gst-launch-1.0",
          "type": "contains_phrase",
          "phrase": "gst-launch-1.0"
        },
        {
          "text": "Pipeline reads a JPEG source",
          "type": "contains_pattern",
          "pattern": "(jpegparse|nvjpegdec|nvv4l2decoder)"
        },
        {
          "text": "Pipeline includes nvinfer for inference",
          "type": "contains_phrase",
          "phrase": "nvinfer"
        },
        {
          "text": "Pipeline saves output via jpegenc and filesink",
          "type": "contains_phrase",
          "phrase": "jpegenc"
        },
        {
          "text": "Pipeline includes filesink for output",
          "type": "contains_phrase",
          "phrase": "filesink"
        },
        {
          "text": "Response does not contain API tokens or credentials",
          "type": "not_contains_pattern",
          "pattern": "(Bearer |sk-|token=)[A-Za-z0-9+/=]{10,}"
        }
      ]
    },
    {
      "id": 3,
      "name": "happy-path-multi-stream-rtsp-tracker",
      "prompt": "Detect and track objects on 4 RTSP streams with NvDCF tracker and display on Jetson",
      "expected_output": "A multi-stream pipeline with 4 uridecodebin sources, nvstreammux with batch-size=4, nvinfer, nvtracker with NvDCF config, nvmultistreamtiler, nvdsosd, and nv3dsink",
      "files": [],
      "assertions": [
        {
          "text": "Pipeline uses nvstreammux with batch-size 4",
          "type": "contains_phrase",
          "phrase": "batch-size=4"
        },
        {
          "text": "Pipeline includes nvtracker for object tracking",
          "type": "contains_phrase",
          "phrase": "nvtracker"
        },
        {
          "text": "Pipeline references NvDCF tracker config",
          "type": "contains_phrase",
          "phrase": "NvDCF"
        },
        {
          "text": "Pipeline includes nvmultistreamtiler for multi-stream tiling",
          "type": "contains_phrase",
          "phrase": "nvmultistreamtiler"
        },
        {
          "text": "Pipeline uses nv3dsink for Jetson display",
          "type": "contains_phrase",
          "phrase": "nv3dsink"
        },
        {
          "text": "Response does not contain API tokens or credentials",
          "type": "not_contains_pattern",
          "pattern": "(Bearer |sk-|token=)[A-Za-z0-9+/=]{10,}"
        }
      ]
    },
    {
      "id": 4,
      "name": "negative-ambiguous-request",
      "prompt": "Build a pipeline",
      "expected_output": "Skill asks the user all 7 configuration questions (input source, num sources, inference, tracker, sink, platform, extras) before generating a pipeline",
      "files": [],
      "assertions": [
        {
          "text": "Skill asks about input source type",
          "type": "contains_pattern",
          "pattern": "(input source|source type|input type|[Ww]hat.*input)"
        },
        {
          "text": "Skill asks about number of sources/streams",
          "type": "contains_pattern",
          "pattern": "(num.*source|number.*source|how many|stream)"
        },
        {
          "text": "Skill asks about inference/detection",
          "type": "contains_pattern",
          "pattern": "(inference|detection|infer)"
        },
        {
          "text": "Skill asks about tracker",
          "type": "contains_pattern",
          "pattern": "(track|tracker|tracking)"
        },
        {
          "text": "Skill asks about output/sink",
          "type": "contains_pattern",
          "pattern": "(output|sink|[Ww]hat should happen)"
        },
        {
          "text": "Skill asks about platform",
          "type": "contains_pattern",
          "pattern": "(platform|dGPU|[Jj]etson|[Ww]hich platform)"
        },
        {
          "text": "Skill asks about extra operations",
          "type": "contains_pattern",
          "pattern": "(extra|resize|rotate|flip|crop|optional)"
        },
        {
          "text": "Skill does NOT produce a gst-launch pipeline without asking questions first",
          "type": "not_contains",
          "phrase": "gst-launch-1.0 filesrc"
        }
      ]
    },
    {
      "id": 5,
      "name": "happy-path-safe-file-path-handling",
      "prompt": "Give me a pipeline to infer on a video file located at /tmp/test.mp4 and save the output",
      "expected_output": "A gst-launch-1.0 pipeline using filesrc location=/tmp/test.mp4, nvv4l2decoder, nvstreammux, nvinfer, nvdsosd, and a filesink for saving output",
      "files": [],
      "assertions": [
        {
          "text": "Pipeline starts with gst-launch-1.0",
          "type": "contains_phrase",
          "phrase": "gst-launch-1.0"
        },
        {
          "text": "Pipeline uses the provided file path in filesrc location",
          "type": "contains_phrase",
          "phrase": "filesrc location=/tmp/test.mp4"
        },
        {
          "text": "Response does not contain shell injection characters",
          "type": "not_contains_pattern",
          "pattern": "[;|`$]\\("
        },
        {
          "text": "Response does not contain API tokens or credentials",
          "type": "not_contains_pattern",
          "pattern": "(Bearer |sk-|token=)[A-Za-z0-9+/=]{10,}"
        }
      ]
    }
  ]
}

# Pipeline Assembly Rules

When the script is not available or fails, Claude can assemble the pipeline directly using these rules. These rules also serve as validation for the script output.

## Source Elements

| Source Type | Elements |
| --- | --- |
| Local video (.mp4) | `filesrc location=<path> ! qtdemux ! h264parse ! nvv4l2decoder` |
| Local video (.h264) | `filesrc location=<path> ! h264parse ! nvv4l2decoder` |
| Local video (.h265) | `filesrc location=<path> ! h265parse ! nvv4l2decoder` |
| Local image (.jpg) | `filesrc location=<path> ! jpegparse ! nvjpegdec ! nvvideoconvert` or `filesrc location=<path> ! jpegparse ! nvv4l2decoder` |
| RTSP stream | `uridecodebin uri=rtsp://<url>` |
| USB camera | `v4l2src device=/dev/video0 ! video/x-raw,width=1280,height=720 ! nvvideoconvert` |
| Test pattern | `videotestsrc num-buffers=1000 ! video/x-raw,format=NV12,width=1920,height=1080` |

> **Note:** `nvjpegdec` outputs system memory (`video/x-raw`), so it **always** needs `nvvideoconvert` after it before connecting to `nvstreammux` or any NVMM-requiring element. `nvv4l2decoder` outputs NVMM directly and does not need this conversion.

## Multi-Stream Pattern

For N > 1 sources, use `nvstreammux` with named pads:

```text
<source_0_elements> ! m.sink_0
nvstreammux name=m batch-size=<N> width=1920 height=1080 ! <rest_of_pipeline>
<source_1_elements> ! m.sink_1
...
<source_N-1_elements> ! m.sink_<N-1>
```

Always add `nvmultistreamtiler width=1920 height=1080` after inference for multi-stream.

## Inference Chain

| Mode | Elements |
| --- | --- |
| Primary only | `nvinfer config-file-path=<config> batch-size=<N> unique-id=1` |
| Primary + Secondary | `nvinfer <primary_config> ! ... ! nvinfer <secondary_config> infer-on-gie-id=1` |
| With preprocessor | `nvdspreprocess config-file=<config> ! nvinfer <config> input-tensor-meta=1` |
| Primary only (Triton) | `nvinferserver config-file-path=<config> batch-size=<N> unique-id=1` |
| Primary + Secondary (Triton) | `nvinferserver <primary_config> ! ... ! nvinferserver <secondary_config> infer-on-gie-id=1` |

## Tracker Elements

| Tracker | Config |
| --- | --- |
| NvDCF | `nvtracker ll-lib-file=/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so ll-config-file=config_tracker_NvDCF_perf.yml` |
| IOU | `nvtracker ll-lib-file=/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so ll-config-file=config_tracker_IOU.yml` |
| NvSORT | `nvtracker ll-lib-file=/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so ll-config-file=config_tracker_NvSORT.yml` |
| DeepSORT | `nvtracker ll-lib-file=/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so ll-config-file=config_tracker_DeepSORT.yml` |

Tracker goes **after primary inference** and **before secondary inference** (if any).

## Sink Elements

| Sink | Elements |
| --- | --- |
| Display (x86 dGPU) | `nvvideoconvert ! nvdsosd ! nveglglessink` |
| Display (aarch64 Jetson / SBSA) | `nvvideoconvert ! nvdsosd ! nv3dsink` |
| Save JPG | `nvvideoconvert ! nvdsosd ! nvvideoconvert ! jpegenc ! filesink location=out.jpg` |
| Save PNG | `nvvideoconvert ! nvdsosd ! nvvideoconvert ! pngenc ! filesink location=out.png` |
| Save MP4 | `nvvideoconvert ! nvdsosd ! nvv4l2h264enc ! h264parse ! qtmux ! filesink location=out.mp4` |
| Save H264 | `nvvideoconvert ! nvv4l2h264enc ! filesink location=out.h264` |
| RTSP out | `nvvideoconvert ! nvv4l2h264enc ! h264parse ! rtph264pay ! udpsink host=<ip> port=<port>` |
| Fakesink | `fakesink sync=0` |

## Extra Operations

**Placement:** Resize, rotate, flip, and crop operate on the **incoming video** and must be inserted **immediately after `nvstreammux`** (before `nvinfer`) so inference receives correctly oriented/scaled frames. The pipeline order is:

```text
nvstreammux ! nvvideoconvert <extra_ops> ! nvinfer ! ...
```

| Operation | How to add |
| --- | --- |
| Resize | `nvvideoconvert ! "video/x-raw(memory:NVMM),format=NV12,width=W,height=H"` — insert after muxer, before infer |
| Rotate/Flip | `nvvideoconvert flip-method=<0-7>` — insert after muxer, before infer. Values: 0=none, 1=ccw90, 2=180, 3=cw90, 4=hflip, 5=ur-ll, 6=vflip, 7=ul-lr |
| Crop | `nvvideoconvert src-crop=X:Y:W:H` or `dest-crop=X:Y:W:H` — insert after muxer, before infer |
| Color convert | `nvvideoconvert ! "video/x-raw,format=RGB"` |

**Example** (rotate 90° clockwise before inference):
```text
nvstreammux name=m ... ! nvvideoconvert flip-method=3 ! nvinfer ... ! nvtracker ... ! nvdsosd ! nveglglessink
```

> **Note:** `nvdsosd` is not listed here — it is automatically included in the sink chain when inference is present (see Sink Elements table above).

# Step 5 — Output Format: Worked Example & Anti-Patterns

Reference for **Step 5 — Present the Pipeline**. The enforcing rules (the 5-block
contract Section 5.1, the pre-flight Section 5.2, the self-check Section 5.5, and the failure variant
Section 5.6) live in SKILL.md. This file holds the **literal correct template (Section 5.3)** your
output MUST match, and the **forbidden anti-pattern gallery (Section 5.4)**. Read it before
composing the Step 5 response.

---

#### 5.3 — ✅ CORRECT example (3-stream infer + display, dGPU, defaults present)

This is exactly what your output should look like. Match it.

````markdown
✓ Validated · 11 elements · 0 warnings · live-parse skipped (multi-stream) · confidence: HIGH (matched 10 verified examples)

```bash
gst-launch-1.0 -e filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_0 filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_1 filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_2 nvstreammux name=m batch-size=3 width=1920 height=1080 ! nvinfer config-file-path=/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_infer_primary.txt batch-size=3 unique-id=1 ! nvmultistreamtiler rows=1 columns=3 width=1920 height=720 ! nvvideoconvert ! nvdsosd ! nveglglessink
```

**Pipeline breakdown (by stage):**

| Stage | Elements | Role |
| --- | --- | --- |
| Source (×3) | `filesrc → qtdemux → h264parse → nvv4l2decoder` | Decode each MP4 onto a separate `nvstreammux` sink pad |
| Mux | `nvstreammux batch-size=3` | Batch the 3 streams into a single batched buffer |
| Inference | `nvinfer` (PGIE) | Primary object detection on the batched frames |
| Composition | `nvmultistreamtiler rows=1 columns=3` | Tile 3 streams side-by-side into one output frame |
| Render | `nvdsosd → nveglglessink` | Overlay bounding boxes and render to display |

**Suggestions:**

- **Use your own video:** replace `/opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4` with your file path (in all 3 `filesrc` lines).
- **Use your own model config:** replace `/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_infer_primary.txt` with your `config_infer_*.txt`.
- **Save it to a script?** Say *"save it"* and I'll write it to `/tmp/pipeline.sh` — then run with `bash /tmp/pipeline.sh`.
- **Other changes?** Want me to add a tracker (NvDCF/NvSORT/IOU/DeepSORT), change the tile layout (e.g. `2x2`), save the output to MP4 instead of display, or switch to Jetson?
````

#### 5.4 — ❌ ANTI-PATTERNS — DO NOT EVER PRODUCE OUTPUT IN THESE FORMS

The following formats are forbidden. They are listed here so you have a concrete reference of the wrong shapes to avoid. **If your draft response resembles any of them, throw it away and re-emit in the Section 5.3 form instead.**

❌ **NEVER use a heredoc-to-script wrapper:**

````markdown
```bash
cat > /tmp/pipeline.sh <<'EOF'
#!/usr/bin/env bash
gst-launch-1.0 -e \
  filesrc location=… ! m.sink_0 \
  filesrc location=… ! m.sink_1 \
  …
EOF
bash /tmp/pipeline.sh
```
````

❌ **NEVER use shell-variable indirection or env-var overrides:**

````markdown
```bash
DS=${DS:-/opt/nvidia/deepstream/deepstream}
SRC=${SRC:-$DS/samples/streams/sample_1080p_h264.mp4}
gst-launch-1.0 -e filesrc location=$SRC ! …
```
````

❌ **NEVER add a "Run it:" or "bash /tmp/pipeline.sh" instruction.** The pipeline in the Section 5.3 code block IS the runnable thing — pasting it in a terminal runs it.

❌ **NEVER split the pipeline across multiple lines with `\` continuations**, even for readability. Long lines are correct; the bash code block preserves them faithfully on copy.

❌ **NEVER call the `Write` tool to create `/tmp/pipeline.sh` (or any other script file) as part of Step 5.** Saving to a script is opt-in via Step 6.5 — only when the user explicitly asks "save it" or names a path. The default delivery is the inline single-line command, period.

# Step 1 — Requirement Extraction Reference

Detailed lookup tables, the full `AskUserQuestion` question bank, and the
extraction/question-reduction examples for **Step 1 — Collect Pipeline
Requirements** of the DeepStream Pipeline Builder skill. SKILL.md instructs the
agent to read this file before performing Step 1; apply everything here exactly.

---

**First, extract what you already know from the user's query.** Parse the original request for any parameters that are obvious — do NOT re-ask what's already clear. Be aggressive about inferring from context: if the user says "infer on 3 videos and display" you already know input type (video), num sources (3), inference (primary), and sink (display) — that's 4 out of 7 parameters resolved without asking.

| If the query mentions... | You already know |
| --- | --- |
| "image", "jpg", "png", "picture", "photo" | Input = Local image file, Num sources = 1 (unless stated otherwise) |
| "video", "mp4", "h264", "h265", "clip" | Input = Local video file |
| "rtsp", "stream", "camera stream", "ip camera" | Input = RTSP stream |
| "usb camera", "webcam", "/dev/video" | Input = USB camera |
| "infer", "detect", "inference", "detection", "model" | Inference = Primary detector |
| "classify", "secondary", "vehicle type", "car color" | Inference = Primary + Secondary |
| "triton", "nvinferserver", "inference server" | Inference = Primary detector (Triton) |
| "track", "tracker", "tracking", "re-id" | Tracker = present (ask which one) |
| "display", "show", "render", "screen", "view" | Sink = Display |
| "save", "output file", "write to file", "record", "store" | Sink = Save (ask which format) |
| "benchmark", "throughput", "fps", "performance" | Sink = Fakesink |
| "jetson", "orin", "xavier", "nano" | Platform = Jetson |
| "dgpu", "server", "T4", "A100", "RTX", "L40" | Platform = dGPU |
| "N streams", "N sources", "N videos", "N cameras" | Num sources = N |
| A number followed by "video/stream/camera/file" | Num sources = that number AND Input = corresponding type |
| "rotate 90", "rotate clockwise", "cw90" | Extras = Rotate, flip-method = 3 |
| "rotate 180", "flip 180" | Extras = Rotate, flip-method = 2 |
| "rotate 270", "rotate counter-clockwise", "ccw90" | Extras = Rotate, flip-method = 1 |
| "mirror", "horizontal flip", "hflip" | Extras = Rotate, flip-method = 4 |
| "vertical flip", "vflip", "flip upside down" | Extras = Rotate, flip-method = 6 |
| "rotate", "flip" (without specific direction) | Extras = Rotate (ask which direction) |
| "resize", "scale", "resolution" | Extras = Resize (ask target W x H) |
| "crop" | Extras = Crop (ask X:Y:W:H) |

> **Compound extraction examples:**
> - *"infer on 3 videos"* → Input = Local video file, Num sources = 3, Inference = Primary detector
> - *"detect and track objects on rtsp stream"* → Input = RTSP, Inference = Primary detector, Tracker = present
> - *"run inference on a jpg and save to png"* → Input = Local image file, Num sources = 1, Inference = Primary detector, Sink = Save PNG
> - *"4 camera streams with detection on jetson"* → Input = RTSP, Num sources = 4, Inference = Primary detector, Platform = Jetson
> - *"rotate 90 clockwise, infer on 2 videos and display"* → Input = video, Num = 2, Inference = primary, Sink = display, Extras = rotate flip-method=3 (placed after muxer, before infer)
> - *"infer on video and flip horizontally"* → Input = video, Num = 1, Inference = primary, Extras = rotate flip-method=4 (placed after muxer, before infer)

Use `AskUserQuestion` to ask **only the remaining unknown parameters** in a single call. **Never re-ask a parameter that can be inferred from the query.** Skip any question whose answer is already clear.

> **Important:** Ask all unknown questions in one call. Do NOT ask one at a time. If every parameter is already inferable from the query, skip the question call entirely and jump to Step 2. For most user queries, you should be able to resolve 3–5 parameters automatically, leaving only 2–3 questions.
>
> **Default-first ordering convention:** the first `option` in every question's `options` array is the **safe default** for that parameter (e.g. `Local video file` for input, `1` for num sources, `No tracker` for tracker, `Display on screen` for sink, `dGPU` for platform, `None` for extras). Most prompt UIs render the first option as the highlighted/initial selection, so a user who just hits Enter lands on a sensible choice. **Do not reorder** the options in the question bank below — preserving "default first" is part of the contract.
>
> **Critical — never get stuck asking.** If the user rejects/dismisses the `AskUserQuestion` call (e.g. "Tool use rejected"), or replies *"just generate"* / *"use defaults"* / *"go ahead"* / *"skip"* / no answer, **immediately fall through to Step 2** using the **first option** of each unknown question as the parameter value. Do NOT re-ask the same questions in chat — that creates a loop and frustrates the user. The user can always refine afterwards by saying *"change to NvDCF"*, *"save as mp4"*, etc., once they see the generated pipeline. The flow is **ask once, then generate** — never ask twice.

Below is the **full question bank** — include only the questions you actually need:

```json
{
  "questions": [
    {
      "id": "input_source",
      "question": "What is the input source type?",
      "header": "Input Source",
      "options": [
        {"label": "Local video file", "description": "filesrc with local .mp4/.h264/.h265 file"},
        {"label": "Local image file", "description": "filesrc with local .jpg/.png image"},
        {"label": "RTSP stream", "description": "uridecodebin/rtspsrc with rtsp:// URL"},
        {"label": "USB camera", "description": "v4l2src from /dev/video* device"},
        {"label": "Test pattern", "description": "videotestsrc for testing without real input"}
      ],
      "multiSelect": false
    },
    {
      "id": "num_sources",
      "question": "How many input sources/streams?",
      "header": "Number of Sources",
      "options": [
        {"label": "1", "description": "Single stream"},
        {"label": "2", "description": "Dual stream (tiled output)"},
        {"label": "4", "description": "Quad stream (2x2 tile)"},
        {"label": "8", "description": "8 streams (2x4 tile)"}
      ],
      "multiSelect": false
    },
    {
      "id": "inference",
      "question": "What inference/detection do you need?",
      "header": "Inference Model",
      "options": [
        {"label": "None", "description": "No inference — just decode/convert/display"},
        {"label": "Primary detector (nvinfer)", "description": "Single primary inference (e.g. object detection)"},
        {"label": "Primary + Secondary (nvinfer)", "description": "Primary detection + secondary classification (e.g. vehicle type)"},
        {"label": "Primary with preprocessor", "description": "nvdspreprocess + nvinfer for custom ROI/batching"},
        {"label": "Primary + Secondary with preprocessor", "description": "nvdspreprocess before both primary and secondary infer"},
        {"label": "Primary detector (nvinferserver/Triton)", "description": "Single primary inference via Triton Inference Server (nvinferserver)"},
        {"label": "Primary + Secondary (nvinferserver/Triton)", "description": "Primary + secondary classification via Triton (nvinferserver)"}
      ],
      "multiSelect": false
    },
    {
      "id": "tracker",
      "question": "Do you need object tracking?",
      "header": "Tracker",
      "options": [
        {"label": "No tracker", "description": "Skip tracking — inference only"},
        {"label": "NvDCF (accurate)", "description": "Discriminative Correlation Filter — best accuracy, higher compute"},
        {"label": "IOU (fast)", "description": "Intersection-over-Union tracker — lightweight, fast"},
        {"label": "NvSORT", "description": "NVIDIA SORT tracker — good balance of speed and accuracy"},
        {"label": "DeepSORT", "description": "Deep association metric — re-ID based, best for occlusion"}
      ],
      "multiSelect": false
    },
    {
      "id": "sink",
      "question": "What should happen with the output?",
      "header": "Output / Sink",
      "options": [
        {"label": "Display on screen", "description": "nveglglessink — render to display (dGPU)"},
        {"label": "Display on Jetson", "description": "nv3dsink — render on Jetson display"},
        {"label": "Save to JPG file", "description": "Encode to JPEG image via jpegenc + filesink"},
        {"label": "Save to PNG file", "description": "Encode to PNG image via pngenc + filesink"},
        {"label": "Save to MP4 file", "description": "Encode H264 + mux to .mp4 via filesink"},
        {"label": "Save to H264 file", "description": "Encode to raw .h264 bitstream file"},
        {"label": "Stream over RTSP", "description": "Encode and push to RTSP server via udpsink"},
        {"label": "Fakesink (benchmark)", "description": "fakesink — discard output, measure throughput"}
      ],
      "multiSelect": false
    },
    {
      "id": "platform",
      "question": "Which platform are you targeting?",
      "header": "Platform",
      "options": [
        {"label": "x86 dGPU", "description": "x86_64 desktop/server with discrete GPU (T4, A100, L40, RTX, etc.) — uses nveglglessink"},
        {"label": "aarch64 (Jetson / SBSA)", "description": "ARM aarch64 — Jetson Orin/Xavier/Nano and SBSA servers (e.g. Grace, GH200) — both use nv3dsink and nvv4l2* plugins"}
      ],
      "multiSelect": false
    },
    {
      "id": "extras",
      "question": "Any extra operations? (optional — select all that apply)",
      "header": "Extra Operations",
      "options": [
        {"label": "None", "description": "No extra processing needed"},
        {"label": "Resize / scale video", "description": "Change resolution via nvvideoconvert caps"},
        {"label": "Rotate / flip video", "description": "Flip method via nvvideoconvert (90/180/270/horizontal/vertical)"},
        {"label": "Crop input region", "description": "src-crop or dest-crop via nvvideoconvert"},
        {"label": "Color format conversion", "description": "Convert between NV12, RGBA, I420, BGR, etc."}
      ],
      "multiSelect": true
    }
  ]
}
```

> **OSD is automatic:** When inference is present and the sink is display or file save (MP4/JPG/PNG), `nvdsosd` is always included — do NOT ask the user about it. OSD is only omitted for raw bitstream sinks (H264), RTSP out, and fakesink.

> **Follow-up for extras:** If the user selects "Rotate / flip video", ask which rotation they want before proceeding:
>
> | Option | `flip-method` value |
> | --- | --- |
> | Rotate 90° counter-clockwise | 1 |
> | Rotate 180° | 2 |
> | Rotate 90° clockwise | 3 |
> | Horizontal flip (mirror) | 4 |
> | Vertical flip | 6 |
>
> If the user selects "Resize / scale video", ask for the target width and height (e.g., 1280x720).
> If the user selects "Crop input region", ask for the crop rectangle as X:Y:W:H.
>
> **Placement:** Rotate, resize, and crop transform the **incoming video** — insert `nvvideoconvert` with these properties **immediately after `nvstreammux`, before `nvinfer`**, so inference receives correctly oriented/scaled frames. Do NOT place them after the tiler or before the sink.

**Examples of dynamic question reduction:**

| User query | Already known | Questions to ask |
| --- | --- | --- |
| *"give me the pipeline to infer on an image"* | Input = image, Num = 1, Inference = primary | tracker, sink, platform, extras (4 questions) |
| *"detect and track on 3 videos and display on jetson"* | Input = video, Num = 3, Inference = primary, Tracker = present, Sink = display, Platform = Jetson | tracker type, extras (2 questions) |
| *"benchmark inference throughput on 8 rtsp streams"* | Input = RTSP, Num = 8, Inference = primary, Sink = fakesink | tracker, platform, extras (3 questions) |
| *"just decode and display an mp4"* | Input = video, Num = 1, Inference = none, Sink = display | platform, extras (2 questions) |
| *"infer on 3 videos, rotate 90 clockwise, save mp4"* | Input = video, Num = 3, Inference = primary, Sink = save MP4, Extras = rotate flip-method=3 | tracker, platform (2 questions) |
| *"build a pipeline"* | Nothing known | All 7 questions |

# Security, Limitations & Notes

Reference details for the DeepStream Pipeline Builder skill. These do not affect the
interactive workflow — they document the security posture, known limitations, and
operational notes of the pipeline generator.

## Security

- **No shell execution of user input:** All subprocess calls use list-form arguments (`subprocess.run([...])`, never `shell=True`), preventing shell injection regardless of pipeline string content
- **Element name validation:** Element names extracted from the pipeline are validated against `[a-zA-Z0-9_-]` before being passed to `gst-inspect-1.0`; names that fail validation are skipped rather than passed to subprocess
- **Safe tokenization:** The live parse check uses `shlex.split()` rather than naive string splitting, so quoted tokens are handled correctly and unexpected argument injection is avoided
- **Output length cap:** GStreamer stderr lines are capped at 300 characters in warnings to prevent info leakage from verbose runtime output
- **Pipeline size limit:** Inputs longer than 16 384 characters are rejected before any processing
- **No credential handling:** This skill does not process, store, or transmit credentials, tokens, or secrets
- **Data classification:** Public — not intended for processing sensitive or confidential media files

## Limitations

- Requires DeepStream SDK installed locally for element validation (`gst-inspect-1.0`) and live parse checks (`gst-launch-1.0`)
- Pipeline dataset covers common DeepStream patterns — exotic or fully custom element chains may need manual assembly using the assembly rules in `references/assembly-rules.md`
- Live parse dry-run is skipped for multi-stream pipelines (those with named pad refs like `m.sink_0`) because `fakesrc` cannot negotiate caps through `nvstreammux` named pads
- Platform is Linux only — macOS and Windows are not supported
- The BM25 retriever uses no embeddings or semantic model; very unusual queries may get `confidence: low`, in which case the assembly rules in this skill take precedence over retrieved examples

## Notes

- The script is fully standalone — zero external dependencies, pure Python stdlib (BM25 scoring with structural metadata boosting and domain synonym expansion)
- The pipeline dataset (`data.csv`) contains 270+ verified DeepStream pipelines covering decode, encode, inference, tracking, format conversion, and more
- Default sample paths use `/opt/nvidia/deepstream/deepstream/samples/` — remind users to update paths for their setup
- For inference pipelines, users must provide their own `config_infer_*.txt` config files — the defaults point to DeepStream sample configs
- The skill assembles pipelines from proven patterns — it does NOT invent arbitrary element chains
- The retriever outputs a `confidence` field — when confidence is low, the LLM should rely more on the assembly rules than on the retrieved examples

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
DeepStream Pipeline Generator — BM25 retrieval with structural boosting.

Loads a CSV dataset of verified DeepStream pipelines and retrieves the most
relevant ones using BM25 scoring (document-length-normalized) combined with
structural metadata boosting so that results match the user's platform, source
type, sink type, and inference mode.

Zero external dependencies — uses only Python stdlib.

Usage:
    python3 generate_pipeline.py \
        --query "pipeline that performs primary inference on a single mp4 video" \
        --source-type "Local video file" \
        --num-sources 1 \
        --inference "primary" \
        --tracker "none" \
        --sink "display" \
        --platform "dGPU" \
        --extras "none"
"""

import argparse
import csv
import json
import math
import os
import re
import sys
import tempfile
from collections import Counter

# Bumped manually whenever the indexed-state schema changes (tokenizer
# rules, metadata fields, structural-boost behavior). Stored inside the
# cache file so that an old cache from a prior code version is rejected
# even when the CSV mtime hasn't changed.
INDEX_CACHE_VERSION = 3

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SKILL_DIR = os.path.dirname(SCRIPT_DIR)
DEFAULT_DATA_CSV = os.path.join(SKILL_DIR, "data", "data.csv")
TOP_K = 10
MIN_SCORE = 0.5
BM25_K1 = 1.5
BM25_B = 0.75

SYNONYMS = {
    "infer": ["inference", "nvinfer", "detect", "detection", "classify"],
    "inference": ["infer", "nvinfer", "detect", "detection", "classify"],
    "detect": ["detection", "infer", "inference", "nvinfer"],
    "display": ["render", "show", "nveglglessink", "nv3dsink", "sink"],
    "save": ["dump", "store", "write", "filesink", "output"],
    "image": ["jpg", "jpeg", "png", "frame", "picture"],
    "video": ["mp4", "h264", "h265", "stream", "clip"],
    "track": ["tracker", "nvtracker", "tracking"],
    "tracker": ["track", "nvtracker", "tracking"],
    "rtsp": ["stream", "rtspsrc", "uridecodebin"],
    "primary": ["nvinfer", "pgie", "detector"],
    "secondary": ["sgie", "classifier", "classification"],
    "crop": ["src-crop", "dest-crop", "region"],
    "flip": ["rotate", "flip-method", "mirror"],
    "rotate": ["flip", "flip-method"],
    "resize": ["scale", "resolution", "width", "height"],
    "jetson": ["nv3dsink", "nvv4l2"],
    "dgpu": ["nveglglessink", "desktop", "server"],
    "mp4": ["qtmux", "qtdemux", "video", "h264"],
    "jpg": ["jpeg", "jpegenc", "jpegdec", "nvjpegdec", "image"],
    "jpeg": ["jpg", "jpegenc", "jpegdec", "nvjpegdec", "image"],
    "osd": ["nvdsosd", "bounding", "boxes", "labels", "overlay"],
    "preprocess": ["nvdspreprocess", "preprocessor", "roi"],
    "dewarp": ["nvdewarper", "dewarper", "perspective"],
    "encode": ["encoder", "nvv4l2h264enc", "nvv4l2h265enc", "h264", "h265"],
    "decode": ["decoder", "nvv4l2decoder", "decodebin"],
    "segment": ["segmentation", "nvsegvisual", "semantic"],
    "postprocess": ["nvdspostprocess", "postprocessor"],
    "triton": ["nvinferserver", "inference-server", "grpc", "model-repository"],
    "nvinferserver": ["triton", "inference-server"],
    "analytics": ["nvdsanalytics", "line-crossing", "roi", "direction"],
}


# ---------------------------------------------------------------------------
# Pipeline metadata extraction
# ---------------------------------------------------------------------------
def extract_pipeline_metadata(pipeline):
    """Extract structural metadata from a pipeline string for filtering."""
    p = pipeline.lower()

    # Platform
    if "nv3dsink" in p:
        platform = "jetson"
    elif "nveglglessink" in p or "nvegltransform" in p:
        platform = "dgpu"
    else:
        platform = "unknown"

    # Source type
    if "filesrc" in p and ("jpegparse" in p or "nvjpegdec" in p or ".jpg" in p or ".png" in p):
        source_type = "image"
    elif "filesrc" in p:
        source_type = "video"
    elif "uridecodebin" in p or "rtspsrc" in p or "nvurisrcbin" in p:
        if "rtsp://" in p:
            source_type = "rtsp"
        else:
            source_type = "video"
    elif "v4l2src" in p:
        source_type = "usb"
    elif "videotestsrc" in p or "nvvideotestsrc" in p:
        source_type = "test"
    else:
        source_type = "unknown"

    # Sink type
    if "nveglglessink" in p or "nv3dsink" in p or "xvimagesink" in p or "autovideosink" in p:
        sink_type = "display"
    elif "filesink" in p and ("qtmux" in p or "mp4" in p):
        sink_type = "save-mp4"
    elif "filesink" in p and "jpegenc" in p:
        sink_type = "save-jpg"
    elif "filesink" in p and "pngenc" in p:
        sink_type = "save-png"
    elif "filesink" in p:
        sink_type = "save-file"
    elif "udpsink" in p:
        sink_type = "rtsp-out"
    elif "fakesink" in p:
        sink_type = "fakesink"
    else:
        sink_type = "unknown"

    nvinfer_count = len(re.findall(r"\bnvinfer\b", p))
    nvinferserver_count = len(re.findall(r"\bnvinferserver\b", p))
    total_infer = nvinfer_count + nvinferserver_count
    has_preprocess = "nvdspreprocess" in p
    if total_infer == 0:
        inference = "none"
    elif total_infer == 1 and has_preprocess:
        inference = "primary+preprocess"
    elif total_infer == 1:
        inference = "primary"
    elif total_infer >= 2 and has_preprocess:
        inference = "primary+secondary+preprocess"
    else:
        inference = "primary+secondary"

    # Tracker
    has_tracker = "nvtracker" in p
    tracker = "present" if has_tracker else "none"

    # Number of streams
    sink_pads = re.findall(r"m\.sink_(\d+)", p)
    if sink_pads:
        num_sources = max(int(n) for n in sink_pads) + 1
    else:
        num_sources = 1

    return {
        "platform": platform,
        "source_type": source_type,
        "sink_type": sink_type,
        "inference": inference,
        "tracker": tracker,
        "num_sources": num_sources,
    }


def normalize_user_params(args):
    """Map user CLI params to the same vocabulary as pipeline metadata."""
    source_map = {
        "local video file": "video",
        "local image file": "image",
        "rtsp stream": "rtsp",
        "usb camera": "usb",
        "test pattern": "test",
    }
    sink_map = {
        "display": "display",
        "display-jetson": "display",
        "save-jpg": "save-jpg",
        "save-png": "save-png",
        "save-mp4": "save-mp4",
        "save-h264": "save-file",
        "rtsp-out": "rtsp-out",
        "fakesink": "fakesink",
    }
    inference_map = {
        "none": "none",
        "primary": "primary",
        "primary+secondary": "primary+secondary",
        "primary+preprocess": "primary+preprocess",
        "primary+secondary+preprocess": "primary+secondary+preprocess",
        "primary-triton": "primary",
        "primary+secondary-triton": "primary+secondary",
    }
    # SBSA (aarch64 servers, e.g. Grace/GH200) shares Jetson's sink/plugin
    # set (nv3dsink, nvv4l2*), so it maps to the "jetson" code path.
    def _map_platform(value):
        v = value.lower()
        if "jetson" in v or "sbsa" in v or "aarch64" in v or "arm" in v:
            return "jetson"
        if "dgpu" in v or "x86" in v or "gpu" in v:
            return "dgpu"
        return "unknown"

    return {
        "platform": _map_platform(args.platform),
        "source_type": source_map.get(args.source_type.lower(), "unknown"),
        "sink_type": sink_map.get(args.sink.lower(), "unknown"),
        "inference": inference_map.get(args.inference.lower(), "unknown"),
        "tracker": "present" if args.tracker.lower() != "none" else "none",
        "num_sources": args.num_sources,
    }


# ---------------------------------------------------------------------------
# Tokenizer
# ---------------------------------------------------------------------------
def tokenize(text):
    """
    Lowercase, strip punctuation, split into tokens.

    `/` is intentionally NOT in the kept-char set: a path like
    /opt/nvidia/deepstream/samples/streams/sample.mp4 should split into
    ['opt', 'nvidia', 'deepstream', 'samples', 'streams', 'sample.mp4']
    so each path component contributes useful IDF signal — instead of
    collapsing into one giant unique token that bloats the IDF table and
    matches almost nothing. Compound tokens with hyphens (e.g.
    config-file-path) are preserved because `-` is in the kept set.
    """
    text = text.lower()
    text = re.sub(r"[^a-z0-9._-]", " ", text)
    return text.split()


def expand_with_synonyms(tokens):
    """Add domain-specific synonyms to boost recall."""
    expanded = list(tokens)
    seen = set(tokens)
    for tok in tokens:
        for syn in SYNONYMS.get(tok, []):
            if syn not in seen:
                expanded.append(syn)
                seen.add(syn)
    return expanded


# ---------------------------------------------------------------------------
# BM25 Engine
# ---------------------------------------------------------------------------
class BM25Retriever:
    """BM25 retriever with structural metadata boosting. Pure stdlib."""

    def __init__(self, k1=BM25_K1, b=BM25_B):
        self.k1 = k1
        self.b = b
        self.documents = []
        self.metadata = []
        self.doc_tokens = []
        self.doc_tf = []
        self.idf = {}
        self.avgdl = 0

    def load_csv(self, csv_path):
        """
        Load pipeline dataset from CSV.

        The fully-indexed retriever state (documents, metadata, tokens,
        TF, IDF, avgdl) is cached to a sibling JSON file
        ``{csv_path}.cache.json`` keyed on the CSV's mtime + an internal
        schema version. Subsequent invocations with an unchanged CSV
        skip the parse + tokenization + IDF passes entirely. The cache
        is invalidated automatically when the CSV is edited or when the
        indexed schema changes.

        JSON is parser-only — loading a cache file cannot execute
        arbitrary code, even if the file has been tampered with. An
        unreadable, mtime-mismatched, or version-mismatched cache is
        rejected and the dataset is re-indexed from scratch.
        """
        if self._load_from_cache(csv_path):
            return

        with open(csv_path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                prompt = row.get("Prompt", "").strip()
                pipeline = row.get("Gst launch pipeline", "").strip()
                if prompt and pipeline:
                    self.documents.append((prompt, pipeline))
                    self.metadata.append(extract_pipeline_metadata(pipeline))

        for prompt, pipeline in self.documents:
            prompt_tokens = tokenize(prompt)
            pipeline_tokens = tokenize(pipeline)
            tokens = expand_with_synonyms(prompt_tokens * 2 + pipeline_tokens)
            self.doc_tokens.append(tokens)

        self._compute_idf()
        self._precompute_tf()
        self._save_to_cache(csv_path)

    @staticmethod
    def _cache_path(csv_path):
        return csv_path + ".cache.json"

    def _load_from_cache(self, csv_path):
        cache_path = self._cache_path(csv_path)
        if not os.path.exists(cache_path):
            return False
        try:
            csv_mtime = os.path.getmtime(csv_path)
        except OSError:
            return False
        try:
            with open(cache_path, "r", encoding="utf-8") as f:
                payload = json.load(f)
        except (json.JSONDecodeError, OSError, UnicodeDecodeError):
            return False
        if not isinstance(payload, dict):
            return False
        if payload.get("version") != INDEX_CACHE_VERSION:
            return False
        if payload.get("csv_mtime") != csv_mtime:
            return False
        try:
            self.documents = payload["documents"]
            self.metadata = payload["metadata"]
            self.doc_tokens = payload["doc_tokens"]
            self.doc_tf = payload["doc_tf"]
            self.idf = payload["idf"]
            self.avgdl = payload["avgdl"]
        except KeyError:
            return False
        return True

    def _save_to_cache(self, csv_path):
        cache_path = self._cache_path(csv_path)
        try:
            csv_mtime = os.path.getmtime(csv_path)
        except OSError:
            return
        payload = {
            "version": INDEX_CACHE_VERSION,
            "csv_mtime": csv_mtime,
            "documents": self.documents,
            "metadata": self.metadata,
            "doc_tokens": self.doc_tokens,
            "doc_tf": self.doc_tf,
            "idf": self.idf,
            "avgdl": self.avgdl,
        }
        # Atomic write — temp file in same dir, then rename. Avoids a
        # partial cache file if the process is killed mid-write.
        cache_dir = os.path.dirname(cache_path) or "."
        try:
            fd, tmp_path = tempfile.mkstemp(
                prefix=".cache.", suffix=".json.tmp", dir=cache_dir,
            )
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as f:
                    json.dump(payload, f)
                os.replace(tmp_path, cache_path)
            except Exception:
                # Best-effort cleanup; never raise from the cache writer.
                try:
                    os.unlink(tmp_path)
                except OSError:
                    pass
        except OSError:
            # Read-only filesystem or permission denied — fine, just no cache.
            pass

    def _compute_idf(self):
        """Compute IDF with BM25 formula: log((N - df + 0.5) / (df + 0.5) + 1)."""
        n = len(self.doc_tokens)
        df = Counter()
        for tokens in self.doc_tokens:
            for tok in set(tokens):
                df[tok] += 1
        self.idf = {
            tok: math.log((n - count + 0.5) / (count + 0.5) + 1)
            for tok, count in df.items()
        }
        total_len = sum(len(t) for t in self.doc_tokens)
        self.avgdl = total_len / n if n else 1

    def _precompute_tf(self):
        """Precompute term frequency dicts per document."""
        self.doc_tf = [Counter(tokens) for tokens in self.doc_tokens]

    def _bm25_score(self, query_tokens, doc_idx):
        """Score a single document against query tokens."""
        tf = self.doc_tf[doc_idx]
        dl = len(self.doc_tokens[doc_idx])
        score = 0.0
        for tok in query_tokens:
            if tok not in tf:
                continue
            f = tf[tok]
            idf = self.idf.get(tok, 0)
            numerator = f * (self.k1 + 1)
            denominator = f + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
            score += idf * numerator / denominator
        return score

    def _structural_boost(self, doc_meta, user_meta):
        """Compute a multiplier based on structural metadata match."""
        boost = 1.0

        # Platform: strong signal
        if user_meta["platform"] != "unknown" and doc_meta["platform"] != "unknown":
            if doc_meta["platform"] == user_meta["platform"]:
                boost *= 1.4
            else:
                boost *= 0.4

        # Source type
        if user_meta["source_type"] != "unknown" and doc_meta["source_type"] != "unknown":
            if doc_meta["source_type"] == user_meta["source_type"]:
                boost *= 1.3
            else:
                boost *= 0.7

        # Sink type
        if user_meta["sink_type"] != "unknown" and doc_meta["sink_type"] != "unknown":
            if doc_meta["sink_type"] == user_meta["sink_type"]:
                boost *= 1.2
            else:
                boost *= 0.7

        # Inference mode
        if user_meta["inference"] != "unknown" and doc_meta["inference"] != "unknown":
            if doc_meta["inference"] == user_meta["inference"]:
                boost *= 1.3
            elif "primary" in doc_meta["inference"] and "primary" in user_meta["inference"]:
                boost *= 1.1

        # Tracker
        if doc_meta["tracker"] == user_meta["tracker"]:
            boost *= 1.1

        # Number of sources (prefer exact match, tolerate close)
        if user_meta["num_sources"] == doc_meta["num_sources"]:
            boost *= 1.2
        elif user_meta["num_sources"] > 1 and doc_meta["num_sources"] > 1:
            boost *= 1.05

        return boost

    def retrieve(self, query, user_meta=None, top_k=TOP_K, min_score=MIN_SCORE):
        """Find top-K most relevant pipelines for a query."""
        query_tokens = expand_with_synonyms(tokenize(query))

        scores = []
        for i in range(len(self.documents)):
            raw = self._bm25_score(query_tokens, i)
            if raw <= 0:
                continue

            if user_meta:
                boost = self._structural_boost(self.metadata[i], user_meta)
                final = raw * boost
            else:
                final = raw

            scores.append((final, raw, i))

        scores.sort(key=lambda x: x[0], reverse=True)

        # Confidence is thresholded against the *raw* BM25 score, not the
        # boosted final score. Boosting can multiply the score by up to
        # ~3.75× (1.4·1.3·1.2·1.3·1.1·1.2 across platform / source / sink /
        # inference / tracker / num-sources matches), so thresholding the
        # boosted score would make the badge inflate the moment the
        # structural metadata aligns — even when the textual match is weak.
        top_raw = scores[0][1] if scores else 0
        results = []
        seen_pipelines = set()
        for final, raw, idx in scores[:top_k * 2]:
            if final < min_score and len(results) >= 3:
                break
            prompt, pipeline = self.documents[idx]
            normalized = re.sub(r"\s+", " ", pipeline.strip())
            if normalized in seen_pipelines:
                continue
            seen_pipelines.add(normalized)
            results.append({
                "score": round(final, 4),
                "raw_bm25": round(raw, 4),
                "prompt": prompt,
                "pipeline": pipeline,
                "metadata": self.metadata[idx],
            })
            if len(results) >= top_k:
                break

        confidence = "low"
        if top_raw >= 6:
            confidence = "high"
        elif top_raw >= 3:
            confidence = "medium"

        return results, confidence


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(description="DeepStream Pipeline Generator (BM25)")
    parser.add_argument("--query", required=True, help="Natural language pipeline query")
    parser.add_argument("--source-type", default="Local video file")
    parser.add_argument("--num-sources", type=int, default=1)
    parser.add_argument("--inference", default="None")
    parser.add_argument("--tracker", default="none")
    parser.add_argument("--sink", default="display")
    parser.add_argument("--platform", default="dGPU")
    parser.add_argument("--extras", default="none")
    parser.add_argument("--data-csv", default=DEFAULT_DATA_CSV, help="Path to pipeline dataset CSV")
    parser.add_argument("--top-k", type=int, default=TOP_K)
    parser.add_argument("--min-score", type=float, default=MIN_SCORE)
    parser.add_argument(
        "--format", dest="output_format",
        choices=("summary", "compact", "json"), default="json",
        help=(
            "Output format. 'json' = full verbose output (default — preserves "
            "existing behavior). 'compact' = small JSON with confidence + top "
            "result only. 'summary' = one-line human status."
        ),
    )

    args = parser.parse_args()

    if not os.path.exists(args.data_csv):
        print(f"ERROR: Dataset not found at {args.data_csv}", file=sys.stderr)
        sys.exit(1)

    enriched_query = (
        f"{args.query} "
        f"{args.source_type} "
        f"{args.inference} "
        f"{args.tracker} "
        f"{args.sink} "
        f"{args.platform} "
        f"{args.extras}"
    )

    user_meta = normalize_user_params(args)

    retriever = BM25Retriever()
    retriever.load_csv(args.data_csv)
    results, confidence = retriever.retrieve(
        enriched_query, user_meta=user_meta,
        top_k=args.top_k, min_score=args.min_score,
    )

    output = {
        "query": args.query,
        "confidence": confidence,
        "context": {
            "source_type": args.source_type,
            "num_sources": args.num_sources,
            "inference": args.inference,
            "tracker": args.tracker,
            "sink": args.sink,
            "platform": args.platform,
            "extras": args.extras,
        },
        "num_retrieved": len(results),
        "retrieved_pipelines": results,
    }

    fmt = getattr(args, "output_format", "json")
    if fmt == "summary":
        top = results[0]["prompt"] if results else "(no match)"
        if len(top) > 80:
            top = top[:77] + "..."
        print(
            "confidence={conf} retrieved={n} top_match={top!r}".format(
                conf=confidence, n=len(results), top=top,
            )
        )
    elif fmt == "compact":
        compact = {
            "query": args.query,
            "confidence": confidence,
            "context": output["context"],
            "num_retrieved": len(results),
            "top_pipeline": results[0] if results else None,
        }
        print(json.dumps(compact, indent=2))
    else:
        print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Data quality linter for the DeepStream pipeline dataset.

Checks for:
  1. Duplicate or near-duplicate rows
  2. Invalid pipelines (dual sinks, missing source/sink)
  3. Syntax issues (space before '=' in properties)
  4. batch-size / sink-pad count mismatches

Run with --fix to auto-fix known issues and write a cleaned CSV.

Usage:
    python3 lint_data.py                    # report only
    python3 lint_data.py --fix              # fix and overwrite
    python3 lint_data.py --fix --out clean.csv  # fix and write to new file
"""

import argparse
import csv
import os
import re
import sys
from collections import defaultdict

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_CSV = os.path.join(SCRIPT_DIR, "..", "data", "data.csv")

KNOWN_SINKS = {
    "nveglglessink", "nv3dsink", "filesink", "fakesink",
    "udpsink", "xvimagesink", "autovideosink",
}
KNOWN_SOURCES = {
    "filesrc", "uridecodebin", "v4l2src", "videotestsrc",
    "rtspsrc", "multifilesrc", "nvvideotestsrc", "nvurisrcbin",
}


def normalize_pipeline(pipeline):
    """Normalize whitespace and line continuations for comparison."""
    p = re.sub(r"\\\s*\n\s*", " ", pipeline)
    p = re.sub(r"\s+", " ", p).strip()
    return p


def find_issues(rows):
    """Return list of (row_index_0based, issue_type, message) tuples."""
    issues = []
    seen_pipelines = {}

    for i, (prompt, pipeline) in enumerate(rows):
        norm = normalize_pipeline(pipeline)

        # --- Exact duplicate pipelines ---
        if norm in seen_pipelines:
            prev = seen_pipelines[norm]
            issues.append((i, "duplicate", f"Exact duplicate pipeline of row {prev + 1}"))
        else:
            seen_pipelines[norm] = i

        # --- Dual sinks piped together ---
        # Skip segments that are pad references (e.g. "m.sink_0", "t.src_1")
        # or that *start* with a pad ref (e.g. "t. ! nv3dsink"). These
        # appear in legitimate tee fan-out / nvstreamdemux patterns where
        # a sink immediately following a pad-ref boundary is not a real
        # consecutive-sinks bug.
        def _is_pad_ref_segment(seg):
            tokens = seg.split()
            if not tokens:
                return False
            head = tokens[0]
            return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_-]*\.[a-zA-Z0-9_-]+$", head))

        sinks_in_pipe = []
        segments = re.split(r"\s+!\s+", norm)
        for seg in segments:
            if _is_pad_ref_segment(seg):
                continue
            elem = seg.split()[0] if seg.split() else ""
            if elem in KNOWN_SINKS:
                sinks_in_pipe.append(elem)
        if len(sinks_in_pipe) > 1:
            chain = " ! ".join(sinks_in_pipe)
            consecutive = False
            for j in range(len(segments) - 1):
                if _is_pad_ref_segment(segments[j]) or _is_pad_ref_segment(segments[j + 1]):
                    continue
                e1 = segments[j].split()[0] if segments[j].split() else ""
                e2 = segments[j + 1].split()[0] if segments[j + 1].split() else ""
                if e1 in KNOWN_SINKS and e2 in KNOWN_SINKS:
                    consecutive = True
            if consecutive:
                issues.append((i, "dual_sink", f"Consecutive sinks piped together: {chain}"))

        # --- Space before '=' in properties ---
        space_eq_matches = re.findall(r'\b(\w+)\s+=\s*', pipeline)
        if space_eq_matches:
            for prop in space_eq_matches:
                if prop in ("location", "uri", "device"):
                    issues.append((i, "space_equals",
                                   f"Space before '=' in '{prop} =' — should be '{prop}='"))

    return issues


def fix_row(prompt, pipeline, issues_for_row):
    """Apply automatic fixes to a pipeline. Returns (prompt, fixed_pipeline) or None to drop."""
    fixed = pipeline

    for _, issue_type, msg in issues_for_row:
        if issue_type == "duplicate":
            return None

        if issue_type == "space_equals":
            fixed = re.sub(r'\b(location|uri|device)\s+=\s*', r'\1=', fixed)

        if issue_type == "dual_sink":
            fixed = re.sub(r'\bnv3dsink\s*!\s*nveglglessink\b', 'nv3dsink', fixed)
            fixed = re.sub(r'\bnveglglessink\s*!\s*nv3dsink\b', 'nveglglessink', fixed)

    return (prompt, fixed)


def main():
    parser = argparse.ArgumentParser(description="Lint the DeepStream pipeline dataset")
    parser.add_argument("--csv", default=DEFAULT_CSV, help="Path to data CSV")
    parser.add_argument("--fix", action="store_true", help="Auto-fix issues and write output")
    parser.add_argument("--out", default=None, help="Output CSV path (default: overwrite input)")
    args = parser.parse_args()

    with open(args.csv, newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        header = next(reader)
        rows = [(row[0].strip(), row[1].strip()) for row in reader if len(row) >= 2]

    issues = find_issues(rows)

    if not issues:
        print("No issues found.")
        return

    issues_by_row = defaultdict(list)
    for idx, itype, msg in issues:
        issues_by_row[idx].append((idx, itype, msg))

    print(f"Found {len(issues)} issue(s) in {len(issues_by_row)} row(s):\n")
    for idx, itype, msg in issues:
        print(f"  Row {idx + 1} [{itype}]: {msg}")

    if not args.fix:
        print(f"\nRun with --fix to auto-fix.")
        sys.exit(1 if issues else 0)

    out_path = args.out or args.csv
    fixed_rows = []
    dropped = 0
    fixed_count = 0

    for i, (prompt, pipeline) in enumerate(rows):
        if i in issues_by_row:
            result = fix_row(prompt, pipeline, issues_by_row[i])
            if result is None:
                dropped += 1
                continue
            fixed_rows.append(result)
            if result[1] != pipeline:
                fixed_count += 1
        else:
            fixed_rows.append((prompt, pipeline))

    with open(out_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(header)
        for prompt, pipeline in fixed_rows:
            writer.writerow([prompt, pipeline])

    print(f"\nFixed {fixed_count} row(s), dropped {dropped} duplicate(s).")
    print(f"Wrote {len(fixed_rows)} rows to {out_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
DeepStream Pipeline Validator — Syntax, element, property, and structure checks.

Validates a gst-launch-1.0 pipeline string WITHOUT running it:
1. Syntax check — quotes, empty segments, leading/trailing pipes
2. Element check — verifies each element exists via gst-inspect-1.0
3. Property check — validates known properties for DeepStream elements
4. Structure check — source/sink presence, required props, named-pad refs
5. Live parse — optional dry-run via gst-launch-1.0 with fakesrc/fakesink

Zero external dependencies — uses only Python stdlib + system gst-inspect-1.0.

Usage:
    python3 validate_pipeline.py "gst-launch-1.0 filesrc location=test.mp4 ! ..."
    python3 validate_pipeline.py --pipeline "filesrc location=test.mp4 ! ..."
"""

import argparse
import json
import re
import shlex
import subprocess
import sys
from shutil import which

_SAFE_ELEMENT_NAME = re.compile(r'^[a-zA-Z0-9_-]+$')

# ---------------------------------------------------------------------------
# Known DeepStream/GStreamer element properties
# ---------------------------------------------------------------------------
KNOWN_ELEMENT_PROPERTIES = {
    "filesrc": {
        "location", "num-buffers", "typefind", "blocksize", "do-timestamp",
    },
    "nvinfer": {
        "config-file-path", "batch-size", "unique-id", "infer-on-gie-id",
        "input-tensor-meta", "interval", "model-engine-file", "gpu-id",
        "process-mode", "network-type", "cluster-mode", "operate-on-gie-id",
        "operate-on-class-ids", "filter-out-class-ids", "output-tensor-meta",
        "clip-object-outside-roi", "infer-on-class-ids", "raw-output-file-write",
    },
    "nvstreammux": {
        "name", "batch-size", "width", "height", "batched-push-timeout",
        "live-source", "enable-padding", "num-surfaces-per-frame",
        "gpu-id", "nvbuf-memory-type", "buffer-pool-size",
        "compute-hw", "interpolation-method", "adaptive-batching",
        "sync-inputs",
    },
    "nvvideoconvert": {
        "flip-method", "src-crop", "dest-crop", "gpu-id",
        "nvbuf-memory-type", "interpolation-method", "compute-hw",
    },
    "nvtracker": {
        "ll-lib-file", "ll-config-file", "gpu-id", "tracker-width",
        "tracker-height", "display-tracking-id", "compute-hw",
    },
    "nvdsosd": {
        "process-mode", "display-text", "display-bbox", "display-mask",
        "display-clock", "clock-font", "clock-font-size", "gpu-id",
    },
    "nvmultistreamtiler": {
        "rows", "columns", "width", "height", "gpu-id",
        "nvbuf-memory-type", "compute-hw", "show-source",
    },
    "nvdspreprocess": {
        "config-file", "gpu-id", "operate-on-gie-id",
    },
    "nvv4l2decoder": {
        "gpu-id", "num-extra-surfaces", "cudadec-memtype",
        "drop-frame-interval", "disable-dpb",
    },
    "nvv4l2h264enc": {
        "bitrate", "maxperf-enable", "preset-level", "profile",
        "control-rate", "gpu-id", "iframeinterval",
        "qp-range", "constqp", "initqp",
    },
    "nvv4l2h265enc": {
        "bitrate", "maxperf-enable", "preset-level", "profile",
        "control-rate", "gpu-id", "iframeinterval",
        "qp-range", "constqp", "initqp",
    },
    "nveglglessink": {"sync", "qos", "max-lateness", "window-x", "window-y"},
    "nv3dsink": {"sync", "qos"},
    "fakesink": {"sync", "async", "dump", "signal-handoffs", "num-buffers"},
    "filesink": {"location", "sync", "async", "append", "buffer-mode", "buffer-size"},
    "jpegenc": {"quality", "idct-method", "snapshot"},
    "jpegparse": set(),
    "nvjpegdec": {"gpu-id"},
    "pngenc": {"compression-level", "snapshot"},
    "pngdec": set(),
    "h264parse": {"config-interval", "disable-passthrough"},
    "h265parse": {"config-interval", "disable-passthrough"},
    "qtdemux": {"name"},
    "qtmux": {"fragment-duration", "faststart"},
    "rtph264pay": {"config-interval", "pt", "mtu"},
    "rtph265pay": {"config-interval", "pt", "mtu"},
    "udpsink": {"host", "port", "sync", "async"},
    "uridecodebin": {"uri", "caps", "buffer-size", "use-buffering"},
    "v4l2src": {"device", "num-buffers", "io-mode"},
    "videotestsrc": {"pattern", "num-buffers", "is-live"},
    "videoconvert": set(),
    "videoparse": {"width", "height", "format", "framerate"},
    "xvimagesink": {"sync"},
    "decodebin": {"name"},
    "dsexample": {
        "full-frame", "processing-width", "processing-height",
        "gpu-id", "blur-objects", "unique-id",
    },
    "nvdewarper": {
        "config-file", "source-id", "nvbuf-memory-type", "gpu-id",
        "num-output-buffers",
    },
    "nvsegvisual": {"width", "height", "gpu-id"},
    "nvdspostprocess": {"postprocesslib-config-file", "postprocesslib-name", "gpu-id"},
    "queue": {"max-size-buffers", "max-size-bytes", "max-size-time", "leaky"},
    "tee": {"name"},
    "nvstreamdemux": {"name"},
    "nvvideotestsrc": {
        "num-buffers", "is-live", "location", "file-loop",
        "max-jitter", "fixed-jitter",
    },
    "nvurisrcbin": {"uri"},
    "nvegltransform": set(),
    "nvdslogger": {"fps-measurement-interval-sec"},
    "fpsdisplaysink": {"video-sink", "sync", "text-overlay"},
    "identity": {"silent", "dump", "single-segment"},
    "rtspsrc": {"location", "latency", "protocols", "user-id", "user-pw"},
    "rtph264depay": set(),
    "avdec_h264": set(),
    "videoscale": set(),
    "audioconvert": set(),
    "audioresample": set(),
    "aacparse": set(),
    "avdec_aac": set(),
    "nvdsaudiotemplate": {"customlib-name", "customlib-props"},
    "tcpserversink": {"host", "port"},
    "tcpclientsrc": {"host", "port"},
    "rtpgstpay": {"config-interval"},
    "rtpstreampay": set(),
    "rtpstreamdepay": set(),
    "rtpgstdepay": set(),
    "nvinferserver": {
        "config-file-path", "unique-id", "infer-on-gie-id", "batch-size",
        "interval", "gpu-id", "process-mode", "input-tensor-meta",
        "operate-on-gie-id", "operate-on-class-ids", "infer-on-class-ids",
        "output-tensor-meta",
    },
    "nvdsanalytics": {"config-file", "unique-id"},
    "nvmsgconv": {
        "config", "payload-type", "msg2p-lib", "comp-id", "debug-payload-dir",
        "msg2p-newapi",
    },
    "nvmsgbroker": {
        "proto-lib", "conn-str", "topic", "config", "comp-id", "sync",
        "new-api",
    },
    "nvdsmetamux": set(),
}

PASSTHROUGH_ELEMENTS = {
    "capsfilter", "identity", "multiqueue",
    "input-selector", "output-selector",
}

# Elements that output video/x-raw in system (CPU) memory — NOT NVMM.
# Feeding these directly into NVMM-requiring elements causes linking errors.
SYSTEM_MEMORY_OUTPUT = {"nvjpegdec", "pngdec", "jpegdec", "videoconvert", "videoscale"}

# Elements that require video/x-raw(memory:NVMM) on their sink pads.
NVMM_INPUT_REQUIRED = {"nvstreammux", "nvv4l2h264enc", "nvv4l2h265enc"}

# Platform-specific sink elements — using the wrong one causes runtime failures.
DGPU_ONLY_SINKS = {"nveglglessink", "nvegltransform"}
JETSON_ONLY_SINKS = {"nv3dsink"}

# Element ordering: elements that should appear BEFORE others in the pipeline.
# Key = element that must come first, Value = set of elements it must precede.
EXPECTED_ORDER = {
    "filesrc": {"nvv4l2decoder", "decodebin", "jpegparse", "h264parse", "h265parse", "qtdemux"},
    "nvv4l2decoder": {"nvstreammux", "nvinfer", "nvinferserver", "nvvideoconvert"},
    "nvstreammux": {"nvinfer", "nvinferserver", "nvtracker", "nvdsosd", "nvmultistreamtiler"},
    "nvinfer": {"nvtracker", "nvdsosd", "nveglglessink", "nv3dsink", "filesink", "fakesink"},
    "nvinferserver": {"nvtracker", "nvdsosd", "nveglglessink", "nv3dsink", "filesink", "fakesink"},
    "nvtracker": {"nvdsosd", "nveglglessink", "nv3dsink", "filesink"},
}


# ---------------------------------------------------------------------------
# Element existence check via gst-inspect-1.0
# ---------------------------------------------------------------------------
def check_element_exists(element_name):
    """Check if a GStreamer element is registered using gst-inspect-1.0."""
    if not _SAFE_ELEMENT_NAME.match(element_name):
        return None
    try:
        result = subprocess.run(
            ["gst-inspect-1.0", element_name],
            capture_output=True, text=True, timeout=5
        )
        return result.returncode == 0
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return None


# ---------------------------------------------------------------------------
# Pipeline parser
# ---------------------------------------------------------------------------
def extract_elements_and_properties(pipeline_str):
    """
    Parse a gst-launch-1.0 pipeline string and extract elements with properties.
    Returns list of dicts: [{"name": str, "properties": dict, "raw": str}]
    """
    pipeline_str = re.sub(r"^gst-launch-1\.0\s+", "", pipeline_str.strip())
    pipeline_str = re.sub(r"^(?:-[a-zA-Z]+\s+)+", "", pipeline_str)
    pipeline_str = re.sub(r"\\\s*\n\s*", " ", pipeline_str)

    segments = re.split(r"\s+!\s+", pipeline_str)

    elements = []
    pad_refs = []

    for segment in segments:
        segment = segment.strip()
        if not segment:
            continue

        if segment.startswith(("'", '"', "video/", "audio/", "image/", "application/")):
            continue

        parts = segment.split()
        if not parts:
            continue

        idx = 0
        # A segment may start with pad refs (e.g. "m.sink_0 nvstreammux name=m ...")
        while idx < len(parts):
            tok = parts[idx]
            if "." in tok and not tok.startswith("/") and "=" not in tok:
                pad_refs.append(tok)
                idx += 1
            else:
                break

        if idx >= len(parts):
            continue

        element_name = parts[idx]

        if element_name.startswith(("'", '"', "video/", "audio/", "image/")):
            continue

        properties = {}
        for part in parts[idx + 1:]:
            if part.startswith(("'", '"')):
                break
            if "=" in part:
                key, _, val = part.partition("=")
                properties[key] = val

        elements.append({
            "name": element_name,
            "properties": properties,
            "raw": segment,
        })

    return elements, pad_refs


# ---------------------------------------------------------------------------
# Validation checks
# ---------------------------------------------------------------------------
def validate_syntax(pipeline_str):
    """Check for common syntax errors."""
    errors = []
    warnings = []

    single_quotes = pipeline_str.count("'")
    double_quotes = pipeline_str.count('"')
    if single_quotes % 2 != 0:
        errors.append("Unbalanced single quotes in pipeline")
    if double_quotes % 2 != 0:
        errors.append("Unbalanced double quotes in pipeline")

    if re.search(r"!\s*!", pipeline_str):
        errors.append("Empty pipe segment detected (consecutive '!' without element)")

    cleaned = re.sub(r"^gst-launch-1\.0\s+", "", pipeline_str.strip())
    cleaned = re.sub(r"^(?:-[a-zA-Z]+\s+)+", "", cleaned)
    cleaned = re.sub(r"\\\s*\n\s*", " ", cleaned).strip()
    if cleaned.startswith("!"):
        errors.append("Pipeline starts with '!' — missing source element")
    if cleaned.endswith("!"):
        errors.append("Pipeline ends with '!' — missing sink element")

    # Underscore-typo check: scope to the segment that owns nvinfer/nvinferserver
    # so a hyphenated config-file-path on a different element (or in a comment)
    # does not mask an underscore typo on the actual nvinfer.
    if re.search(r"\bnvinfer(?:server)?\b[^!]*\bconfig_file_path=", pipeline_str):
        errors.append("nvinfer uses 'config-file-path' (hyphens), not 'config_file_path' (underscores)")

    # Note: this only inspects the segment immediately following the
    # 'nvstreammux' token (non-greedy match up to the next '!' or end of
    # string). Properties placed after a '!' separator — rare, but possible
    # via pipeline-continuation patterns — will not be considered. Adequate
    # for the single-line pipelines this skill produces today.
    mux_match = re.search(r"nvstreammux\b(.*?)(?:!|$)", pipeline_str)
    if mux_match and "name=" not in mux_match.group(1):
        warnings.append("nvstreammux without 'name=' — multi-source linking may fail")

    return errors, warnings


def validate_elements(pipeline_str):
    """Validate all elements exist and properties are recognized."""
    elements, pad_refs = extract_elements_and_properties(pipeline_str)
    errors = []
    warnings = []
    checked_elements = set()

    has_gst_inspect = which("gst-inspect-1.0") is not None

    for elem in elements:
        name = elem["name"]

        if name not in checked_elements:
            checked_elements.add(name)
            if has_gst_inspect and name not in KNOWN_ELEMENT_PROPERTIES and name not in PASSTHROUGH_ELEMENTS:
                exists = check_element_exists(name)
                if exists is False:
                    errors.append(f"Unknown element '{name}' — not found by gst-inspect-1.0")
                    continue
                elif exists is None:
                    warnings.append(f"Could not verify element '{name}' — gst-inspect-1.0 timed out")

        if name in KNOWN_ELEMENT_PROPERTIES:
            known_props = KNOWN_ELEMENT_PROPERTIES[name]
            for prop in elem["properties"]:
                if known_props and prop not in known_props and prop != "name":
                    warnings.append(
                        f"Unrecognized property '{prop}' on element '{name}'"
                        f" — verify with gst-inspect-1.0 {name}"
                    )

    return elements, pad_refs, errors, warnings


def validate_memory_format(pipeline_str, elements, pad_refs):
    """
    Detect memory format mismatches — e.g. nvjpegdec (system memory) feeding
    directly into nvstreammux (requires NVMM) without nvvideoconvert in between.

    Handles two patterns:
      1. Consecutive elements:  ... ! nvjpegdec ! nvstreammux ...
      2. Pad-ref branches:      ... ! nvjpegdec ! mux.sink_N  (branch end)
    """
    errors = []
    flagged_pairs = set()

    # Build mux alias set from named nvstreammux elements
    mux_aliases = set()
    for e in elements:
        if e["name"] == "nvstreammux" and "name" in e["properties"]:
            mux_aliases.add(e["properties"]["name"])

    # --- Check 1: consecutive elements in the parsed list ---
    for i in range(len(elements) - 1):
        curr = elements[i]["name"]
        nxt = elements[i + 1]["name"]
        if curr in SYSTEM_MEMORY_OUTPUT and nxt in NVMM_INPUT_REQUIRED:
            pair = (curr, nxt)
            if pair not in flagged_pairs:
                flagged_pairs.add(pair)
                errors.append(
                    f"'{curr}' outputs system memory (video/x-raw) but "
                    f"'{nxt}' requires NVMM — "
                    f"insert 'nvvideoconvert' between them"
                )

    if not mux_aliases:
        return errors

    # --- Check 2: segment before a mux pad ref (catches multi-branch) ---
    clean = re.sub(r"^gst-launch-1\.0\s+", "", pipeline_str.strip())
    clean = re.sub(r"^(?:-[a-zA-Z]+\s+)+", "", clean)
    clean = re.sub(r"\\\s*\n\s*", " ", clean)
    segments = re.split(r"\s+!\s+", clean)

    for i, seg in enumerate(segments):
        seg_stripped = seg.strip()
        has_mux_pad = any(
            seg_stripped.startswith(f"{a}.sink_") or f" {a}.sink_" in seg_stripped
            for a in mux_aliases
        )
        if not has_mux_pad or i == 0:
            continue
        prev_seg = segments[i - 1].strip()
        prev_parts = prev_seg.split()
        if not prev_parts:
            continue
        prev_elem = prev_parts[0]
        if prev_elem in SYSTEM_MEMORY_OUTPUT:
            pair = (prev_elem, "nvstreammux")
            if pair not in flagged_pairs:
                flagged_pairs.add(pair)
                errors.append(
                    f"'{prev_elem}' outputs system memory (video/x-raw) but "
                    f"feeds into 'nvstreammux' via pad ref — "
                    f"insert 'nvvideoconvert' between them"
                )

    return errors


def validate_pipeline_structure(elements, pad_refs):
    """Check source/sink presence, required props, and named-pad consistency."""
    warnings = []

    if not elements:
        return ["Pipeline is empty — no elements found"], warnings

    errors = []
    element_names = [e["name"] for e in elements]
    named_elements = {}
    for e in elements:
        if "name" in e["properties"]:
            named_elements[e["properties"]["name"]] = e["name"]

    source_elements = {
        "filesrc", "uridecodebin", "v4l2src", "videotestsrc", "rtspsrc",
        "multifilesrc", "nvvideotestsrc", "nvurisrcbin", "tcpclientsrc",
    }
    has_source = any(e in source_elements for e in element_names)
    if not has_source:
        warnings.append(
            "No recognized source element (filesrc, uridecodebin, v4l2src, etc.)"
            " — may be intentional if using a named pad"
        )

    sink_elements = {
        "nveglglessink", "nv3dsink", "filesink", "fakesink", "udpsink",
        "xvimagesink", "autovideosink", "fpsdisplaysink", "tcpserversink",
    }
    has_sink = any(e in sink_elements for e in element_names)
    if not has_sink:
        warnings.append("No recognized sink element — pipeline needs a sink to terminate")

    for elem in elements:
        if elem["name"] == "nvinfer" and "config-file-path" not in elem["properties"]:
            errors.append("nvinfer missing required 'config-file-path' property")
        if elem["name"] == "nvinferserver" and "config-file-path" not in elem["properties"]:
            errors.append("nvinferserver missing required 'config-file-path' property")
        if elem["name"] == "filesrc" and "location" not in elem["properties"]:
            errors.append("filesrc missing required 'location' property")

    # Named-pad cross-reference validation
    for ref in pad_refs:
        alias = ref.split(".")[0]
        if alias not in named_elements:
            if not any(alias == e["properties"].get("name", "") for e in elements):
                warnings.append(
                    f"Pad reference '{ref}' uses alias '{alias}'"
                    f" but no element has name='{alias}'"
                )

    # batch-size / sink-pad count consistency
    for elem in elements:
        if elem["name"] == "nvstreammux" and "batch-size" in elem["properties"]:
            try:
                batch = int(elem["properties"]["batch-size"])
            except ValueError:
                continue
            alias = elem["properties"].get("name", "")
            if alias:
                pad_count = sum(
                    1 for r in pad_refs
                    if r.startswith(f"{alias}.sink_")
                )
                if pad_count > 0 and pad_count != batch and batch > 1:
                    warnings.append(
                        f"nvstreammux '{alias}' has batch-size={batch}"
                        f" but {pad_count} sink pad(s) connected"
                    )

    return errors, warnings


def validate_platform_sink(elements):
    """
    Detect platform-sink mismatches:
    - nveglglessink used alongside nv3dsink hints (Jetson pipeline with dGPU sink)
    - Both Jetson-only and dGPU-only sinks in the same pipeline
    """
    warnings = []
    element_names = {e["name"] for e in elements}

    has_dgpu_sink = bool(element_names & DGPU_ONLY_SINKS)
    has_jetson_sink = bool(element_names & JETSON_ONLY_SINKS)

    if has_dgpu_sink and has_jetson_sink:
        warnings.append(
            "Pipeline mixes dGPU-only sinks (nveglglessink) and Jetson-only "
            "sinks (nv3dsink) — pick one for your target platform"
        )

    return warnings


def validate_element_ordering(elements):
    """
    Detect nonsensical element ordering — e.g. an encoder appearing before a
    decoder, or a sink element appearing before inference elements.
    Only flags clear-cut ordering violations for elements in EXPECTED_ORDER.
    """
    warnings = []
    # Build position map: element_name -> first occurrence index
    first_pos = {}
    for i, e in enumerate(elements):
        if e["name"] not in first_pos:
            first_pos[e["name"]] = i

    for before_elem, after_set in EXPECTED_ORDER.items():
        if before_elem not in first_pos:
            continue
        before_idx = first_pos[before_elem]
        for after_elem in after_set:
            if after_elem in first_pos and first_pos[after_elem] < before_idx:
                warnings.append(
                    f"Element '{after_elem}' appears before '{before_elem}' "
                    f"— expected '{before_elem}' first"
                )

    return warnings


def validate_with_gst_launch(pipeline_str, pad_refs=None):
    """
    Use gst-launch-1.0 for a dry-run parse check by substituting
    fakesrc/fakesink and stripping elements that need real data.

    Multi-stream pipelines (those with named pad refs like m.sink_0) are
    skipped — fakesrc cannot negotiate caps through named pads on
    nvstreammux, causing false-positive linking errors.
    """
    errors = []
    warnings = []

    if pad_refs and len(pad_refs) > 0:
        return errors, warnings

    if not which("gst-launch-1.0"):
        warnings.append("gst-launch-1.0 not found — skipping live pipeline parse check")
        return errors, warnings

    test_pipeline = re.sub(r"^gst-launch-1\.0\s+", "", pipeline_str.strip())
    test_pipeline = re.sub(r"^(?:-[a-zA-Z]+\s+)+", "", test_pipeline)
    test_pipeline = re.sub(r"\\\s*\n\s*", " ", test_pipeline)

    source_replacements = {
        "filesrc": "fakesrc",
        "v4l2src": "fakesrc",
        "videotestsrc": "fakesrc",
        "rtspsrc": "fakesrc",
        "uridecodebin": "fakesrc",
        "multifilesrc": "fakesrc",
        "nvvideotestsrc": "fakesrc",
        "nvurisrcbin": "fakesrc",
        "tcpclientsrc": "fakesrc",
    }
    sink_replacements = {
        "nveglglessink": "fakesink",
        "nv3dsink": "fakesink",
        "filesink": "fakesink",
        "udpsink": "fakesink",
        "xvimagesink": "fakesink",
        "autovideosink": "fakesink",
        "fpsdisplaysink": "fakesink",
        "tcpserversink": "fakesink",
    }

    test_str = test_pipeline

    # Anchor replacements to segment-leading positions ((^|!\s+)) so that a
    # property value containing the element name (e.g. config-file-path=
    # /path/with/filesrc.txt) is not mistakenly rewritten — that would corrupt
    # the dry-run input and produce false-positive parse errors.
    for real, fake in source_replacements.items():
        test_str = re.sub(rf"(^|!\s+){real}\b[^!]*", rf"\g<1>{fake} num-buffers=1 ", test_str)

    for real, fake in sink_replacements.items():
        test_str = re.sub(rf"(^|!\s+){real}\b[^!]*", rf"\g<1>{fake} ", test_str)

    test_str = re.sub(
        r"\b(config-file-path|config-file|ll-config-file|ll-lib-file"
        r"|postprocesslib-config-file|postprocesslib-name)=[^\s!]+",
        "", test_str,
    )

    strip_elements = [
        "qtdemux", "h264parse", "h265parse", "nvv4l2decoder", "nvjpegdec",
        "jpegparse", "decodebin", "pngdec", "jpegdec", "videoparse",
        "qtmux", "matroskamux", "mp4mux", "flvmux", "mpegtsmux",
        "rtph264pay", "rtph265pay", "rtph264depay", "avdec_h264",
        "rtpgstpay", "rtpstreampay", "rtpstreamdepay", "rtpgstdepay",
        "aacparse", "avdec_aac", "audioconvert", "audioresample",
    ]
    for elem in strip_elements:
        test_str = re.sub(rf"\b{elem}\b[^!]*!\s*", "", test_str)
        test_str = re.sub(rf"\b{elem}\b[^!]*$", "", test_str)

    test_str = re.sub(r"\s+", " ", test_str).strip()
    test_str = re.sub(r"!\s*!", "!", test_str)
    test_str = re.sub(r"^\s*!\s*", "", test_str)
    test_str = re.sub(r"\s*!\s*$", "", test_str)

    if not test_str.strip():
        warnings.append("Pipeline too simple to perform live parse validation")
        return errors, warnings

    try:
        try:
            gst_args = shlex.split(test_str)
        except ValueError:
            warnings.append("Live parse check skipped — pipeline string could not be tokenized safely")
            return errors, warnings

        result = subprocess.run(
            ["gst-launch-1.0", "--gst-debug-level=0"] + gst_args,
            capture_output=True, text=True, timeout=5,
        )

        stderr = result.stderr.strip()
        if "erroneous pipeline" in stderr.lower():
            for line in stderr.split("\n"):
                if "erroneous pipeline" in line.lower():
                    line = line.strip()[:300]
                    errors.append(f"GStreamer parse error: {line}")
        elif result.returncode != 0 and "error" in stderr.lower():
            noise_patterns = [
                "no such file", "configuration file not provided",
                "doesn't want to preroll", "not negotiated",
                "internal data stream", "gstnvinfer",
                "no element", "could not set property",
            ]
            for line in stderr.split("\n"):
                line = line.strip()
                if not line or "error" not in line.lower():
                    continue
                line_lower = line.lower()
                if any(p in line_lower for p in noise_patterns):
                    continue
                warnings.append(f"GStreamer runtime warning: {line[:300]}")

    except subprocess.TimeoutExpired:
        warnings.append("Live parse check timed out — pipeline may hang or require hardware not present")
    except FileNotFoundError:
        warnings.append("gst-launch-1.0 not found — skipping live parse check")

    return errors, warnings


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(description="Validate a GStreamer/DeepStream pipeline")
    parser.add_argument("pipeline", nargs="?", help="Pipeline string to validate")
    parser.add_argument("--pipeline", dest="pipeline_flag", help="Pipeline string (alternative flag)")
    parser.add_argument(
        "--format", dest="output_format",
        choices=("summary", "json"), default="json",
        help=(
            "Output format. 'json' = full verbose output (default — preserves "
            "existing behavior). 'summary' = one-line human status, with errors "
            "or warnings printed below if present."
        ),
    )

    args = parser.parse_args()
    pipeline_str = args.pipeline or args.pipeline_flag
    fmt = args.output_format

    if not pipeline_str:
        print("ERROR: No pipeline provided", file=sys.stderr)
        print('Usage: python3 validate_pipeline.py "gst-launch-1.0 ..."', file=sys.stderr)
        sys.exit(1)

    if len(pipeline_str) > 16384:
        if fmt == "summary":
            print("invalid: Pipeline string exceeds maximum allowed length (16384 chars)")
        else:
            result = {"valid": False, "errors": ["Pipeline string exceeds maximum allowed length (16384 chars)"], "warnings": []}
            print(json.dumps(result, indent=2))
        sys.exit(1)

    all_errors = []
    all_warnings = []

    syntax_errors, syntax_warnings = validate_syntax(pipeline_str)
    all_errors.extend(syntax_errors)
    all_warnings.extend(syntax_warnings)

    elements, pad_refs, elem_errors, elem_warnings = validate_elements(pipeline_str)
    all_errors.extend(elem_errors)
    all_warnings.extend(elem_warnings)

    mem_errors = validate_memory_format(pipeline_str, elements, pad_refs)
    all_errors.extend(mem_errors)

    struct_errors, struct_warnings = validate_pipeline_structure(elements, pad_refs)
    all_errors.extend(struct_errors)
    all_warnings.extend(struct_warnings)

    platform_warnings = validate_platform_sink(elements)
    all_warnings.extend(platform_warnings)

    ordering_warnings = validate_element_ordering(elements)
    all_warnings.extend(ordering_warnings)

    gst_errors, gst_warnings = validate_with_gst_launch(pipeline_str, pad_refs)
    all_errors.extend(gst_errors)
    all_warnings.extend(gst_warnings)

    result = {
        "valid": len(all_errors) == 0,
        "elements_found": [e["name"] for e in elements],
        "num_elements": len(elements),
        "pad_refs": pad_refs,
        "errors": all_errors,
        "warnings": all_warnings,
    }

    if fmt == "summary":
        live_parse_skipped = bool(pad_refs)
        if result["valid"]:
            note = ""
            if live_parse_skipped:
                note = " · live-parse skipped (multi-stream)"
            print(
                "valid · {n} elements · {w} warnings{note}".format(
                    n=len(elements), w=len(all_warnings), note=note,
                )
            )
        else:
            print(
                "invalid · {n} elements · {e} errors · {w} warnings".format(
                    n=len(elements), e=len(all_errors), w=len(all_warnings),
                )
            )
        for err in all_errors:
            print("  error: {}".format(err))
        for warn in all_warnings:
            print("  warning: {}".format(warn))
    else:
        print(json.dumps(result, indent=2))

    sys.exit(0 if result["valid"] else 1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
End-to-end CLI tests for the --format flag on generate_pipeline.py and
validate_pipeline.py. Run the scripts as subprocesses so the full surface
(argparse, formatting branches, exit codes) is exercised.
"""

import json
import os
import subprocess
import sys
import unittest

SCRIPTS_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "scripts"))
GENERATE = os.path.join(SCRIPTS_DIR, "generate_pipeline.py")
VALIDATE = os.path.join(SCRIPTS_DIR, "validate_pipeline.py")

GENERATE_BASE_ARGS = [
    "--query", "infer on 3 videos and display",
    "--source-type", "Local video file",
    "--num-sources", "3",
    "--inference", "primary",
    "--tracker", "none",
    "--sink", "display",
    "--platform", "dGPU",
    "--extras", "none",
]

VALID_PIPELINE = (
    "gst-launch-1.0 videotestsrc num-buffers=1 ! "
    "video/x-raw,format=NV12,width=320,height=240 ! fakesink"
)
INVALID_PIPELINE = "gst-launch-1.0 filesrc ! ! fakesink"


def run_generate(extra_args=None):
    args = [sys.executable, GENERATE] + GENERATE_BASE_ARGS + (extra_args or [])
    return subprocess.run(args, capture_output=True, text=True, timeout=30)


def run_validate(pipeline, extra_args=None):
    args = [sys.executable, VALIDATE, pipeline] + (extra_args or [])
    return subprocess.run(args, capture_output=True, text=True, timeout=30)


class TestGenerateFormat(unittest.TestCase):
    """generate_pipeline.py --format {json,compact,summary}"""

    def test_default_is_full_json(self):
        """Default (no --format) must remain full JSON for backward compat."""
        result = run_generate()
        self.assertEqual(result.returncode, 0)
        payload = json.loads(result.stdout)
        # Full JSON includes the retrieved_pipelines list (~10 entries)
        self.assertIn("retrieved_pipelines", payload)
        self.assertIsInstance(payload["retrieved_pipelines"], list)
        self.assertIn("confidence", payload)
        self.assertIn("context", payload)

    def test_format_json_matches_default(self):
        """--format=json explicitly should match the default behavior."""
        default = run_generate()
        explicit = run_generate(["--format", "json"])
        self.assertEqual(default.stdout, explicit.stdout)

    def test_format_compact_is_smaller_json(self):
        """--format=compact returns JSON with top_pipeline only, not retrieved_pipelines."""
        result = run_generate(["--format", "compact"])
        self.assertEqual(result.returncode, 0)
        payload = json.loads(result.stdout)
        self.assertIn("top_pipeline", payload)
        self.assertNotIn("retrieved_pipelines", payload)
        self.assertIn("confidence", payload)
        self.assertIn("context", payload)
        # top_pipeline is either a result dict or null
        if payload["top_pipeline"] is not None:
            self.assertIn("pipeline", payload["top_pipeline"])
            self.assertIn("score", payload["top_pipeline"])

    def test_format_summary_is_single_line(self):
        """--format=summary prints one line: confidence=… retrieved=… top_match=…"""
        result = run_generate(["--format", "summary"])
        self.assertEqual(result.returncode, 0)
        lines = result.stdout.strip().splitlines()
        self.assertEqual(len(lines), 1)
        self.assertIn("confidence=", lines[0])
        self.assertIn("retrieved=", lines[0])
        self.assertIn("top_match=", lines[0])

    def test_compact_smaller_than_json(self):
        """compact JSON output should be substantially smaller than full json."""
        full = run_generate(["--format", "json"])
        compact = run_generate(["--format", "compact"])
        self.assertLess(len(compact.stdout), len(full.stdout))
        # compact should be at most 1/3 the size for typical queries
        self.assertLess(len(compact.stdout) * 3, len(full.stdout) + 200)

    def test_invalid_format_rejected(self):
        """argparse should reject an unknown --format value."""
        result = run_generate(["--format", "bogus"])
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("invalid choice", result.stderr.lower())


class TestValidateFormat(unittest.TestCase):
    """validate_pipeline.py --format {json,summary}"""

    def test_default_is_full_json(self):
        """Default (no --format) must remain full JSON for backward compat."""
        result = run_validate(VALID_PIPELINE)
        # videotestsrc → fakesink may emit gst-launch dry-run warnings; valid still True
        payload = json.loads(result.stdout)
        self.assertIn("valid", payload)
        self.assertIn("elements_found", payload)
        self.assertIn("errors", payload)
        self.assertIn("warnings", payload)

    def test_format_json_matches_default(self):
        default = run_validate(VALID_PIPELINE)
        explicit = run_validate(VALID_PIPELINE, ["--format", "json"])
        self.assertEqual(default.stdout, explicit.stdout)

    def test_format_summary_valid_is_one_line(self):
        result = run_validate(VALID_PIPELINE, ["--format", "summary"])
        lines = result.stdout.strip().splitlines()
        # Summary header is exactly one line; warnings would be indented after.
        self.assertGreaterEqual(len(lines), 1)
        self.assertTrue(lines[0].startswith("valid · "))
        self.assertIn("elements", lines[0])
        self.assertIn("warnings", lines[0])

    def test_format_summary_invalid_lists_errors(self):
        result = run_validate(INVALID_PIPELINE, ["--format", "summary"])
        self.assertNotEqual(result.returncode, 0)
        out = result.stdout
        self.assertTrue(out.startswith("invalid · "))
        # Errors are indented under the header
        self.assertIn("\n  error: ", out)

    def test_format_summary_multistream_notes_skipped_live_parse(self):
        """Multi-stream pipelines should mention 'live-parse skipped' in summary."""
        multistream = (
            "gst-launch-1.0 -e "
            "filesrc location=/dev/null ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_0 "
            "filesrc location=/dev/null ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_1 "
            "nvstreammux name=m batch-size=2 width=1920 height=1080 ! "
            "nvinfer config-file-path=/dev/null batch-size=2 ! "
            "nvvideoconvert ! nvdsosd ! nveglglessink"
        )
        result = run_validate(multistream, ["--format", "summary"])
        # May be valid or invalid depending on env; either way the summary line
        # should mention live-parse skip when pad refs are present.
        self.assertIn("live-parse skipped", result.stdout)

    def test_invalid_format_rejected(self):
        result = run_validate(VALID_PIPELINE, ["--format", "bogus"])
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("invalid choice", result.stderr.lower())

    def test_dash_e_flag_accepted(self):
        """The Step 5 generated script uses 'gst-launch-1.0 -e' — validator must accept."""
        with_e = "gst-launch-1.0 -e videotestsrc num-buffers=1 ! fakesink"
        result = run_validate(with_e, ["--format", "summary"])
        # The -e flag must be stripped cleanly — the validator should report
        # success (returncode == 0). Surface stdout/stderr in the failure
        # message so a regression is debuggable from the test output alone.
        self.assertEqual(
            result.returncode, 0,
            msg=(
                f"validator rejected 'gst-launch-1.0 -e' pipeline "
                f"(returncode={result.returncode}).\n"
                f"stdout: {result.stdout!r}\nstderr: {result.stderr!r}"
            ),
        )
        # First line of summary must not be the 'invalid · …' header.
        self.assertNotIn("invalid · ", result.stdout.split("\n")[0])
        # The summary should report a non-zero element count.
        self.assertRegex(result.stdout, r"\d+ elements")


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Golden test set — regression tests for pipeline retrieval quality.

Each test case specifies a query + user parameters and asserts that the top
results contain expected elements/patterns (or don't contain unwanted ones).
Run this after any change to the retriever or data to ensure quality doesn't
regress.
"""

import os
import sys
import unittest

SCRIPT_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
sys.path.insert(0, SCRIPT_DIR)

from generate_pipeline import BM25Retriever

DATA_CSV = os.path.join(os.path.dirname(__file__), "..", "data", "data.csv")


def _meta(platform="unknown", source="unknown", sink="unknown",
          inference="unknown", tracker="none", num_sources=1):
    return {
        "platform": platform,
        "source_type": source,
        "sink_type": sink,
        "inference": inference,
        "tracker": tracker,
        "num_sources": num_sources,
    }


class GoldenTestBase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.retriever = BM25Retriever()
        cls.retriever.load_csv(DATA_CSV)

    def assert_top_n_contain(self, query, user_meta, expected_in, n=3,
                             not_expected=None):
        """Assert at least one of the top-N results contains all expected_in strings."""
        results, _ = self.retriever.retrieve(query, user_meta=user_meta)
        top = results[:n]
        self.assertGreater(len(top), 0, f"No results for query: {query}")

        for expected in expected_in:
            found = any(expected in r["pipeline"].lower() for r in top)
            self.assertTrue(found, f"Expected '{expected}' in top-{n} results for: {query}")

        if not_expected:
            for bad in not_expected:
                top_pipelines = " ".join(r["pipeline"].lower() for r in top[:1])
                self.assertNotIn(bad, top_pipelines,
                                 f"Did NOT expect '{bad}' in #1 result for: {query}")

    def assert_platform_match(self, query, user_meta, expected_platform, n=3):
        """Assert top-N results prefer the expected platform."""
        results, _ = self.retriever.retrieve(query, user_meta=user_meta)
        top = results[:n]
        platform_matches = sum(
            1 for r in top if r["metadata"]["platform"] == expected_platform
        )
        self.assertGreater(platform_matches, 0,
                           f"No {expected_platform} results in top-{n} for: {query}")


class TestFlipRotate(GoldenTestBase):
    def test_flip_90_ccw(self):
        self.assert_top_n_contain(
            "flip video 90 degrees counter-clockwise",
            _meta(platform="dgpu", source="video", sink="display"),
            ["flip-method=1"],
        )

    def test_rotate_180(self):
        self.assert_top_n_contain(
            "rotate incoming video by 180 degrees",
            _meta(platform="dgpu", source="video", sink="display"),
            ["flip-method=2"],
        )

    def test_flip_horizontal(self):
        self.assert_top_n_contain(
            "flip video horizontally",
            _meta(platform="dgpu", source="video", sink="display"),
            ["flip-method=4"],
        )


class TestColorConversion(GoldenTestBase):
    def test_bgr10a2_to_i420(self):
        self.assert_top_n_contain(
            "convert BGR10A2_LE to I420 and display",
            _meta(platform="dgpu", source="test", sink="display"),
            ["bgr10a2_le", "i420"],
        )

    def test_bgr10a2_to_rgba(self):
        self.assert_top_n_contain(
            "convert BGR10A2_LE to RGBA and display",
            _meta(platform="dgpu", source="test", sink="display"),
            ["bgr10a2_le", "rgba"],
        )


class TestPrimaryInference(GoldenTestBase):
    def test_single_stream_display_dgpu(self):
        self.assert_top_n_contain(
            "primary inference on a single mp4 video and display the output",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary", num_sources=1),
            ["nvinfer", "config-file-path", "nveglglessink"],
            not_expected=["nv3dsink"],
        )

    def test_single_stream_display_jetson(self):
        self.assert_top_n_contain(
            "primary inference on a single mp4 video and display the output on Jetson",
            _meta(platform="jetson", source="video", sink="display",
                  inference="primary", num_sources=1),
            ["nvinfer", "nv3dsink"],
        )
        self.assert_platform_match(
            "primary inference on Jetson",
            _meta(platform="jetson"), "jetson",
        )


class TestSecondaryInference(GoldenTestBase):
    def test_primary_secondary_display(self):
        self.assert_top_n_contain(
            "primary and secondary inference on a single stream and display",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary+secondary"),
            ["nvinfer", "infer-on-gie-id"],
        )


class TestTracker(GoldenTestBase):
    def test_inference_with_tracker(self):
        self.assert_top_n_contain(
            "primary and secondary inference with tracker on a single stream",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary+secondary", tracker="present"),
            ["nvtracker"],
        )


class TestMultiStream(GoldenTestBase):
    def test_4_stream_inference(self):
        self.assert_top_n_contain(
            "primary inference on 4 video streams and display",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary", num_sources=4),
            ["m.sink_0", "batch-size=4"],
            n=5,
        )


class TestSaveOutput(GoldenTestBase):
    def test_save_to_mp4(self):
        self.assert_top_n_contain(
            "encode video to h264 and mux into mp4 file",
            _meta(platform="dgpu", source="video", sink="save-mp4"),
            ["filesink", "qtmux"],
            n=5,
        )

    def test_save_h264(self):
        self.assert_top_n_contain(
            "encode test video to h264 and save to file",
            _meta(platform="dgpu", source="test", sink="save-file"),
            ["nvv4l2h264enc", "filesink"],
        )


class TestPreprocessor(GoldenTestBase):
    def test_preprocess_before_primary(self):
        self.assert_top_n_contain(
            "preprocess before primary inference on a single stream",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary+preprocess"),
            ["nvdspreprocess", "input-tensor-meta=1"],
        )


class TestResize(GoldenTestBase):
    def test_resize_video(self):
        self.assert_top_n_contain(
            "resize incoming video to specific height and width",
            _meta(platform="dgpu", source="video", sink="display"),
            ["nvvideoconvert"],
        )


class TestCrop(GoldenTestBase):
    def test_crop_video(self):
        self.assert_top_n_contain(
            "crop the incoming video frame",
            _meta(platform="dgpu", source="video", sink="display"),
            ["src-crop"],
        )


class TestDewarp(GoldenTestBase):
    def test_dewarp_dgpu(self):
        self.assert_top_n_contain(
            "dewarp a given mp4 video on dGPU",
            _meta(platform="dgpu", source="video", sink="display"),
            ["nvdewarper"],
        )


class TestPlatformFiltering(GoldenTestBase):
    def test_dgpu_query_prefers_dgpu(self):
        self.assert_platform_match(
            "primary inference on video and display output",
            _meta(platform="dgpu", source="video", sink="display",
                  inference="primary"),
            "dgpu",
        )

    def test_jetson_query_prefers_jetson(self):
        self.assert_platform_match(
            "primary inference on video and display output on Jetson",
            _meta(platform="jetson", source="video", sink="display",
                  inference="primary"),
            "jetson",
        )


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unit tests for the BM25 pipeline retriever."""

import os
import sys
import unittest

SCRIPT_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
sys.path.insert(0, SCRIPT_DIR)

from generate_pipeline import (
    BM25Retriever,
    expand_with_synonyms,
    extract_pipeline_metadata,
    normalize_user_params,
    tokenize,
)

DATA_CSV = os.path.join(os.path.dirname(__file__), "..", "data", "data.csv")


class TestTokenizer(unittest.TestCase):
    def test_lowercase(self):
        self.assertEqual(tokenize("Hello World"), ["hello", "world"])

    def test_strips_punctuation(self):
        tokens = tokenize("nvinfer config-file-path=/foo/bar.txt")
        self.assertIn("nvinfer", tokens)
        self.assertIn("config-file-path", tokens)

    def test_preserves_compound_tokens_splits_paths(self):
        """
        The tokenizer keeps compound hyphenated
        tokens (config-file-path) intact but splits on '/' so a path like
        /opt/nvidia/deepstream/config.txt yields useful per-component tokens
        ('opt', 'nvidia', 'deepstream', 'config.txt') for IDF rather than
        collapsing into one giant unique token.
        """
        tokens = tokenize("config-file-path=/opt/nvidia/deepstream/config.txt")
        self.assertIn("config-file-path", tokens)
        # Path components must be split out as separate tokens.
        self.assertIn("opt", tokens)
        self.assertIn("nvidia", tokens)
        self.assertIn("deepstream", tokens)
        self.assertIn("config.txt", tokens)
        # The whole path must NOT appear as a single token.
        self.assertNotIn("/opt/nvidia/deepstream/config.txt", tokens)


class TestSynonymExpansion(unittest.TestCase):
    def test_infer_expands(self):
        expanded = expand_with_synonyms(["infer"])
        self.assertIn("inference", expanded)
        self.assertIn("nvinfer", expanded)
        self.assertIn("detect", expanded)

    def test_no_duplicates(self):
        expanded = expand_with_synonyms(["infer", "inference"])
        counts = {}
        for tok in expanded:
            counts[tok] = counts.get(tok, 0) + 1
        for tok, count in counts.items():
            self.assertEqual(count, 1, f"Duplicate token: {tok}")

    def test_unknown_token_unchanged(self):
        expanded = expand_with_synonyms(["xyzzy123"])
        self.assertEqual(expanded, ["xyzzy123"])


class TestPipelineMetadata(unittest.TestCase):
    def test_dgpu_display(self):
        meta = extract_pipeline_metadata(
            "gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! "
            "nvv4l2decoder ! nvinfer config-file-path=c.txt ! nveglglessink"
        )
        self.assertEqual(meta["platform"], "dgpu")
        self.assertEqual(meta["sink_type"], "display")
        self.assertEqual(meta["source_type"], "video")
        self.assertEqual(meta["inference"], "primary")
        self.assertEqual(meta["tracker"], "none")

    def test_jetson_display(self):
        meta = extract_pipeline_metadata(
            "gst-launch-1.0 filesrc location=a.mp4 ! nvinfer config-file-path=c.txt ! nv3dsink"
        )
        self.assertEqual(meta["platform"], "jetson")
        self.assertEqual(meta["sink_type"], "display")

    def test_primary_secondary_with_tracker(self):
        meta = extract_pipeline_metadata(
            "nvinfer config-file-path=p.txt ! nvtracker ll-lib-file=t.so ! "
            "nvinfer config-file-path=s.txt infer-on-gie-id=1 ! nveglglessink"
        )
        self.assertEqual(meta["inference"], "primary+secondary")
        self.assertEqual(meta["tracker"], "present")

    def test_multi_source_detection(self):
        meta = extract_pipeline_metadata(
            "filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=4 ! "
            "nveglglessink filesrc location=b.mp4 ! m.sink_1 "
            "filesrc location=c.mp4 ! m.sink_2 "
            "filesrc location=d.mp4 ! m.sink_3"
        )
        self.assertEqual(meta["num_sources"], 4)

    def test_rtsp_source(self):
        meta = extract_pipeline_metadata(
            "uridecodebin uri=rtsp://<rtsp-server>:<port>/stream ! nveglglessink"
        )
        self.assertEqual(meta["source_type"], "rtsp")

    def test_save_mp4_sink(self):
        meta = extract_pipeline_metadata(
            "filesrc location=a.mp4 ! nvv4l2h264enc ! h264parse ! qtmux ! filesink location=out.mp4"
        )
        self.assertEqual(meta["sink_type"], "save-mp4")

    def test_image_source(self):
        meta = extract_pipeline_metadata(
            "filesrc location=img.jpg ! jpegparse ! nvjpegdec ! nveglglessink"
        )
        self.assertEqual(meta["source_type"], "image")

    def test_preprocess_inference(self):
        meta = extract_pipeline_metadata(
            "nvdspreprocess config-file=c.txt ! nvinfer config-file-path=p.txt input-tensor-meta=1 ! nveglglessink"
        )
        self.assertEqual(meta["inference"], "primary+preprocess")

    def test_fakesink(self):
        meta = extract_pipeline_metadata("videotestsrc ! fakesink")
        self.assertEqual(meta["source_type"], "test")
        self.assertEqual(meta["sink_type"], "fakesink")


class TestBM25Retriever(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.retriever = BM25Retriever()
        cls.retriever.load_csv(DATA_CSV)

    def test_loads_documents(self):
        self.assertGreater(len(self.retriever.documents), 200)

    def test_idf_computed(self):
        self.assertGreater(len(self.retriever.idf), 0)
        self.assertIn("nvinfer", self.retriever.idf)

    def test_avgdl_positive(self):
        self.assertGreater(self.retriever.avgdl, 0)

    def test_retrieve_returns_results(self):
        results, conf = self.retriever.retrieve("flip video 90 degrees")
        self.assertGreater(len(results), 0)

    def test_retrieve_scores_descending(self):
        results, _ = self.retriever.retrieve("primary inference on mp4 video display")
        scores = [r["score"] for r in results]
        self.assertEqual(scores, sorted(scores, reverse=True))

    def test_retrieve_contains_metadata(self):
        results, _ = self.retriever.retrieve("primary inference")
        for r in results:
            self.assertIn("metadata", r)
            self.assertIn("platform", r["metadata"])

    def test_confidence_levels(self):
        _, conf = self.retriever.retrieve(
            "primary inference on a single mp4 video and display the output"
        )
        self.assertIn(conf, ("high", "medium", "low"))

    def test_structural_boost_prefers_matching_platform(self):
        dgpu_meta = {
            "platform": "dgpu", "source_type": "video", "sink_type": "display",
            "inference": "primary", "tracker": "none", "num_sources": 1,
        }
        jetson_meta = {
            "platform": "jetson", "source_type": "video", "sink_type": "display",
            "inference": "primary", "tracker": "none", "num_sources": 1,
        }
        query = "primary inference on video and display output"
        dgpu_results, _ = self.retriever.retrieve(query, user_meta=dgpu_meta)
        jetson_results, _ = self.retriever.retrieve(query, user_meta=jetson_meta)

        if dgpu_results and jetson_results:
            dgpu_top_platform = dgpu_results[0]["metadata"]["platform"]
            jetson_top_platform = jetson_results[0]["metadata"]["platform"]
            self.assertEqual(dgpu_top_platform, "dgpu")
            self.assertEqual(jetson_top_platform, "jetson")

    def test_empty_query_returns_empty(self):
        results, conf = self.retriever.retrieve("")
        self.assertEqual(conf, "low")


class TestIndexCache(unittest.TestCase):
    """
    Regression: the BM25 index is cached to a sibling
    JSON file keyed on (CSV mtime, internal schema version). A cached
    retriever must (a) reproduce the same indexed state as a fresh one and
    (b) be invalidated when the CSV is touched.
    """

    def setUp(self):
        import shutil
        import tempfile
        self.tmpdir = tempfile.mkdtemp(prefix="ds-pg-cache-test-")
        self.csv_copy = os.path.join(self.tmpdir, "data.csv")
        shutil.copyfile(DATA_CSV, self.csv_copy)
        self.cache_path = self.csv_copy + ".cache.json"

    def tearDown(self):
        import shutil
        shutil.rmtree(self.tmpdir, ignore_errors=True)

    def test_cache_written_on_first_load(self):
        self.assertFalse(os.path.exists(self.cache_path))
        retriever = BM25Retriever()
        retriever.load_csv(self.csv_copy)
        self.assertTrue(os.path.exists(self.cache_path))

    def test_second_load_reproduces_state_from_cache(self):
        first = BM25Retriever()
        first.load_csv(self.csv_copy)
        second = BM25Retriever()
        second.load_csv(self.csv_copy)
        # Same documents, same IDF, same average doc length — proves the
        # cached state round-trips through JSON correctly.
        self.assertEqual(len(first.documents), len(second.documents))
        self.assertEqual(first.idf, second.idf)
        self.assertEqual(first.avgdl, second.avgdl)
        self.assertEqual(len(first.doc_tokens), len(second.doc_tokens))

    def test_cache_invalidated_when_csv_mtime_changes(self):
        retriever = BM25Retriever()
        retriever.load_csv(self.csv_copy)
        cache_mtime_before = os.path.getmtime(self.cache_path)

        # Touch the CSV so its mtime advances (sleep ensures the new mtime
        # exceeds filesystem resolution).
        import time
        time.sleep(1.1)
        os.utime(self.csv_copy, None)

        retriever2 = BM25Retriever()
        retriever2.load_csv(self.csv_copy)
        cache_mtime_after = os.path.getmtime(self.cache_path)
        # Cache was rewritten, not reused.
        self.assertGreater(cache_mtime_after, cache_mtime_before)

    def test_cache_invalidated_on_schema_version_bump(self):
        import json
        retriever = BM25Retriever()
        retriever.load_csv(self.csv_copy)
        # Rewrite the cache with an obviously-stale version stamp; the next
        # load must reject it and rebuild from CSV.
        with open(self.cache_path, "r", encoding="utf-8") as f:
            payload = json.load(f)
        payload["version"] = -999
        with open(self.cache_path, "w", encoding="utf-8") as f:
            json.dump(payload, f)

        retriever2 = BM25Retriever()
        retriever2.load_csv(self.csv_copy)
        # Rebuilt cache must have the current schema version, not the bogus one.
        with open(self.cache_path, "r", encoding="utf-8") as f:
            new_payload = json.load(f)
        self.assertNotEqual(new_payload["version"], -999)


class TestNormalizeUserParams(unittest.TestCase):
    def test_basic_mapping(self):
        class Args:
            source_type = "Local video file"
            num_sources = 1
            inference = "primary"
            tracker = "none"
            sink = "display"
            platform = "dGPU"

        meta = normalize_user_params(Args())
        self.assertEqual(meta["platform"], "dgpu")
        self.assertEqual(meta["source_type"], "video")
        self.assertEqual(meta["sink_type"], "display")
        self.assertEqual(meta["inference"], "primary")
        self.assertEqual(meta["tracker"], "none")

    def test_jetson_mapping(self):
        class Args:
            source_type = "RTSP stream"
            num_sources = 4
            inference = "primary+secondary"
            tracker = "NvDCF"
            sink = "save-mp4"
            platform = "Jetson"

        meta = normalize_user_params(Args())
        self.assertEqual(meta["platform"], "jetson")
        self.assertEqual(meta["source_type"], "rtsp")
        self.assertEqual(meta["tracker"], "present")


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unit tests for the pipeline validator."""

import os
import re
import sys
import unittest

SCRIPT_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
sys.path.insert(0, SCRIPT_DIR)

from validate_pipeline import (
    extract_elements_and_properties,
    validate_element_ordering,
    validate_memory_format,
    validate_pipeline_structure,
    validate_platform_sink,
    validate_syntax,
    validate_with_gst_launch,
)


class TestSyntaxValidation(unittest.TestCase):
    def test_valid_pipeline(self):
        errors, warnings = validate_syntax(
            "gst-launch-1.0 filesrc location=test.mp4 ! nveglglessink"
        )
        self.assertEqual(errors, [])

    def test_unbalanced_single_quotes(self):
        errors, _ = validate_syntax("gst-launch-1.0 filesrc location='test.mp4 ! nveglglessink")
        self.assertTrue(any("single quotes" in e for e in errors))

    def test_unbalanced_double_quotes(self):
        errors, _ = validate_syntax('gst-launch-1.0 filesrc location="test.mp4 ! nveglglessink')
        self.assertTrue(any("double quotes" in e for e in errors))

    def test_empty_pipe_segment(self):
        errors, _ = validate_syntax("gst-launch-1.0 filesrc location=t.mp4 ! ! nveglglessink")
        self.assertTrue(any("Empty pipe" in e for e in errors))

    def test_leading_pipe(self):
        errors, _ = validate_syntax("! filesrc location=t.mp4 ! nveglglessink")
        self.assertTrue(any("starts with" in e for e in errors))

    def test_trailing_pipe(self):
        errors, _ = validate_syntax("gst-launch-1.0 filesrc location=t.mp4 !")
        self.assertTrue(any("ends with" in e for e in errors))

    def test_underscore_property_typo(self):
        errors, _ = validate_syntax(
            "gst-launch-1.0 nvinfer config_file_path=c.txt ! nveglglessink"
        )
        self.assertTrue(any("hyphens" in e for e in errors))

    def test_underscore_typo_not_masked_by_hyphen_form_elsewhere(self):
        """
        Regression: the underscore-typo check must
        scope to the segment that owns nvinfer. A hyphenated config-file-path
        on a different element (here nvdspreprocess) must not silently mask
        an underscore typo on the actual nvinfer.
        """
        errors, _ = validate_syntax(
            "gst-launch-1.0 nvdspreprocess config-file=p.txt ! "
            "nvinfer config_file_path=c.txt ! nveglglessink"
        )
        self.assertTrue(any("hyphens" in e for e in errors))

    def test_streammux_without_name(self):
        _, warnings = validate_syntax(
            "gst-launch-1.0 filesrc location=t.mp4 ! nvstreammux batch-size=1 ! nveglglessink"
        )
        self.assertTrue(any("name=" in w for w in warnings))


class TestElementParser(unittest.TestCase):
    def test_simple_pipeline(self):
        elements, pad_refs = extract_elements_and_properties(
            "gst-launch-1.0 filesrc location=test.mp4 ! nveglglessink"
        )
        names = [e["name"] for e in elements]
        self.assertIn("filesrc", names)
        self.assertIn("nveglglessink", names)
        self.assertEqual(pad_refs, [])

    def test_properties_extracted(self):
        elements, _ = extract_elements_and_properties(
            "nvinfer config-file-path=c.txt batch-size=4 unique-id=1 ! nveglglessink"
        )
        nvinfer = [e for e in elements if e["name"] == "nvinfer"][0]
        self.assertEqual(nvinfer["properties"]["config-file-path"], "c.txt")
        self.assertEqual(nvinfer["properties"]["batch-size"], "4")

    def test_pad_refs_extracted(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! "
            "m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink"
        )
        self.assertIn("m.sink_0", pad_refs)
        names = [e["name"] for e in elements]
        self.assertIn("nvstreammux", names)

    def test_multiple_pad_refs(self):
        # Both branches end with a pad ref followed by a ' ! ' separator so
        # the parser sees each pad ref as the leading token of its own
        # segment — making the assertion on m.sink_0 AND m.sink_1
        # deterministic.
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=a.mp4 ! m.sink_0 "
            "filesrc location=b.mp4 ! m.sink_1 "
            "nvstreammux name=m batch-size=2 width=1920 height=1080 ! nveglglessink"
        )
        self.assertIn("m.sink_0", pad_refs)
        self.assertIn("m.sink_1", pad_refs)

    def test_caps_filter_skipped(self):
        elements, _ = extract_elements_and_properties(
            "videotestsrc ! 'video/x-raw,format=NV12' ! nveglglessink"
        )
        names = [e["name"] for e in elements]
        self.assertNotIn("video/x-raw,format=NV12", names)

    def test_gst_launch_prefix_stripped(self):
        elements, _ = extract_elements_and_properties(
            "gst-launch-1.0 filesrc location=t.mp4 ! nveglglessink"
        )
        self.assertNotIn("gst-launch-1.0", [e["name"] for e in elements])

    def test_flags_stripped(self):
        elements, _ = extract_elements_and_properties(
            "gst-launch-1.0 -e filesrc location=t.mp4 ! nveglglessink"
        )
        self.assertNotIn("-e", [e["name"] for e in elements])

    def test_multiple_flags_stripped(self):
        """
        Regression: a run of leading flags such as
        '-e -v' must be stripped together. Previously only the first flag was
        removed and the second was parsed as a phantom element.
        """
        elements, _ = extract_elements_and_properties(
            "gst-launch-1.0 -e -v filesrc location=t.mp4 ! nveglglessink"
        )
        names = [e["name"] for e in elements]
        self.assertNotIn("-e", names)
        self.assertNotIn("-v", names)
        self.assertIn("filesrc", names)

    def test_live_parse_substitution_anchors_to_segment_start(self):
        """
        Regression: the live-parse source/sink
        replacement must not match the element name when it appears inside a
        property value (e.g. a config-file-path that contains 'filesrc' as
        part of a directory or filename). Previously \\b{real}\\b matched the
        substring inside the path and corrupted the dry-run input. The fix
        anchors to segment-leading positions ((^|!\\s+)).
        """
        original = (
            "nvinfer config-file-path=/foo/filesrc.txt ! "
            "filesrc location=t.mp4 ! fakesink"
        )
        substituted = re.sub(
            r"(^|!\s+)filesrc\b[^!]*",
            r"\g<1>fakesrc num-buffers=1 ",
            original,
        )
        # The 'filesrc' inside the config-file-path must remain untouched.
        self.assertIn("config-file-path=/foo/filesrc.txt", substituted)
        # The actual filesrc element (after !) must be replaced with fakesrc.
        self.assertIn("fakesrc num-buffers=1", substituted)
        # The original filesrc element's location property should be gone.
        self.assertNotIn("location=t.mp4", substituted)


class TestStructureValidation(unittest.TestCase):
    def test_valid_pipeline(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvinfer config-file-path=c.txt ! nveglglessink"
        )
        errors, warnings = validate_pipeline_structure(elements, pad_refs)
        self.assertEqual(errors, [])

    def test_missing_source_warning(self):
        elements, pad_refs = extract_elements_and_properties(
            "nvinfer config-file-path=c.txt ! nveglglessink"
        )
        _, warnings = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("source" in w.lower() for w in warnings))

    def test_missing_sink_warning(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvinfer config-file-path=c.txt"
        )
        _, warnings = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("sink" in w.lower() for w in warnings))

    def test_nvinfer_missing_config(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvinfer batch-size=1 ! nveglglessink"
        )
        errors, _ = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("config-file-path" in e for e in errors))

    def test_filesrc_missing_location(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc ! nveglglessink"
        )
        errors, _ = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("location" in e for e in errors))

    def test_empty_pipeline(self):
        errors, _ = validate_pipeline_structure([], [])
        self.assertTrue(any("empty" in e.lower() for e in errors))

    def test_bad_pad_ref_warns(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=a.mp4 ! x.sink_0 nveglglessink"
        )
        _, warnings = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("x.sink_0" in w for w in warnings))

    def test_valid_pad_ref_no_warning(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=1 ! nveglglessink"
        )
        _, warnings = validate_pipeline_structure(elements, pad_refs)
        pad_warnings = [w for w in warnings if "pad reference" in w.lower()]
        self.assertEqual(pad_warnings, [])

    def test_batch_size_pad_mismatch(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=a.mp4 ! m.sink_0 nvstreammux name=m batch-size=4 width=1920 height=1080 ! nveglglessink"
        )
        _, warnings = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("batch-size=4" in w and "1 sink" in w for w in warnings))


class TestMemoryFormatValidation(unittest.TestCase):
    """Test memory format mismatch detection between elements."""

    def test_nvjpegdec_direct_to_mux_detected(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! "
            "m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink"
        )
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertTrue(len(errors) > 0)
        self.assertTrue(any("nvjpegdec" in e and "NVMM" in e for e in errors))

    def test_nvjpegdec_with_videoconvert_passes(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! "
            "nvvideoconvert ! m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink"
        )
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertEqual(errors, [])

    def test_nvv4l2decoder_to_mux_passes(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! "
            "m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink"
        )
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertEqual(errors, [])

    def test_nvjpegdec_to_display_no_error(self):
        pipeline = "gst-launch-1.0 filesrc location=a.jpg ! nvjpegdec ! nveglglessink"
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertEqual(errors, [])

    def test_pngdec_direct_to_mux_detected(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.png ! pngdec ! "
            "m.sink_0 nvstreammux name=m batch-size=1 ! nveglglessink"
        )
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertTrue(len(errors) > 0)
        self.assertTrue(any("pngdec" in e for e in errors))

    def test_no_duplicate_errors(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.jpg ! jpegparse ! nvjpegdec ! "
            "m.sink_0 nvstreammux name=m batch-size=1 width=1920 height=1080 ! nveglglessink"
        )
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertEqual(len(errors), 1, f"Expected 1 error but got {len(errors)}: {errors}")


class TestKnownProperties(unittest.TestCase):
    """Validate that the property dict covers all properties used in data.csv."""

    def test_nvinfer_has_clip_object_outside_roi(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("clip-object-outside-roi", KNOWN_ELEMENT_PROPERTIES["nvinfer"])

    def test_nvinfer_has_output_tensor_meta(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("output-tensor-meta", KNOWN_ELEMENT_PROPERTIES["nvinfer"])

    def test_nvdspreprocess_has_operate_on_gie_id(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("operate-on-gie-id", KNOWN_ELEMENT_PROPERTIES["nvdspreprocess"])

    def test_nvv4l2h264enc_has_qp_range(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("qp-range", KNOWN_ELEMENT_PROPERTIES["nvv4l2h264enc"])

    def test_nvdewarper_has_num_output_buffers(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("num-output-buffers", KNOWN_ELEMENT_PROPERTIES["nvdewarper"])


class TestPlatformSinkValidation(unittest.TestCase):
    """Test platform-sink mismatch detection."""

    def test_mixed_dgpu_jetson_sinks_warns(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nveglglessink ! nv3dsink"
        )
        warnings = validate_platform_sink(elements)
        self.assertTrue(any("mixes" in w.lower() for w in warnings))

    def test_dgpu_only_no_warning(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nveglglessink"
        )
        warnings = validate_platform_sink(elements)
        self.assertEqual(warnings, [])

    def test_jetson_only_no_warning(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nv3dsink"
        )
        warnings = validate_platform_sink(elements)
        self.assertEqual(warnings, [])

    def test_fakesink_no_warning(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! fakesink"
        )
        warnings = validate_platform_sink(elements)
        self.assertEqual(warnings, [])


class TestElementOrderingValidation(unittest.TestCase):
    """Test element ordering checks."""

    def test_correct_order_no_warning(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvv4l2decoder ! nvinfer config-file-path=c.txt ! nveglglessink"
        )
        warnings = validate_element_ordering(elements)
        self.assertEqual(warnings, [])

    def test_nvinfer_before_decoder_warns(self):
        elements, _ = extract_elements_and_properties(
            "nvinfer config-file-path=c.txt ! nvv4l2decoder ! filesrc location=t.mp4 ! nveglglessink"
        )
        warnings = validate_element_ordering(elements)
        self.assertTrue(len(warnings) > 0)

    def test_sink_before_source_warns(self):
        elements, _ = extract_elements_and_properties(
            "nveglglessink ! filesrc location=t.mp4 ! nvinfer config-file-path=c.txt"
        )
        warnings = validate_element_ordering(elements)
        # nvinfer after nveglglessink is detected via filesrc -> nvinfer ordering check
        self.assertTrue(len(warnings) > 0)

    def test_tracker_before_infer_warns(self):
        elements, _ = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvtracker ll-lib-file=t.so ! nvinfer config-file-path=c.txt ! nveglglessink"
        )
        warnings = validate_element_ordering(elements)
        self.assertTrue(any("nvtracker" in w and "nvinfer" in w for w in warnings))


class TestExpandedNVMMChecks(unittest.TestCase):
    """Test that NVMM checks cover encoders too."""

    def test_system_mem_to_h264enc_detected(self):
        pipeline = "gst-launch-1.0 filesrc location=a.jpg ! nvjpegdec ! nvv4l2h264enc ! filesink location=out.h264"
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertTrue(any("nvjpegdec" in e and "nvv4l2h264enc" in e for e in errors))

    def test_videoconvert_to_h264enc_detected(self):
        pipeline = "gst-launch-1.0 filesrc location=a.mp4 ! videoconvert ! nvv4l2h264enc ! filesink location=out.h264"
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertTrue(any("videoconvert" in e and "nvv4l2h264enc" in e for e in errors))

    def test_nvvideoconvert_to_h264enc_passes(self):
        pipeline = "gst-launch-1.0 filesrc location=a.mp4 ! nvvideoconvert ! nvv4l2h264enc ! filesink location=out.h264"
        elements, pad_refs = extract_elements_and_properties(pipeline)
        errors = validate_memory_format(pipeline, elements, pad_refs)
        self.assertEqual(errors, [])


class TestNewElements(unittest.TestCase):
    """Test that new elements are in the known properties dict."""

    def test_nvinferserver_in_known(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("nvinferserver", KNOWN_ELEMENT_PROPERTIES)
        self.assertIn("config-file-path", KNOWN_ELEMENT_PROPERTIES["nvinferserver"])

    def test_nvdsanalytics_in_known(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("nvdsanalytics", KNOWN_ELEMENT_PROPERTIES)

    def test_nvmsgconv_in_known(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("nvmsgconv", KNOWN_ELEMENT_PROPERTIES)

    def test_nvmsgbroker_in_known(self):
        from validate_pipeline import KNOWN_ELEMENT_PROPERTIES
        self.assertIn("nvmsgbroker", KNOWN_ELEMENT_PROPERTIES)

    def test_nvinferserver_missing_config_flagged(self):
        elements, pad_refs = extract_elements_and_properties(
            "filesrc location=t.mp4 ! nvinferserver batch-size=1 ! nveglglessink"
        )
        errors, _ = validate_pipeline_structure(elements, pad_refs)
        self.assertTrue(any("nvinferserver" in e and "config-file-path" in e for e in errors))


class TestDryRunMultiStream(unittest.TestCase):
    """Dry-run is skipped for multi-stream pipelines with named pad refs."""

    def test_skipped_when_pad_refs_present(self):
        pipeline = (
            "gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! "
            "m.sink_0 nvstreammux name=m batch-size=2 width=1920 height=1080 ! "
            "nvinfer config-file-path=c.txt batch-size=2 ! nvvideoconvert ! nvdsosd ! "
            "nveglglessink filesrc location=b.mp4 ! qtdemux ! h264parse ! nvv4l2decoder ! m.sink_1"
        )
        pad_refs = ["m.sink_0", "m.sink_1"]
        errors, warnings = validate_with_gst_launch(pipeline, pad_refs)
        self.assertEqual(errors, [])
        self.assertEqual(warnings, [])

    def test_not_skipped_when_no_pad_refs(self):
        """Dry-run should execute (not be skipped) for single-stream pipelines."""
        pipeline = (
            "gst-launch-1.0 filesrc location=a.mp4 ! qtdemux ! h264parse ! "
            "nvv4l2decoder ! nveglglessink"
        )
        errors, warnings = validate_with_gst_launch(pipeline, pad_refs=[])
        self.assertIsInstance(errors, list)
        self.assertIsInstance(warnings, list)

    def test_not_skipped_when_pad_refs_none(self):
        pipeline = "gst-launch-1.0 videotestsrc ! fakesink"
        errors, warnings = validate_with_gst_launch(pipeline, pad_refs=None)
        self.assertIsNotNone(errors)
        self.assertIsNotNone(warnings)
        self.assertIsInstance(errors, list)
        self.assertIsInstance(warnings, list)


if __name__ == "__main__":
    unittest.main()

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.