reCamera Gimbal Tracking on IQ-9075 with QIM SDK Pipelines

Building a camera system that tracks things on its own usually means writing separate custom code for AI inference, hardware acceleration, and messaging, then piecing it all together. The result is a fragile setup that's hard to build, fine-tune, and maintain. Qualcomm's Intelligent Multimedia (QIM) SDK addresses this by exposing on-device AI inference, hardware-accelerated video processing, and even downstream messaging as native GStreamer plugins, letting a full perception pipeline be expressed as a single, declarative graph of standard elements rather than bespoke integration code. This project uses that approach to solve a common problem: reliably keeping a moving subject in frame without a human operator, by using the QIM SDK's GStreamer elements to run inference on-device and translate detections into physical camera motion, all within one composable pipeline.

Hardware

We are using a Qualcomm Dragonwing IQ-9075 EVK with a dedicated NPU for on-device AI (up to 100 TOPS AI performance).

We are using the reCamera Gimbal 2002w, featuring a dual-axis motorized mount powered by two brushless motors that support 360-degree yaw and 180-degree pitch motion. The reCamera can stream image data via standard protocols such as RTSP, HTTP, WebSocket, and TCP/UDP. Additionally, it supports MQTT for connecting with other IoT devices.

Setup Dragonwing IQ-9075 EVK

We utilized Qualcomm Launcher to flash Ubuntu 24.04 onto the IQ-9075 EVK.

Qualcomm Launcher enables configuration for network settings and SSH access.

Below are the installation steps for the QIM SDK.

1.Add Qualcomm PPA

$ sudo apt update
$ sudo add-apt-repository ppa:ubuntu-qcom-iot/qcom-ppa

2.Install GStreamer core components

$ sudo apt install -y gstreamer1.0-tools gstreamer1.0-plugins-good \
  gstreamer1.0-plugins-base gstreamer1.0-libav

3.Install Qualcomm hardware-accelerated GStreamer plugins

$ sudo apt install -y gstreamer1.0-plugins-qcom

4.Install AI runtime and inference components

$ sudo apt install -y tensorflow-lite-qcom-apps libqnn-dev libsnpe-dev

Verify that Qualcomm-specific GStreamer plugins are available after installation.

$ gst-inspect-1.0 | grep qti

qtibatch: qtibatch: Batching stream buffers
qtimetamux: qtimetamux: Meta muxer
qtimetatransform: qtimetatransform: Meta Transform
qtimlaclassification: qtimlaclassification: Machine Learning audio classification
qtimlaconverter: qtimlaconverter: Machine Learning Audio Converter
qtimldemux: qtimldemux: Batching stream buffers
qtimlmetaextractor: qtimlmetaextractor: Video mlmeta extractor
qtimlmetaparser: qtimlmetaparser: Meta parser
qtimlpostprocess: qtimlpostprocess: Machine Learning postprocess
qtimlqnn: qtimlqnn: QNN based ML plugin
qtimlsnpe: qtimlsnpe: SNPE Machine Learning
qtimltflite: qtimltflite: TFLite Machine Learning
qtimlvclassification: qtimlvclassification: Machine Learning image classification
qtimlvconverter: qtimlvconverter: Machine Learning Video Converter
qtimlvdetection: qtimlvdetection: Machine Learning image object detection
qtimlvpose: qtimlvpose: Machine Learning Pose
qtimlvsegmentation: qtimlvsegmentation: Machine Learning image segmentation
qtimlvsuperresolution: qtimlvsuperresolution: Machine Learning image super resolution
qtimsgpub: qtimsgpub: Message Publisher Client
qtimsgsub: qtimsgsub: Message Subscriber Client
qtiobjtracker: qtiobjtracker: Object Tracker
qtiqmmfsrc: qtiqmmfsrc: QMMF Video Source
qtiredissink: qtiredissink: QTI Redis Sink Element
qtirestrictedzonedbg: qtirestrictedzonedbg: Restricted Zone Filter
qtirtspbin: qtirtspbin: RTSP streaming Bin
qtismartvencbin: qtismartvencbin: Smart Video Encode Bin
qtisocketsink: qtisocketsink: QTI Socket Sink Element
qtisocketsrc: qtisocketsrc: QTI Socket Source Element
qtiuridecodebin: qtiuridecodebin: Generic bin
qtivcomposer: qtivcomposer: Video composer
qtivideotemplate: qtivideotemplate: Video template
qtivoverlay: qtivoverlay: Video Overlay
qtivsplit: qtivsplit: Video stream splitter
qtivtransform: qtivtransform: Video transformer

Setup reCamera Gimbal

