Step-by-Step: Swarm Interface

Here we will go through step-by-step how to interface with a swarm of Crazyflies and make all the copters of the swarm hover and fly simultaneously in a square shape using the Swarm() class of the cflib. For this tutorial you will need a swarm (2 or more) of Crazyflies with the latest firmware version installed and a positioning system (Lighthouse, Loco or MoCap) that is able to provide data for the position estimation of the Crazyflies. You can also use the Flow Deck, but keep in mind that you should command relative movements of each Crazyflie and due to its nature it may lead to accumulative errors and unexpected behavior over time.

Prerequisites

We will assume that you already have completed the following steps before you start with the tutorial:

Have some basic experience with Python
Followed the Crazyflie getting started guide
Set up the positioning system of your choice
Read the high level commander, swarm and SyncCrazyflie documentation.

Get the script started

Since you should have installed cflib in the previous step by step tutorial, you’re all ready to go now. Open up a new python script called swarm_rectangle.py. First you will start by adding the following import to the script:

import time

import cflib.crtp
from cflib.crazyflie.swarm import CachedCfFactory
from cflib.crazyflie.swarm import Swarm

uris = {
    'radio://0/20/2M/E7E7E7E701',
    'radio://0/20/2M/E7E7E7E702',
    'radio://0/20/2M/E7E7E7E703',
    'radio://0/20/2M/E7E7E7E704',
    # Add more URIs if you want more copters in the swarm
    # URIs in a swarm using the same radio must also be on the same channel
}

if __name__ == '__main__':
    cflib.crtp.init_drivers()
    factory = CachedCfFactory(rw_cache='./cache')
    with Swarm(uris, factory=factory) as swarm:

This will import all the necessary modules and open the necessary links for communication with all the Crazyflies of the swarm. Note that the URIs in a swarm using the same radio must also be on the same channel. Swarm is a wrapper class which facilitates the execution of functions given by the user for each copter and can execute them in parallel or sequentially. Each Crazyflie is treated as a SyncCrazyflie instance and as the first argument in swarm wide actions. There is no need to worry about threads since they are handled internally. To reduce connection time, the factory is chosen to be instance of the CachedCfFactory class that will cache the Crazyflie objects in the ./cache directory.

The radio addresses of the copters are defined in the uris list and you can add more if you want.

Step 1: Light Check

In order to verify everything is setup and working properly a light check will be performed. During this check, all the copters will light up red for a short period of time and then return to normal. This is achieved by setting the parameter led.bitmask to 255 which results in all the LEDs of each copter lighting up simultaneously.

Add the helper functions activate_led_bit_mask, deactivate_led_bit_mask and the function light_check above __main__:

def activate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 255)

def deactivate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 0)

def light_check(scf):
    activate_led_bit_mask(scf)
    time.sleep(2)
    deactivate_led_bit_mask(scf)

light_check will light up a copter red for 2 seconds and then return them to normal.

Below ... Swarm(...) in __main__, execute the light check for each copter:

    swarm.parallel_safe(light_check)

The light_check() is going to be called through the parallel_safe() method which will execute it for all Crazyflies in the swarm, in parallel. One thread per Crazyflie is started to execute the function. The threads are joined at the end and if one or more of the threads raised an exception this function will also raise an exception.

import time

import cflib.crtp
from cflib.crazyflie.swarm import CachedCfFactory
from cflib.crazyflie.swarm import Swarm


def activate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 255)

def deactivate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 0)

def light_check(scf):
    activate_led_bit_mask(scf)
    time.sleep(2)
    deactivate_led_bit_mask(scf)
    time.sleep(2)


uris = {
    'radio://0/20/2M/E7E7E7E701',
    'radio://0/20/2M/E7E7E7E702',
    'radio://0/20/2M/E7E7E7E703',
    'radio://0/20/2M/E7E7E7E704',
    # Add more URIs if you want more copters in the swarm
    # URIs in a swarm using the same radio must also be on the same channel
}

if __name__ == '__main__':
    cflib.crtp.init_drivers()
    factory = CachedCfFactory(rw_cache='./cache')
    with Swarm(uris, factory=factory) as swarm:
        swarm.parallel_safe(light_check)

If everything is working properly, you can move to the next step.

Step 2: Security Before Flying

Before executing any take off and flight manoeuvres, the copters need to make sure that they have a precise enough position estimation. Otherwise it will take off anyway and it is very likely to crash. This is done through reset_estimators() by resetting the internal position estimator of each copter and waiting until the variance of the position estimation drops below a certain threshold.

with Swarm(uris, factory=factory) as swarm:
        swarm.parallel_safe(light_check)
        swarm.reset_estimators()

Step 3: Taking off and Landing Sequentially

Make sure that your positioning system of your choice is set up and ready to go. Now we are going to execute the first take off and landing using the high level commander. The high level commander (more information here) is a class that handles all the high level commands like takeoff, landing, hover, go to position and others. The high level commander is usually preferred since it needs less communication and provides more autonomy for the Crazyflie. It is always on, but just in a lower priority so you just need to execute the take off and land commands through the below functions:

