Cannot access video stream via RTSP or Olympe QR scanning

ANAFI Ai
1

Description:
I’m working with an ANAFI Ai and trying to scan QR codes from the live video feed using OpenCV + Olympe. The same script worked on another drone

What I’ve tried:

  • RTSP via OpenCV/FFMPEG:
    Using cv2.VideoCapture("rtsp://192.168.42.1:554/live", cv2.CAP_FFMPEG) times out after ~30 seconds with:
VIDEOIO(FFMPEG): backend is generally available but can't be used to capture by name
Stream timeout triggered after 30000 ms
  • RTSP via GStreamer (gst-launch-1.0):
    Command:
gst-launch-1.0 -v rtspsrc location=rtsp://192.168.42.1:554/live protocols=tcp latency=200 ! \
  rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! autovideosink

Fails with Could not receive message. (Timeout while waiting for server response).

  • Olympe streaming (drone.streaming.get_latest_frame()):
    After enabling media streaming with:
drone(VideoEnable(1)).wait()
drone(VideoEnableChanged(enabled=1)).wait(_timeout=5)
drone.streaming.start()

I get an AttributeError: 'Pdraw' object has no attribute 'is_running' if I check .is_running.
If I drop that and call .start() directly, I don’t get errors, but I also don’t receive frames (window stays black / get_latest_frame() returns None).

Environment:

  • Linux (Ubuntu with conda Python 3.12)
  • Olympe installed from pip
  • OpenCV 4.x with FFMPEG and GStreamer enabled
  • Connected via Wi-Fi direct (legacy mode shows in logs)

Logs:

  • RTSP: Timeout during SETUP stage (Could not receive message)
  • Olympe: Connection fine, telemetry received, but no video frames.
my code:
import olympe
from olympe.messages.ardrone3.Piloting import TakeOff, Landing, moveBy
from olympe.messages.ardrone3.PilotingState import FlyingStateChanged
from olympe.messages.ardrone3.PilotingSettings import MaxTilt

import time
import threading
import cv2
import keyboard
import os
import numpy as np

DRONE_IP = os.environ.get("DRONE_IP", "192.168.42.1")
DRONE_RTSP_PORT = os.environ.get("DRONE_RTSP_PORT", "554")
RTSP_URL = f"rtsp://{DRONE_IP}:{DRONE_RTSP_PORT}/live"

drone_should_land = False
drone_should_hover = False


def keyboard_listener():
    global drone_should_land, drone_should_hover
    while True:
        if keyboard.is_pressed("q"):
            print("[KEYBOARD] Emergency land triggered (q pressed)!")
            drone_should_land = True
            break
        elif keyboard.is_pressed("s"):
            print("[KEYBOARD] Hover triggered (s pressed)!")
            drone_should_hover = True


def safe_move(drone, dx, dy, dz, dpsi):
    global drone_should_land, drone_should_hover
    if drone_should_land:
        print("[MOVE] Land requested. Skipping movement.")
        return
    if drone_should_hover:
        print("[MOVE] Hover requested. Holding position...")
        drone(FlyingStateChanged(state="hovering", _timeout=5)).wait()
        while not drone_should_land:
            time.sleep(0.5)
        return

    print(f"[MOVE] Executing moveBy: dx={dx}, dy={dy}, dz={dz}, dpsi={dpsi}")
    move = drone(
        moveBy(dx, dy, dz, dpsi)
        >> FlyingStateChanged(state="hovering", _timeout=10)
    ).wait()
    if not move.success():
        print("[MOVE] Movement failed!")


