dronemaster 1.2.1

A modern alternative to the robomaster SDK

Dronemaster

a modern alternative to the Robomaster SDK for controlling DJI Tello EDU drones.
This sdk is a result of my frustration with the official robomaster sdk and how other libraries handle specific things.
It is based on the official tello 3.0 protocol documentation, but is extended with undocumented features that were discovered by experiments.
It harnesses the power of pythons async capabilities and also allows to control multiple drones at once.

Basics

This section describes how the drone works, so you are able to efficiently work with this sdk and avoid common pitfalls.
Firstly the drone can operate in two basic modes which can be selected with the slider on the side of the extension module:

• ap-mode: the drone connects to a preexisting wifi
• sta-mode: the drone creates its own wifi

When operating in ap-mode and using static dhcp entries, the drone sometimes receives a wrong ip

The protocol is based on udp, so packet-loss due to an unstable connection might lead to timeouts (this sdk tries its best to mitigate this, but it is not always successful).
Sometimes the error is not due to the programmer or the sdk, but simply due to the drone itself.
The drone has been observed as reporting ok to a takeoff-request but then staying firmly on the ground even with enough charge, so also account for such failures.

Specs

The drone is equipped with some sensors that might be interesting.

• gyro: to measure roll, pitch and yaw in degrees
• accelerometer: to measure the acceleration in x, y and z direction relative to the current orientation. Note that you should remove gravity from these by using the current orientation.
• barometer: to estimate the elevation of the drone in meters
• downwards time-of-flight sensor: to accurately measure the height above ground
• forwards time-of-flight sensor: to measure the distance to an object (located on the extension module)
• internal temperature: measures the internal temperature (above ~100°C it no longer takes off and above ~105°C it shuts down)
• front camera: positioned at an angle of ~13° downwards that records 720P or 480P in H264
• downwards camera: black and white camera without an IR-filter
• velocity calculation: the internal chip calculates the velocity in x, y and z direction in global space. It uses the downwards camera, so if there is not enough detail in its picture these speeds are 0
• mission-pad detection: this is currently not usable with this sdk

Installing

To install this sdk, simply use pip
pip install dronemaster
or when using the simulator
pip install dronemaster[simulator]

Simulator

This sdk provides a crude simulator so anybody without a drone is able to test parts of their program.
First run pip install dronemaster[simulator] to install the required dependencies.
Then start the simulator with python3 -m dronemaster [IP] where [IP] is the ip-address to listen on.
You can now connect to your own ip as if it would be a drone.

Api overview

The exact behavior of each function is documented in docstrings. This section should only give a very basic overview.

Logging

The sdk uses the python logging module, so you are able to configure it to your liking.

• DEBUG logs exactly what raw packets are sent and received (even repeated ones)
• INFO only logs successful commands and their response
• WARN logs unexpected data
• ERROR logs invalid answers to commands and timeouts

To access the logger, use dronemaster.command_logger

Connecting

First, create the dronemaster.Drone object with the target-ip.
Then call the drone.initialize() function to connect with it.
You should periodically call the drone.keepalive() function to stop the drone from automatically landing.

Flying

To control the flight, use the flight object received by drone.flight.
To learn more, visit the examples or inspect it in your IDE

Extension module

Access the top-rgb-led via drone.rgb and the matrix via drone.matrix

Configuring the video

To configure the video (like resolution, fps, bitrate and selected camera) use drone.video

Configuring the drone

You can configure the drone by using functions like drone.set_own_wifi(ssid, passwd) or drone.set_connecting_wifi(ssid, passwd)

Receiving state information

The last state object is always available at drone.last_state
Register a callback to be called on a new packet with drone.state_subscribe(async_callback) where async_callback is an async function which receives one parameter with type dronemaster.DroneState

Examples

These are the two most important ones. See more here

Simple flight

import asyncio
import dronemaster

import logging
logging.basicConfig(format="%(asctime)s [%(name)s] [%(levelname)s] %(message)s") # dronemaster uses the logging module internally
dronemaster.command_logger.setLevel(logging.WARNING) # do not log raw commands
# DEBUG logs exactly what commands are sent and received (even repeated ones)
# INFO only logs successful commands and their response
# WARN logs unexpected data
# ERROR logs invalid answers to commands and timeouts

async def main():
    # create the drone, but don't connect
    ep_drone = dronemaster.Drone("127.0.0.1")

    # mark the drone as active and connect to it
    await ep_drone.initialize()

    serial = await ep_drone.serial_number()
    battery = await ep_drone.battery()
    print(f"Connected with drone '{serial}' with {battery}%")

    ep_flight = ep_drone.flight

    # takeoff and wait for completion (up to 20s)
    await ep_flight.takeoff()

    # fly forwards 20cm
    await ep_flight.forward(20, timeout=7)

    # land again
    await ep_flight.land()