Please refer to the reCamera Gimbal Getting Started Guide for instructions on assembly and setup. Out of the box, the device features a pre-installed operating system and a Node-RED dashboard accessible via the reCamera's IP address. Node-RED is a flow-based visual programming environment designed to simplify the integration of hardware, APIs, and web services. We switched the reCamera from its default access point mode to join the IQ-9075's network using the dashboard's network settings.

Data Acquisition

To get started, we registered an account on Edge Impulse Studio and created a new project. We then navigated to the Data Acquisition page, clicked the Connect a device link, and selected Connect to your computer to use a laptop camera. Using this setup, we collected 1, 222 images across various distances and lighting conditions, which we subsequently labeled via the Labeling Queue tab.

Model Training

For model development, we need to create an impulse, a custom pipeline integrating image processing and machine learning models. Navigate to the Impulse Design > Create Impulse page, select Add a processing block, and choose Image to preprocess and normalize image data. On the same page, select Add a learning block and choose Object Detection (Images). We are using a 320x320 image size. Then, click the Save Impulse button.

For feature generation, go to the Impulse > Image page set the Color depth parameter as RGB and click on the Save parameters button which redirects to another page where we should click on the Generate Feature button.

To train the model, navigate to the Impulse > ObjectDetection page. The Object detection settings we selected are displayed below.

We opted for the basic spatial and color space augmentations in the Advanced training settings.

We chose the latest YOLO-Pro model, ideal for detecting hands of varying sizes due to different distances from the camera, and capable of running efficiently on the IQ-9075's NPU. Click on the Save & train button to begin training.

Once training is finished, the training performance is shown as follows.

Model Deployment

To run inference on the IQ-9075's NPU, we selected the Qualcomm IM SDK GStreamer plugin (AARCH64, QNN) as the deployment target.

We selected the Quantized (int8) option for Model Optimizations.

Click the Build button to compile and download the Edge Impulse GStreamer plugin library bundle.

Application development

We replaced the default Node-RED flow on the reCamera dashboard that handles device initialization, video streaming, and gimbal actuation in a single graph. An inject node fires once at startup to run LED Init, setting the onboard LED to a known starting state. The camera node captures video at 1280x720 resolution and 30fps, feeding a dedicated stream node that publishes the feed over RTSP for external consumption. In parallel, an MQTT subscriber node listens on the "cmd" topic for incoming yaw and pitch angle commands published by the tracking system running on the IQ-9075 EVK.

Each incoming command first passes through a Validate function node, which checks the message before it's allowed to affect gimbal motion. Below is the validation code used, which can be added in the Edit dialog.

Once validated, the message is split two ways: one branch sends the yaw and pitch values directly to the Set Motors Pitch/Yaw node to physically reposition the gimbal, while the other triggers a 250ms pulse that blinks the onboard LED as a visual heartbeat confirming a valid command was received and acted on. The complete flow is shown below.

Unpack the downloaded Edge Impulse GStreamer plugin library bundle directly into the application directory on the IQ-9075 EVK. This bundle provides the shared library inside the lib folder alongside a config.json file required for application execution.

hands-qim-gstreamer-eim-qnn-v38-impulse-#11
├── config.json
├── lib
│   ├── libgstedgeimpulse-820595-hands.d
│   ├── libgstedgeimpulse-820595-hands.so
│   └── version.txt

We implemented a GStreamer pipeline that pulls video over RTSP, decodes it in hardware, crops and downsizes it to a 320×320 square, runs the Edge Impulse hand-detection model, draws the results, and re-encodes as JPEG for anything polling the appsink. The GStreamer pipeline description is provided below.

rtspsrc location=rtsp://192.168.3.10:554/live latency=0 protocols=tcp name=src
src. ! 
rtph264depay ! h264parse ! v4l2h264dec ! 
video/x-raw,format=NV12,width=1280,height=720 ! 
videocrop left=280 right=280 top=0 bottom=0 ! 
videoscale ! video/x-raw,format=NV12,width=320,height=320 ! videoconvert ! 
video/x-raw,format=RGB !
edgeimpulsevideoinfer_820595-hands threshold="40.minScore=0.5 debug=true" !
edgeimpulseoverlay_820595-hands show-labels=false !
videoconvert ! jpegenc quality=80 !
appsink name=jpegsink emit-signals=true sync=false max-buffers=1 drop=true

qtimsgpub name=msgpub protocol=mqtt host={MQTT_BROKER} port={MQTT_PORT} sync=false async=false

The GStreamer MQTT element operates outside that chain, publishing independently whenever it receives a signal from the application. The high-level pipeline is illustrated in the visualization below.