def scan_qr_for_20_seconds(drone):
    print("[QR] Connecting to drone video stream...")
    cap = cv2.VideoCapture(RTSP_URL, cv2.CAP_FFMPEG)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 3)

    if not cap.isOpened():
        print("[QR] Failed to open video stream")
        return []

    detector = cv2.QRCodeDetector()
    qr_found = []
    start_time = time.time()

    while time.time() - start_time < 20:
        ret, frame = cap.read()
        if not ret:
            continue

        data, bbox, _ = detector.detectAndDecode(frame)

        if bbox is not None and len(bbox) > 0:
            bbox = np.int32(bbox)
            for i in range(len(bbox[0])):
                pt1 = tuple(bbox[0][i])
                pt2 = tuple(bbox[0][(i + 1) % len(bbox[0])])
                cv2.line(frame, pt1, pt2, (0, 255, 0), 2)
            if data:
                cv2.putText(frame, data, (int(bbox[0][0][0]), int(bbox[0][0][1]) - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)

        if data and data not in [q[0] for q in qr_found]:
            print(f"[QR] QR Code detected: {data}")
            qr_found.append((data, time.time() - start_time))
            break

        # ✅ Show live drone video feed with QR overlay
        cv2.imshow("Drone QR Debug View", frame)
        if cv2.waitKey(1) & 0xFF == ord("q") or drone_should_land:
            break

    cap.release()
    cv2.destroyAllWindows()
    return qr_found


def main():
    global drone_should_land, drone_should_hover

    print("[MAIN] Connecting to drone...")
    drone = olympe.Drone(DRONE_IP)
    drone.connect()

    print("[MAIN] Starting keyboard listener...")
    listener_thread = threading.Thread(target=keyboard_listener, daemon=True)
    listener_thread.start()

    # Step 5: Scan for QR
    print("[MAIN] Scanning for QR code...")
    qr_found = scan_qr_for_20_seconds(drone)

    if qr_found:
        print(f"[MAIN] QR code(s) detected: {qr_found}")
    else:
        print("[MAIN] No QR code found within time limit.")

   
    print("[MAIN] Drone landed.")

    drone.disconnect()
 

if __name__ == "__main__":
    main()
2

Hello!

First, I do not get where you pick that is_running from ; it is actually not from Pdraw API and I can’t see from where you would have picked that
Then, I do not see in your code where you instance your Renderer. Without it, you won’t have any streaming rendering
If you’re still on Olympe 7.7, I suggest you check the example doc/examples/streaming.py

# Start video streaming
self.drone.streaming.start()
self.renderer = PdrawRenderer(pdraw=self.drone.streaming)
3

Thank you for pointing that out! You were right — is_running isn’t part of the Pdraw API, and initially I also wasn’t instantiating a renderer.

In the end, I solved it by switching away from relying on is_running and the default PdrawRenderer. Instead, I followed the doc/examples/streaming.py pattern and registered streaming callbacks (raw_cb for decoded YUV frames, h264_cb for stats, flush_raw_cb for cleanup). From the raw frames I convert YUV → BGR with OpenCV and run my QR detector there.

I also switched the drone to H.265 (HEVC), installed the extra ffmpeg codecs (libavcodec-extra), and added a “first-frame probe” to confirm decoding starts. Finally, I handle the OpenCV preview window entirely from the main thread, and only stop the pipeline from there after a timeout — this avoided the freezes and Pdraw.destroy() errors I was hitting earlier.

So in short: removing the unsupported is_running, moving to the callback approach instead of the renderer, and adjusting the shutdown order fixed the streaming and let me do live QR detection successfully. Thanks again for steering me back to the example!


Also, here is my sample code for a simple QR detection. Can you give me your feedback if you have time?

#!/usr/bin/env python3
import csv
import os
import queue
import tempfile
import threading
import time

import cv2
import numpy as np
import olympe

# Quieter logs
olympe.log.update_config({"loggers": {"olympe": {"level": "WARNING"}}})

DRONE_IP = os.environ.get("DRONE_IP", "192.168.42.1")
DRONE_RTSP_PORT = os.environ.get("DRONE_RTSP_PORT")  # e.g., "554"


class QRScanWithPreview:
    """
    - NO takeoff/movement
    - OpenCV preview window with QR overlay
    - Scans for full timeout_s seconds, collects ALL detections
    - Press 'q' to stop early
    """

    def __init__(self, timeout_s: int = 40, show_window: bool = True, window_name: str = "ANAFI QR"):
        self.drone = olympe.Drone(DRONE_IP)
        self.tempd = tempfile.mkdtemp(prefix="olympe_streaming_qr_")
        print(f"[INFO] Output dir: {self.tempd}")

        # Optional stats file
        self.h264_frame_stats = []
        self.h264_stats_file = open(os.path.join(self.tempd, "h264_stats.csv"), "w+")
        self.h264_stats_writer = csv.DictWriter(self.h264_stats_file, ["fps", "bitrate"])
        self.h264_stats_writer.writeheader()

        # Decoded frames pipeline
        self.frame_queue = queue.Queue()
        self.processing_thread = threading.Thread(target=self._yuv_frame_processing, daemon=True)

        # Shared preview buffer (shown from main thread)
        self.last_bgr = None
        self.last_lock = threading.Lock()

        self.running = False
        self.timeout_s = timeout_s
        self.qr_detector = cv2.QRCodeDetector()
        self.detected = set()                 # <- collect all QR strings
        self.stop_event = threading.Event()
        self._timeout_timer = None

        self.show_window = show_window
        self.window_name = window_name
        self.first_frame_evt = threading.Event()

    # ---------- lifecycle ----------

    def start(self):
        assert self.drone.connect(retry=3), "Failed to connect to drone"
        if DRONE_RTSP_PORT:
            self.drone.streaming.server_addr = f"{DRONE_IP}:{DRONE_RTSP_PORT}"

        # Optional: record stream
        self.drone.streaming.set_output_files(
            video=os.path.join(self.tempd, "streaming.mp4"),
            metadata=os.path.join(self.tempd, "streaming_metadata.json"),
        )

        # Register callbacks
        self.drone.streaming.set_callbacks(
            raw_cb=self._yuv_frame_cb,
            h264_cb=self._h264_frame_cb,
            flush_raw_cb=self._flush_cb,
        )

        self.drone.streaming.start()
        self.running = True
        self.processing_thread.start()

        if not self.first_frame_evt.wait(timeout=5.0):
            print("[ERR] No decoded YUV frames in 5s. Check codec install or switch camera to H.265.")

        # Watchdog timeout
        self._timeout_timer = threading.Timer(self.timeout_s, self._on_timeout)
        self._timeout_timer.daemon = True
        self._timeout_timer.start()

    def stop(self):
        if self._timeout_timer is not None:
            self._timeout_timer.cancel()

        # signal worker to finish
        self.running = False

        # Clear callbacks first (prevents new frames during teardown)
        try:
            self.drone.streaming.set_callbacks(raw_cb=None, h264_cb=None, flush_raw_cb=None)
        except Exception:
            pass

        time.sleep(0.3)  # small drain

        if self.processing_thread.is_alive():
            self.processing_thread.join(timeout=2.0)

        # Drain any queued frames
        try:
            while not self.frame_queue.empty():
                self.frame_queue.get_nowait().unref()
        except Exception:
            pass

        # Stop streaming
        try:
            self.drone.streaming.stop()
        except Exception:
            pass
        time.sleep(0.3)

        # Disconnect / close
        try:
            self.drone.disconnect()
        finally:
            try:
                self.h264_stats_file.close()
            except Exception:
                pass

        # Close window from main thread
        if self.show_window:
            try:
                cv2.destroyWindow(self.window_name)
            except Exception:
                pass

    # ---------- callbacks ----------

    def _yuv_frame_cb(self, yuv_frame):
        yuv_frame.ref()
        self.frame_queue.put_nowait(yuv_frame)
        self.first_frame_evt.set()

    def _flush_cb(self, stream):
        if stream.get("vdef_format") != olympe.VDEF_I420:
            return True
        try:
            while not self.frame_queue.empty():
                self.frame_queue.get_nowait().unref()
        except Exception:
            pass
        return True

    def _h264_frame_cb(self, h264_frame):
        # 1s sliding window bitrate/FPS 
        _, frame_size = h264_frame.as_ctypes_pointer()
        info = h264_frame.info()
        ts = info["ntp_raw_timestamp"]
        if not bool(info["is_sync"]):
            while self.h264_frame_stats and (self.h264_frame_stats[0][0] + 1e6) < ts:
                self.h264_frame_stats.pop(0)
            self.h264_frame_stats.append((ts, frame_size))
            fps = len(self.h264_frame_stats)
            bitrate = 8 * sum(sz for _, sz in self.h264_frame_stats)
            self.h264_stats_writer.writerow({"fps": fps, "bitrate": bitrate})

    # ---------- processing (worker thread) ----------

    def _yuv_frame_processing(self):
        cvt_map = {
            olympe.VDEF_I420: cv2.COLOR_YUV2BGR_I420,
            olympe.VDEF_NV12: cv2.COLOR_YUV2BGR_NV12,
        }

        while self.running and not self.stop_event.is_set():
            try:
                yuv_frame = self.frame_queue.get(timeout=0.1)
            except queue.Empty:
                continue

            try:
                fmt = yuv_frame.format()
                if fmt not in cvt_map:
                    continue

                yuv_np = yuv_frame.as_ndarray()
                bgr = cv2.cvtColor(yuv_np, cvt_map[fmt])

                # Detect QR (keep scanning; don't stop here)
                data, bbox, _ = self.qr_detector.detectAndDecode(bgr)
                if bbox is not None and len(bbox) > 0:
                    bb = np.int32(bbox)
                    for i in range(len(bb[0])):
                        p1 = tuple(bb[0][i]); p2 = tuple(bb[0][(i+1) % len(bb[0])])
                        cv2.line(bgr, p1, p2, (0, 255, 0), 2)
                if data:
                    if data not in self.detected:
                        self.detected.add(data)
                        print(f"[QR] Detected: {data}")
                    cv2.putText(bgr, data, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 255, 0), 2)

                # Publish latest frame for preview (main thread shows it)
                if self.show_window:
                    with self.last_lock:
                        self.last_bgr = bgr
            finally:
                yuv_frame.unref()

    # ---------- helpers ----------

    def _on_timeout(self):
        self.stop_event.set()

    def scan(self):
        self.start()
        start = time.time()
        try:
            while not self.stop_event.is_set():
                # Preview / UI loop in main thread
                if self.show_window:
                    with self.last_lock:
                        frame = None if self.last_bgr is None else self.last_bgr.copy()
                    if frame is not None:
                        cv2.imshow(self.window_name, frame)
                        if cv2.waitKey(1) & 0xFF == ord('q'):
                            self.stop_event.set()
                            break
                # Timeout check (redundant to timer, but cheap)
                if time.time() - start >= self.timeout_s:
                    self.stop_event.set()
                time.sleep(0.01)
        finally:
            self.stop()
        return list(self.detected)


if __name__ == "__main__":
    scanner = QRScanWithPreview(timeout_s=40, show_window=True)
    results = scanner.scan()
    if results:
        print(f"[RESULT] QR codes detected: {results}")
    else:
        print("[RESULT] No QR detected within timeout.")
4

would be great if you could help me figure out how to detect smaller QR codes (probably there is a better approach than the one that i used)

5

This topic was automatically closed 3 days after the last reply. New replies are no longer allowed.