deepx-ai/dx_stream/dx-agent-stream-build-pipeline

Skill: Build dx_stream Pipeline Application

This skill document is sufficient. Read this FIRST before exploring source code.

Only read source files if this document lacks the specific information you need.

Overview

Build a GStreamer pipeline application for real-time video processing on DEEPX NPU. This skill covers all 5 pipeline categories with complete templates and patterns.

Output Isolation (MUST FOLLOW)

All AI-generated pipeline applications MUST be created under dx-agent-dev/, NOT in production directories. This prevents accidental modification of existing pipelines.

Session Directory

dx-agent-dev/<YYYYMMDD-HHMMSS>_<model>_<pipeline_category>/
├── session.json          # Build metadata
├── README.md             # How to run this pipeline
├── pipeline.py           # Python pipeline script
├── run_<app>.sh           # Shell script wrapper
└── config/               # Pipeline-specific configs (optional)

session.json Template

{
  "session_id": "<YYYYMMDD-HHMMSS>_<agent>_<model>_<category>",
  "created_at": "<ISO 8601 with local timezone — use datetime.now().astimezone().isoformat(timespec='seconds')>",
  "model": "<DX inference model name — e.g. 'yolo26n'>",
  // ⚠ HARD ERROR: This is the DXNN model name deployed (e.g., "yolo26n", "EfficientNet_Lite0").
  // NOT the AI agent name. Writing "claude-sonnet-4.6", "gpt-4.1", or any AI model name
  // here will cause test_session_json_model_is_dx_model to FAIL.
  "pipeline_category": "<single_model|multi_model|cascaded|parallel|broker> (use underscores, NOT hyphens)",
  "task": "<Object Detection|Pose Estimation|Segmentation|Classification>",
  "postprocess_lib": "<libpostprocess_xxx.so>",
  "tracker": "<dxtracker|none>",
  "input_size": "<WxH e.g. 640x640>",
  "status": "complete",
  "notes": "<any relevant notes>"
}

`session_id` MUST contain the agent identifier (R81). Format: YYYYMMDD-HHMMSS_<agent>_<model>_<task> where <agent> is one of: claude, codex, copilot, cursor, opencode.

// CORRECT:  "session_id": "20260429-155626_copilot_yolo26n_cascaded"
// WRONG:    "session_id": "20260429-155626_yolo26n_cascaded"  ← missing agent identifier

`created_at` MUST include timezone offset. Do NOT use datetime.now().strftime(...) without appending timezone — downstream ISO 8601 parsers will reject timestamps without an offset.

# Correct — includes +09:00 or equivalent local offset:
"created_at": datetime.now().astimezone().isoformat(timespec='seconds')
# Wrong — missing timezone:
"created_at": datetime.now().strftime('%Y-%m-%dT%H:%M:%S')

When to Use Production Path

Only create files in production directories when the user EXPLICITLY says:

• "Add this to the production codebase"
• "Create this in the examples directory"
• "Make this a permanent addition"

Default behavior: ALWAYS use dx-agent-dev/.

Usage

Invoke with /dx-agent-stream-build-pipeline or ask the dx-stream-builder agent. Specify the pipeline category (single-model, multi-model, cascaded, parallel, or broker), the model name, and the input source type.

Template Compliance (MANDATORY)

The templates in this document are battle-tested against the actual dx_stream runtime. Generated code MUST follow these templates exactly.

Rules

1. Follow the template structure — use the template as the baseline for correctness.

User-requested additions and modifications are allowed. Do NOT add unrequested features (no custom logging format, no FPS overlay unless asked)

1. Default: single output sink — fpsdisplaysink OR fakesink. If the user

requests file output, replace the display sink with videoconvert ! x264enc bitrate=4000 speed-preset=ultrafast tune=zerolatency ! h264parse ! mp4mux ! filesink location=...

1. `--output` / `tee` is RECOMMENDED for detection pipelines — Detection

pipelines SHOULD include the --output argument in parse_args() and the tee code path for simultaneous display+recording, unless the user explicitly excludes it. When included, use the tee name=t pattern: `` tee name=t \ t. ! queue ! videoconvert ! fpsdisplaysink sync=false \ t. ! queue ! videoconvert ! x264enc bitrate=4000 speed-preset=ultrafast tune=zerolatency ! h264parse ! mp4mux ! filesink location=output.mp4 ` **CRITICAL: x264enc MUST include tune=zerolatency`** — see Pitfall #14

1. English log messages — all logger.info/logger.error messages in English,

matching the style of existing examples

1. `--headless` flag — always include it in pipeline.py. Additionally check

DISPLAY env as fallback (see template)

1. `run_<app>.sh` MUST delegate to `pipeline.py` — the shell wrapper MUST invoke

python pipeline.py with --input and --model arguments. Embedding gst-launch-1.0 inline in run_<app>.sh is PROHIBITED (see Anti-Patterns below). The test test_run_script_invokes_pipeline enforces this and will fail on inline gst-launch.

1. Path resolution — run_<app>.sh uses DX_STREAM_ROOT (2 levels up from

dx-agent-dev/<session>/), NOT the production 3-level-up pattern. See the run_<app>.sh template below for the exact code

Anti-Patterns (NEVER Do)