def take_off(scf):
    commander= scf.cf.high_level_commander

    commander.takeoff(1.0, 2.0)
    time.sleep(3)

def land(scf):
    commander= scf.cf.high_level_commander

    commander.land(0.0, 2.0)
    time.sleep(2)

    commander.stop()

def hover_sequence(scf):
    take_off(scf)
    land(scf)

Initially, we want only one copter at a time executing the hover_sequence so it is going to be called through the sequential() method of the Swarm in the following way:

swarm.sequential(hover_sequence)

Leading to the following code:

import time

import cflib.crtp
from cflib.crazyflie.swarm import CachedCfFactory
from cflib.crazyflie.swarm import Swarm


def activate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 255)

def deactivate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 0)

def light_check(scf):
    activate_led_bit_mask(scf)
    time.sleep(2)
    deactivate_led_bit_mask(scf)
    time.sleep(2)

def take_off(scf):
    commander= scf.cf.high_level_commander

    commander.takeoff(1.0, 2.0)
    time.sleep(3)

def land(scf):
    commander= scf.cf.high_level_commander

    commander.land(0.0, 2.0)
    time.sleep(2)

    commander.stop()

def hover_sequence(scf):
    take_off(scf)
    land(scf)

uris = {
    'radio://0/20/2M/E7E7E7E701',
    'radio://0/20/2M/E7E7E7E702',
    'radio://0/20/2M/E7E7E7E703',
    'radio://0/20/2M/E7E7E7E704',
    # Add more URIs if you want more copters in the swarm
    # URIs in a swarm using the same radio must also be on the same channel
}

if __name__ == '__main__':
    cflib.crtp.init_drivers()
    factory = CachedCfFactory(rw_cache='./cache')
    with Swarm(uris, factory=factory) as swarm:
        print('Connected to Crazyflies')
        swarm.parallel_safe(light_check)
        swarm.reset_estimators()

        swarm.sequential(hover_sequence)

After executing it you will see all copters performing the light check and then each copter take off, hover and land. This process is repeated for all copters in the swarm.

Step 4: Taking off and Landing in Sync

If you want to take off and land in sync, you can use the parallel_safe() method of the Swarm class.

    swarm.parallel_safe(hover_sequence)

Now the same action is happening but for all the copters in parallel.

Step 5: Performing a square shape flight

To make the swarm fly in a square shape, we will use the go_to method of the high level commander. Each copter independently executes 4 relative movements (+X, +Y, -X, -Y) tracing a 1×1 m square and returning to its own starting position. Since the movements are relative, all copters fly the same square path simultaneously regardless of where they started, and you should see the entire swarm move in sync.

def run_square_sequence(scf):
    box_size = 1
    flight_time = 2

    commander= scf.cf.high_level_commander

    commander.go_to(box_size, 0, 0, 0, flight_time, relative=True)
    time.sleep(flight_time)

    commander.go_to(0, box_size, 0, 0, flight_time, relative=True)
    time.sleep(flight_time)

    commander.go_to(-box_size, 0, 0, 0, flight_time, relative=True)
    time.sleep(flight_time)

    commander.go_to(0, -box_size, 0, 0, flight_time, relative=True)
    time.sleep(flight_time)

Keep in mind that the go_to() command does not block the code so you have to wait as long as the flight time of each movement to continue to the next one.

Since we want these maneuvers to be synchronized, the parallel_safe() method is chosen to execute the sequence, in similar fashion with the above steps. You can include the sequence execution in the main code of the swarm in the following way:

    swarm.parallel_safe(take_off)
    swarm.parallel_safe(run_square_sequence)
    swarm.parallel_safe(land)

Make sure that your script looks similar to the following and execute it:

import time

import cflib.crtp
from cflib.crazyflie.swarm import CachedCfFactory
from cflib.crazyflie.swarm import Swarm


def activate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 255)

def deactivate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 0)

def light_check(scf):
    ...

def take_off(scf):
    ...

def land(scf):
    ...

def run_square_sequence(scf: SyncCrazyflie):
    ...

uris = [
    'radio://0/20/2M/E7E7E7E701',
    'radio://0/20/2M/E7E7E7E702',
    'radio://0/20/2M/E7E7E7E703',
    'radio://0/20/2M/E7E7E7E704',
    # Add more URIs if you want more copters in the swarm
    # URIs in a swarm using the same radio must also be on the same channel
]

if __name__ == '__main__':
    cflib.crtp.init_drivers()
    factory = CachedCfFactory(rw_cache='./cache')
    with Swarm(uris, factory=factory) as swarm:
        print('Connected to Crazyflies')
        swarm.parallel_safe(light_check)
        swarm.reset_estimators()

        swarm.parallel_safe(take_off)
        swarm.parallel_safe(run_square_sequence)
        swarm.parallel_safe(land)

