Get photos taken by ANAFI in real-time

ANAFI USA / ANAFI Thermal / ANAFI 4K
1

I refer to the following page, but I can’t get the image.
(URL: https://forum.developer.parrot.com/t/get-real-time-photos-from-the-anafi-in-sphinx/9966)

The program is as follows

from olympe.messages.camera import (
    set_camera_mode,
    set_photo_mode,
    take_photo,
    photo_progress,
)
import olympe
import os
import re
import requests
import shutil
import tempfile
import xml.etree.ElementTree as ET

import cv2
import numpy as np

# Drone IP
ANAFI_IP = "192.168.42.1"

# Drone web server URL
ANAFI_URL = "http://{}/".format(ANAFI_IP)

# Drone media web API URL
ANAFI_MEDIA_API_URL = ANAFI_URL + "api/v1/media/medias/"

XMP_TAGS_OF_INTEREST = (
    "CameraRollDegree",
    "CameraPitchDegree",
    "CameraYawDegree",
    "CaptureTsUs",
    # NOTE: GPS metadata is only present if the drone has a GPS fix
    # (i.e. they won't be present indoor)
    "GPSLatitude",
    "GPSLongitude",
    "GPSAltitude",
)

def take_photo_burst(drone):
    # take a photo burst and get the associated media_id
    photo_saved = drone(photo_progress(result="photo_saved", _policy="wait"))
    drone(take_photo(cam_id=0)).wait()
    photo_saved.wait()
    media_id = photo_saved.received_events().last().args["media_id"]

    # download the photos associated with this media id
    media_info_response = requests.get(ANAFI_MEDIA_API_URL + media_id)
    media_info_response.raise_for_status()
    download_dir = tempfile.mkdtemp()
    for resource in media_info_response.json()["resources"]:
        image_response = requests.get(ANAFI_URL + resource["url"], stream=True)
        download_path = os.path.join(download_dir, resource["resource_id"])
        mage_response.raise_for_status()
        with open(download_path, "wb") as image_file:
            shutil.copyfileobj(image_response.raw, image_file)

        # parse the xmp metadata
        with open(download_path, "rb") as image_file:
            image_data = image_file.read()

            image = cv2.imread(str(image_file))
            cv2.namedWindow('image', cv2.WINDOW_NORMAL)
            cv2.imshow("Photo", image)
            cv2.waitKey(0)
            cv2.destroyAllWindows()

            image_xmp_start = image_data.find(b"<x:xmpmeta")
            image_xmp_end = image_data.find(b"</x:xmpmeta")
            image_xmp = ET.fromstring(image_data[image_xmp_start : image_xmp_end + 12])
            for image_meta in image_xmp[0][0]:
                xmp_tag = re.sub(r"{[^}]*}", "", image_meta.tag)
                xmp_value = image_meta.text
                # only print the XMP tags we are interested in
                if xmp_tag in XMP_TAGS_OF_INTEREST:
                    print(resource["resource_id"], xmp_tag, xmp_value)


def setup_photo_burst_mode(drone):
drone(set_camera_mode(cam_id=0, value="photo")).wait()
# For the file_format: jpeg is the only available option
# dng is not supported in burst mode
drone(
    set_photo_mode(
        cam_id=0,
        mode="burst",
        format="rectilinear",
        file_format="jpeg",
        burst="burst_14_over_1s",
        bracketing="preset_1ev",
        capture_interval=0.0,
        )
    ).wait()


def main(drone):
    drone.connection()
    setup_photo_burst_mode(drone)
    take_photo_burst(drone)


if __name__ == "__main__":
    with olympe.Drone(ANAFI_IP) as drone:
        main(drone)

I want to get a “.jpg” file.