• Using x264enc without tune=zerolatency (see Pitfall #14)
• Using relative model paths in dxinfer model-path= (must be absolute)
• Copying SRC_DIR calculation from production scripts in dx_stream/pipelines/
• Writing log messages in Korean or other non-English languages
• Adding unrequested features that complicate the pipeline (e.g., FPS overlay, auto-recording)
• Embedding `gst-launch-1.0` inline in `run_<app>.sh` — PROHIBITED. The run script

MUST delegate to python pipeline.py, not call gst-launch-1.0 directly. Example of the correct pattern: ```bash # CORRECT — delegates to pipeline.py: source "$DX_STREAM_ROOT/venv-dx_stream/bin/activate" python pipeline.py --input "$INPUT_VIDEO" --model "$MODEL_PATH" \ --postprocess-lib /usr/local/share/gstdxstream/lib/{POSTPROCESS_LIB}

# ANTI-PATTERN (PROHIBITED) — inline gst-launch embedding: gst-launch-1.0 urisourcebin uri=file://$INPUT_VIDEO ! decodebin ! dxpreprocess ... ```

• Omitting `dxrate` after RTSP decodebin — always add dxrate max-rate=30 in the

RTSP source branch. Example: urisourcebin uri=rtsp://... ! decodebin ! dxrate max-rate=30 ! dxpreprocess ...

• `created_at` without timezone offset — always use datetime.now().astimezone().isoformat(timespec='seconds')

Phase 0: Prerequisites Check

Before starting the build workflow, verify:

Working directory note: All Phase 0 commands assume you are in the dx_stream/ root.

The sanity check script lives TWO levels up: ../../scripts/sanity_check.sh

(NOT scripts/sanity_check.sh — that path does not exist from dx_stream/).

NEVER pipe through | tail or | head — they mask the exit code and may hide failures.

1. dx-runtime: bash ../../scripts/sanity_check.sh --dx_rt

• ← MANDATORY: run from `dx_stream/` root with the `../../` prefix. Do NOT truncate the path.
• FAIL → bash ../../install.sh --all --exclude-app --exclude-stream --skip-uninstall --venv-reuse
• Re-run sanity_check.sh — must PASS after install
• If still failing → STOP (unconditional). User instructions to continue do NOT override this.

If NPU hardware init failure ("Device initialization failed"): tell the user a cold boot / system reboot is required, then STOP. NEVER proceed with code generation while sanity check is failing. NEVER mark this check as "done" when it actually failed.

1. GStreamer plugins: gst-inspect-1.0 dxinfer — FAIL → ./install.sh && ./build.sh
2. Postprocess libraries: ls /usr/local/share/gstdxstream/lib/libpostprocess_*.so
3. Python venv (for pipeline.py): The pydxs module is only available inside

the dx_stream virtual environment. Verify it: ``bash source venv-dx_stream/bin/activate && python3 -c "import pydxs; print('pydxs OK')" ``

• FAIL → ./install.sh && ./build.sh (creates venv-dx_stream and installs pydxs)
• If venv exists but pydxs not importable → source venv-dx_stream/bin/activate && pip install -e .
• IMPORTANT: The README.md template MUST document this venv activation step

so users know to activate it before running pipeline.py (see pitfall #15)

1. DX-H1 V-NPU (optional): Build V-NPU support with ./build.sh --dxvnpu.

On Windows, DEEPX_VNPU_DIR must identify the installed V-NPU SDK. Confirm optional codec registration with gst-inspect-1.0 dxvnpudec and gst-inspect-1.0 dxvnpuenc before generating a hardware-codec pipeline.

Prerequisites

• DX-RT installed and NPU detected (dxrt-cli -s)
• GStreamer plugin registered (gst-inspect-1.0 dxinfer)
• Model file downloaded (./setup.sh --model="<model>.dxnn")

Quick Reference

DX-H1 V-NPU Usage

• Use dxvnpudec and dxvnpuenc only in standard raw-video codec branches; they

are outside the application/x-dxvideoraw multi-stream domain.

• Set dxinfer backend=dxvnpu only for a V-NPU-specific inference deployment.

Use backend=auto when the pipeline must retain fallback behavior.

• dxpreprocess, dxscale, and dxconvert may use a VNPU transform kernel for

image resize and color conversion. Do not make that optional acceleration a pipeline requirement.

Supported Models (14)

Core Pipeline Pattern

source → dxpreprocess → queue → dxinfer → queue → dxpostprocess → queue → dxtracker → queue → dxosd → sink

DxTracker is RECOMMENDED for object-detection pipelines. Pipelines whose

task is object detection (yolo*, ssd, scrfd, efficient-det, etc.) SHOULD include

dxtracker between dxpostprocess and dxosd by default. Use the config file

../../dx_stream/configs/tracker_config.json (relative to the session directory).

The user may explicitly exclude tracker if not needed for their use case.

Template: pipeline.py (Python GStreamer Pipeline)

#!/usr/bin/env python3
"""dx_stream pipeline: {TASK} with {MODEL}."""
import argparse
import logging
import os
from pathlib import Path
import sys
try:
    import pydxs  # noqa: F401 — confirms dx_stream venv is active
except ImportError:
    print("Error: pydxs not found. Activate the dx_stream venv first:", file=sys.stderr)
    print("  source ../../venv-dx_stream/bin/activate", file=sys.stderr)
    sys.exit(1)
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
logger = logging.getLogger(__name__)