Step 6: Performing a flight with different arguments

You can also feed different arguments to each Crazyflie in the swarm. The parallel_safe(), parallel(), and sequential() methods accept an optional args_dict parameter which is a dictionary that maps each URI to a list of extra arguments to pass to the per-Crazyflie function.

You create this dictionary yourself, with the URI (radio address) as the key and a list of arguments as the value:

my_args = {
    URI0: [optional_param0_cf0, optional_param1_cf0],
    URI1: [optional_param0_cf1, optional_param1_cf1],
    ...
}

swarm.parallel_safe(my_function, args_dict=my_args)

The arguments in each list are appended to the scf argument when the function is called, so my_function(scf, optional_param0, optional_param1) is what gets executed for each copter.

In the example below, four copters are placed in a square (pay attention to the order of the Crazyflies) and each one executes a different movement sequence. The dictionary seq_args maps each URI to its own sequence, which is then passed as the sequence argument to run_sequence.

# The layout of the positions (1m apart from each other):
#   <------ 1 m ----->
#   0                1
#          ^              ^
#          |Y             |
#          |              |
#          +------> X    1 m
#                         |
#                         |
#   3               2     .


uris = [
    'radio://0/20/2M/E7E7E7E701',
    'radio://0/20/2M/E7E7E7E702',
    'radio://0/20/2M/E7E7E7E703',
    'radio://0/20/2M/E7E7E7E704',
]

h = 0.0 # remain constant height similar to take off height
x0, y0 = +1.0, +1.0
x1, y1 = -1.0, -1.0

#    x   y   z  time
sequence0 = [
    (x1, y0, h, 3.0),
    (x0, y1, h, 3.0),
    (x0,  0, h, 3.0),
]

sequence1 = [
    (x0, y0, h, 3.0),
    (x1, y1, h, 3.0),
    (.0, y1, h, 3.0),
]

sequence2 = [
    (x0, y1, h, 3.0),
    (x1, y0, h, 3.0),
    (x1,  0, h, 3.0),
]

sequence3 = [
    (x1, y1, h, 3.0),
    (x0, y0, h, 3.0),
    (.0, y0, h, 3.0),
]

seq_args = {
    uris[0]: [sequence0],
    uris[1]: [sequence1],
    uris[2]: [sequence2],
    uris[3]: [sequence3],
}

def run_sequence(scf: SyncCrazyflie, sequence):
    cf = scf.cf

    for arguments in sequence:
        commander = scf.cf.high_level_commander

        x, y, z = arguments[0], arguments[1], arguments[2]
        duration = arguments[3]

        print('Setting position {} to cf {}'.format((x, y, z), cf.link_uri))
        commander.go_to(x, y, z, 0, duration, relative=True)
        time.sleep(duration)

And in the main code of the swarm, you can execute the sequence as follows:

swarm.parallel_safe(run_sequence, args_dict=seq_args)

The final script is going to look like this:

import time

import cflib.crtp
from cflib.crazyflie.swarm import CachedCfFactory
from cflib.crazyflie.swarm import Swarm
from cflib.crazyflie.syncCrazyflie import SyncCrazyflie


def activate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 255)

def deactivate_led_bit_mask(scf):
    scf.cf.param.set_value('led.bitmask', 0)

def light_check(scf):
    ...

def take_off(scf):
    ...

def land(scf):
    ...


def run_square_sequence(scf):
    ...

uris = ...

# The layout of the positions (1m apart from each other):
#   <------ 1 m ----->
#   0                1
#          ^              ^
#          |Y             |
#          |              |
#          +------> X    1 m
#                         |
#                         |
#   3               2     .


h = 0.0 # remain constant height similar to take off height
x0, y0 = +1.0, +1.0
x1, y1 = -1.0, -1.0

#    x   y   z  time
sequence0 = ...

sequence1 = ...

sequence2 = ...

sequence3 = ...

seq_args = ...

def run_sequence(scf: SyncCrazyflie, sequence):
    ...


if __name__ == '__main__':
    cflib.crtp.init_drivers()
    factory = CachedCfFactory(rw_cache='./cache')
    with Swarm(uris, factory=factory) as swarm:
        print('Connected to Crazyflies')
        swarm.parallel_safe(light_check)

        swarm.reset_estimators()

        swarm.parallel_safe(take_off)
        swarm.parallel_safe(run_square_sequence)
        swarm.parallel_safe(run_sequence, args_dict=seq_args)
        swarm.parallel_safe(land)

You’re done! The full code of this tutorial can be found in the example/step-by-step/ folder.

What is next?

Now you are able to control a swarm of Crazyflies and you can experiment with different behaviors for each one of them while maintaining the functionality, simplicity of working with just one since the parallelism is handled internally and you can just focus on creating awesome applications! For more examples and inspiration on the Swarm functionality, you can check out the examples/swarm/ folder of the cflib.