Hl Build Pipeline App

Skill: Build GStreamer Pipeline Application

Build a complete GStreamer pipeline app for real-time video processing on Hailo-8/8L/10H.

When This Skill Is Loaded

• User wants real-time video processing (detection, pose, segmentation)
• User mentions: GStreamer, pipeline, stream, FPS, real-time video, tracking
• User needs a video app with high throughput rather than AI understanding

Reference Implementations

The canonical pipeline app is detection/. Other examples: pose_estimation/, instance_segmentation/, face_recognition/.

Do NOT read these source files. This SKILL.md contains all patterns needed to build any pipeline app. The sections below cover: basic pipelines, frame overlays, custom backgrounds, pose extraction, detection data, and subclassing existing pipeline classes.

Minimum Context for Any Pipeline App

Read this SKILL.md (full file, single read) + common_pitfalls.md. That's it. Build immediately.

Build Process

Step 1: Create App Directory

Create the app directory:

hailo_apps/python/<type>/<app_name>/
├── app.yaml              # App manifest (required)
├── run.sh                # Launch wrapper
├── __init__.py
├── <app_name>.py         # Main app
└── README.md             # Usage documentation (REQUIRED — never skip)

Create app.yaml with type: pipeline and run.sh wrapper. Do NOT register in defines.py or resources_config.yaml.

Step 2: Build Main App

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst

import hailo  # Required for detection/landmark extraction in callbacks

from hailo_apps.python.core.common.hailo_logger import get_logger
from hailo_apps.python.core.common.core import resolve_hef_path, handle_list_models_flag
from hailo_apps.python.core.common.parser import get_pipeline_parser
# If your app uses resolve_hef_path with an app name, register it in defines.py.
# Otherwise use a local string constant:
# APP_NAME = "my_pipeline_app"
from hailo_apps.python.core.gstreamer.gstreamer_app import GStreamerApp, app_callback_class
from hailo_apps.python.core.gstreamer.gstreamer_helper_pipelines import (
    SOURCE_PIPELINE,
    INFERENCE_PIPELINE,
    INFERENCE_PIPELINE_WRAPPER,
    DISPLAY_PIPELINE,
    TRACKER_PIPELINE,
    USER_CALLBACK_PIPELINE,
    QUEUE,
)
from hailo_apps.python.core.common.buffer_utils import (
    get_caps_from_pad,
    get_numpy_from_buffer,
)

logger = get_logger(__name__)

APP_NAME = "my_pipeline_app"


class UserAppCallback(app_callback_class):
    """Custom callback class for per-frame state."""
    def __init__(self):
        super().__init__()
        self.detection_count = 0


def app_callback(element, buffer, user_data):
    """Per-frame callback — runs on every GStreamer buffer."""
    # Access detections from buffer
    # user_data.detection_count += len(detections)
    return Gst.FlowReturn.OK


class MyPipelineApp(GStreamerApp):
    def __init__(self, app_callback, user_data, parser=None):
        parser = parser or get_pipeline_parser()
        handle_list_models_flag(parser, APP_NAME)
        args = parser.parse_args()
        super().__init__(args, user_data)

        self.hef_path = resolve_hef_path(args.hef_path, APP_NAME, self.arch)
        logger.info("HEF: %s", self.hef_path)

    def get_pipeline_string(self):
        return (
            SOURCE_PIPELINE(self.video_source, self.arch)
            + " ! "
            + INFERENCE_PIPELINE(
                hef_path=self.hef_path,
                batch_size=self.batch_size,
            )
            + " ! "
            + USER_CALLBACK_PIPELINE()
            + " ! "
            + DISPLAY_PIPELINE(video_sink=self.video_sink, sync=self.sync)
        )


def main():
    user_data = UserAppCallback()
    app = MyPipelineApp(app_callback, user_data)
    app.run()


if __name__ == "__main__":
    main()

Step 4: Validate

python3 .github/scripts/validate_app.py hailo_apps/python/pipeline_apps/my_pipeline_app --smoke-test

Critical Conventions

0. USB camera input: Always use --input usb for USB cameras — the framework auto-detects the correct device. NEVER hardcode /dev/video0 — that is often the integrated webcam, not the USB camera. If you need a specific device, run v4l2-ctl --list-devices first.
1. CLI parser: get_pipeline_parser() (NOT get_standalone_parser())
2. Pipeline composition: Use helper functions — SOURCE_PIPELINE, INFERENCE_PIPELINE, DISPLAY_PIPELINE
3. Callback: app_callback(element, buffer, user_data) — never call user_data.increment()
4. Resolution preservation: Use INFERENCE_PIPELINE_WRAPPER for full-res display
5. Tracking: TRACKER_PIPELINE() for ByteTrack
6. Cascaded inference: CROPPER_PIPELINE() for crop → second model
7. VAAPI: Add QUEUE("vaapi_queue") + vaapi_convert_pipeline for HW decode

Common Patterns

Frame Overlay Pattern (use_frame + OpenCV Drawing)

When you need to draw on frames (overlays, game graphics, custom visualizations), use the use_frame pattern:

1. Enable use_frame in your callback class

class UserAppCallback(app_callback_class):
    def __init__(self):
        super().__init__()
        self.use_frame = True  # Enables frame access in callback