if __name__ == '__main__':
    asyncio.run(main())

Video receiving

import asyncio
import dronemaster
import cv2

import logging
logging.basicConfig(format="%(asctime)s [%(name)s] [%(levelname)s] %(message)s") # dronemaster uses the logging module internally
dronemaster.command_logger.setLevel(logging.WARNING) # do not log raw commands

async def main():
    # create the drone, but don't connect
    ep_drone = dronemaster.Drone("127.0.0.1")

    # mark the drone as active and connect to it
    await ep_drone.initialize()

    serial = await ep_drone.serial_number()
    battery = await ep_drone.battery()
    print(f"Connected with drone '{serial}' with {battery}%")

    await ep_drone.video.set_video_port(11112) # per default, port 11111 is used

    await ep_drone.video.streamon()

    # uncomment to use the downwards black and white camera
    #await ep_drone.video.downvision(True)

    # listen on all ips on port 11112
    # the drone sends the video in H264 format with an I-Frame every few seconds
    cap = cv2.VideoCapture("udp://0.0.0.0:11112")

    while True:
        if cap.isOpened():
            ret, frame = cap.read()
            cv2.imshow('live stream', frame)
            await asyncio.sleep(1/60) # important to allow the sdk to read new commands
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    
    cap.release()
    cv2.destroyAllWindows()

    await ep_drone.video.streamoff()
    

if __name__ == '__main__':
    asyncio.run(main())

Api documentation

• dronemaster.Drone
  • DroneState
    • pitch
    • roll
    • yaw
    • vgx
    • vgy
    • vgz
    • bat
    • templ
    • temph
    • tof
    • h
    • time
    • agx
    • agy
    • agz
    • baro
    • last_update
    • delta
  • Drone
    • state_subscribe
    • initialize
    • serial_number
    • battery
    • flight_time
    • sdk_version
    • get_hardware
    • wifi_serial
    • wifi_version
    • ext_tof
    • tof
    • keepalive
    • set_own_wifi
    • set_connecting_wifi
    • send_raw_command
    • reboot
    • disconnect
  • Video
    • streamon
    • streamoff
    • downvision
    • setfps
    • setbitrate
    • setresolution
    • set_video_port
  • Flight
    • takeoff
    • forward
    • back
    • up
    • down
    • left
    • right
    • clockwise
    • counterclockwise
    • land
    • stop
    • emergency
    • motoron
    • motoroff
    • flip
    • throwfly
    • speed
    • rc
  • RGBLed
    • set
    • pulse
    • flash
  • Matrix
    • set_brightness
    • set_pattern

dronemaster.Drone

DroneState Objects

class DroneState(TypedDict)

The current state of the drone, reported ~10x per second.

Notes:

• the velocities are calculated by using the downwards camera, so when flying low or above featureless ground, it may be wrongly reported as 0

pitch

pitch in degrees

roll

roll in degrees

yaw

yaw in degrees

vgx

velocity in x-direction (forwards/backwards) in dm/s

vgy

velocity in y-direction (left/right) in dm/s

vgz

velocity in z-direction (up/down) in dm/s

bat

battery percentage

templ

the lower range of the internal temperature sensor in °C

temph

the upper range of the internal temperature sensor in °C

tof

the vertical distance to ground in cm. reported as 10 if out of range

h

the calculated height, note that this only works in flight

time

number of seconds the drone has been flying

agx

acceleration in x-direction in cm/s²

agy

acceleration in y-direction in cm/s²

agz

acceleration in z-direction in cm/s²

baro

Height above sea level as reported by the barometer in m

last_update

unix timestamp of the last update

delta

time in seconds between the last two state packets

Drone Objects

class Drone()

The main class to control the drone.

Example:

drone = Drone(DRONE_IP)
await drone.initialize()

# calls my_callback_function ~10x per second
drone.state_subscribe(my_callback_function)

# you should call this every few seconds to know if the drone has disconnected
await drone.keepalive()

# everything flight-related
flight = drone.flight

# control the rgb-led on top of the extension-module
rgb = drone.rgb

# control the extension-modules led-matrix
matrix = drone.matrix

# everything video-related
video = drone.video

state_subscribe

def state_subscribe(callable: Callable[[DroneState], Coroutine[Any, Any,
                                                               None]])

Adds the callable to the list of functions to be called when a new state is available

Arguments:

• callable: The async function to be called when a new state-packet is received

initialize

async def initialize() -> None

Initializes the drone connection and optionally starts the communication channel.

This function must be called before any other

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 2.5s

serial_number

async def serial_number() -> str

Queries the drone for its serial number in the format [A-Z0-9]{14}

Raises:

• ProtocolError: raised when not receiving the correct format
• TimeoutError: raised when not answering after 5s

Returns:

the serial number in the format [A-Z0-9]{14}

battery

async def battery() -> int

Queries the drone for its current battery charce percentage from 0-100

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

the battery percentage

flight_time

async def flight_time() -> int

Gets the time the motors have been running

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

the motor-time in seconds

sdk_version

async def sdk_version() -> int

Gets current sdk version (sometimes reported as 20 or 30)

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

the sdk version

get_hardware

async def get_hardware() -> Literal["TELLO", "RMTT"]

Checks if the drone is connected to the extension module

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

RMTT if connected to the extension module, else TELLO

wifi_serial

async def wifi_serial() -> str

Gets the wifi serial number

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

the wifi serial number

wifi_version

async def wifi_version() -> str

Gets the wifi version

Raises:

• ProtocolError: raised when not receiving the battery percentage
• TimeoutError: raised when not answering after 5s

Returns:

the wifi version in the format wifivx.x.x.x

ext_tof

async def ext_tof()

Reads the horizontal distance of the drone to the front.

Raises:

• ProtocolError: raised when not receiving data in the correct format
• TimeoutError: raised when not answering after 2.5s

Returns:

the horizontal distance to the front in mm or None if out of range

tof

async def tof() -> Optional[int]

Reads the vertical distance of the drone to ground.

Raises:

• ProtocolError: raised when not receiving data in the correct format
• TimeoutError: raised when not answering after 2.5s

Returns:

the vertical distance to ground in mm or None if out of range

keepalive

async def keepalive() -> bool

Sends a "ping" packet to the drone to stop it from landing automatically, but only if there is no currently waiting command

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 2.5s

Returns:

True when the ping was successful, False when it was skipped

set_own_wifi

async def set_own_wifi(ssid: str, pwd: str) -> None

Sets the ssid/password to use when the drone creates its own wifi.

Note that the drone adds a RMTT- or TELLO- prefix to the ssid. The drone reboots after 3s, so you have to manually reconnect

Arguments:

• ssid: the drone-wifis ssid
• pwd: the password

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

set_connecting_wifi

async def set_connecting_wifi(ssid: str, pwd: str) -> None

Sets the ssid/password of an existing wifi to connect to.

Note that the drone sometimes receives other/incorrect ip-adresses. The drone reboots after 3s, so you have to manually reconnect

Arguments:

• ssid: the wifis ssid
• pwd: the password

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

send_raw_command

async def send_raw_command(command,
                           wait_for_answer: bool = True,
                           timeout: float = 1) -> Optional[str]

Sends a raw command to the drone. Intended for debugging/manual mode

Arguments:

• wait_for_answer: if the code expects an answer
• timeout: when expecting an answer, how long to wait

Raises:

• ProtocolError: should never happen
• TimeoutError: raised when not answering after 5 * timeout

Returns:

the answer or None if no answer is expected

reboot

def reboot() -> None

Reboots the drone and closes the connection, so you have to manually reconnect with it

disconnect

def disconnect() -> None

disconnect from the drone

Video Objects

class Video(Module)

streamon

async def streamon() -> None

Starts the stream so the drone sends the H264 stream to port 11111

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

streamoff

async def streamoff() -> None

Stops the video stream

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

downvision

async def downvision(on: bool) -> None

Switches the used camera.

Arguments:

• on: if the downwards camera should be used

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

setfps

async def setfps(fps: Literal["high", "middle", "low"]) -> None

Sets the streams fps to

• high -> 30fps
• middle -> 15fps
• low -> 5fps

Arguments:

• fps: the requested fps

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

setbitrate

async def setbitrate(bitrate: Literal["auto", 1, 2, 3, 4, 5]) -> None

Sets the streams maximum bitrate to auto or the requested Mbps

Arguments:

• bitrate: the bitrate in Mbps or auto

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

setresolution

async def setresolution(resolution: Literal["high", "low"]) -> None

Sets the video streams resolution

• high -> 720P
• low -> 480P

Arguments:

• resolution: the requested resolution

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

set_video_port

async def set_video_port(port: int) -> None

Sets the port the drone should send the video stream

Arguments:

• port: the port to send the stream to [1024,65535]

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

Flight Objects

class Flight(Module)

takeoff

