How To Use the Raspberry Pi Python SDK

Where to get the code

Pull down the newest Python SDK from the GitHub repo!

You can follow our Getting Started with Raspberry Pi instructions for more details.

Asynchronous

API and Shell (Pinging)

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import SpheroRvrTargets



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program demonstrates how to use the echo command, which sends data to RVR and RVR returns

        the same data. Echo can be used to check to see if RVR is connected and awake.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    echo_response = await rvr.echo(

        data=[0, 1, 2],

        target=SpheroRvrTargets.primary.value

    )

    print('Echo response 1: ', echo_response)


    echo_response = await rvr.echo(

        data=[4, 5, 6],

        target=SpheroRvrTargets.secondary.value

    )

    print('Echo response 2: ', echo_response)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


API and Shell (Pinging) with REST API

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import RestfulAsyncDal

from sphero_sdk import SpheroRvrTargets



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=RestfulAsyncDal(

        domain='10.211.2.21'# Add your raspberry-pi's IP address here

        port=2010  # The port opened by the npm server is always 2010

    )

)



async def main():

    """ This program demonstrates how to use the echo command, which sends data to RVR and has RVR returns

        the same data. Echo can be used to check to see if RVR is connected and awake.  In order to test it,

        a node.js server must be running on the raspberry-pi connected to RVR.  This code is meant to be

        executed from a separate computer.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    echo_response = await rvr.echo(

        data=[0, 1, 2],

        target=SpheroRvrTargets.primary.value

    )

    print('Echo response 1: ', echo_response)


    echo_response = await rvr.echo(

        data=[4, 5, 6],

        target=SpheroRvrTargets.secondary.value

    )

    print('Echo response 2: ', echo_response)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Color Sensor (Color Detection)

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import RvrStreamingServices



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def color_detected_handler(color_detected_data):

    print('Color detection data response: ', color_detected_data)



async def main():

    """ This program demonstrates how to use the color sensor on RVR (located on the down side of RVR, facing the floor)

        to report colors detected. To exit program, press <CTRL-C>

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.enable_color_detection(is_enabled=True)

    await rvr.sensor_control.add_sensor_data_handler(

        service=RvrStreamingServices.color_detection,

        handler=color_detected_handler

    )

    await rvr.sensor_control.start(interval=250)


    while True:

        await asyncio.sleep(1)



if __name__ == '__main__':

    try:

        asyncio.ensure_future(

            main()

        )

        loop.run_forever()


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.enable_color_detection(is_enabled=False),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


Color Sensor (Color Detection) with REST API

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import RestfulAsyncDal

from sphero_sdk import RvrStreamingServices



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=RestfulAsyncDal(

        domain='10.211.2.21'# Add your raspberry-pi's IP address here

        port=2010  # The port opened by the npm server is always 2010

    )

)



async def color_detected_handler(color_detected_data):

    print('Color detection data response: ', color_detected_data)



