python

ESPP Python Library

This folder contains some test python scripts for loading the espp shared objects built within the espp/lib folder into Python.

Table of Contents

• ESPP Python Library
  • Scripts
  • Setup
    • Install Python Requirements
  • Quick Start

Scripts

This section gives a brief overview of what the scripts in this folder do.

• task.py: This script is a simple example of how to use the espp library to create a task. It demonstrates how to create a task, run it, and handle its results. Note: unlike python multithreading, this task is actually run in a separate thread.
• timer.py: This script demonstrates how to use the espp library to create a timer. It shows how to create a timer, start it, and handle its expiration. The timer runs in a separate thread and calls a callback function when it expires. Note: unlike python multithreading, this timer is actually run in a separate thread.
• udp_client.py and udp_server.py: These scripts demonstrate how to use the espp library to create a UDP client and server. The server listens for incoming UDP packets and prints them to the console, while the client sends UDP packets to the server.
• socket_reactor.py: Demonstrates espp.SocketReactor - a single select() event loop plus a thread pool that services many receiver sockets instead of one thread per socket. It multiplexes two UDP echo receivers on one reactor and round-trips a datagram through each.
• socket_reactor_test.py: Self-checking test for the espp.SocketReactor Python binding (lifecycle, UDP receiver + echo, sender info, None responses, multiple receivers, dynamic remove()). Exits 0 on full pass, 1 on any failure. Mirrors the C++ pc/tests/socket_reactor.cpp.
• rtsp_client.py and rtsp_server.py: These scripts demonstrate how to use the espp library to create an RTSP client and server. The server streams MJPEG video from a webcam or display capture. The camera path captures live microphone audio by default, while display capture keeps the simulated audio tone unless --audio-source microphone is selected. The client receives the stream, validates audio delivery, plays the audio in real time, and displays the video in a window. Note: these scripts require additional 3rd party libraries such as opencv, mss, zeroconf, and sounddevice.
• rtsp_client_multitrack.py and rtsp_server_multitrack.py: These scripts exercise the generic multitrack RTSP APIs. The multitrack server includes audio by default; camera mode uses live microphone capture and the other video sources use the simulated tone unless --audio-source microphone is selected. Pass --no-audio to disable the audio track. The multitrack client plays audio by default when running with a UI; use --no-audio-playback to disable it or --play-audio --headless to exercise playback without opening the video window.
• rtps_host.py: A pure-stdlib host-side RTPS harness for discovering an ESPP RtpsParticipant, printing SPDP/SEDP metadata, and optionally publishing or receiving standard CDR-over-RTPS UInt32 user-data samples (routed by writer GUID via SEDP) without needing Python bindings. It follows endpoint-advertised user-data multicast locators, joining matching subscribed-topic multicast groups dynamically. Run python rtps_host.py --self-test to validate the wire-format encoders/decoders against the firmware with no network I/O.
• rtps_pubsub.py, rtps_publisher.py, rtps_subscriber.py: RTPS tests that use the espp Python library (espp.RtpsParticipant + espp.CdrWriter/ CdrReader) rather than the pure-stdlib harness. rtps_pubsub.py is self-contained (a publisher and subscriber in one process; exits 0 if samples were received). rtps_publisher.py / rtps_subscriber.py are standalone and interoperate with each other, with the C++ rtps_publisher/rtps_subscriber in ../pc, and with rtps_host.py. They take an optional [topic] [advertised_ipv4] (a real interface IP is needed for cross-host discovery). Note: RTPS discovery is multicast, so these require a multicast-capable network.
• cobs_demo.py: Demonstration of ESPP COBS functionality with native Python data types. Shows ESPP encoding/decoding, cross-library compatibility with the cobs-python library, and practical usage examples. Includes design differences explanation and validation against the reference implementation.

Important Design Differences: The ESPP COBS implementation differs from other COBS libraries:

• Delimiters: ESPP automatically adds 0x00 delimiters to encoded packets, while other libraries may not
• Empty Packets: ESPP ignores empty packets (length = 0) for performance, while other libraries may encode them
• Compatibility: The demo scripts show how to handle these differences for cross-library integration

Setup

All scripts need the espp python package installed. Released versions come from PyPI:

pip install espp

If you're working from this repository (e.g. developing the bindings or running the scripts against unreleased changes), use an editable install from the repository root instead - it builds the same package from the local sources:

pip install -e ..   # from this folder (or `pip install -e .` from the repo root)

Note: after changing C++ code (components or bindings), re-run the editable install to rebuild the extension; the build is incremental. Pure-python changes are picked up automatically.

Note: The COBS demo script (cobs_demo.py) additionally requires the cobs library for cross-validation with the reference implementation.

Install Python Requirements

Some tests (e.g. rtsp_client.py, rtsp_server.py) make use of 3rd party libraries such as zeroconf, opencv, mss, sounddevice to facilitate mDNS discovery, image display / webcam capture, display capture, and audio playback. For these tests, you will need to install the requirements.txt:

# create the virtual environment
python3 -m venv env
# activate it
source env/bin/activate
# install the requirements
pip install -r requirements.txt

Afterwards, you only need to source the environment and run the test in question:

source env/bin/activate
# now you can run the tests
python3 rtsp_server.py

Quick Start

To run a test, you can use the following commands:

python3 <test_name>.py
# e.g.
python3 task.py
# or 
python3 udp_client.py
# or discover / test RTPS from a host machine
python3 rtps_host.py --advertised-address <your-host-ip>

Note: the udp_client.py script requires a running instance of the udp_server.py script. To run the server, use the following command from another terminal:

python3 udp_server.py

For the default ESP RTPS example, the host harness now defaults to the responder side of the request/response test, so it will subscribe to espp/rtps_example/request and echo values back on espp/rtps_example/response:

python3 rtps_host.py --advertised-address 192.168.1.50

When the ESP RTPS example is configured for per-topic multicast, the host harness will automatically join the discovered request-topic multicast group and send response samples using the discovered response-reader locators.

To act as the initiator instead, swap the topics and enable periodic publishing:

python3 rtps_host.py --advertised-address 192.168.1.50 \
  --subscribe-topic espp/rtps_example/response \
  --publish-topic espp/rtps_example/request \
  --publish-value 42 \
  --publish-interval 1.0 \
  --no-echo-received