The tracking system running on the Qualcomm IQ-9075 is implemented as a Python script that builds a GStreamer pipeline using Qualcomm's IM SDK plugins alongside Edge Impulse's own GStreamer elements. The edgeimpulsevideoinfer plugin runs the inference, which uses the IQ-9075's NPU for fast, hardware-accelerated detection. Detection results are overlaid on the video via the edgeimpulseoverlay plugin and delivered as structured JSON messages on the pipeline's own GStreamer bus, avoiding any manual parsing of raw inference buffers.

A GStreamer bus handler reads these messages, filters detections by confidence, and selects the hand closest to the previously tracked position to keep the system locked onto a single target even when multiple hands appear in frame. The selected target's pixel offset from frame center is smoothed and converted into an angular error using the camera's field of view, then passed through independent PID controllers for the yaw and pitch axes to produce smooth, damped gimbal movements without oscillation. These angle commands are published back to the reCamera over MQTT using the IM SDK's qtimsgpub plugin, closing the loop between detection and actuation.

In addition to driving the gimbal, the annotated video is served as a live MJPEG stream over HTTP, allowing the detection results to be viewed in a browser in real time. The complete tracker code is given below.

import json
import os
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# GST_PLUGIN_PATH must be set before Gst.init()
EI_DEPLOY_DIR = os.path.expanduser(
    "./hands-qim-gstreamer-eim-qnn-v38-impulse-#11"
)
EI_LIB_PATH = os.path.join(EI_DEPLOY_DIR, "lib")
os.environ["GST_PLUGIN_PATH"] = EI_LIB_PATH + ":" + os.environ.get("GST_PLUGIN_PATH", "")

import gi

gi.require_version("Gst", "1.0")
from gi.repository import GLib, Gst  # noqa: E402

# Config
RECAMERA_IP = '192.168.3.10'
RTSP_PATH = f"rtsp://{RECAMERA_IP}:554/live"

MQTT_BROKER = RECAMERA_IP
MQTT_PORT = 1883
MQTT_TOPIC = "cmd"

# reCamera image stream resolution
SOURCE_W, SOURCE_H = 1280, 720
# model input dimension
FRAME_W, FRAME_H = 320, 320
_CROP_SIDE = min(SOURCE_W, SOURCE_H)        
_CROP_LR = (SOURCE_W - _CROP_SIDE) // 2    
_CROP_TB = (SOURCE_H - _CROP_SIDE) // 2     

# From config.json
EI_PLUGIN_VARIANT = "820595-hands"
EI_MODEL_INPUT_SIZE = 320  

VIDEO_INFER_ELEMENT = f"edgeimpulsevideoinfer_{EI_PLUGIN_VARIANT}"
OVERLAY_ELEMENT = f"edgeimpulseoverlay_{EI_PLUGIN_VARIANT}"

EI_THRESHOLD = "40.minScore=0.5"

# Camera field of view -- calculated from OV5647 sensor specs:
#   sensor width  = 2592px * 1.4um = 3.629mm
#   sensor height = 1944px * 1.4um = 2.722mm
#   FOV = 2*atan(dimension / (2*focal_length)), f=3.46mm
HFOV_DEG = 55.4 * (720 / 1280)  
VFOV_DEG = 42.9                

CONF_THRESHOLD_PCT = 70.0     
                             
# direction sign
YAW_SIGN = -1.0
PITCH_SIGN = 1.0
EMA_ALPHA = 0.15        
DEAD_ZONE_PX = 4
MAX_DELTA_DEG = 4.0     
PUBLISH_INTERVAL_S = 1.0 / 6.0   

# Per-axis PID gains. Error is expressed in degrees
YAW_KP, YAW_KI, YAW_KD = 0.35, 0.0, 0.15
PITCH_KP, PITCH_KI, PITCH_KD = 0.35, 0.0, 0.15
PID_INTEGRAL_MAX = MAX_DELTA_DEG * 2.0         

YAW_MIN, YAW_MAX = 30.0, 330.0                  
YAW_CENTER = (YAW_MIN + YAW_MAX) / 2.0         
PITCH_MIN, PITCH_MAX = 10.0, 120.0
PITCH_CENTER = (PITCH_MIN + PITCH_MAX) / 2.0  

TARGET_LOST_TIMEOUT_S = 1.0

DEBUG_PRINT_RAW = False   

state = {
    "yaw": YAW_CENTER,
    "pitch": PITCH_CENTER,
    "locked_label": None,
    "smoothed_cx": None,
    "smoothed_cy": None,
    "last_pub": 0.0,
    "last_seen": 0.0,
}
_debug_frames_left = {"n": 8}  