async def main():

    """ This program uses the color sensor on RVR (located on the down side of RVR, facing the floor) to report colors detected.

        To exit program, press <CTRL-C>

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.enable_color_detection(is_enabled=True)

    await rvr.sensor_control.add_sensor_data_handler(

        service=RvrStreamingServices.color_detection,

        handler=color_detected_handler

    )

    await rvr.sensor_control.start(interval=250)


    while True:

        await asyncio.sleep(1)



if __name__ == '__main__':

    try:

        asyncio.ensure_future(

            main()

        )

        loop.run_forever()


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.enable_color_detection(is_enabled=False),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()

Drive - Raw Motors

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import RawMotorModesEnum



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program has RVR drive around in different directions.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.reset_yaw()


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.forward.value,

        left_speed=128# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.forward.value,

        right_speed=128  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.reverse.value,

        left_speed=64# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.reverse.value,

        right_speed=64  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.reverse.value,

        left_speed=128# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.forward.value,

        right_speed=128  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.forward.value,

        left_speed=128# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.forward.value,

        right_speed=128  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.off.value,

        left_speed=0# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.off.value,

        right_speed=0  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Drive with Heading

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import DriveFlagsBitmask



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program has RVR drive around in different directions using the function drive_with_heading.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.reset_yaw()


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.drive_reverse.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=90# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=270# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=0# Valid heading values are 0-359

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Drive with Heading - Reverse Mode

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import DriveFlagsBitmask



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program has RVR drive around in different directions using the function drive_with_heading.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.reset_yaw()


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=90# Valid heading values are 0-359

        flags=DriveFlagsBitmask.drive_reverse.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=0# Valid heading values are 0-359

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Drive with Helper

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program has RVR drive with how to drive RVR using the drive control helper.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.drive_control.reset_heading()


    await rvr.drive_control.drive_forward_seconds(

        speed=64,

        heading=0# Valid heading values are 0-359

        time_to_drive=1

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_control.drive_backward_seconds(

        speed=64,

        heading=0# Valid heading values are 0-359

        time_to_drive=1

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_control.turn_left_degrees(

        heading=0# Valid heading values are 0-359

        amount=90

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Drive with Helper - Roll

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program has RVR drive with how to drive RVR using the drive control helper.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.drive_control.reset_heading()


    await rvr.drive_control.roll_start(

        speed=64,

        heading=90

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_control.roll_stop(heading=270)


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


Drive with REST API

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import RestfulAsyncDal

from sphero_sdk import DriveFlagsBitmask



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=RestfulAsyncDal(

        domain='10.211.2.21'# Add your raspberry-pi's IP address here

        port=2010  # The port opened by the npm server is always 2010

    )

)



async def main():

    """ This program has RVR drive using the Rest API.  In order to test it, a node.js server must be running on the

        raspberry-pi connected to RVR.  This code is meant to be executed from a separate computer.


        Note:

            To give RVR time to drive, we call asyncio.sleep(...); if we did not have these calls, the program would

            go on and execute all the statements and exit without the driving ever taking place.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.reset_yaw()


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.drive_reverse.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=90# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=128# Valid speed values are 0-255

        heading=270# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.drive_with_heading(

        speed=0# Valid heading values are 0-359

        heading=0# Valid heading values are 0-359

        flags=DriveFlagsBitmask.none.value

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()

Infrared - Broadcast IR

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import InfraredCodes

from sphero_sdk import RawMotorModesEnum



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program sets up RVR to communicate with another robot, e.g. BOLT, capable of infrared communication.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.start_robot_to_robot_infrared_broadcasting(

        far_code=InfraredCodes.one.value,

        near_code=InfraredCodes.zero.value

    )


    for i in range(2):

        await rvr.raw_motors(

            left_mode=RawMotorModesEnum.forward.value,

            left_speed=64# Valid speed values are 0-255

            right_mode=RawMotorModesEnum.forward.value,

            right_speed=64  # Valid speed values are 0-255

        )


        # Delay to allow RVR to drive

        await asyncio.sleep(2)


    await rvr.stop_robot_to_robot_infrared_broadcasting()


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.stop_robot_to_robot_infrared_broadcasting(),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


Infrared - Broadcast IR with Helper

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import InfraredCodes

from sphero_sdk import RawMotorModesEnum



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program sets up RVR to communicate with another robot, e.g. BOLT, capable of infrared communication.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.infrared_control.start_infrared_broadcasting(

        far_code=InfraredCodes.one,

        near_code=InfraredCodes.zero

    )


    await rvr.raw_motors(

        left_mode=RawMotorModesEnum.forward.value,

        left_speed=64# Valid speed values are 0-255

        right_mode=RawMotorModesEnum.forward.value,

        right_speed=64  # Valid speed values are 0-255

    )


    # Delay to allow RVR to drive

    await asyncio.sleep(4)


    await rvr.infrared_control.stop_infrared_broadcasting()


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.stop_robot_to_robot_infrared_broadcasting(),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


Infrared - Listen For and Send IR

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def infrared_message_received_handler(infrared_message):

    print('Infrared message response: ', infrared_message)