async def takeoff() -> None

Tries to takeoff

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 20s

forward

async def forward(dist: int, timeout: float = 5) -> None

The drone flies dist cm forwards.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

back

async def back(dist: int, timeout: float = 5) -> None

The drone flies dist cm backwards.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

up

async def up(dist: int, timeout: float = 5) -> None

The drone flies dist cm upwards.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

down

async def down(dist: int, timeout: float = 5) -> None

The drone flies dist cm downwards.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

left

async def left(dist: int, timeout: float = 5) -> None

The drone flies dist cm to the left.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

right

async def right(dist: int, timeout: float = 5) -> None

The drone flies dist cm to the right.

Arguments:

• dist: distance in cm in the range [20, 500]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

clockwise

async def clockwise(angle: int, timeout: float = 5) -> None

The drone rotates angle degrees clockwise.

Arguments:

• angle: angle in degrees in the range [1, 360]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

counterclockwise

async def counterclockwise(angle: int, timeout: float = 5) -> None

The drone rotates angle degrees counterclockwise.

Arguments:

• angle: angle in degrees in the range [1, 360]
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

land

async def land() -> None

Tries to land the drone.

Raises:

• ProtocolError: raised when not receiving ok (eg. the drone is not in the air)
• TimeoutError: raised when not answering after 20s

stop

async def stop() -> None

Immediately stops the drones movement and hovers

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

emergency

async def emergency() -> None

Immediately stops all motors and lets the drone fall out of the sky

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

motoron

async def motoron() -> None

Enters motoron-mode

this is a low speed motor rotation mode to reduce the internal temperature in order to avoid overheating

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

motoroff

async def motoroff() -> None

Exits motoron-mode

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

flip

async def flip(direction: Literal["l", "r", "f", "b"],
               timeout: float = 5) -> None

Flips the drone forwards, backwards, left or right

Arguments:

• direction: the first character of the direction to flip
• timeout: the timeout in seconds

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after timeout

throwfly

async def throwfly() -> None

Enables the throwfly mode.

Throw the drone within 5s horizontally to launch the drone

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

speed

async def speed(speed: int) -> None

Sets the speed to use when using flying commands except the rc command

Arguments:

• speed: the speed the drone is allowed to fly in cm/s [10,100]

Raises:

• ProtocolError: raised when not receiving ok
• TimeoutError: raised when not answering after 5s

rc

def rc(roll: int, pitch: int, throttle: int, yaw: int) -> None

Sends movement like you would do on a rc-controller

Arguments:

• roll: The roll (left-right) in the range [-100,100]
• pitch: The pitch (forwards-backwards) in the range [-100,100]
• throttle: The throttle (up-down) in the range [-100,100]
• yaw: The yaw (rotation) in the range [-100,100]

RGBLed Objects

class RGBLed(Module)

set

async def set(color: Tuple[int, int, int]) -> None

Sets the top-led color

Arguments:

• color: The R,G,B values in the range [0,255]

Raises:

• ProtocolError: raised when not receiving led ok
• TimeoutError: raised when not answering after 2.5s

pulse

async def pulse(color: Tuple[int, int, int], frequency: float) -> None

Pulses the top-led color

Arguments:

• color: The R,G,B values in the range [0,255]
• frequency: The frequency to pulse in Hz in range [0.1,2.5]

Raises:

• ProtocolError: raised when not receiving led ok
• TimeoutError: raised when not answering after 2.5s

flash

async def flash(color1: Tuple[int, int, int], color2: Tuple[int, int, int],
                frequency: float) -> None

Flashes the top-led color between color1 and color2

Arguments:

• color1: The R,G,B values in the range [0,255]
• color2: The R,G,B values in the range [0,255]
• frequency: The frequency to pulse in Hz in range [0.1,10]

Raises:

• ProtocolError: raised when not receiving led ok
• TimeoutError: raised when not answering after 2.5s

Matrix Objects

class Matrix(Module)

set_brightness

async def set_brightness(brightness: int) -> None

Sets the brigthness of the 8x8 led-matrix

Arguments:

• brightness: the brightness in the range [0,255]

Raises:

• ProtocolError: raised when not receiving mled ok
• TimeoutError: raised when not answering after 2.5s

set_pattern

async def set_pattern(pattern: str) -> None

Sets the pattern of the 8x8 led-matrix, starting at the topmost row and going to the right

Arguments:

• pattern: the pattern string, consisting of r for red, b for blue, p for purple, 0 for off, limited to a length of max 64

Raises:

• ProtocolError: raised when not receiving mled ok
• TimeoutError: raised when not answering after 2.5s