class PID:
    def __init__(self, kp, ki, kd, out_limit, integral_limit, d_filter_alpha=0.3):
        self.kp, self.ki, self.kd = kp, ki, kd
        self.out_limit = out_limit
        self.integral_limit = integral_limit
        self.d_filter_alpha = d_filter_alpha
        self.integral = 0.0
        self.prev_error = None
        self.prev_time = None
        self.filtered_derivative = 0.0

    def reset(self):
        self.integral = 0.0
        self.prev_error = None
        self.prev_time = None
        self.filtered_derivative = 0.0

    def update(self, error, now):
        dt = 0.0 if self.prev_time is None else max(1e-3, now - self.prev_time)

        self.integral += error * dt
        self.integral = max(-self.integral_limit, min(self.integral_limit, self.integral))

        if self.prev_error is not None and dt > 0:
            raw_derivative = (error - self.prev_error) / dt
            self.filtered_derivative = (
                self.d_filter_alpha * raw_derivative
                + (1 - self.d_filter_alpha) * self.filtered_derivative
            )

        output = self.kp * error + self.ki * self.integral + self.kd * self.filtered_derivative
        output = max(-self.out_limit, min(self.out_limit, output))

        self.prev_error = error
        self.prev_time = now
        return output


yaw_pid = PID(YAW_KP, YAW_KI, YAW_KD, MAX_DELTA_DEG, PID_INTEGRAL_MAX)
pitch_pid = PID(PITCH_KP, PITCH_KI, PITCH_KD, MAX_DELTA_DEG, PID_INTEGRAL_MAX)

class Detection:
    __slots__ = ("label", "confidence", "x", "y", "w", "h")

    def __init__(self, label, confidence, x, y, w, h):
        self.label = label
        self.confidence = confidence
        self.x, self.y, self.w, self.h = x, y, w, h

    @property
    def cx(self):
        return self.x + self.w / 2

    @property
    def cy(self):
        return self.y + self.h / 2


def parse_bus_result(result_json_str: str):
    try:
        data = json.loads(result_json_str)
    except (json.JSONDecodeError, TypeError):
        return []

    boxes = data.get("bounding_boxes") or []

    if boxes:
        print("RAW EI RESULT:", result_json_str)
        _debug_frames_left["n"] -= 1

    dets = []
    for b in boxes:
        try:
            confidence = float(b.get("value", 0.0)) * 100.0  
            if confidence < CONF_THRESHOLD_PCT:
                continue  
            dets.append(
                Detection(
                    label=b.get("label", "?"),
                    confidence=confidence,
                    x=float(b["x"]),
                    y=float(b["y"]),
                    w=float(b["width"]),
                    h=float(b["height"]),
                )
            )
        except (KeyError, TypeError, ValueError):
            continue
    return dets


def pick_target(detections):
    if not detections:
        return None

    if state["smoothed_cx"] is not None:
        candidates = detections
        if state["locked_label"] is not None:
            same_label = [d for d in detections if d.label == state["locked_label"]]
            if same_label:
                candidates = same_label

        def dist(d):
            return (d.cx - state["smoothed_cx"]) ** 2 + (d.cy - state["smoothed_cy"]) ** 2

        return min(candidates, key=dist)

    return max(detections, key=lambda d: d.confidence)


# Pipeline
Gst.init(None)

PIPELINE_DESC = f"""
rtspsrc location={RTSP_PATH} latency=0 protocols=tcp name=src
src. ! rtph264depay ! h264parse ! v4l2h264dec !
  video/x-raw,format=NV12,width={SOURCE_W},height={SOURCE_H} !
  videocrop left={_CROP_LR} right={_CROP_LR} top={_CROP_TB} bottom={_CROP_TB} !
  videoscale ! video/x-raw,format=NV12,width={FRAME_W},height={FRAME_H} !
  videoconvert ! video/x-raw,format=RGB !
  {VIDEO_INFER_ELEMENT} threshold="{EI_THRESHOLD} debug=true" !
  {OVERLAY_ELEMENT} show-labels=false !
  videoconvert ! jpegenc quality=80 !
  appsink name=jpegsink emit-signals=true sync=false max-buffers=1 drop=true

qtimsgpub name=msgpub protocol=mqtt host={MQTT_BROKER} port={MQTT_PORT} sync=false async=false
"""

pipeline = Gst.parse_launch(PIPELINE_DESC)
msgpub = pipeline.get_by_name("msgpub")

_latest_frame = {"data": None}
_frame_lock = threading.Lock()