async def main():

    """ This program sets up RVR to communicate with another robot, e.g. BOLT, capable of infrared communication.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.on_robot_to_robot_infrared_message_received_notify(handler=infrared_message_received_handler)


    await rvr.enable_robot_infrared_message_notify(is_enabled=True)


    infrared_code = 3

    strength = 64


    while True:

        await rvr.send_infrared_message(

            infrared_code=infrared_code,

            front_strength=strength,

            left_strength=strength,

            right_strength=strength,

            rear_strength=strength

        )


        print('Infrared message sent with code: {0}'.format(infrared_code))


        await asyncio.sleep(2)



if __name__ == '__main__':

    try:

        asyncio.ensure_future(

            main()

        )

        loop.run_forever()


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.stop_robot_to_robot_infrared_broadcasting(),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


Infrared - Listen For and Send IR with Helper

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import SerialAsyncDal

from sphero_sdk import InfraredCodes



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def infrared_message_received_handler(infrared_message):

    print('Infrared message response: ', infrared_message)



async def main():

    """ This program sets up RVR to communicate with another robot, e.g. BOLT, capable of infrared communication.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.infrared_control.listen_for_infrared_message(handler=infrared_message_received_handler)


    codes = [

        InfraredCodes.zero,

        InfraredCodes.one,

        InfraredCodes.two,

        InfraredCodes.three

    ]


    while True:

        await rvr.infrared_control.send_infrared_messages(

            messages=codes,

            strength=64

        )


        print('Infrared message sent with codes: {0}'.format([code.value for code in codes]))


        await asyncio.sleep(2)



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.stop_robot_to_robot_infrared_broadcasting(),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


Infrared - Listen For and Send IR with REST API

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import RestfulAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=RestfulAsyncDal(

        domain='10.211.2.21'# Add your raspberry-pi's IP address here

        port=2010  # The port opened by the npm server is always 2010

    )

)



async def infrared_message_received_handler(response):

    print('Response data for IR message received:',response)



async def main():

    """ This program sets up RVR to communicate with another robot, e.g. BOLT, capable of infrared communication.


        To try this out, write a script for your other robot that a) broadcasts on the corresponding

        channel that RVR is set to listen to [in this case channel 0] and b) listens on the channel which

        RVR sends messages on [in this case channel 3]

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.on_robot_to_robot_infrared_message_received_notify(handler=infrared_message_received_handler)


    await rvr.enable_robot_infrared_message_notify(is_enabled=True)


    infrared_code = 3

    strength = 64


    while True:

        await rvr.send_infrared_message(

            infrared_code=infrared_code,

            front_strength=strength,

            left_strength=strength,

            right_strength=strength,

            rear_strength=strength

        )


        print('Infrared message sent with code: {0}'.format(infrared_code))


        await asyncio.sleep(2)



if __name__ == '__main__':

    try:

        asyncio.ensure_future(

            main()

        )

        loop.run_forever()


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            asyncio.gather(

                rvr.stop_robot_to_robot_infrared_broadcasting(),

                rvr.close()

            )

        )


    finally:

        if loop.is_running():

            loop.close()


LEDs - Set All LEDs

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import Colors

from sphero_sdk import RvrLedGroups

from sphero_sdk import SerialAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program demonstrates how to set the all the LEDs.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.set_all_leds(

        led_group=RvrLedGroups.all_lights.value,

        led_brightness_values=[color for _ in range(10) for color in Colors.off.value]

    )


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.set_all_leds(

        led_group=RvrLedGroups.all_lights.value,

        led_brightness_values=[color for x in range(10) for color in [0, 255, 0]]

    )


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


LEDs - Set All LEDs with Helper

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import Colors

from sphero_sdk import SerialAsyncDal



loop = asyncio.get_event_loop()


rvr = SpheroRvrAsync(

    dal=SerialAsyncDal(

        loop

    )

)



async def main():

    """ This program demonstrates how to set the all the LEDs of RVR using the LED control helper.

    """


    await rvr.wake()


    # Give RVR time to wake up

    await asyncio.sleep(2)


    await rvr.led_control.turn_leds_off()


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.led_control.set_all_leds_color(color=Colors.yellow)


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.led_control.turn_leds_off()


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.led_control.set_all_leds_rgb(red=255, green=144, blue=0)


    # Delay to show LEDs change

    await asyncio.sleep(1)


    await rvr.close()



if __name__ == '__main__':

    try:

        loop.run_until_complete(

            main()

        )


    except KeyboardInterrupt:

        print('\nProgram terminated with keyboard interrupt.')


        loop.run_until_complete(

            rvr.close()

        )


    finally:

        if loop.is_running():

            loop.close()


LEDs - Set Multiple LEDs

import os

import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../')))


import asyncio

from sphero_sdk import SpheroRvrAsync

from sphero_sdk import Colors