CRITICAL: Setting use_frame = True in the callback class alone is NOT enough when subclassing a pipeline class (e.g., GStreamerPoseEstimationApp). GStreamerApp.__init__() overwrites user_data.use_frame from the CLI default (False). You MUST also force it in the app class:

class MyApp(GStreamerPoseEstimationApp):
    def __init__(self, app_callback, user_data, parser=None):
        super().__init__(app_callback, user_data, parser)
        self.options_menu.use_frame = True  # starts display process
        user_data.use_frame = True          # enables frame extraction

Without this, set_frame() calls are silently ignored and only the raw camera feed is shown.

2. Get the frame in the callback

import cv2
import hailo
from hailo_apps.python.core.common.buffer_utils import get_caps_from_pad, get_numpy_from_buffer

def app_callback(element, buffer, user_data):
    pad = element.get_static_pad("src")
    format, width, height = get_caps_from_pad(pad)

    frame = None
    if user_data.use_frame and format and width and height:
        # Signature: get_numpy_from_buffer(buffer, format, width, height)
        # Returns RGB numpy array (H, W, 3)
        frame = get_numpy_from_buffer(buffer, format, width, height)

3. Draw with OpenCV, convert RGB→BGR, then set_frame()

    if user_data.use_frame and frame is not None:
        # Draw on the frame (frame is RGB from GStreamer)
        cv2.circle(frame, (x, y), 10, (0, 255, 0), -1)
        cv2.putText(frame, "Hello", (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 255), 2)

        # CRITICAL: Convert RGB → BGR before set_frame (OpenCV expects BGR)
        frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
        user_data.set_frame(frame)

    return Gst.FlowReturn.OK

Key rules:

• get_numpy_from_buffer(buffer, format, width, height) — NOT (buffer, pad, format)
• Frame comes in RGB from GStreamer
• Must convert to BGR with cv2.cvtColor(frame, cv2.COLOR_RGB2BGR) before set_frame()
• Always check user_data.use_frame and that frame is not None
• Pass --use-frame on CLI to enable (or set self.use_frame = True in callback class)

Custom Background Pattern (games, virtual scenes)

When the user wants a custom background image (not the live camera feed), use background.copy() — never blend the camera feed with the background via addWeighted or similar.

# ✅ CORRECT — background only, no camera feed visible
if self.background is not None:
    output = self.background.copy()
else:
    output = np.zeros_like(frame)

# Draw game elements on output...
# Draw hand/body markers from pose data on output...

output = cv2.cvtColor(output, cv2.COLOR_RGB2BGR)
user_data.set_frame(output)

# ❌ WRONG — blends camera feed, user sees themselves + background = confusing
output = cv2.addWeighted(self.background, 0.4, frame, 0.6, 0)

Rule: If the user provides a background image or asks for a virtual scene, the camera feed is used only for pose/detection data extraction — it must NOT appear in the rendered output. The frame from get_numpy_from_buffer() is still needed to extract detections, but the display output should be background.copy() with game elements drawn on top.

Detection Data Extraction in Callbacks

Getting ROI and Detections

import hailo

def app_callback(element, buffer, user_data):
    roi = hailo.get_roi_from_buffer(buffer)
    detections = roi.get_objects_typed(hailo.HAILO_DETECTION)

    for detection in detections:
        label = detection.get_label()        # e.g., "person", "car"
        confidence = detection.get_confidence()  # 0.0 - 1.0
        bbox = detection.get_bbox()          # Normalized bounding box
        # bbox.xmin(), bbox.ymin(), bbox.width(), bbox.height() — all normalized [0,1]

Getting Track IDs (requires TRACKER_PIPELINE in pipeline)

        track_id = 0
        track = detection.get_objects_typed(hailo.HAILO_UNIQUE_ID)
        if len(track) == 1:
            track_id = track[0].get_id()

Getting Pose Landmarks (pose estimation models)

        landmarks = detection.get_objects_typed(hailo.HAILO_LANDMARKS)
        if landmarks:
            points = landmarks[0].get_points()
            # Each point has .x() and .y() — normalized to bounding box
            # Convert to pixel coordinates:
            for point in points:
                pixel_x = int((point.x() * bbox.width() + bbox.xmin()) * frame_width)
                pixel_y = int((point.y() * bbox.height() + bbox.ymin()) * frame_height)

Hailo Detection Types Reference

Reusing Existing Pipeline Classes

For apps that extend existing pipelines (e.g., a pose estimation game), subclass the domain-specific pipeline class instead of the base GStreamerApp:

from hailo_apps.python.pipeline_apps.pose_estimation.pose_estimation_pipeline import (
    GStreamerPoseEstimationApp,
)

class MyPoseGame(GStreamerPoseEstimationApp):
    """Inherits full pose pipeline: SOURCE → INFERENCE → TRACKER → USER_CALLBACK → DISPLAY"""
    pass  # Pipeline is already configured — just write your callback

def app_callback(element, buffer, user_data):
    # All pose detection data is available here
    ...

def main():
    user_data = UserAppCallback()
    app = MyPoseGame(app_callback, user_data)  # No pipeline config needed
    app.run()

Available pipeline classes to subclass:

When subclassing, you get the full pipeline for free — just provide your custom app_callback and app_callback_class.