def _on_new_jpeg_sample(sink):
    sample = sink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.OK
    buf = sample.get_buffer()
    ok, mapinfo = buf.map(Gst.MapFlags.READ)
    if ok:
        with _frame_lock:
            _latest_frame["data"] = bytes(mapinfo.data)
        buf.unmap(mapinfo)
    return Gst.FlowReturn.OK


jpegsink = pipeline.get_by_name("jpegsink")
jpegsink.connect("new-sample", _on_new_jpeg_sample)


class MJPEGHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Age", "0")
        self.send_header("Cache-Control", "no-cache, private")
        self.send_header("Pragma", "no-cache")
        self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame")
        self.end_headers()
        try:
            while True:
                with _frame_lock:
                    frame = _latest_frame["data"]
                if frame is not None:
                    self.wfile.write(b"--frame\r\n")
                    self.wfile.write(b"Content-Type: image/jpeg\r\n")
                    self.wfile.write(f"Content-Length: {len(frame)}\r\n\r\n".encode())
                    self.wfile.write(frame)
                    self.wfile.write(b"\r\n")
                time.sleep(1.0 / 15.0)
        except (BrokenPipeError, ConnectionResetError):
            pass

    def log_message(self, format, *args):
        pass


def _run_http_server():
    server = ThreadingHTTPServer(("0.0.0.0", 8080), MJPEGHandler)
    server.serve_forever()


threading.Thread(target=_run_http_server, daemon=True).start()


def handle_detections(detections):
    target = pick_target(detections)

    now = time.time()
    if target is None:
        if now - state["last_seen"] > TARGET_LOST_TIMEOUT_S:
            yaw_pid.reset()
            pitch_pid.reset()
        return

    state["last_seen"] = now
    state["locked_label"] = target.label

    if state["smoothed_cx"] is None:
        state["smoothed_cx"], state["smoothed_cy"] = target.cx, target.cy
    else:
        state["smoothed_cx"] = EMA_ALPHA * target.cx + (1 - EMA_ALPHA) * state["smoothed_cx"]
        state["smoothed_cy"] = EMA_ALPHA * target.cy + (1 - EMA_ALPHA) * state["smoothed_cy"]

    dx = state["smoothed_cx"] - FRAME_W / 2
    dy = state["smoothed_cy"] - FRAME_H / 2

    if abs(dx) < DEAD_ZONE_PX:
        dx = 0.0
    if abs(dy) < DEAD_ZONE_PX:
        dy = 0.0

    if now - state["last_pub"] < PUBLISH_INTERVAL_S:
        return

    yaw_error = YAW_SIGN * (dx / FRAME_W) * HFOV_DEG
    pitch_error = PITCH_SIGN * (dy / FRAME_H) * VFOV_DEG

    yaw_delta = yaw_pid.update(yaw_error, now)
    pitch_delta = pitch_pid.update(pitch_error, now)

    state["yaw"] = max(YAW_MIN, min(YAW_MAX, state["yaw"] + yaw_delta))
    state["pitch"] = max(PITCH_MIN, min(PITCH_MAX, state["pitch"] + pitch_delta))

    payload = json.dumps(
        {"yaw_angle": f"{state['yaw']:.1f}", "pitch_angle": round(state["pitch"], 1)}
    )
    msgpub.emit("add-publish", MQTT_TOPIC, payload)
    state["last_pub"] = now


def on_bus_message(bus, message, loop):
    t = message.type
    if t == Gst.MessageType.EOS:
        loop.quit()
    elif t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"GStreamer error: {err} ({debug})")
        loop.quit()
    elif t == Gst.MessageType.ELEMENT:
        structure = message.get_structure()
        if structure is not None and structure.get_name() == "edge-impulse-video-inference-result":
            result_json_str = structure.get_string("result")
            if result_json_str:
                detections = parse_bus_result(result_json_str)
                handle_detections(detections)
    return True


loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", on_bus_message, loop)

pipeline.set_state(Gst.State.PLAYING)
print("Annotated MJPEG stream: http://<IQ-9075-IP>:8080/")

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    pipeline.set_state(Gst.State.NULL)

Live Demo

The live demo shows the application tracking a hand in real time. Some oscillation is still visible in the frame, indicating the PID gains need further tuning, but the system remains usable in its current state.

Conclusion

This project demonstrates that pairing NPU-accelerated inference with a PID-based control loop yields smooth, real-time reCamera Gimbal tracking. The main benefits are faster development through reusable GStreamer elements, lower latency from on-device NPU inference, and a maintainable pipeline that's easy to extend. Future directions include leveraging the Edge Impulse output's unused object-tracking field for persistent multi-target tracking, retargeting the pipeline to other detection classes, and adding further actuators for more complex camera behaviors.