#include
#include
#include
#include
#include
#include
#include
#include
#include
#include "span_caster.h"
#include "espp.hpp"
namespace py = pybind11;
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! AUTOGENERATED CODE !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
// // Autogenerated code below! Do not edit!
// // Autogenerated code end
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! AUTOGENERATED CODE END !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
void py_init_module_espp(py::module &m) {
// using namespace espp; // NON!
// create an Error class which is really just std::error_code
py::class_(
m, "Error", py::module_local()) // should be module_local so other modules can have
// std::error_code if they need to
.def(py::init())
.def("__repr__",
[](const std::error_code &self) { return fmt::format("Error({})", self.message()); })
.def("__bool__", [](const std::error_code &self) { return bool(self); });
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! AUTOGENERATED CODE !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
// // Autogenerated code below! Do not edit!
//////////////////// ////////////////////
auto pyClassBaseComponent =
py::class_(m, "BaseComponent", py::dynamic_attr(),
"/ Base class for all components\n/ Provides a logger and "
"some basic logging configuration")
.def("get_name", &espp::BaseComponent::get_name,
"/ Get the name of the component\n/ \\return A const reference to the name of the "
"component\n/ \note This is the tag of the logger")
.def("set_log_tag", &espp::BaseComponent::set_log_tag, py::arg("tag"),
"/ Set the tag for the logger\n/ \\param tag The tag to use for the logger")
.def("get_log_level", &espp::BaseComponent::get_log_level,
"/ Get the log level for the logger\n/ \\return The verbosity level of the "
"logger\n/ \\sa Logger::Verbosity\n/ \\sa Logger::set_verbosity")
.def("set_log_level", &espp::BaseComponent::set_log_level, py::arg("level"),
"/ Set the log level for the logger\n/ \\param level The verbosity level to use for "
"the logger\n/ \\sa Logger::Verbosity\n/ \\sa Logger::set_verbosity")
.def("set_log_verbosity", &espp::BaseComponent::set_log_verbosity, py::arg("level"),
"/ Set the log verbosity for the logger\n/ \\param level The verbosity level to use "
"for the logger\n/ \note This is a convenience method that calls set_log_level\n/ "
"\\sa set_log_level\n/ \\sa Logger::Verbosity\n/ \\sa Logger::set_verbosity")
.def("get_log_verbosity", &espp::BaseComponent::get_log_verbosity,
"/ Get the log verbosity for the logger\n/ \\return The verbosity level of the "
"logger\n/ \note This is a convenience method that calls get_log_level\n/ \\sa "
"get_log_level\n/ \\sa Logger::Verbosity\n/ \\sa Logger::get_verbosity")
.def("set_log_rate_limit", &espp::BaseComponent::set_log_rate_limit,
py::arg("rate_limit"),
"/ Set the rate limit for the logger\n/ \\param rate_limit The rate limit to use "
"for the logger\n/ \note Only calls to the logger that have _rate_limit suffix will "
"be rate limited\n/ \\sa Logger::set_rate_limit");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassCobs =
py::class_(
m, "Cobs", py::dynamic_attr(),
"*\n * @brief COBS (Consistent Overhead Byte Stuffing) encoder/decoder\n *\n * Provides "
"single-packet encoding and decoding using the COBS algorithm\n * with 0 as the "
"delimiter.\n * COBS encoding can add at most ceil(n/254) + 1 bytes overhead, plus 1 "
"byte\n * for the delimiter.\n * COBS changes the size of the packet by at least 1 byte, "
"so it's not possible to encode in\n * place. MAX_BLOCK_SIZE = 254 is the maximum number "
"of non-zero bytes in an encoded block.\n *\n * @see "
"https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing\n")
.def(py::init()) // implicit default constructor
.def_static("max_encoded_size", &espp::Cobs::max_encoded_size, py::arg("payload_len"),
"*\n * @brief Calculate maximum encoded size for a given payload length\n "
" *\n * @param payload_len Length of input data\n * @return Maximum "
"number of bytes needed for encoding (including delimiter)\n")
.def_static("encode_packet",
py::overload_cast(&espp::Cobs::encode_packet),
py::arg("data"),
"*\n * @brief Encode a single packet\n *\n * @param data Input data to "
"encode\n * @return Encoded data with COBS encoding and delimiter\n")
.def_static(
"encode_packet",
py::overload_cast(
&espp::Cobs::encode_packet),
py::arg("data"), py::arg("output"),
"*\n * @brief Encode a single packet to existing buffer\n *\n * @param data "
"Input data to encode\n * @param output Output buffer span (must be at least "
"max_encoded_size)\n * @return Number of bytes written to output\n")
.def_static("max_decoded_size", &espp::Cobs::max_decoded_size, py::arg("encoded_len"),
"*\n * @brief Calculate maximum decoded size for a given encoded length\n "
" *\n * @param encoded_len Length of COBS-encoded data\n * @return "
"Maximum number of bytes needed for decoding (accounts for delimiter)\n")
.def_static("decode_packet",
py::overload_cast(&espp::Cobs::decode_packet),
py::arg("encoded_data"),
"*\n * @brief Decode a single packet from COBS-encoded data\n *\n * "
"@param encoded_data COBS-encoded data\n * @return Decoded packet data, or "
"empty if invalid\n")
.def_static("decode_packet",
py::overload_cast(
&espp::Cobs::decode_packet),
py::arg("encoded_data"), py::arg("output"),
"*\n * @brief Decode a single packet to existing buffer\n *\n * @param "
"encoded_data COBS-encoded data\n * @param output Output buffer span (must "
"be at least max_decoded_size)\n * @return Number of bytes written to "
"output, or 0 if decoding failed\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassCobsStreamDecoder =
py::class_(
m, "CobsStreamDecoder", py::dynamic_attr(),
"*\n * @brief Streaming decoder for multiple COBS-encoded packets\n *\n * Useful for "
"processing incoming data streams where packets may arrive\n * in fragments or multiple "
"packets may arrive together.\n")
.def(py::init())
.def("add_data",
py::overload_cast(&espp::CobsStreamDecoder::add_data),
py::arg("data"),
"*\n * @brief Add encoded data to the decoder buffer\n *\n * @param data New "
"encoded data span\n")
.def("add_data",
py::overload_cast(&espp::CobsStreamDecoder::add_data),
py::arg("data"),
"*\n * @brief Add encoded data to the decoder buffer (move semantics)\n *\n * "
"@param data New encoded data vector (will be moved)\n")
.def("extract_packet", &espp::CobsStreamDecoder::extract_packet,
"*\n * @brief Try to extract the next complete packet. Removes the extracted data "
"from the buffer.\n *\n * @return Decoded packet data, or empty if no complete "
"packet found\n")
.def("remaining_data", &espp::CobsStreamDecoder::remaining_data,
"*\n * @brief Access remaining unprocessed data for debug purposes\n *\n * "
"@return Const reference to buffered data that hasn't been processed yet\n")
.def("buffer_size", &espp::CobsStreamDecoder::buffer_size,
"*\n * @brief Get the size of buffered data\n *\n * @return Number of bytes "
"currently buffered\n")
.def("clear", &espp::CobsStreamDecoder::clear,
"*\n * @brief Clear all buffered data\n");
auto pyClassCobsStreamEncoder =
py::class_(
m, "CobsStreamEncoder", py::dynamic_attr(),
"*\n * @brief Streaming encoder for multiple packets\n *\n * Useful for batching "
"multiple packets together for transmission\n * or for building up data to send in "
"chunks.\n")
.def(py::init())
.def("add_packet",
py::overload_cast(&espp::CobsStreamEncoder::add_packet),
py::arg("data"),
"*\n * @brief Add a packet to be encoded\n *\n * @param data Packet data "
"span\n")
.def("add_packet",
py::overload_cast(&espp::CobsStreamEncoder::add_packet),
py::arg("data"),
"*\n * @brief Add a packet to be encoded (move semantics)\n *\n * @param data "
"Packet data vector (will be moved)\n")
.def("get_encoded_data", &espp::CobsStreamEncoder::get_encoded_data,
"*\n * @brief Get all encoded data as a single buffer for debug purposes\n *\n "
" * @return All encoded packets concatenated, const reference\n")
.def("extract_data", py::overload_cast(&espp::CobsStreamEncoder::extract_data),
py::arg("max_size"),
"*\n * @brief Extract encoded data up to a maximum size\n *\n * @param "
"max_size Maximum number of bytes to extract\n * @return Encoded data up to "
"max_size bytes\n")
.def("extract_data",
py::overload_cast(&espp::CobsStreamEncoder::extract_data),
py::arg("output"), py::arg("max_size"),
"*\n * @brief Extract encoded data directly to a buffer\n *\n * @param output "
"Output buffer to write data to\n * @param max_size Maximum number of bytes to "
"extract\n * @return Number of bytes actually written to output\n")
.def("buffer_size", &espp::CobsStreamEncoder::buffer_size,
"*\n * @brief Get the current buffer size\n *\n * @return Number of bytes "
"currently buffered\n")
.def("clear", &espp::CobsStreamEncoder::clear,
"*\n * @brief Clear all buffered data\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRgb =
py::class_(m, "Rgb", py::dynamic_attr(),
"*\n * @brief Class representing a color using RGB color space.\n")
.def_readwrite("r", &espp::Rgb::r, "/< Red value in [0, 1]")
.def_readwrite("g", &espp::Rgb::g, "/< Green value in [0, 1]")
.def_readwrite("b", &espp::Rgb::b, "/< Blue value in [0, 1]")
.def(py::init())
.def(py::init(), py::arg("r"), py::arg("g"),
py::arg("b"),
"*\n * @brief Construct an Rgb object from the provided rgb values.\n * @note "
"If provided values outside the range [0,1], it will rescale them to\n * be "
"within the range [0,1] by dividing by 255.\n * @param r Floating point value for "
"the red channel, should be in range [0,\n * 1]\n * @param g Floating "
"point value for the green channel, should be in range\n * [0, 1]\n * "
"@param b Floating point value for the blue channel, should be in range\n * "
" [0, 1]\n")
.def(py::init(), py::arg("rgb"),
"*\n * @brief Copy-construct an Rgb object from the provided object.\n * @note "
"If provided values outside the range [0,1], it will rescale them to\n * be "
"within the range [0,1] by dividing by 255.\n * @param rgb Rgb struct containing "
"the values to copy.\n")
.def(py::init(), py::arg("hsv"),
"*\n * @brief Construct an Rgb object from the provided Hsv object.\n * @note "
"This calls hsv.rgb() on the provided object, which means that invalid\n * "
"HSV data (not in the ranges [0,360], [0,1], and [0,1]) could lead to\n * "
"bad RGB data. The Rgb constructor will automatically convert the\n * "
"values to be in the proper range, but the perceived color will be\n * "
"changed.\n * @param hsv Hsv object to copy.\n")
.def(py::init(), py::arg("hex"),
"*\n * @brief Construct an Rgb object from the provided hex value.\n * @param "
"hex Hex value to convert to RGB. The hex value should be in the\n * "
"format 0xRRGGBB.\n")
.def("__add__", &espp::Rgb::operator+, py::arg("rhs"),
"*\n * @brief Perform additive color blending (averaging)\n * @param rhs Other "
"color to add to this color to create the resultant color\n * @return Resultant "
"color from blending this color with the \\p rhs color.\n")
.def("__iadd__", &espp::Rgb::operator+=, py::arg("rhs"),
"*\n * @brief Perform additive color blending (averaging)\n * @param rhs Other "
"color to add to this color\n")
.def("__eq__", &espp::Rgb::operator==, py::arg("rhs"))
.def("__ne__", &espp::Rgb::operator!=, py::arg("rhs"))
.def("hsv", &espp::Rgb::hsv,
"*\n * @brief Get a HSV representation of this RGB color.\n * @return An HSV "
"object containing the HSV representation.\n")
.def("hex", &espp::Rgb::hex,
"*\n * @brief Get the hex representation of this RGB color.\n * @return The hex "
"representation of this RGB color.\n");
auto pyClassHsv =
py::class_(m, "Hsv", py::dynamic_attr(),
"*\n * @brief Class representing a color using HSV color space.\n")
.def_readwrite("h", &espp::Hsv::h, "/< Hue in [0, 360]")
.def_readwrite("s", &espp::Hsv::s, "/< Saturation in [0, 1]")
.def_readwrite("v", &espp::Hsv::v, "/< Value in [0, 1]")
.def(py::init())
.def(py::init(), py::arg("h"), py::arg("s"),
py::arg("v"),
"*\n * @brief Construct a Hsv object from the provided values.\n * @param h Hue "
"- will be clamped to be in range [0, 360]\n * @param s Saturation - will be "
"clamped to be in range [0, 1]\n * @param v Value - will be clamped to be in "
"range [0, 1]\n")
.def(py::init(), py::arg("hsv"),
"*\n * @brief Copy-construct the Hsv object\n * @param hsv Object to copy "
"from.\n")
.def(py::init(), py::arg("rgb"),
"*\n * @brief Construct Hsv object from Rgb object. Calls rgb.hsv() to perform\n "
" * the conversion.\n * @param rgb The Rgb object to convert and copy.\n")
.def("__eq__", &espp::Hsv::operator==, py::arg("rhs"))
.def("__ne__", &espp::Hsv::operator!=, py::arg("rhs"))
.def("rgb", &espp::Hsv::rgb,
"*\n * @brief Get a RGB representation of this HSV color.\n * @return An RGB "
"object containing the RGB representation.\n");
m.def("color_code", py::overload_cast(espp::color_code), py::arg("rgb"),
"\n(C++ auto return type)");
m.def("color_code", py::overload_cast(espp::color_code), py::arg("hsv"),
"\n(C++ auto return type)");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassEventManager =
py::class_(
m, "EventManager", py::dynamic_attr(),
"*\n * @brief Singleton class for managing events. Provides mechanisms for\n * "
"anonymous publish / subscribe interactions - enabling one to one,\n * one to "
"many, many to one, and many to many data distribution with\n * loose coupling "
"and low overhead. Each topic runs a thread for that\n * topic's subscribers, "
"executing all the callbacks in sequence and\n * then going to sleep again until "
"new data is published.\n *\n * @note In c++ objects, it's recommended to call the\n * "
" add_publisher/add_subscriber functions in the class constructor and\n * then "
"to call the remove_publisher/remove_subscriber functions in the\n * class "
"destructor.\n *\n * @note It is recommended (unless you are only interested in events "
"and not\n * data or are only needing to transmit actual strings) to use a\n * "
" serialization library (such as espp::serialization - which wraps\n * alpaca) to "
"serialize your data structures to string when publishing\n * and then deserialize "
"your data from string in the subscriber\n * callbacks.\n *\n * \\section "
"event_manager_ex1 Event Manager Example\n * \\snippet event_manager_example.cpp event "
"manager example\n")
.def_static("get", &espp::EventManager::get,
"*\n * @brief Get the singleton instance of the EventManager.\n * "
"@return A reference to the EventManager singleton.\n",
py::return_value_policy::reference)
.def("add_publisher", &espp::EventManager::add_publisher, py::arg("topic"),
py::arg("component"),
"*\n * @brief Register a publisher for \\p component on \\p topic.\n * @param "
"topic Topic name for the data being published.\n * @param component Name of the "
"component publishing data.\n * @return True if the publisher was added, False if "
"it was already\n * registered for that component.\n")
.def("add_subscriber",
py::overload_cast(
&espp::EventManager::add_subscriber),
py::arg("topic"), py::arg("component"), py::arg("callback"),
py::arg("stack_size_bytes") = 8192,
"*\n * @brief Register a subscriber for \\p component on \\p topic.\n * @param "
"topic Topic name for the data being subscribed to.\n * @param component Name of "
"the component publishing data.\n * @param callback The event_callback_fn to be "
"called when receicing data on\n * \\p topic.\n * @param "
"stack_size_bytes The stack size in bytes to use for the subscriber\n * @note The "
"stack size is only used if a subscriber is not already registered\n * for "
"that topic. If a subscriber is already registered for that topic,\n * the "
"stack size is ignored.\n * @return True if the subscriber was added, False if it "
"was already\n * registered for that component.\n")
.def("add_subscriber",
py::overload_cast(
&espp::EventManager::add_subscriber),
py::arg("topic"), py::arg("component"), py::arg("callback"), py::arg("task_config"),
"*\n * @brief Register a subscriber for \\p component on \\p topic.\n * @param "
"topic Topic name for the data being subscribed to.\n * @param component Name of "
"the component publishing data.\n * @param callback The event_callback_fn to be "
"called when receicing data on\n * \\p topic.\n * @param task_config The "
"task configuration to use for the subscriber.\n * @note The task_config is only "
"used if a subscriber is not already\n * registered for that topic. If a "
"subscriber is already registered for\n * that topic, the task_config is "
"ignored.\n * @return True if the subscriber was added, False if it was already\n "
" * registered for that component.\n")
.def("publish", &espp::EventManager::publish, py::arg("topic"), py::arg("data"),
"*\n * @brief Publish \\p data on \\p topic.\n * @param topic Topic to publish "
"data on.\n * @param data Data to publish, within a vector container.\n * "
"@return True if \\p data was successfully published to \\p topic, False\n * "
" otherwise. Publish will not occur (and will return False) if\n * "
"there are no subscribers for this topic.\n")
.def("remove_publisher", &espp::EventManager::remove_publisher, py::arg("topic"),
py::arg("component"),
"*\n * @brief Remove \\p component's publisher for \\p topic.\n * @param topic "
"The topic that \\p component was publishing on.\n * @param component The "
"component for which the publisher was registered.\n * @return True if the "
"publisher was removed, False if it was not\n * registered.\n")
.def("remove_subscriber", &espp::EventManager::remove_subscriber, py::arg("topic"),
py::arg("component"),
"*\n * @brief Remove \\p component's subscriber for \\p topic.\n * @param topic "
"The topic that \\p component was subscribing to.\n * @param component The "
"component for which the subscriber was registered.\n * @return True if the "
"subscriber was removed, False if it was not\n * registered.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassFtpServer =
py::class_(m, "FtpServer", py::dynamic_attr(),
"/ \\brief A class that implements a FTP server.")
.def(py::init(),
py::arg("ip_address"), py::arg("port"), py::arg("root"),
"/ \\brief A class that implements a FTP server.\n/ \note The IP Address is not "
"currently used to select the right\n/ interface, but is instead passed to "
"the FtpClientSession so that\n/ it can be used in the PASV command.\n/ "
"\\param ip_address The IP address to listen on.\n/ \\param port The port to listen "
"on.\n/ \\param root The root directory of the FTP server.")
.def("start", &espp::FtpServer::start,
"/ \\brief Start the FTP server.\n/ Bind to the port and start accepting "
"connections.\n/ \\return True if the server was started, False otherwise.",
py::call_guard())
.def("stop", &espp::FtpServer::stop, "/ \\brief Stop the FTP server.",
py::call_guard());
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassLogger = py::class_(
m, "Logger", py::dynamic_attr(),
"*\n * @brief Logger provides a wrapper around nicer / more robust formatting than\n * "
"standard ESP_LOG* macros with the ability to change the log level at\n * run-time. Logger "
"currently is a light wrapper around libfmt (future\n * std::format).\n *\n * To save on "
"code size, the logger has the ability to be compiled out based on\n * the log level set in "
"the sdkconfig. This means that if the log level is set to\n * ERROR, all debug, info, and "
"warn logs will be compiled out. This is done by\n * checking the log level at compile time "
"and only compiling in the functions\n * that are needed.\n *\n * The logger can also be "
"compiled with support for cursor commands. This allows\n * the logger to move the cursor "
"up, down, clear the line, clear the screen, and\n * move the cursor to a specific position. "
"This can be useful for creating\n * various types of interactive output or to maintian "
"context with long-running\n * logs.\n *\n * \\section logger_ex1 Basic Example\n * "
"\\snippet logger_example.cpp Logger example\n * \\section logger_ex2 Threaded Logging and "
"Verbosity Example\n * \\snippet logger_example.cpp MultiLogger example\n * \\section "
"logger_ex3 Cursor Commands Example\n * \\snippet logger_example.cpp Cursor Commands "
"example\n");
{ // inner classes & enums of Logger
auto pyEnumVerbosity =
py::enum_(
pyClassLogger, "Verbosity", py::arithmetic(),
"*\n * Verbosity levels for the logger, in order of increasing priority.\n")
.value("debug", espp::Logger::Verbosity::DEBUG, "*< Debug level verbosity.")
.value("info", espp::Logger::Verbosity::INFO, "*< Info level verbosity.")
.value("warn", espp::Logger::Verbosity::WARN, "*< Warn level verbosity.")
.value("error", espp::Logger::Verbosity::ERROR, "*< Error level verbosity.")
.value("none", espp::Logger::Verbosity::NONE,
"*< No verbosity - logger will not print anything.");
auto pyClassLogger_ClassConfig =
py::class_(pyClassLogger, "Config", py::dynamic_attr(),
"*\n * @brief Configuration struct for the logger.\n")
.def(py::init(
[](std::string_view tag = std::string_view(), bool include_time = {true},
std::chrono::duration rate_limit = std::chrono::duration(0),
espp::Logger::Verbosity level = espp::Logger::Verbosity::WARN) {
auto r_ctor_ = std::make_unique();
r_ctor_->tag = tag;
r_ctor_->include_time = include_time;
r_ctor_->rate_limit = rate_limit;
r_ctor_->level = level;
return r_ctor_;
}),
py::arg("tag") = std::string_view(), py::arg("include_time") = bool{true},
py::arg("rate_limit") = std::chrono::duration(0),
py::arg("level") = espp::Logger::Verbosity::WARN)
.def_readwrite("tag", &espp::Logger::Config::tag,
"*< The TAG that will be prepended to all logs.")
.def_readwrite("include_time", &espp::Logger::Config::include_time,
"*< Include the time in the log.")
.def_readwrite(
"rate_limit", &espp::Logger::Config::rate_limit,
"*< The rate limit for the logger. Optional, if 0\n");
m.def("sgn", py::overload_cast(espp::sgn), py::arg("x"),
"*\n * @brief Get the sign of a number (+1, 0, or -1)\n * @param x Value to get the sign "
"of\n * @return Sign of x: -1 if x < 0, 0 if x == 0, or +1 if x > 0\n");
m.def("lerp", espp::lerp, py::arg("a"), py::arg("b"), py::arg("t"),
"*\n * @brief Linear interpolation between two values.\n * @param a First value.\n * "
"@param b Second value.\n * @param t Interpolation factor in the range [0, 1].\n * @return "
"Linear interpolation between a and b.\n");
m.def("inv_lerp", espp::inv_lerp, py::arg("a"), py::arg("b"), py::arg("v"),
"*\n * @brief Compute the inverse lerped value.\n * @param a First value (usually the "
"lower of the two).\n * @param b Second value (usually the higher of the two).\n * @param "
"v Value to inverse lerp (usually a value between a and b).\n * @return Inverse lerp "
"value, the factor of v between a and b in the range [0,\n * 1] if v is between a "
"and b, 0 if v == a, or 1 if v == b. If a == b,\n * 0 is returned. If v is outside "
"the range [a, b], the value is\n * extrapolated linearly (i.e. if v < a, the "
"value is less than 0, if v\n * > b, the value is greater than 1).\n");
m.def(
"piecewise_linear", espp::piecewise_linear, py::arg("points"), py::arg("x"),
"*\n * @brief Compute the piecewise linear interpolation between a set of points.\n * @param "
"points Vector of points to interpolate between. The vector should be\n * "
"sorted by the first value in the pair. The first value in the\n * pair is the "
"x value and the second value is the y value. The x\n * values should be "
"unique. The function will interpolate between\n * the points using linear "
"interpolation. If x is less than the\n * first x value, the first y value is "
"returned. If x is greater\n * than the last x value, the last y value is "
"returned. If x is\n * between two x values, the y value is interpolated "
"between the\n * two y values.\n * @param x Value to interpolate at. Should be "
"a value from the first\n * distribution of the points (the domain). If x is "
"outside the domain\n * of the points, the value returned will be clamped to the "
"first or\n * last y value.\n * @return Interpolated value at x.\n");
m.def("round", espp::round, py::arg("x"),
"*\n * @brief Round x to the nearest integer.\n * @param x Floating point value to be "
"rounded.\n * @return Nearest integer to x.\n");
m.def("fast_ln", espp::fast_ln, py::arg("x"),
"*\n * @brief fast natural log function, ln(x).\n * @note This speed hack comes from:\n * "
" https://gist.github.com/LingDong-/7e4c4cae5cbbc44400a05fba65f06f23\n * @param x Value to "
"take the natural log of.\n * @return ln(x)\n");
m.def("fast_sin", espp::fast_sin, py::arg("angle"),
"*\n * @brief Fast approximation of sin(angle) (radians).\n * @note \\p Angle must be in "
"the range [0, 2PI].\n * @param angle Angle in radians [0, 2*PI]\n * @return Approximation "
"of sin(value)\n");
m.def("fast_cos", espp::fast_cos, py::arg("angle"),
"*\n * @brief Fast approximation of cos(angle) (radians).\n * @note \\p Angle must be in "
"the range [0, 2PI].\n * @param angle Angle in radians [0, 2*PI]\n * @return Approximation "
"of cos(value)\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassGaussian = py::class_(
m, "Gaussian", py::dynamic_attr(),
"*\n * @brief Implements a gaussian function\n * "
"\\f$y(t)=\\alpha\\exp(-\\frac{(t-\\beta)^2}{2\\gamma^2})\\f$.\n * @details Alows you to "
"store the alpha, beta, and gamma coefficients as well\n * as update them "
"dynamically.\n *\n * \\section gaussian_ex1 Example\n * \\snippet math_example.cpp gaussian "
"example\n * \\section gaussian_ex2 Fade-In/Fade-Out Example\n * \\snippet math_example.cpp "
"gaussian fade in fade out example\n");
{ // inner classes & enums of Gaussian
auto pyClassGaussian_ClassConfig =
py::class_(
pyClassGaussian, "Config", py::dynamic_attr(),
"*\n * @brief Configuration structure for initializing the gaussian.\n")
.def(py::init([](float gamma = float(), float alpha = {1.0f}, float beta = {0.5f}) {
auto r_ctor_ = std::make_unique();
r_ctor_->gamma = gamma;
r_ctor_->alpha = alpha;
r_ctor_->beta = beta;
return r_ctor_;
}),
py::arg("gamma") = float(), py::arg("alpha") = float{1.0f},
py::arg("beta") = float{0.5f})
.def_readwrite(
"gamma", &espp::Gaussian::Config::gamma,
"/< Slope of the gaussian, range [0, 1]. 0 is more of a thin spike from 0 up to")
.def_readwrite("alpha", &espp::Gaussian::Config::alpha,
"/< Max amplitude of the gaussian output, defautls to 1.0.")
.def_readwrite(
"beta", &espp::Gaussian::Config::beta,
"/< Beta value for the gaussian, default to be symmetric at 0.5 in range [0,1].")
.def("__eq__", &espp::Gaussian::Config::operator==, py::arg("rhs"));
} // end of inner classes & enums of Gaussian
pyClassGaussian.def(py::init())
.def("__call__", &espp::Gaussian::operator(), py::arg("t"),
"*\n * @brief Evaluate the gaussian at \\p t.\n * @note Convienience wrapper around "
"the at() method.\n * @param t The evaluation parameter, [0, 1].\n * @return The "
"gaussian evaluated at \\p t.\n")
.def("update", &espp::Gaussian::update, py::arg("config"),
"*\n * @brief Update the gaussian configuration.\n * @param config The new "
"configuration.\n")
.def("set_config", &espp::Gaussian::set_config, py::arg("config"),
"*\n * @brief Set the configuration of the gaussian.\n * @param config The new "
"configuration.\n")
.def("get_config", &espp::Gaussian::get_config,
"*\n * @brief Get the current configuration of the gaussian.\n * @return The "
"current configuration.\n")
.def("get_gamma", &espp::Gaussian::get_gamma,
"*\n * @brief Get the gamma value.\n * @return The gamma value.\n")
.def("get_alpha", &espp::Gaussian::get_alpha,
"*\n * @brief Get the alpha value.\n * @return The alpha value.\n")
.def("get_beta", &espp::Gaussian::get_beta,
"*\n * @brief Get the beta value.\n * @return The beta value.\n")
.def("set_gamma", &espp::Gaussian::set_gamma, py::arg("gamma"),
"*\n * @brief Set the gamma value.\n * @param gamma The new gamma value.\n")
.def("set_alpha", &espp::Gaussian::set_alpha, py::arg("alpha"),
"*\n * @brief Set the alpha value.\n * @param alpha The new alpha value.\n")
.def("set_beta", &espp::Gaussian::set_beta, py::arg("beta"),
"*\n * @brief Set the beta value.\n * @param beta The new beta value.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRangeMapper_int = py::class_(
m, "RangeMapper_int", py::dynamic_attr(),
"*\n * @brief Template class for converting a value from an uncentered [minimum,\n * "
"maximum] range into a centered output range (default [-1,1]). If\n * provided a "
"non-zero deadband, it will convert all values within\n * [center-deadband, "
"center+deadband] to be the configured\n * output_center (default 0).\n *\n * "
"The RangeMapper can be optionally configured to invert the input,\n * so that it "
"will compute the input w.r.t. the configured min/max of\n * the input range when "
"mapping to the output range - this will mean\n * that a values within the ranges "
"[minimum, minimum+deadband] and\n * [maximum-deadband, maximum] will all map to the "
"output_center and\n * the input center will map to both output_max and output_min\n "
"* depending on the sign of the input.\n *\n * @tparam T Numeric type to use for the "
"input and output values.\n *\n * @note When inverting the input range, you are introducing "
"a discontinuity\n * between the input distribution and the output distribution at "
"the\n * input center. Noise around the input's center value will create\n * "
"oscillations in the output which will jump between output maximum\n * and output "
"minimum. Therefore it is advised to use \\p invert_input\n * sparignly, and to set "
"the values robustly.\n *\n * The RangeMapper can be optionally configured to invert "
"the output,\n * so that after converting from the input range to the output range,\n "
"* it will flip the sign on the output.\n *\n * \\section range_mapper_ex1 Example\n "
"* \\snippet math_example.cpp range_mapper example\n");
{ // inner classes & enums of RangeMapper_int
auto pyClassRangeMapper_ClassConfig =
py::class_(
pyClassRangeMapper_int, "Config", py::dynamic_attr(),
"*\n * @brief Configuration for the input uncentered range with optional\n * "
"values for the centered output range, default values of 0 output center\n * and 1 "
"output range provide a default output range between [-1, 1].\n")
.def(py::init([](int center = int(), int center_deadband = 0, int minimum = int(),
int maximum = int(), int range_deadband = 0, int output_center = 0,
int output_range = 1, bool invert_output = false) {
auto r_ctor_ = std::make_unique();
r_ctor_->center = center;
r_ctor_->center_deadband = center_deadband;
r_ctor_->minimum = minimum;
r_ctor_->maximum = maximum;
r_ctor_->range_deadband = range_deadband;
r_ctor_->output_center = output_center;
r_ctor_->output_range = output_range;
r_ctor_->invert_output = invert_output;
return r_ctor_;
}),
py::arg("center") = int(), py::arg("center_deadband") = 0,
py::arg("minimum") = int(), py::arg("maximum") = int(),
py::arg("range_deadband") = 0, py::arg("output_center") = 0,
py::arg("output_range") = 1, py::arg("invert_output") = false)
.def_readwrite("center", &espp::RangeMapper::Config::center,
"*< Center value for the input range.")
.def_readwrite("center_deadband", &espp::RangeMapper::Config::center_deadband,
"*< Deadband amount around (+-) the center for which output will be 0.")
.def_readwrite("minimum", &espp::RangeMapper::Config::minimum,
"*< Minimum value for the input range.")
.def_readwrite("maximum", &espp::RangeMapper::Config::maximum,
"*< Maximum value for the input range.")
.def_readwrite("range_deadband", &espp::RangeMapper::Config::range_deadband,
"*< Deadband amount around the minimum and maximum for which output "
"will\n be min/max output.")
.def_readwrite("output_center", &espp::RangeMapper::Config::output_center,
"*< The center for the output. Default 0.")
.def_readwrite(
"output_range", &espp::RangeMapper::Config::output_range,
"*< The range (+/-) from the center for the output. Default 1. @note Will\n "
" be passed through std::abs() to ensure it is positive.")
.def_readwrite("invert_output", &espp::RangeMapper::Config::invert_output,
"*< Whether to invert the output (default False). @note If True will "
"flip the sign\n of the output after converting from "
"the input distribution.");
} // end of inner classes & enums of RangeMapper_int
pyClassRangeMapper_int.def(py::init())
.def("get_center_deadband", &espp::RangeMapper::get_center_deadband,
"*\n * @brief Return the configured deadband around the center of the input\n * "
" distribution\n * @return Deadband around the center of the input distribution for "
"this\n * range mapper.\n")
.def("get_minimum", &espp::RangeMapper::get_minimum,
"*\n * @brief Return the configured minimum of the input distribution\n * @return "
"Minimum of the input distribution for this range mapper.\n")
.def("get_maximum", &espp::RangeMapper::get_maximum,
"*\n * @brief Return the configured maximum of the input distribution\n * @return "
"Maximum of the input distribution for this range mapper.\n")
.def(
"get_range", &espp::RangeMapper::get_range,
"*\n * @brief Return the configured range of the input distribution\n * @note Always "
"positive.\n * @return Range of the input distribution for this range mapper.\n")
.def("get_range_deadband", &espp::RangeMapper::get_range_deadband,
"*\n * @brief Return the configured deadband around the min/max of the input\n * "
" distribution\n * @return Deadband around the min/max of the input distribution "
"for this\n * range mapper.\n")
.def("get_output_center", &espp::RangeMapper::get_output_center,
"*\n * @brief Return the configured center of the output distribution\n * @return "
"Center of the output distribution for this range mapper.\n")
.def("get_output_range", &espp::RangeMapper::get_output_range,
"*\n * @brief Return the configured range of the output distribution\n * @note "
"Always positive.\n * @return Range of the output distribution for this range "
"mapper.\n")
.def("get_output_min", &espp::RangeMapper::get_output_min,
"*\n * @brief Return the configured minimum of the output distribution\n * @return "
"Minimum of the output distribution for this range mapper.\n")
.def("get_output_max", &espp::RangeMapper::get_output_max,
"*\n * @brief Return the configured maximum of the output distribution\n * @return "
"Maximum of the output distribution for this range mapper.\n")
.def("set_center_deadband", &espp::RangeMapper::set_center_deadband, py::arg("deadband"),
"*\n * @brief Set the deadband around the center of the input distribution.\n * "
"@param deadband The deadband to use around the center of the input\n * "
"distribution.\n * @note The deadband must be non-negative.\n * @note The deadband "
"is applied around the center value of the input\n * distribution.\n")
.def("set_range_deadband", &espp::RangeMapper::set_range_deadband, py::arg("deadband"),
"*\n * @brief Set the deadband around the min/max of the input distribution.\n * "
"@param deadband The deadband to use around the min/max of the input\n * "
"distribution.\n * @note The deadband must be non-negative.\n * @note The deadband "
"is applied around the min/max values of the input\n * distribution.\n")
.def("map", &espp::RangeMapper::map, py::arg("v"),
"*\n * @brief Map a value \\p v from the input distribution into the configured\n * "
" output range (centered, default [-1,1]).\n * @param v Value from the "
"(possibly uncentered and possibly inverted -\n * defined by the previously "
"configured Config) input distribution\n * @return Value within the centered output "
"distribution.\n")
.def("unmap", &espp::RangeMapper::unmap, py::arg("v"),
"*\n * @brief Unmap a value \\p v from the configured output range (centered,\n * "
" default [-1,1]) back into the input distribution.\n * @param v Value from the "
"centered output distribution.\n * @return Value within the input distribution.\n");
auto pyClassRangeMapper_float = py::class_(
m, "RangeMapper_float", py::dynamic_attr(),
"*\n * @brief Template class for converting a value from an uncentered [minimum,\n * "
"maximum] range into a centered output range (default [-1,1]). If\n * provided a "
"non-zero deadband, it will convert all values within\n * [center-deadband, "
"center+deadband] to be the configured\n * output_center (default 0).\n *\n * "
"The RangeMapper can be optionally configured to invert the input,\n * so that it "
"will compute the input w.r.t. the configured min/max of\n * the input range when "
"mapping to the output range - this will mean\n * that a values within the ranges "
"[minimum, minimum+deadband] and\n * [maximum-deadband, maximum] will all map to the "
"output_center and\n * the input center will map to both output_max and output_min\n "
"* depending on the sign of the input.\n *\n * @tparam T Numeric type to use for the "
"input and output values.\n *\n * @note When inverting the input range, you are introducing "
"a discontinuity\n * between the input distribution and the output distribution at "
"the\n * input center. Noise around the input's center value will create\n * "
"oscillations in the output which will jump between output maximum\n * and output "
"minimum. Therefore it is advised to use \\p invert_input\n * sparignly, and to set "
"the values robustly.\n *\n * The RangeMapper can be optionally configured to invert "
"the output,\n * so that after converting from the input range to the output range,\n "
"* it will flip the sign on the output.\n *\n * \\section range_mapper_ex1 Example\n "
"* \\snippet math_example.cpp range_mapper example\n");
{ // inner classes & enums of RangeMapper_float
auto pyClassRangeMapper_ClassConfig =
py::class_(
pyClassRangeMapper_float, "Config", py::dynamic_attr(),
"*\n * @brief Configuration for the input uncentered range with optional\n * "
"values for the centered output range, default values of 0 output center\n * and 1 "
"output range provide a default output range between [-1, 1].\n")
.def(py::init([](float center = float(), float center_deadband = 0,
float minimum = float(), float maximum = float(),
float range_deadband = 0, float output_center = 0,
float output_range = 1, bool invert_output = false) {
auto r_ctor_ = std::make_unique();
r_ctor_->center = center;
r_ctor_->center_deadband = center_deadband;
r_ctor_->minimum = minimum;
r_ctor_->maximum = maximum;
r_ctor_->range_deadband = range_deadband;
r_ctor_->output_center = output_center;
r_ctor_->output_range = output_range;
r_ctor_->invert_output = invert_output;
return r_ctor_;
}),
py::arg("center") = float(), py::arg("center_deadband") = 0,
py::arg("minimum") = float(), py::arg("maximum") = float(),
py::arg("range_deadband") = 0, py::arg("output_center") = 0,
py::arg("output_range") = 1, py::arg("invert_output") = false)
.def_readwrite("center", &espp::RangeMapper::Config::center,
"*< Center value for the input range.")
.def_readwrite("center_deadband", &espp::RangeMapper::Config::center_deadband,
"*< Deadband amount around (+-) the center for which output will be 0.")
.def_readwrite("minimum", &espp::RangeMapper::Config::minimum,
"*< Minimum value for the input range.")
.def_readwrite("maximum", &espp::RangeMapper::Config::maximum,
"*< Maximum value for the input range.")
.def_readwrite("range_deadband", &espp::RangeMapper::Config::range_deadband,
"*< Deadband amount around the minimum and maximum for which output "
"will\n be min/max output.")
.def_readwrite("output_center", &espp::RangeMapper::Config::output_center,
"*< The center for the output. Default 0.")
.def_readwrite(
"output_range", &espp::RangeMapper::Config::output_range,
"*< The range (+/-) from the center for the output. Default 1. @note Will\n "
" be passed through std::abs() to ensure it is positive.")
.def_readwrite("invert_output", &espp::RangeMapper::Config::invert_output,
"*< Whether to invert the output (default False). @note If True will "
"flip the sign\n of the output after converting from "
"the input distribution.");
} // end of inner classes & enums of RangeMapper_float
pyClassRangeMapper_float.def(py::init())
.def("get_center_deadband", &espp::RangeMapper::get_center_deadband,
"*\n * @brief Return the configured deadband around the center of the input\n * "
" distribution\n * @return Deadband around the center of the input distribution for "
"this\n * range mapper.\n")
.def("get_minimum", &espp::RangeMapper::get_minimum,
"*\n * @brief Return the configured minimum of the input distribution\n * @return "
"Minimum of the input distribution for this range mapper.\n")
.def("get_maximum", &espp::RangeMapper::get_maximum,
"*\n * @brief Return the configured maximum of the input distribution\n * @return "
"Maximum of the input distribution for this range mapper.\n")
.def(
"get_range", &espp::RangeMapper::get_range,
"*\n * @brief Return the configured range of the input distribution\n * @note Always "
"positive.\n * @return Range of the input distribution for this range mapper.\n")
.def("get_range_deadband", &espp::RangeMapper::get_range_deadband,
"*\n * @brief Return the configured deadband around the min/max of the input\n * "
" distribution\n * @return Deadband around the min/max of the input distribution "
"for this\n * range mapper.\n")
.def("get_output_center", &espp::RangeMapper::get_output_center,
"*\n * @brief Return the configured center of the output distribution\n * @return "
"Center of the output distribution for this range mapper.\n")
.def("get_output_range", &espp::RangeMapper::get_output_range,
"*\n * @brief Return the configured range of the output distribution\n * @note "
"Always positive.\n * @return Range of the output distribution for this range "
"mapper.\n")
.def("get_output_min", &espp::RangeMapper::get_output_min,
"*\n * @brief Return the configured minimum of the output distribution\n * @return "
"Minimum of the output distribution for this range mapper.\n")
.def("get_output_max", &espp::RangeMapper::get_output_max,
"*\n * @brief Return the configured maximum of the output distribution\n * @return "
"Maximum of the output distribution for this range mapper.\n")
.def("set_center_deadband", &espp::RangeMapper::set_center_deadband,
py::arg("deadband"),
"*\n * @brief Set the deadband around the center of the input distribution.\n * "
"@param deadband The deadband to use around the center of the input\n * "
"distribution.\n * @note The deadband must be non-negative.\n * @note The deadband "
"is applied around the center value of the input\n * distribution.\n")
.def("set_range_deadband", &espp::RangeMapper::set_range_deadband, py::arg("deadband"),
"*\n * @brief Set the deadband around the min/max of the input distribution.\n * "
"@param deadband The deadband to use around the min/max of the input\n * "
"distribution.\n * @note The deadband must be non-negative.\n * @note The deadband "
"is applied around the min/max values of the input\n * distribution.\n")
.def("map", &espp::RangeMapper::map, py::arg("v"),
"*\n * @brief Map a value \\p v from the input distribution into the configured\n * "
" output range (centered, default [-1,1]).\n * @param v Value from the "
"(possibly uncentered and possibly inverted -\n * defined by the previously "
"configured Config) input distribution\n * @return Value within the centered output "
"distribution.\n")
.def("unmap", &espp::RangeMapper::unmap, py::arg("v"),
"*\n * @brief Unmap a value \\p v from the configured output range (centered,\n * "
" default [-1,1]) back into the input distribution.\n * @param v Value from the "
"centered output distribution.\n * @return Value within the input distribution.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassVector2d_int =
py::class_(
m, "Vector2d_int", py::dynamic_attr(),
"*\n * @brief Container representing a 2 dimensional vector.\n *\n * Provides "
"getters/setters, index operator, and vector / scalar math\n * utilities.\n *\n * "
"\\section vector_ex1 Example\n * \\snippet math_example.cpp vector2d example\n")
.def(py::init(), py::arg("x") = 0, py::arg("y") = 0,
"*\n * @brief Constructor for the vector, defaults to 0,0.\n * @param x The "
"starting X value.\n * @param y The starting Y value.\n")
.def(py::init(), py::arg("other"),
"*\n * @brief Vector copy constructor.\n * @param other Vector to copy.\n")
.def("magnitude", &espp::Vector2d::magnitude,
"*\n * @brief Returns vector magnitude: ||v||.\n * @return The magnitude.\n")
.def("magnitude_squared", &espp::Vector2d::magnitude_squared,
"*\n * @brief Returns vector magnitude squared: ||v||^2.\n * @return The "
"magnitude squared.\n")
.def(
"x", [](espp::Vector2d &self) { return self.x(); },
"*\n * @brief Getter for the x value.\n * @return The current x value.\n")
.def("x", py::overload_cast(&espp::Vector2d::x), py::arg("v"),
"*\n * @brief Setter for the x value.\n * @param v New value for \\c x.\n")
.def(
"y", [](espp::Vector2d &self) { return self.y(); },
"*\n * @brief Getter for the y value.\n * @return The current y value.\n")
.def("y", py::overload_cast(&espp::Vector2d::y), py::arg("v"),
"*\n * @brief Setter for the y value.\n * @param v New value for \\c y.\n")
.def(
"__lt__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) < 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__le__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) == 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__ge__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) >= 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__gt__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) > 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def("__eq__", &espp::Vector2d::operator==, py::arg("other"),
"*\n * @brief Equality operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return True if the vectors are equal, False "
"otherwise.\n")
.def("__getitem__", &espp::Vector2d::operator[], py::arg("index"),
"*\n * @brief Index operator for vector elements.\n * @note Returns a mutable "
"reference to the element.\n * @param index The index to return.\n * @return "
"Mutable reference to the element at \\p index.\n")
.def(
"__neg__", [](espp::Vector2d &self) { return self.operator-(); },
"*\n * @brief Negate the vector.\n * @return The new vector which is the "
"negative.\n")
.def("__sub__",
py::overload_cast(&espp::Vector2d::operator-,
py::const_),
py::arg("rhs"),
"*\n * @brief Return a new vector which is the provided vector subtracted from\n "
" * this vector.\n * @param rhs The vector to subtract from this vector.\n "
" * @return Resultant vector subtraction.\n")
.def("__isub__", &espp::Vector2d::operator-=, py::arg("rhs"),
"*\n * @brief Return the provided vector subtracted from this vector.\n * "
"@param rhs The vector to subtract from this vector.\n * @return Resultant vector "
"subtraction.\n")
.def("__add__", &espp::Vector2d::operator+, py::arg("rhs"),
"*\n * @brief Return a new vector, which is the addition of this vector and the\n "
" * provided vector.\n * @param rhs The vector to add to this vector.\n "
"* @return Resultant vector addition.\n")
.def("__iadd__", &espp::Vector2d::operator+=, py::arg("rhs"),
"*\n * @brief Return the vector added with the provided vector.\n * @param rhs "
"The vector to add to this vector.\n * @return Resultant vector addition.\n")
.def("__mul__", &espp::Vector2d::operator*, py::arg("v"),
"*\n * @brief Return a scaled version of the vector, multiplied by the provided\n "
" * value.\n * @param v Value the vector should be multiplied by.\n * "
"@return Resultant scaled vector.\n")
.def("__imul__", &espp::Vector2d::operator*=, py::arg("v"),
"*\n * @brief Return the vector multiplied by the provided value.\n * @param v "
"Value the vector should be scaled by.\n * @return Resultant scaled vector.\n")
.def("__truediv__",
py::overload_cast(&espp::Vector2d::operator/, py::const_),
py::arg("v"),
"*\n * @brief Return a scaled version of the vector, divided by the provided\n "
"* value.\n * @param v Value the vector should be divided by.\n * "
"@return Resultant scaled vector.\n")
.def("__truediv__",
py::overload_cast(&espp::Vector2d::operator/,
py::const_),
py::arg("v"),
"*\n * @brief Return a scaled version of the vector, divided by the provided\n "
"* vector value. Scales x and y independently.\n * @param v Vector values "
"the vector should be divided by.\n * @return Resultant scaled vector.\n")
.def("__itruediv__", py::overload_cast(&espp::Vector2d::operator/=),
py::arg("v"),
"*\n * @brief Return the vector divided by the provided value.\n * @param v "
"Value the vector should be divided by.\n * @return Resultant scaled vector.\n")
.def("__itruediv__",
py::overload_cast(&espp::Vector2d::operator/=),
py::arg("v"),
"*\n * @brief Return the vector divided by the provided vector values.\n * "
"@param v Vector of values the vector should be divided by.\n * @return Resultant "
"scaled vector.\n")
.def("dot", &espp::Vector2d::dot, py::arg("other"),
"*\n * @brief Dot product of this vector with another vector.\n * @param other "
"The second vector\n * @return The dot product (x1*x2 + y1*y2)\n")
.def("normalized", &espp::Vector2d::normalized,
"*\n * @brief Return normalized (unit length) version of the vector.\n * "
"@return The normalized vector.\n");
auto pyClassVector2d_float =
py::class_(
m, "Vector2d_float", py::dynamic_attr(),
"*\n * @brief Container representing a 2 dimensional vector.\n *\n * Provides "
"getters/setters, index operator, and vector / scalar math\n * utilities.\n *\n * "
"\\section vector_ex1 Example\n * \\snippet math_example.cpp vector2d example\n")
.def(py::init(), py::arg("x") = 0, py::arg("y") = 0,
"*\n * @brief Constructor for the vector, defaults to 0,0.\n * @param x The "
"starting X value.\n * @param y The starting Y value.\n")
.def(py::init(), py::arg("other"),
"*\n * @brief Vector copy constructor.\n * @param other Vector to copy.\n")
.def("magnitude", &espp::Vector2d::magnitude,
"*\n * @brief Returns vector magnitude: ||v||.\n * @return The magnitude.\n")
.def("magnitude_squared", &espp::Vector2d::magnitude_squared,
"*\n * @brief Returns vector magnitude squared: ||v||^2.\n * @return The "
"magnitude squared.\n")
.def(
"x", [](espp::Vector2d &self) { return self.x(); },
"*\n * @brief Getter for the x value.\n * @return The current x value.\n")
.def("x", py::overload_cast(&espp::Vector2d::x), py::arg("v"),
"*\n * @brief Setter for the x value.\n * @param v New value for \\c x.\n")
.def(
"y", [](espp::Vector2d &self) { return self.y(); },
"*\n * @brief Getter for the y value.\n * @return The current y value.\n")
.def("y", py::overload_cast(&espp::Vector2d::y), py::arg("v"),
"*\n * @brief Setter for the y value.\n * @param v New value for \\c y.\n")
.def(
"__lt__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) < 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__le__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) == 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__ge__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) >= 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def(
"__gt__",
[](const espp::Vector2d &self, const espp::Vector2d &other) -> bool {
auto cmp = [&self](auto &&other) -> bool { return self.operator(other) > 0; };
return cmp(other);
},
py::arg("other"),
"*\n * @brief Spaceship operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return -1 if this vector is less than \\p other, "
"0 if they are equal, 1 if\n * this vector is greater than \\p other.\n")
.def("__eq__", &espp::Vector2d::operator==, py::arg("other"),
"*\n * @brief Equality operator for comparing two vectors.\n * @param other The "
"vector to compare against.\n * @return True if the vectors are equal, False "
"otherwise.\n")
.def("__getitem__", &espp::Vector2d::operator[], py::arg("index"),
"*\n * @brief Index operator for vector elements.\n * @note Returns a mutable "
"reference to the element.\n * @param index The index to return.\n * @return "
"Mutable reference to the element at \\p index.\n")
.def(
"__neg__", [](espp::Vector2d &self) { return self.operator-(); },
"*\n * @brief Negate the vector.\n * @return The new vector which is the "
"negative.\n")
.def("__sub__",
py::overload_cast(&espp::Vector2d::operator-,
py::const_),
py::arg("rhs"),
"*\n * @brief Return a new vector which is the provided vector subtracted from\n "
" * this vector.\n * @param rhs The vector to subtract from this vector.\n "
" * @return Resultant vector subtraction.\n")
.def("__isub__", &espp::Vector2d::operator-=, py::arg("rhs"),
"*\n * @brief Return the provided vector subtracted from this vector.\n * "
"@param rhs The vector to subtract from this vector.\n * @return Resultant vector "
"subtraction.\n")
.def("__add__", &espp::Vector2d::operator+, py::arg("rhs"),
"*\n * @brief Return a new vector, which is the addition of this vector and the\n "
" * provided vector.\n * @param rhs The vector to add to this vector.\n "
"* @return Resultant vector addition.\n")
.def("__iadd__", &espp::Vector2d::operator+=, py::arg("rhs"),
"*\n * @brief Return the vector added with the provided vector.\n * @param rhs "
"The vector to add to this vector.\n * @return Resultant vector addition.\n")
.def("__mul__", &espp::Vector2d::operator*, py::arg("v"),
"*\n * @brief Return a scaled version of the vector, multiplied by the provided\n "
" * value.\n * @param v Value the vector should be multiplied by.\n * "
"@return Resultant scaled vector.\n")
.def("__imul__", &espp::Vector2d::operator*=, py::arg("v"),
"*\n * @brief Return the vector multiplied by the provided value.\n * @param v "
"Value the vector should be scaled by.\n * @return Resultant scaled vector.\n")
.def("__truediv__",
py::overload_cast(&espp::Vector2d::operator/, py::const_),
py::arg("v"),
"*\n * @brief Return a scaled version of the vector, divided by the provided\n "
"* value.\n * @param v Value the vector should be divided by.\n * "
"@return Resultant scaled vector.\n")
.def("__truediv__",
py::overload_cast(&espp::Vector2d::operator/,
py::const_),
py::arg("v"),
"*\n * @brief Return a scaled version of the vector, divided by the provided\n "
"* vector value. Scales x and y independently.\n * @param v Vector values "
"the vector should be divided by.\n * @return Resultant scaled vector.\n")
.def("__itruediv__", py::overload_cast(&espp::Vector2d::operator/=),
py::arg("v"),
"*\n * @brief Return the vector divided by the provided value.\n * @param v "
"Value the vector should be divided by.\n * @return Resultant scaled vector.\n")
.def("__itruediv__",
py::overload_cast(&espp::Vector2d::operator/=),
py::arg("v"),
"*\n * @brief Return the vector divided by the provided vector values.\n * "
"@param v Vector of values the vector should be divided by.\n * @return Resultant "
"scaled vector.\n")
.def("dot", &espp::Vector2d::dot, py::arg("other"),
"*\n * @brief Dot product of this vector with another vector.\n * @param other "
"The second vector\n * @return The dot product (x1*x2 + y1*y2)\n")
.def("normalized", &espp::Vector2d::normalized,
"*\n * @brief Return normalized (unit length) version of the vector.\n * "
"@return The normalized vector.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassNdef = py::class_(
m, "Ndef", py::dynamic_attr(),
"*\n * @brief implements serialization & deserialization logic for NFC Data\n * "
"Exchange Format (NDEF) records which can be stored on and\n * transmitted from NFC "
"devices.\n *\n * @details NDEF records can be composed the following way:\n * "
"@code{.unparsed}\n * Bit 7 6 5 4 3 2 1 0\n * "
"------ ------ ------ ------ ------ ------ ------ ------\n * [ MB ] [ ME ] [ CF ] "
" [ SR ] [ IL ] [ TNF ]\n * [ TYPE LENGTH (may "
"be 0) ]\n * [ PAYLOAD LENGTH (1B or 4B, see SR) ]\n * "
" [ ID LENGTH (if IL) ]\n * [ "
" RECORD TYPE (if TYPE LENGTH > 0) ]\n * [ ID (if "
"IL) ]\n * [ PAYLOAD (payload length bytes)]\n "
"* @endcode\n *\n * The first byte (Flags) has these bits:\n * * Bits 0-3: TNF - Type "
"Name Format - describes record type (see TNF class)\n * * Bit 3: IL - ID Length - "
"indicates if the ID Length Field is present or not\n * * Bit 4: SR - Short Record - set to "
"1 if the payload length field is 1 byte (8\n * bits / 0-255) or less, otherwise the "
"payload length is 4 bytes\n * * Bit 5: CF - Chunk Flag - indicates if this is the first "
"record chunk or a\n * middle record chunk, set to 0 for the first record of the "
"message and\n * for subsequent records set to 1.\n * * Bit 6: ME - Message End - 1 "
"indicates if this is the last record in the\n * message\n * * Bit 7: MB - Message "
"Begin - 1 indicates if this is the first record in the\n * message\n *\n * @note Some "
"information about NDEF can be found:\n * * "
"https://www.maskaravivek.com/post/understanding-the-format-of-ndef-messages/\n * * "
"https://ndeflib.readthedocs.io/en/stable/records/bluetooth.html\n * * "
"https://developer.android.com/reference/android/nfc/NdefMessage\n * * "
"https://www.oreilly.com/library/view/beginning-nfc/9781449324094/ch04.html\n * * "
"https://learn.adafruit.com/adafruit-pn532-rfid-nfc/ndef\n *\n");
{ // inner classes & enums of Ndef
auto pyEnumTNF =
py::enum_(
pyClassNdef, "TNF", py::arithmetic(),
"*\n * @brief Type Name Format (TNF) field is a 3-bit value that describes the\n * "
" record type.\n *\n * Some Common TNF::WELL_KNOWN record type strings:\n "
"* * Text (T)\n * * URI (U)\n * * Smart Poster (Sp)\n * * Alternative "
"Carrier (ac)\n * * Handover Carrier (Hc)\n * * Handover Request (Hr)\n * "
"* Handover Select (Hs)\n")
.value("empty", espp::Ndef::TNF::EMPTY, "/< Record is empty")
.value("well_known", espp::Ndef::TNF::WELL_KNOWN,
"/< Type field contains a well-known RTD type name")
.value("mime_media", espp::Ndef::TNF::MIME_MEDIA,
"/< Type field contains a media type (RFC 2046)")
.value("absolute_uri", espp::Ndef::TNF::ABSOLUTE_URI,
"/< Type field contains an absolute URI (RFC 3986)")
.value("external_type", espp::Ndef::TNF::EXTERNAL_TYPE,
"/< Type field Contains an external type name")
.value("unknown", espp::Ndef::TNF::UNKNOWN,
"/< Payload type is unknown, type length must be 0.")
.value("unchanged", espp::Ndef::TNF::UNCHANGED,
"/< Indicates the payload is an intermediate or final chunk of a chunked NDEF")
.value("reserved", espp::Ndef::TNF::RESERVED,
"/< Reserved by the NFC forum for future use");
auto pyEnumUic =
py::enum_(
pyClassNdef, "Uic", py::arithmetic(),
"*\n * URI Identifier Codes (UIC), See Table A-3 at\n * "
"https://www.oreilly.com/library/view/beginning-nfc/9781449324094/apa.html\n * and "
"https://learn.adafruit.com/adafruit-pn532-rfid-nfc/ndef\n")
.value("none", espp::Ndef::Uic::NONE, "/< Exactly as written")
.value("http_www", espp::Ndef::Uic::HTTP_WWW, "/< http://www.")
.value("https_www", espp::Ndef::Uic::HTTPS_WWW, "/< https://www.")
.value("http", espp::Ndef::Uic::HTTP, "/< http://")
.value("https", espp::Ndef::Uic::HTTPS, "/< https://")
.value("tel", espp::Ndef::Uic::TEL, "/< tel:")
.value("mailto", espp::Ndef::Uic::MAILTO, "/< mailto:")
.value("ftp_anon", espp::Ndef::Uic::FTP_ANON, "/< ftp://anonymous:anonymous@")
.value("ftp_ftp", espp::Ndef::Uic::FTP_FTP, "/< ftp://ftp.")
.value("ftps", espp::Ndef::Uic::FTPS, "/< ftps://")
.value("sftp", espp::Ndef::Uic::SFTP, "/< sftp://")
.value("smb", espp::Ndef::Uic::SMB, "/< smb://")
.value("nfs", espp::Ndef::Uic::NFS, "/< nfs://")
.value("ftp", espp::Ndef::Uic::FTP, "/< ftp://")
.value("dav", espp::Ndef::Uic::DAV, "/< dav://")
.value("news", espp::Ndef::Uic::NEWS, "/< news:")
.value("telnet", espp::Ndef::Uic::TELNET, "/< telnet://")
.value("imap", espp::Ndef::Uic::IMAP, "/< imap:")
.value("rstp", espp::Ndef::Uic::RSTP, "/< rtsp://")
.value("urn", espp::Ndef::Uic::URN, "/< urn:")
.value("pop", espp::Ndef::Uic::POP, "/< pop:")
.value("sip", espp::Ndef::Uic::SIP, "/< sip:")
.value("sips", espp::Ndef::Uic::SIPS, "/< sips:")
.value("tftp", espp::Ndef::Uic::TFTP, "/< tftp:")
.value("btspp", espp::Ndef::Uic::BTSPP, "/< btspp://")
.value("btl2_cap", espp::Ndef::Uic::BTL2CAP, "/< btl2cap://")
.value("btgoep", espp::Ndef::Uic::BTGOEP, "/< btgoep://")
.value("tcpobex", espp::Ndef::Uic::TCPOBEX, "/< tcpobex://")
.value("irdaobex", espp::Ndef::Uic::IRDAOBEX, "/< irdaobex://")
.value("file", espp::Ndef::Uic::FILE, "/< file://")
.value("urn_epc_id", espp::Ndef::Uic::URN_EPC_ID, "/< urn:epc:id:")
.value("urn_epc_tag", espp::Ndef::Uic::URN_EPC_TAG, "/< urn:epc:tag:")
.value("urn_epc_pat", espp::Ndef::Uic::URN_EPC_PAT, "/< urn:epc:pat:")
.value("urn_epc_raw", espp::Ndef::Uic::URN_EPC_RAW, "/< urn:epc:raw:")
.value("urn_epc", espp::Ndef::Uic::URN_EPC, "/< urn:epc:")
.value("urn_nfc", espp::Ndef::Uic::URN_NFC, "/< urn:nfc:");
auto pyEnumBtType = py::enum_(pyClassNdef, "BtType", py::arithmetic(),
"*\n * @brief Type of Bluetooth radios.\n")
.value("bredr", espp::Ndef::BtType::BREDR, "/< BT Classic")
.value("ble", espp::Ndef::BtType::BLE, "/< BT Low Energy");
auto pyEnumBtAppearance =
py::enum_(
pyClassNdef, "BtAppearance", py::arithmetic(),
"*\n * @brief Some appearance codes for BLE radios.\n")
.value("unknown", espp::Ndef::BtAppearance::UNKNOWN, "/< Generic Unknown")
.value("phone", espp::Ndef::BtAppearance::PHONE, "/< Generic Phone")
.value("computer", espp::Ndef::BtAppearance::COMPUTER, "/< Generic Computer")
.value("watch", espp::Ndef::BtAppearance::WATCH, "/< Generic Watch")
.value("clock", espp::Ndef::BtAppearance::CLOCK, "/< Generic Clock")
.value("display", espp::Ndef::BtAppearance::DISPLAY, "/< Generic Display")
.value("remote_control", espp::Ndef::BtAppearance::REMOTE_CONTROL,
"/< Generic Remote Control")
.value("generic_hid", espp::Ndef::BtAppearance::GENERIC_HID, "/< Generic HID")
.value("keyboard", espp::Ndef::BtAppearance::KEYBOARD, "/< HID Keyboard")
.value("mouse", espp::Ndef::BtAppearance::MOUSE, "/< HID Mouse")
.value("joystick", espp::Ndef::BtAppearance::JOYSTICK, "/< HID Joystick")
.value("gamepad", espp::Ndef::BtAppearance::GAMEPAD, "/< HID Gamepad")
.value("touchpad", espp::Ndef::BtAppearance::TOUCHPAD, "/< HID Touchpad")
.value("gaming", espp::Ndef::BtAppearance::GAMING, "/< Generic Gaming group");
auto pyEnumCarrierPowerState =
py::enum_(
pyClassNdef, "CarrierPowerState", py::arithmetic(),
"*\n * @brief Power state of a BLE radio.\n * @details Representation of the "
"carrier power state in a Handover Select\n * message.\n")
.value("inactive", espp::Ndef::CarrierPowerState::INACTIVE, "/< Carrier power is off")
.value("active", espp::Ndef::CarrierPowerState::ACTIVE, "/< Carrier power is on")
.value("activating", espp::Ndef::CarrierPowerState::ACTIVATING,
"/< Carrier power is turning on")
.value("unknown", espp::Ndef::CarrierPowerState::UNKNOWN,
"/< Carrier power state is unknown");
auto pyEnumBtEir =
py::enum_(
pyClassNdef, "BtEir", py::arithmetic(),
"*\n * @brief Extended Inquiry Response (EIR) codes for data types in BT and BLE\n "
"* out of band (OOB) pairing NDEF records.\n")
.value(
"flags", espp::Ndef::BtEir::FLAGS,
"/< BT flags: b0: LE limited discoverable mode, b1: LE general discoverable mode,")
.value("uuids_16_bit_partial", espp::Ndef::BtEir::UUIDS_16_BIT_PARTIAL,
"/< Incomplete list of 16 bit service class UUIDs")
.value("uuids_16_bit_complete", espp::Ndef::BtEir::UUIDS_16_BIT_COMPLETE,
"/< Complete list of 16 bit service class UUIDs")
.value("uuids_32_bit_partial", espp::Ndef::BtEir::UUIDS_32_BIT_PARTIAL,
"/< Incomplete list of 32 bit service class UUIDs")
.value("uuids_32_bit_complete", espp::Ndef::BtEir::UUIDS_32_BIT_COMPLETE,
"/< Complete list of 32 bit service class UUIDs")
.value("uuids_128_bit_partial", espp::Ndef::BtEir::UUIDS_128_BIT_PARTIAL,
"/< Incomplete list of 128 bit service class UUIDs")
.value("uuids_128_bit_complete", espp::Ndef::BtEir::UUIDS_128_BIT_COMPLETE,
"/< Complete list of 128 bit service class UUIDs")
.value("short_local_name", espp::Ndef::BtEir::SHORT_LOCAL_NAME,
"/< Shortened Bluetooth Local Name")
.value("long_local_name", espp::Ndef::BtEir::LONG_LOCAL_NAME,
"/< Complete Bluetooth Local Name")
.value("tx_power_level", espp::Ndef::BtEir::TX_POWER_LEVEL,
"/< TX Power level (1 byte), -127 dBm to +127 dBm")
.value("class_of_device", espp::Ndef::BtEir::CLASS_OF_DEVICE, "/< Class of Device")
.value("sp_hash_c192", espp::Ndef::BtEir::SP_HASH_C192, "/< Simple Pairing Hash C-192")
.value("sp_random_r192", espp::Ndef::BtEir::SP_RANDOM_R192,
"/< Simple Pairing Randomizer R-192")
.value("security_manager_tk", espp::Ndef::BtEir::SECURITY_MANAGER_TK,
"/< Security Manager TK Value (LE Legacy Pairing)")
.value("security_manager_flags", espp::Ndef::BtEir::SECURITY_MANAGER_FLAGS,
"/< Flags (1 B), b0: OOB flags field (1 = 00B data present, 0 not), b1: LE "
"Supported")
.value("appearance", espp::Ndef::BtEir::APPEARANCE, "/< Appearance")
.value("mac", espp::Ndef::BtEir::MAC, "/< Bluetooth Device Address")
.value("le_role", espp::Ndef::BtEir::LE_ROLE, "/< LE Role")
.value("sp_hash_c256", espp::Ndef::BtEir::SP_HASH_C256, "/< Simple Pairing Hash C-256")
.value("sp_hash_r256", espp::Ndef::BtEir::SP_HASH_R256,
"/< Simple Pairing Randomizer R-256")
.value("le_sc_confirmation", espp::Ndef::BtEir::LE_SC_CONFIRMATION,
"/< LE Secure Connections Confirmation Value")
.value("le_sc_random", espp::Ndef::BtEir::LE_SC_RANDOM,
"/< LE Secure Connections Random Value");
auto pyEnumBleRole =
py::enum_(
pyClassNdef, "BleRole", py::arithmetic(),
"*\n * @brief Possible roles for BLE records to indicate support for.\n")
.value("peripheral_only", espp::Ndef::BleRole::PERIPHERAL_ONLY,
"/< Radio can only act as a peripheral")
.value("central_only", espp::Ndef::BleRole::CENTRAL_ONLY,
"/< Radio can only act as a central")
.value("peripheral_central", espp::Ndef::BleRole::PERIPHERAL_CENTRAL,
"/< Radio can act as both a peripheral and a central, but prefers peripheral")
.value("central_peripheral", espp::Ndef::BleRole::CENTRAL_PERIPHERAL,
"/< Radio can act as both a peripheral and a central, but prefers central");
auto pyEnumWifiEncryptionType =
py::enum_(
pyClassNdef, "WifiEncryptionType", py::arithmetic(),
"*\n * @brief Types of configurable encryption for WiFi networks\n")
.value("none", espp::Ndef::WifiEncryptionType::NONE, "/< No encryption")
.value("wep", espp::Ndef::WifiEncryptionType::WEP, "/< WEP")
.value("tkip", espp::Ndef::WifiEncryptionType::TKIP, "/< TKIP")
.value("aes", espp::Ndef::WifiEncryptionType::AES, "/< AES");
auto pyEnumWifiAuthenticationType =
py::enum_(
pyClassNdef, "WifiAuthenticationType", py::arithmetic(),
"*\n * @brief WiFi network authentication\n")
.value("open", espp::Ndef::WifiAuthenticationType::OPEN, "/< Open / no security")
.value("wpa_personal", espp::Ndef::WifiAuthenticationType::WPA_PERSONAL,
"/< WPA personal")
.value("shared", espp::Ndef::WifiAuthenticationType::SHARED, "/< Shared key")
.value("wpa_enterprise", espp::Ndef::WifiAuthenticationType::WPA_ENTERPRISE,
"/< WPA enterprise")
.value("wpa2_enterprise", espp::Ndef::WifiAuthenticationType::WPA2_ENTERPRISE,
"/< WPA2 Enterprise")
.value("wpa2_personal", espp::Ndef::WifiAuthenticationType::WPA2_PERSONAL,
"/< WPA2 personal")
.value("wpa_wpa2_personal", espp::Ndef::WifiAuthenticationType::WPA_WPA2_PERSONAL,
"/< Both WPA and WPA2 personal");
auto pyClassNdef_ClassWifiConfig =
py::class_(
pyClassNdef, "WifiConfig", py::dynamic_attr(),
"*\n * @brief Configuration structure for wifi configuration ndef structure.\n")
.def(py::init([](std::string_view ssid = std::string_view(),
std::string_view key = std::string_view(),
espp::Ndef::WifiAuthenticationType authentication =
espp::Ndef::WifiAuthenticationType::WPA2_PERSONAL,
espp::Ndef::WifiEncryptionType encryption =
espp::Ndef::WifiEncryptionType::AES,
uint64_t mac_address = 0xFFFFFFFFFFFF) {
auto r_ctor_ = std::make_unique();
r_ctor_->ssid = ssid;
r_ctor_->key = key;
r_ctor_->authentication = authentication;
r_ctor_->encryption = encryption;
r_ctor_->mac_address = mac_address;
return r_ctor_;
}),
py::arg("ssid") = std::string_view(), py::arg("key") = std::string_view(),
py::arg("authentication") = espp::Ndef::WifiAuthenticationType::WPA2_PERSONAL,
py::arg("encryption") = espp::Ndef::WifiEncryptionType::AES,
py::arg("mac_address") = 0xFFFFFFFFFFFF)
.def_readwrite("ssid", &espp::Ndef::WifiConfig::ssid, "/< SSID for the network")
.def_readwrite("key", &espp::Ndef::WifiConfig::key,
"/< Security key / password for the network")
.def_readwrite("authentication", &espp::Ndef::WifiConfig::authentication,
"/< Authentication type the network")
.def_readwrite("encryption", &espp::Ndef::WifiConfig::encryption,
"/< Encryption type the network uses.")
.def_readwrite("mac_address", &espp::Ndef::WifiConfig::mac_address,
"/< Broadcast MAC address FF:FF:FF:FF:FF:FF");
} // end of inner classes & enums of Ndef
pyClassNdef
.def_readonly_static("handover_version", &espp::Ndef::HANDOVER_VERSION,
"/< Connection Handover version 1.3")
.def(py::init(), py::arg("tnf"),
py::arg("type"), py::arg("payload"),
"*\n * @brief Makes an NDEF record with header and payload.\n * @param tnf The TNF "
"for this packet.\n * @param type String view for the type of this packet\n * "
"@param payload The payload data for the packet\n")
.def_static("make_text", &espp::Ndef::make_text, py::arg("text"),
"*\n * @brief Static function to make an NDEF record for transmitting "
"english\n * text.\n * @param text The text that the NDEF record will "
"hold.\n * @return NDEF record object.\n")
.def_static("make_uri", &espp::Ndef::make_uri, py::arg("uri"),
py::arg("uic") = espp::Ndef::Uic::NONE,
"*\n * @brief Static function to make an NDEF record for loading a URI.\n * "
"@param uri URI for the record to point to.\n * @param uic UIC for the uri - "
"helps shorten the uri text / NDEF record.\n * @return NDEF record object.\n")
.def_static("make_android_launcher", &espp::Ndef::make_android_launcher, py::arg("uri"),
"*\n * @brief Static function to make an NDEF record for launching an Android "
"App.\n * @param uri URI for the android package / app to launch.\n * "
"@return NDEF record object.\n")
.def_static("make_wifi_config", &espp::Ndef::make_wifi_config, py::arg("config"),
"*\n * @brief Create a WiFi credential tag.\n * @param config WifiConfig "
"describing the WiFi network.\n * @return NDEF record object.\n")
.def_static(
"make_collision_resolution_record", &espp::Ndef::make_collision_resolution_record,
py::arg("random_number"),
"\n * @brief Create a collision resolution record.\n * @param random_number Random "
"number to use for the collision resolution.\n * @return NDEF record object.\n")
.def_static(
"make_handover_select", &espp::Ndef::make_handover_select, py::arg("carrier_data_ref"),
"*\n * @brief Create a Handover Select record for a Bluetooth device.\n * @see\n * "
"https://members.nfc-forum.org/apps/group_public/download.php/18688/"
"NFCForum-AD-BTSSP_1_1.pdf\n * @param carrier_data_ref Reference to the carrier data "
"record, which is the\n * record that contains the actual bluetooth data. This "
"should be the\n * same as the id of the carrier data record, such as '0'.\n "
"* @return NDEF record object.\n")
.def_static("make_handover_request", &espp::Ndef::make_handover_request,
py::arg("carrier_data_ref"),
"*\n * @brief Create a Handover request record for a Bluetooth device.\n * "
"@see\n * "
"https://members.nfc-forum.org/apps/group_public/download.php/18688/"
"NFCForum-AD-BTSSP_1_1.pdf\n * @param carrier_data_ref Reference to the "
"carrier data record, which is the\n * record that contains the actual "
"bluetooth data. This should be the\n * same as the id of the carrier "
"data record, such as '0'.\n * @return NDEF record object.\n")
.def_static(
"make_alternative_carrier", &espp::Ndef::make_alternative_carrier, py::arg("power_state"),
py::arg("carrier_data_ref"),
"*\n * @brief Create a Handover Request record for a Bluetooth device.\n * @details "
"See page 18 of https://core.ac.uk/download/pdf/250136576.pdf for more details.\n * "
"@param power_state Power state of the alternative carrier.\n * @param "
"carrier_data_ref Reference to the carrier data record, which is the\n * record "
"that contains the actual bluetooth data. This should be the\n * same as the id "
"of the carrier data record, such as '0'.\n * @return NDEF record object.\n")
.def_static("make_oob_pairing", &espp::Ndef::make_oob_pairing, py::arg("mac_addr"),
py::arg("device_class"), py::arg("name"), py::arg("random_value") = "",
py::arg("confirm_value") = "",
"*\n * @brief Static function to make an NDEF record for BT classic OOB "
"Pairing (Android).\n * @param mac_addr 48 bit MAC Address of the BT radio\n "
"* @note If the address is e.g. f4:12:fa:42:fe:9e then the mac_addr should be\n "
" * 0xf412fa42fe9e.\n * @param device_class The bluetooth device class "
"for this radio.\n * @param name Name of the BT device.\n * @param "
"random_value The Simple pairing randomizer R for the pairing.\n * @param "
"confirm_value The Simple pairing hash C (confirm value) for the\n * "
" pairing.\n * @return NDEF record object.\n")
.def_static(
"make_le_oob_pairing", &espp::Ndef::make_le_oob_pairing, py::arg("mac_addr"),
py::arg("role"), py::arg("name") = "",
py::arg("appearance") = espp::Ndef::BtAppearance::UNKNOWN, py::arg("random_value") = "",
py::arg("confirm_value") = "", py::arg("tk") = "",
"*\n * @brief Static function to make an NDEF record for BLE OOB Pairing (Android).\n "
" * @param mac_addr 48 bit MAC Address of the BLE radio.\n * @note If the address is "
"e.g. f4:12:fa:42:fe:9e then the mac_addr should be\n * 0xf412fa42fe9e.\n * "
"@param role The BLE role of the device (central / peripheral / dual)\n * @param name "
"Name of the BLE device. Optional.\n * @param appearance BtAppearance of the device. "
"Optional.\n * @param random_value The Simple pairing randomizer R for the pairing. "
"(16 bytes, optional)\n * @param confirm_value The Simple pairing hash C (confirm "
"value) for the pairing. (16 bytes,\n * optional)\n * @param tk Temporary key for "
"the pairing (16 bytes, optional)\n * @return NDEF record object.\n")
.def("serialize", &espp::Ndef::serialize, py::arg("message_begin") = true,
py::arg("message_end") = true,
"*\n * @brief Serialize the NDEF record into a sequence of bytes.\n * @param "
"message_begin True if this is the first record in the message.\n * @param "
"message_end True if this is the last record in the message.\n * @return The "
"vector of bytes representing the NDEF record.\n")
.def("payload", &espp::Ndef::payload,
"*\n * @brief Return just the payload as a vector of bytes.\n * @return Payload of "
"the NDEF record as a vector of bytes.\n")
.def("set_id", &espp::Ndef::set_id, py::arg("id"),
"*\n * @brief Set the payload ID of the NDEF record.\n * @param id ID of the NDEF "
"record.\n")
.def("get_id", &espp::Ndef::get_id,
"*\n * @brief Get the ID of the NDEF record.\n * @return ID of the NDEF record.\n")
.def("get_size", &espp::Ndef::get_size,
"*\n * @brief Get the number of bytes needed for the NDEF record.\n * @return Size "
"of the NDEF record (bytes), for serialization.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassPid = py::class_(
m, "Pid", py::dynamic_attr(),
"*\n * @brief Simple PID (proportional, integral, derivative) controller class\n * "
"with integrator clamping, output clamping, and prevention of\n * integrator windup "
"during output saturation. This class is\n * thread-safe, so you can update(), "
"clear(), and change_gains() from\n * multiple threads if needed.\n *\n * \\section "
"pid_ex1 Basic PID Example\n * \\snippet pid_example.cpp pid example\n * \\section pid_ex2 "
"Complex PID Example\n * \\snippet pid_example.cpp complex pid example\n");
{ // inner classes & enums of Pid
auto pyClassPid_ClassConfig =
py::class_(pyClassPid, "Config", py::dynamic_attr(), "")
.def(
py::init([](float kp = float(), float ki = float(), float kd = float(),
float integrator_min = float(), float integrator_max = float(),
float output_min = float(), float output_max = float(),
espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->kp = kp;
r_ctor_->ki = ki;
r_ctor_->kd = kd;
r_ctor_->integrator_min = integrator_min;
r_ctor_->integrator_max = integrator_max;
r_ctor_->output_min = output_min;
r_ctor_->output_max = output_max;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("kp") = float(), py::arg("ki") = float(), py::arg("kd") = float(),
py::arg("integrator_min") = float(), py::arg("integrator_max") = float(),
py::arg("output_min") = float(), py::arg("output_max") = float(),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("kp", &espp::Pid::Config::kp, "*< Proportional gain.")
.def_readwrite(
"ki", &espp::Pid::Config::ki,
"*< Integral gain. @note should not be pre-multiplied by the time constant.")
.def_readwrite(
"kd", &espp::Pid::Config::kd,
"*< Derivative gain. @note should not be pre-divided by the time-constant.")
.def_readwrite(
"integrator_min", &espp::Pid::Config::integrator_min,
"*< Minimum value the integrator can wind down to. @note Operates at the\n "
" same scale as \\p output_min and \\p output_max. Could be 0 "
"or negative.\n Can have different magnitude from "
"integrator_max for asymmetric\n response.")
.def_readwrite(
"integrator_max", &espp::Pid::Config::integrator_max,
"*< Maximum value the integrator can wind up to. @note Operates at the\n "
" same scale as \\p output_min and \\p output_max.")
.def_readwrite("output_min", &espp::Pid::Config::output_min,
"*< Limit the minimum output value. Can be a different magnitude from "
"output\n max for asymmetric output behavior.")
.def_readwrite("output_max", &espp::Pid::Config::output_max,
"*< Limit the maximum output value.")
.def_readwrite("log_level", &espp::Pid::Config::log_level,
"*< Verbosity for the adc logger.");
} // end of inner classes & enums of Pid
pyClassPid.def(py::init())
.def("__call__", &espp::Pid::operator(), py::arg("error"),
"*\n * @brief Update the PID controller with the latest error measurement,\n * "
" getting the output control signal in return.\n *\n * @note Tracks invocation "
"timing to better compute time-accurate\n * integral/derivative signals.\n "
"*\n * @param error Latest error signal.\n * @return The output control signal "
"based on the PID state and error.\n")
.def("get_error", &espp::Pid::get_error,
"*\n * @brief Get the current error (as of the last time update() or operator()\n * "
" were called)\n * @return Most recent error.\n")
.def("get_integrator", &espp::Pid::get_integrator,
"*\n * @brief Get the current integrator (as of the last time update() or\n * "
" operator() were called)\n * @return Most recent integrator value.\n")
.def("get_config", &espp::Pid::get_config,
"*\n * @brief Get the configuration for the PID (gains, etc.).\n * @return Config "
"structure containing gains, etc.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
// Bound before the socket / thread_pool sections, whose bindings use
// QosBand values as default arguments (defaults are converted at def time).
py::enum_(
m, "QosBand",
"*\n * @brief Priority band for queued work. Critical is the most urgent and Low the "
"least;\n * Normal is the default for band-less submissions.\n")
.value("Critical", espp::QosBand::Critical)
.value("High", espp::QosBand::High)
.value("Normal", espp::QosBand::Normal)
.value("Low", espp::QosBand::Low);
//////////////////// ////////////////////
//////////////////// ////////////////////
// Bound before the socket section, whose bindings use std::optional
// parameters.
py::enum_(
m, "Dscp",
"*\n * @brief Standard DiffServ code points (DSCP) for IP traffic marking - the 6-bit "
"field\n * in the IP TOS / Traffic Class byte (RFC 2474). E.g. Dscp.Ef = expedited "
"forwarding\n * for latency-critical flows; Dscp.Cs1 = low-priority data; Dscp.Af41 = "
"high-priority\n * assured forwarding with low drop probability.\n")
.value("Cs0", espp::Dscp::Cs0)
.value("Default", espp::Dscp::Default)
.value("Le", espp::Dscp::Le)
.value("Cs1", espp::Dscp::Cs1)
.value("Af11", espp::Dscp::Af11)
.value("Af12", espp::Dscp::Af12)
.value("Af13", espp::Dscp::Af13)
.value("Cs2", espp::Dscp::Cs2)
.value("Af21", espp::Dscp::Af21)
.value("Af22", espp::Dscp::Af22)
.value("Af23", espp::Dscp::Af23)
.value("Cs3", espp::Dscp::Cs3)
.value("Af31", espp::Dscp::Af31)
.value("Af32", espp::Dscp::Af32)
.value("Af33", espp::Dscp::Af33)
.value("Cs4", espp::Dscp::Cs4)
.value("Af41", espp::Dscp::Af41)
.value("Af42", espp::Dscp::Af42)
.value("Af43", espp::Dscp::Af43)
.value("Cs5", espp::Dscp::Cs5)
.value("VoiceAdmit", espp::Dscp::VoiceAdmit)
.value("Ef", espp::Dscp::Ef)
.value("Cs6", espp::Dscp::Cs6)
.value("Cs7", espp::Dscp::Cs7);
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassSocket =
py::class_(m, "Socket", py::dynamic_attr(),
"*\n * @brief Class for a generic socket with some helper "
"functions for\n * configuring the socket.\n");
{ // inner classes & enums of Socket
auto pyEnumType =
py::enum_(pyClassSocket, "Type", py::arithmetic(), "")
.value("raw", espp::Socket::Type::RAW,
"*< Only IP headers, no TCP or UDP headers as well.")
.value("dgram", espp::Socket::Type::DGRAM, "*< UDP/IP socket - datagram.")
.value("stream", espp::Socket::Type::STREAM, "*< TCP/IP socket - stream.");
auto pyClassSocket_ClassInfo =
py::class_(
pyClassSocket, "Info", py::dynamic_attr(),
"*\n * @brief Storage for socket information (address, port) with convenience\n * "
" functions to convert to/from POSIX structures.\n")
.def(py::init([](std::string address = std::string(), size_t port = size_t()) {
auto r_ctor_ = std::make_unique();
r_ctor_->address = address;
r_ctor_->port = port;
return r_ctor_;
}),
py::arg("address") = std::string(), py::arg("port") = size_t())
.def_readwrite("address", &espp::Socket::Info::address,
"*< IP address of the endpoint as a string.")
.def_readwrite("port", &espp::Socket::Info::port,
"*< Port of the endpoint as an integer.")
.def("init_ipv4", &espp::Socket::Info::init_ipv4, py::arg("addr"), py::arg("prt"),
"*\n * @brief Initialize the struct as an ipv4 address/port combo.\n * "
"@param addr IPv4 address string\n * @param prt port number\n")
.def("ipv4_ptr", &espp::Socket::Info::ipv4_ptr,
"*\n * @brief Gives access to IPv4 sockaddr structure (sockaddr_in) for use\n "
" * with low level socket calls like sendto / recvfrom.\n * @return "
"*sockaddr_in pointer to ipv4 data structure\n")
.def("update", &espp::Socket::Info::update,
"*\n * @brief Will update address and port based on the curent data in raw.\n")
.def("from_sockaddr",
py::overload_cast(
&espp::Socket::Info::from_sockaddr),
py::arg("source_address"),
"*\n * @brief Fill this Info from the provided sockaddr struct.\n * "
"@param &source_address sockaddr info filled out by recvfrom.\n")
.def("from_sockaddr",
py::overload_cast(&espp::Socket::Info::from_sockaddr),
py::arg("source_address"),
"*\n * @brief Fill this Info from the provided sockaddr struct.\n * "
"@param &source_address sockaddr info filled out by recvfrom.\n");
} // end of inner classes & enums of Socket
pyClassSocket
.def("is_valid", &espp::Socket::is_valid,
"*\n * @brief Is the socket valid.\n * @return True if the socket file descriptor "
"is >= 0.\n")
.def_static("is_valid_fd", &espp::Socket::is_valid_fd, py::arg("socket_fd"),
"*\n * @brief Is the socket valid.\n * @param socket_fd Socket file "
"descriptor.\n * @return True if the socket file descriptor is >= 0.\n")
.def("native_handle", &espp::Socket::native_handle,
"*\n * @brief Get the underlying native socket file descriptor / handle.\n * @note "
"Provided so an external event loop (e.g. SocketReactor) can add this\n * "
"socket to a select()/poll() set. The Socket retains ownership of the\n * "
"descriptor; do not close it directly.\n * @return The socket file descriptor.\n")
.def("get_ipv4_info", &espp::Socket::get_ipv4_info,
"*\n * @brief Get the Socket::Info for the socket.\n * @details This will call "
"getsockname() on the socket to get the\n * sockaddr_storage structure, and "
"then fill out the Socket::Info\n * structure.\n * @return Socket::Info "
"for the socket.\n")
.def("set_receive_timeout", &espp::Socket::set_receive_timeout, py::arg("timeout"),
"*\n * @brief Set the receive timeout on the provided socket.\n * @param timeout "
"requested timeout, must be > 0.\n * @return True if SO_RECVTIMEO was successfully "
"set.\n")
.def("set_dscp", &espp::Socket::set_dscp, py::arg("dscp"),
"*\n * @brief Mark this socket's TRANSMITTED packets with a DSCP code point\n * "
" (applied as IP_TOS - RFC 2474). Best-effort: network / driver\n * "
"treatment of outgoing traffic (e.g. Dscp.Ef), NOT local scheduling.\n * @param "
"dscp the espp.Dscp code point to apply.\n * @return True if IP_TOS was "
"successfully set.\n")
.def("get_dscp", &espp::Socket::get_dscp,
"*\n * @brief Get the DSCP code point this socket marks its transmitted packets\n "
"* with (read back from IP_TOS; ECN bits discarded).\n * @return The "
"espp.Dscp code point, or None on failure.\n")
.def("enable_reuse", &espp::Socket::enable_reuse,
"*\n * @brief Allow others to use this address/port combination after we're done\n "
"* with it.\n * @return True if SO_REUSEADDR and SO_REUSEPORT were "
"successfully set.\n")
.def("make_multicast", &espp::Socket::make_multicast, py::arg("time_to_live") = 1,
py::arg("loopback_enabled") = true, py::arg("interface_address") = "",
"*\n * @brief Configure the socket to be multicast (if time_to_live > 0).\n * "
" Sets the IP_MULTICAST_TTL (number of multicast hops allowed) and\n * "
"optionally configures whether this node should receive its own\n * multicast "
"packets (IP_MULTICAST_LOOP).\n * @param time_to_live number of multicast hops "
"allowed (TTL).\n * @param loopback_enabled Whether to receive our own multicast "
"packets.\n * @param interface_address Optional dotted-decimal IPv4 address of the "
"local\n * interface to use for *outgoing* multicast (IP_MULTICAST_IF). When\n "
" * empty or \"0.0.0.0\", the OS chooses its default multicast interface.\n * "
" Set this to the IP of the desired NIC on multi-homed hosts (e.g. to\n * "
" force multicast out a wired interface instead of Wi-Fi).\n * @return True if "
"IP_MULTICAST_TTL and IP_MULTICAST_LOOP were set.\n")
.def("add_multicast_group", &espp::Socket::add_multicast_group, py::arg("multicast_group"),
py::arg("interface_address") = "",
"*\n * @brief If this is a server socket, add it to the provided the multicast\n * "
" group.\n *\n * @note Multicast groups must be Class D addresses "
"(224.0.0.0 to\n * 239.255.255.255)\n *\n * See "
"https://en.wikipedia.org/wiki/Multicast_address for more\n * information.\n "
"* @param multicast_group multicast group to join.\n * @param interface_address "
"Optional dotted-decimal IPv4 address of the local\n * interface on which to "
"join the group (imr_interface) and to use for\n * outgoing multicast "
"(IP_MULTICAST_IF). When empty or \"0.0.0.0\", the OS\n * default interface is "
"used. Set this on multi-homed hosts so the group\n * is joined on the desired "
"NIC (e.g. wired instead of Wi-Fi).\n * @return True if IP_ADD_MEMBERSHIP was "
"successfully set.\n")
.def("select", &espp::Socket::select, py::arg("timeout"),
"*\n * @brief Select on the socket for read events.\n * @param timeout how long to "
"wait for an event.\n * @return number of events that occurred.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassTcpSocket = py::class_(
m, "TcpSocket", py::dynamic_attr(),
"*\n * @brief Class for managing sending and receiving data using TCP/IP. Can be\n * "
" used to create client or server sockets.\n *\n * \\section tcp_ex1 TCP Client Example\n "
"* \\snippet socket_example.cpp TCP Client example\n * \\section tcp_ex2 TCP Server "
"Example\n * \\snippet socket_example.cpp TCP Server example\n *\n * \\section tcp_ex3 TCP "
"Client Response Example\n * \\snippet socket_example.cpp TCP Client Response example\n * "
"\\section tcp_ex4 TCP Server Response Example\n * \\snippet socket_example.cpp TCP Server "
"Response example\n *\n");
{ // inner classes & enums of TcpSocket
auto pyClassTcpSocket_ClassConfig =
py::class_(pyClassTcpSocket, "Config", py::dynamic_attr(),
"*\n * @brief Config struct for the TCP socket.\n")
.def(
py::init([](espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("log_level", &espp::TcpSocket::Config::log_level,
"*< Verbosity level for the TCP socket logger.");
auto pyClassTcpSocket_ClassConnectConfig =
py::class_(
pyClassTcpSocket, "ConnectConfig", py::dynamic_attr(),
"*\n * @brief Config struct for connecting to a remote TCP server.\n")
.def(py::init([](std::string ip_address = std::string(), size_t port = size_t()) {
auto r_ctor_ = std::make_unique();
r_ctor_->ip_address = ip_address;
r_ctor_->port = port;
return r_ctor_;
}),
py::arg("ip_address") = std::string(), py::arg("port") = size_t())
.def_readwrite("ip_address", &espp::TcpSocket::ConnectConfig::ip_address,
"*< Address to send data to.")
.def_readwrite("port", &espp::TcpSocket::ConnectConfig::port,
"*< Port number to send data to.");
auto pyClassTcpSocket_ClassTransmitConfig =
py::class_(
pyClassTcpSocket, "TransmitConfig", py::dynamic_attr(),
"*\n * @brief Config struct for sending data to a remote TCP socket.\n * @note "
"This is only used when waiting for a response from the remote.\n")
.def(py::init([](bool wait_for_response = false, size_t response_size = 0,
espp::Socket::response_callback_fn on_response_callback = nullptr,
std::chrono::duration response_timeout =
std::chrono::duration(0.5f)) {
auto r_ctor_ = std::make_unique();
r_ctor_->wait_for_response = wait_for_response;
r_ctor_->response_size = response_size;
r_ctor_->on_response_callback = on_response_callback;
r_ctor_->response_timeout = response_timeout;
return r_ctor_;
}),
py::arg("wait_for_response") = false, py::arg("response_size") = 0,
py::arg("on_response_callback") = py::none(),
py::arg("response_timeout") = std::chrono::duration(0.5f))
.def_readwrite("wait_for_response", &espp::TcpSocket::TransmitConfig::wait_for_response,
"*< Whether to wait for a response from the remote or not.")
.def_readwrite(
"response_size", &espp::TcpSocket::TransmitConfig::response_size,
"*< If waiting for a response, this is the maximum size response we will receive.")
.def_readwrite("on_response_callback",
&espp::TcpSocket::TransmitConfig::on_response_callback,
"*< If waiting for a\n response, this is an optional "
"handler which is provided the response data.")
.def_readwrite("response_timeout", &espp::TcpSocket::TransmitConfig::response_timeout,
"*< If waiting for a response, this is the maximum timeout to wait.")
.def_static("default", &espp::TcpSocket::TransmitConfig::Default);
} // end of inner classes & enums of TcpSocket
pyClassTcpSocket.def(py::init())
.def("reinit", &espp::TcpSocket::reinit,
"*\n * @brief Reinitialize the socket, cleaning it up if first it is already\n * "
" initalized.\n")
.def("close", &espp::TcpSocket::close, "*\n * @brief Close the socket.\n")
.def("is_connected", &espp::TcpSocket::is_connected,
"*\n * @brief Check if the socket is connected to a remote endpoint.\n * @return "
"True if the socket is connected to a remote endpoint.\n")
.def("connect", &espp::TcpSocket::connect, py::arg("connect_config"),
"*\n * @brief Open a connection to the remote TCP server.\n * @param connect_config "
"ConnectConfig struct describing the server endpoint.\n * @return True if the client "
"successfully connected to the server.\n",
py::call_guard())
.def("get_remote_info", &espp::TcpSocket::get_remote_info,
"*\n * @brief Get the remote endpoint info.\n * @return The remote endpoint info.\n")
.def("transmit",
py::overload_cast(
&espp::TcpSocket::transmit),
py::arg("data"), py::arg("transmit_config") = espp::TcpSocket::TransmitConfig::Default(),
"*\n * @brief Send data to the endpoint already connected to by TcpSocket::connect.\n "
" * Can be configured to block waiting for a response from the remote.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data vector of bytes to send to the remote endpoint.\n * @param "
"transmit_config TransmitConfig struct indicating whether to wait for a\n * "
"response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("transmit",
py::overload_cast(
&espp::TcpSocket::transmit),
py::arg("data"), py::arg("transmit_config") = espp::TcpSocket::TransmitConfig::Default(),
"*\n * @brief Send data to the endpoint already connected to by TcpSocket::connect.\n "
" * Can be configured to block waiting for a response from the remote.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data vector of bytes to send to the remote endpoint.\n * @param "
"transmit_config TransmitConfig struct indicating whether to wait for a\n * "
"response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("transmit",
py::overload_cast(
&espp::TcpSocket::transmit),
py::arg("data"), py::arg("transmit_config") = espp::TcpSocket::TransmitConfig::Default(),
"*\n * @brief Send data to the endpoint already connected to by TcpSocket::connect.\n "
" * Can be configured to block waiting for a response from the remote.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data string view of bytes to send to the remote endpoint.\n * @param "
"transmit_config TransmitConfig struct indicating whether to wait for a\n * "
"response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("transmit",
py::overload_cast(
&espp::TcpSocket::transmit),
py::arg("data"), py::arg("transmit_config") = espp::TcpSocket::TransmitConfig::Default(),
"*\n * @brief Send data to the endpoint already connected to by TcpSocket::connect.\n "
" * Can be configured to block waiting for a response from the remote.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data span of bytes to send to the remote endpoint.\n * @param "
"transmit_config TransmitConfig struct indicating whether to wait for a\n * "
"response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("receive", py::overload_cast(&espp::TcpSocket::receive),
py::arg("data"), py::arg("max_num_bytes"),
"*\n * @brief Call read on the socket, assuming it has already been configured\n * "
" appropriately.\n *\n * @param data Vector of bytes of received data.\n * "
"@param max_num_bytes Maximum number of bytes to receive.\n * @return True if "
"successfully received, False otherwise.\n",
py::call_guard())
.def("receive", py::overload_cast(&espp::TcpSocket::receive),
py::arg("data"), py::arg("max_num_bytes"),
"*\n * @brief Call read on the socket, assuming it has already been configured\n * "
" appropriately.\n * @note This function will block until max_num_bytes are "
"received or the\n * receive timeout is reached.\n * @note The data pointed "
"to by data must be at least max_num_bytes in size.\n * @param data Pointer to buffer "
"to receive data.\n * @param max_num_bytes Maximum number of bytes to receive.\n * "
"@return Number of bytes received.\n",
py::call_guard())
.def("bind", &espp::TcpSocket::bind, py::arg("port"),
"*\n * @brief Bind the socket as a server on \\p port.\n * @param port The port to "
"which to bind the socket.\n * @return True if the socket was bound.\n")
.def("listen", &espp::TcpSocket::listen, py::arg("max_pending_connections"),
"*\n * @brief Listen for incoming client connections.\n * @note Must be called "
"after bind and before accept.\n * @see bind\n * @see accept\n * @param "
"max_pending_connections Max number of allowed pending connections.\n * @return True "
"if socket was able to start listening.\n")
.def("accept", &espp::TcpSocket::accept,
"*\n * @brief Accept an incoming connection.\n * @note Blocks until a connection is "
"accepted.\n * @note Must be called after listen.\n * @note This function will "
"block until a connection is accepted.\n * @return A unique pointer to a "
"TcpClientSession if a connection was\n * accepted, None otherwise.\n",
py::call_guard());
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassUdpSocket = py::class_(
m, "UdpSocket", py::dynamic_attr(),
"*\n * @brief Class for managing sending and receiving data using UDP/IP. Can be\n * "
" used to create client or server sockets.\n *\n * See\n * "
"https://github.com/espressif/esp-idf/tree/master/examples/protocols/sockets/udp_multicast\n "
"* for more information on udp multicast sockets.\n *\n * \\section udp_ex1 UDP Client "
"Example\n * \\snippet socket_example.cpp UDP Client example\n * \\section udp_ex2 UDP "
"Server Example\n * \\snippet socket_example.cpp UDP Server example\n *\n * \\section "
"udp_ex3 UDP Client Response Example\n * \\snippet socket_example.cpp UDP Client Response "
"example\n * \\section udp_ex4 UDP Server Response Example\n * \\snippet socket_example.cpp "
"UDP Server Response example\n *\n * \\section udp_ex5 UDP Multicast Client Example\n * "
"\\snippet socket_example.cpp UDP Multicast Client example\n * \\section udp_ex6 UDP "
"Multicast Server Example\n * \\snippet socket_example.cpp UDP Multicast Server example\n "
"*\n");
{ // inner classes & enums of UdpSocket
auto pyClassUdpSocket_ClassReceiveConfig =
py::class_(pyClassUdpSocket, "ReceiveConfig",
py::dynamic_attr(), "")
.def(py::init([](size_t port = size_t(), size_t buffer_size = size_t(),
bool is_multicast_endpoint = {false},
std::string multicast_group = {""},
std::string multicast_interface = {""},
espp::Socket::receive_callback_fn on_receive_callback = {nullptr},
espp::QosBand band = {espp::QosBand::Normal},
std::optional dscp = {}) {
auto r_ctor_ = std::make_unique();
r_ctor_->port = port;
r_ctor_->buffer_size = buffer_size;
r_ctor_->is_multicast_endpoint = is_multicast_endpoint;
r_ctor_->multicast_group = multicast_group;
r_ctor_->multicast_interface = multicast_interface;
r_ctor_->on_receive_callback = on_receive_callback;
r_ctor_->band = band;
r_ctor_->dscp = dscp;
return r_ctor_;
}),
py::arg("port") = size_t(), py::arg("buffer_size") = size_t(),
py::arg("is_multicast_endpoint") = bool{false},
py::arg("multicast_group") = std::string{""},
py::arg("multicast_interface") = std::string{""},
py::arg("on_receive_callback") = espp::Socket::receive_callback_fn{nullptr},
py::arg("band") = espp::QosBand::Normal,
py::arg("dscp") = std::optional{})
.def_readwrite("port", &espp::UdpSocket::ReceiveConfig::port,
"*< Port number to bind to / receive from.")
.def_readwrite("buffer_size", &espp::UdpSocket::ReceiveConfig::buffer_size,
"*< Max size of data we can receive at one time.")
.def_readwrite("is_multicast_endpoint",
&espp::UdpSocket::ReceiveConfig::is_multicast_endpoint,
"*< Whether this should be a multicast endpoint.")
.def_readwrite("multicast_group", &espp::UdpSocket::ReceiveConfig::multicast_group,
"*< If this is a multicast endpoint, this is the group it belongs to.")
.def_readwrite(
"multicast_interface", &espp::UdpSocket::ReceiveConfig::multicast_interface,
"*< Optional local IPv4 interface address on which to join the multicast group "
"and\n receive its traffic. Empty/\"0.0.0.0\" lets the OS pick the "
"default interface; set it\n on multi-homed hosts to bind multicast "
"to a specific NIC (e.g. wired vs Wi-Fi).")
.def_readwrite("on_receive_callback",
&espp::UdpSocket::ReceiveConfig::on_receive_callback,
"*< Function containing business logic to handle data received.")
.def_readwrite("band", &espp::UdpSocket::ReceiveConfig::band,
"*< Priority band for dispatching this socket's receive handling when "
"registered on an espp.SocketReactor (unused by start_receiving()).")
.def_readwrite("dscp", &espp::UdpSocket::ReceiveConfig::dscp,
"*< Optional espp.Dscp code point (e.g. Dscp.Ef) to mark this "
"socket's TRANSMITTED packets with (applied as IP_TOS by "
"espp.SocketReactor at registration, best-effort).");
auto pyClassUdpSocket_ClassSendConfig =
py::class_(pyClassUdpSocket, "SendConfig", py::dynamic_attr(),
"")
.def(py::init([](std::string ip_address = std::string(), size_t port = size_t(),
bool is_multicast_endpoint = {false},
std::string multicast_interface = {""},
bool wait_for_response = {false}, size_t response_size = {0},
espp::Socket::response_callback_fn on_response_callback = {nullptr},
std::chrono::duration response_timeout =
std::chrono::duration(0.5f)) {
auto r_ctor_ = std::make_unique();
r_ctor_->ip_address = ip_address;
r_ctor_->port = port;
r_ctor_->is_multicast_endpoint = is_multicast_endpoint;
r_ctor_->multicast_interface = multicast_interface;
r_ctor_->wait_for_response = wait_for_response;
r_ctor_->response_size = response_size;
r_ctor_->on_response_callback = on_response_callback;
r_ctor_->response_timeout = response_timeout;
return r_ctor_;
}),
py::arg("ip_address") = std::string(), py::arg("port") = size_t(),
py::arg("is_multicast_endpoint") = bool{false},
py::arg("multicast_interface") = std::string{""},
py::arg("wait_for_response") = bool{false}, py::arg("response_size") = size_t{0},
py::arg("on_response_callback") = espp::Socket::response_callback_fn{nullptr},
py::arg("response_timeout") = std::chrono::duration(0.5f))
.def_readwrite("ip_address", &espp::UdpSocket::SendConfig::ip_address,
"*< Address to send data to.")
.def_readwrite("port", &espp::UdpSocket::SendConfig::port,
"*< Port number to send data to.")
.def_readwrite("is_multicast_endpoint",
&espp::UdpSocket::SendConfig::is_multicast_endpoint,
"*< Whether this should be a multicast endpoint.")
.def_readwrite(
"multicast_interface", &espp::UdpSocket::SendConfig::multicast_interface,
"*< Optional local IPv4 interface address to use for outgoing multicast\n "
" (IP_MULTICAST_IF). Empty/\"0.0.0.0\" lets the OS pick the default "
"interface; set it on\n multi-homed hosts to send multicast out a "
"specific NIC (e.g. wired vs Wi-Fi).")
.def_readwrite("wait_for_response", &espp::UdpSocket::SendConfig::wait_for_response,
"*< Whether to wait for a response from the remote or not.")
.def_readwrite(
"response_size", &espp::UdpSocket::SendConfig::response_size,
"*< If waiting for a response, this is the maximum size response we will receive.")
.def_readwrite("on_response_callback",
&espp::UdpSocket::SendConfig::on_response_callback,
"*< If waiting for a response, this is an optional handler which is "
"provided the\n response data.")
.def_readwrite("response_timeout", &espp::UdpSocket::SendConfig::response_timeout,
"*< If waiting for a response, this is the maximum timeout to wait.");
auto pyClassUdpSocket_ClassConfig =
py::class_(pyClassUdpSocket, "Config", py::dynamic_attr(), "")
.def(
py::init([](espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("log_level", &espp::UdpSocket::Config::log_level,
"*< Verbosity level for the UDP socket logger.");
} // end of inner classes & enums of UdpSocket
pyClassUdpSocket.def(py::init())
.def("stop_receiving", &espp::UdpSocket::stop_receiving,
"/ Stop the receive task, if one is running, and close the socket.",
py::call_guard())
.def("send",
py::overload_cast(
&espp::UdpSocket::send),
py::arg("data"), py::arg("send_config"),
"*\n * @brief Send data to the endpoint specified by the send_config.\n * "
"Can be configured to multicast (within send_config) and can be\n * configured "
"to block waiting for a response from the remote.\n *\n * @note in the case "
"of multicast, it will block only until the first\n * response.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data vector of bytes to send to the remote endpoint.\n * @param send_config "
"SendConfig struct indicating where to send and whether\n * to wait for a "
"response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("send",
py::overload_cast(
&espp::UdpSocket::send),
py::arg("data"), py::arg("send_config"),
"*\n * @brief Send data to the endpoint specified by the send_config.\n * "
"Can be configured to multicast (within send_config) and can be\n * configured "
"to block waiting for a response from the remote.\n *\n * @note in the case "
"of multicast, it will block only until the first\n * response.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data String view of bytes to send to the remote endpoint.\n * @param "
"send_config SendConfig struct indicating where to send and whether\n * to "
"wait for a response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("send",
py::overload_cast(
&espp::UdpSocket::send),
py::arg("data"), py::arg("send_config"),
"*\n * @brief Send data to the endpoint specified by the send_config.\n * "
"Can be configured to multicast (within send_config) and can be\n * configured "
"to block waiting for a response from the remote.\n *\n * @note in the case "
"of multicast, it will block only until the first\n * response.\n *\n "
" * If response is requested, a callback can be provided in\n * "
"send_config which will be provided the response data for\n * processing.\n "
"* @param data std::span of bytes to send to the remote endpoint.\n * @param "
"send_config SendConfig struct indicating where to send and whether\n * to "
"wait for a response.\n * @return True if the data was sent, False otherwise.\n",
py::call_guard())
.def("receive", &espp::UdpSocket::receive, py::arg("max_num_bytes"), py::arg("data"),
py::arg("remote_info"),
"*\n * @brief Call recvfrom on the socket, assuming it has already been\n * "
"configured appropriately.\n *\n * @param max_num_bytes Maximum number of bytes to "
"receive.\n * @param data Vector of bytes of received data.\n * @param remote_info "
"Socket::Info containing the sender's information. This\n * will be populated "
"with the information about the sender.\n * @return True if successfully received, "
"False otherwise.\n",
py::call_guard())
.def("bind", &espp::UdpSocket::bind, py::arg("receive_config"),
"*\n * @brief Bind the socket as a server according to \\p receive_config (bind to\n "
" * the port and, if requested, join the multicast group), without\n * "
"starting any receive thread.\n * @note This is called for you by start_receiving(). "
"Use it directly when\n * driving the socket from an external event loop such "
"as\n * espp::SocketReactor, which reads the socket itself.\n * @param "
"receive_config ReceiveConfig describing the port / multicast setup.\n * Its "
"callback / buffer_size fields are not used by this method.\n * @return True if the "
"socket was bound (and joined the group, if multicast).\n")
.def("start_receiving", &espp::UdpSocket::start_receiving, py::arg("task_config"),
py::arg("receive_config"),
"*\n * @brief Configure a server socket and start a thread to continuously\n * "
" receive and handle data coming in on that socket.\n *\n * @param task_config "
"Task::BaseConfig struct for configuring the receive task.\n * @param receive_config "
"ReceiveConfig struct with socket and callback info.\n * @return True if the socket "
"was created and task was started, False otherwise.\n",
py::call_guard());
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassTask = py::class_(
m, "Task", py::dynamic_attr(),
"*\n * @brief Task provides an abstraction over std::thread which optionally\n * includes "
"memory / priority configuration on ESP systems. It allows users to\n * easily stop the "
"task, and will automatically stop itself if destroyed.\n *\n * There is also a utility "
"function which can be used to get the info for the\n * task of the current context, or for "
"a provided Task object.\n *\n * There is also a helper function to run a lambda on a "
"specific core, which can\n * be used to run a specific function on a specific core, as you "
"might want to\n * do when registering an interrupt driver on a specific core.\n *\n * "
"\\section task_ex1 Basic Task Example\n * \\snippet task_example.cpp Task example\n * "
"\\section task_ex2 Task Watchdog Example\n * \\snippet task_example.cpp task watchdog "
"example\n * \\section task_ex3 Many Task Example\n * \\snippet task_example.cpp ManyTask "
"example\n * \\section task_ex4 Long Running Task Example\n * \\snippet task_example.cpp "
"LongRunningTask example\n * \\section task_ex5 Long Running Task Notified Example "
"(Recommended)\n * \\snippet task_example.cpp LongRunningTaskNotified example\n * \\section "
"task_ex6 Task Info Example\n * \\snippet task_example.cpp Task Info example\n * \\section "
"task_ex7 Task Request Stop Example\n * \\snippet task_example.cpp Task Request Stop "
"example\n * \\section task_ex8 Task Priority and Core Affinity Example\n * \\snippet "
"task_example.cpp Task Priority and Core example\n *\n * \\section run_on_core_ex1 Run on "
"Core Example\n * \\snippet task_example.cpp run on core example\n * \\section "
"run_on_core_ex2 Run on Core (Non-Blocking) Example\n * \\snippet task_example.cpp run on "
"core nonblocking example\n");
{ // inner classes & enums of Task
auto pyClassTask_ClassBaseConfig =
py::class_(
pyClassTask, "BaseConfig", py::dynamic_attr(),
"*\n * @brief Base configuration struct for the Task.\n * @note This is designed "
"to be used as a configuration struct in other classes\n * that may have a "
"Task as a member.\n")
.def(py::init([](std::string name = std::string(), size_t stack_size_bytes = {4096},
size_t priority = {0}, int core_id = {-1},
bool host_realtime = {false}) {
auto r_ctor_ = std::make_unique();
r_ctor_->name = name;
r_ctor_->stack_size_bytes = stack_size_bytes;
r_ctor_->priority = priority;
r_ctor_->core_id = core_id;
r_ctor_->host_realtime = host_realtime;
return r_ctor_;
}),
py::arg("name") = std::string(), py::arg("stack_size_bytes") = size_t{4096},
py::arg("priority") = size_t{0}, py::arg("core_id") = int{-1},
py::arg("host_realtime") = false)
.def_readwrite("name", &espp::Task::BaseConfig::name, "*< Name of the task")
.def_readwrite("stack_size_bytes", &espp::Task::BaseConfig::stack_size_bytes,
"*< Stack Size (B) allocated to the task.")
.def_readwrite("priority", &espp::Task::BaseConfig::priority,
"*< Priority of the task, 0 is lowest priority on ESP / FreeRTOS.")
.def_readwrite("core_id", &espp::Task::BaseConfig::core_id,
"*< Core ID of the task, -1 means it is not pinned to any core.")
.def_readwrite("host_realtime", &espp::Task::BaseConfig::host_realtime,
"*< Opt-in to applying the priority to the OS thread on host "
"platforms (SCHED_FIFO on Linux/macOS; ignored on ESP).");
auto pyClassTask_ClassConfig =
py::class_(
pyClassTask, "Config", py::dynamic_attr(),
"*\n * @brief Configuration struct for the Task.\n * Can be initialized "
"with any of the supported callback function\n * signatures.\n")
.def(py::init(
[](espp::Task::callback_variant callback = espp::Task::callback_variant(),
espp::Task::BaseConfig task_config = espp::Task::BaseConfig(),
espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->callback = callback;
r_ctor_->task_config = task_config;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("callback") = espp::Task::callback_variant(),
py::arg("task_config") = espp::Task::BaseConfig(),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("callback", &espp::Task::Config::callback, "*< Callback function")
.def_readwrite("task_config", &espp::Task::Config::task_config,
"*< Base configuration for the task.")
.def_readwrite("log_level", &espp::Task::Config::log_level,
"*< Log verbosity for the task.");
} // end of inner classes & enums of Task
pyClassTask
.def(py::init())
.def(py::init())
.def("start", &espp::Task::start,
"*\n * @brief Start executing the task.\n *\n * @return True if the task started, "
"False if it was already started.\n",
py::call_guard())
.def("stop", &espp::Task::stop,
"*\n * @brief Stop the task execution.\n * @details This will request the task to "
"stop, notify the condition variable,\n * and (if this calling context is "
"not the task context) join the\n * thread.\n * @return True if the task "
"stopped, False if it was not started / already\n * stopped.\n",
py::call_guard())
.def("is_started", &espp::Task::is_started,
"*\n * @brief Has the task been started or not?\n *\n * @return True if the task "
"is started / running, False otherwise.\n")
.def("is_running", &espp::Task::is_running,
"*\n * @brief Is the task running?\n *\n * @return True if the task is running, "
"False otherwise.\n")
.def("set_priority", &espp::Task::set_priority, py::arg("priority"),
"*\n * @brief Set the priority of the task.\n * @details The new priority is always "
"stored in the task's configuration, so\n * it will be used the next time "
"the task is started. If the task is\n * currently running (ESP only), the "
"change is also applied to the\n * live task immediately via "
"vTaskPrioritySet().\n * @param priority New FreeRTOS priority (0 is lowest priority "
"on ESP /\n * FreeRTOS). It is clamped to [0, configMAX_PRIORITIES - 1] on "
"ESP.\n * @return True if the change was applied to the currently-running task; "
"False\n * if the task is not running (the new value still takes effect "
"the\n * next time the task is started) or the platform does not support\n "
"* changing a live task's priority.\n")
.def("get_configured_priority", &espp::Task::get_configured_priority,
"*\n * @brief Get the priority stored in the task's configuration.\n * @details "
"This is the value set at construction or via set_priority(); it\n * is the "
"priority the task will be started with (and, if the task\n * is running, "
"the priority that was last requested for it).\n * @return The configured priority "
"(0 is lowest; see BaseConfig.priority).\n")
.def(
"set_core_id", &espp::Task::set_core_id, py::arg("core_id"),
"*\n * @brief Set the core affinity (core ID) of the task.\n * @details The new core "
"ID is always stored in the task's configuration, so\n * it will be used the "
"next time the task is started. On the default\n * ESP-IDF FreeRTOS port a "
"running task's core affinity cannot be\n * changed (it is fixed when the "
"task is created), so for an\n * already-started task this only takes effect "
"after a stop()/start().\n * If the underlying FreeRTOS build does provide a "
"runtime\n * core-affinity API (configUSE_CORE_AFFINITY on a multi-core SMP\n "
" * build), the change is applied to the live task immediately.\n * @param "
"core_id Core to pin the task to (0 or 1), or -1 to leave the task\n * "
"unpinned (able to run on any core).\n * @return True if the change was applied to the "
"currently-running task; False\n * if the task is not running or the platform "
"cannot change a live\n * task's core affinity (in which case the new value "
"still takes\n * effect the next time the task is started).\n")
.def(
"get_id", [](espp::Task &self) { return self.get_id(); },
"*\n * @brief Get the ID for this Task's thread / task context.\n * @return ID for "
"this Task's thread / task context.\n * @warning This will only return a valid id if "
"the task is started.\n")
.def_static("get_current_id", &espp::Task::get_current_id,
"*\n * @brief Get the ID for the current thread / task context.\n * @return "
"ID for the current thread / task context.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassTimer = py::class_(
m, "Timer", py::dynamic_attr(),
"/ @brief A timer that can be used to schedule tasks to run at a later time.\n/ @details A "
"timer can be used to schedule a task to run at a later time.\n/ The timer will run "
"in the background and will call the task when\n/ the time is up. The timer can be "
"canceled at any time. A timer\n/ can be configured to run once or to repeat.\n/\n/ "
" The timer uses a task to run in the background. The task will\n/ sleep "
"until the timer is ready to run. When the timer is ready to\n/ run, the task will "
"call the callback function. The callback\n/ function can return True to cancel the "
"timer or False to keep the\n/ timer running. If the timer is configured to repeat, "
"then the\n/ callback function will be called again after the period has\n/ "
" elapsed. If the timer is configured to run once, then the\n/ callback function "
"will only be called once.\n/\n/ The timer can be configured to start automatically "
"when it is\n/ constructed. If the timer is not configured to start\n/ "
"automatically, then the timer can be started by calling start().\n/ The timer can "
"be canceled at any time by calling cancel().\n/\n/ @note The timer uses a task to run in "
"the background, so the timer\n/ callback function will be called in the context of "
"the task. The\n/ timer callback function should not block for a long time because "
"it\n/ will block the task. If the timer callback function blocks for a\n/ long "
"time, then the timer will not be able to keep up with the\n/ period.\n/\n/ @note "
"Timing resolution. On ESP / FreeRTOS the timer waits on the\n/ scheduler, which can "
"only resolve time to a single tick\n/ (1 / CONFIG_FREERTOS_HZ seconds; e.g. 10 ms at "
"the 100 Hz default,\n/ 1 ms at 1000 Hz). A period or delay that is shorter than - or "
"within\n/ a couple of ticks of - the tick period cannot be honored accurately:\n/ "
" it will be rounded up to a whole number of ticks and can jitter by up\n/ to a full "
"tick. The constructor, set_period() and start(delay) log a\n/ warning when the "
"requested period/delay is at or near the tick\n/ period. For sub-tick or highly "
"accurate periodic work, either raise\n/ CONFIG_FREERTOS_HZ or use the esp_timer-based "
"HighResolutionTimer\n/ instead. The timer schedules against an absolute wake-up time "
"(the\n/ k-th callback targets start + k*period), so it does not accumulate\n/ "
"drift even when individual iterations jitter.\n/\n/ \\section timer_ex1 Timer Example 1\n/ "
"\\snippet timer_example.cpp timer example\n/ \\section timer_ex2 Timer Watchdog Example\n/ "
"\\snippet timer_example.cpp timer watchdog example\n/ \\section timer_ex3 Timer Delay "
"Example\n/ \\snippet timer_example.cpp timer delay example\n/ \\section timer_ex4 Oneshot "
"Timer Example\n/ \\snippet timer_example.cpp timer oneshot example\n/ \\section timer_ex5 "
"Timer Cancel Itself Example\n/ \\snippet timer_example.cpp timer cancel itself example\n/ "
"\\section timer_ex6 Oneshot Timer Cancel Itself Then Start again with Delay Example\n/ "
"\\snippet timer_example.cpp timer oneshot restart example\n/ \\section timer_ex7 Timer "
"Update Period Example\n/ \\snippet timer_example.cpp timer update period example\n/ "
"\\section timer_ex8 Timer AdvancedConfig Example\n/ \\snippet timer_example.cpp timer "
"advanced config example");
{ // inner classes & enums of Timer
auto pyClassTimer_ClassConfig =
py::class_(pyClassTimer, "Config", py::dynamic_attr(),
"/ @brief The configuration for the timer.")
.def(py::init([](std::string_view name = std::string_view(),
std::chrono::duration period = std::chrono::duration(),
std::chrono::duration delay = std::chrono::duration(0),
espp::Timer::callback_fn callback = espp::Timer::callback_fn(),
bool auto_start = {true}, size_t stack_size_bytes = {4096},
size_t priority = {0}, int core_id = {-1},
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN) {
auto r_ctor_ = std::make_unique();
r_ctor_->name = name;
r_ctor_->period = period;
r_ctor_->delay = delay;
r_ctor_->callback = callback;
r_ctor_->auto_start = auto_start;
r_ctor_->stack_size_bytes = stack_size_bytes;
r_ctor_->priority = priority;
r_ctor_->core_id = core_id;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("name") = std::string_view(),
py::arg("period") = std::chrono::duration(),
py::arg("delay") = std::chrono::duration(0),
py::arg("callback") = espp::Timer::callback_fn(),
py::arg("auto_start") = bool{true}, py::arg("stack_size_bytes") = size_t{4096},
py::arg("priority") = size_t{0}, py::arg("core_id") = int{-1},
py::arg("log_level") = espp::Logger::Verbosity::WARN)
.def_readwrite("name", &espp::Timer::Config::name, "/< The name of the timer.")
.def_readwrite(
"period", &espp::Timer::Config::period,
"/< The period of the timer. If 0, the timer callback will only be called once.")
.def_readwrite("delay", &espp::Timer::Config::delay,
"/< The delay before the first execution of the timer callback after "
"start() is called.")
.def_readwrite("callback", &espp::Timer::Config::callback,
"/< The callback function to call when the timer expires.")
.def_readwrite("auto_start", &espp::Timer::Config::auto_start,
"/< If True, the timer will start automatically when constructed.")
.def_readwrite("stack_size_bytes", &espp::Timer::Config::stack_size_bytes,
"/< The stack size of the task that runs the timer.")
.def_readwrite("priority", &espp::Timer::Config::priority,
"/< Priority of the timer, 0 is lowest priority on ESP / FreeRTOS.")
.def_readwrite("core_id", &espp::Timer::Config::core_id,
"/< Core ID of the timer, -1 means it is not pinned to any core.")
.def_readwrite("log_level", &espp::Timer::Config::log_level,
"/< The log level for the timer.");
auto pyClassTimer_ClassAdvancedConfig =
py::class_(pyClassTimer, "AdvancedConfig", py::dynamic_attr(),
"/ @brief Advanced configuration for the timer.")
.def(py::init([](std::chrono::duration period = std::chrono::duration(),
std::chrono::duration delay = std::chrono::duration(0),
espp::Timer::callback_fn callback = espp::Timer::callback_fn(),
bool auto_start = {true},
espp::Task::BaseConfig task_config = espp::Task::BaseConfig(),
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN) {
auto r_ctor_ = std::make_unique();
r_ctor_->period = period;
r_ctor_->delay = delay;
r_ctor_->callback = callback;
r_ctor_->auto_start = auto_start;
r_ctor_->task_config = task_config;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("period") = std::chrono::duration(),
py::arg("delay") = std::chrono::duration(0),
py::arg("callback") = espp::Timer::callback_fn(),
py::arg("auto_start") = bool{true},
py::arg("task_config") = espp::Task::BaseConfig(),
py::arg("log_level") = espp::Logger::Verbosity::WARN)
.def_readwrite(
"period", &espp::Timer::AdvancedConfig::period,
"/< The period of the timer. If 0, the timer callback will only be called once.")
.def_readwrite("delay", &espp::Timer::AdvancedConfig::delay,
"/< The delay before the first execution of the timer callback after "
"start() is called.")
.def_readwrite("callback", &espp::Timer::AdvancedConfig::callback,
"/< The callback function to call when the timer expires.")
.def_readwrite("auto_start", &espp::Timer::AdvancedConfig::auto_start,
"/< If True, the timer will start automatically when constructed.")
.def_readwrite("task_config", &espp::Timer::AdvancedConfig::task_config,
"/< The task configuration for the timer.")
.def_readwrite("log_level", &espp::Timer::AdvancedConfig::log_level,
"/< The log level for the timer.");
} // end of inner classes & enums of Timer
pyClassTimer.def(py::init())
.def(py::init())
.def(
"start", [](espp::Timer &self) { return self.start(); },
"/ @brief Start the timer.\n/ @details Starts the timer. Does nothing if the timer is "
"already running.\n/ @return True if the timer was started or is already running, False "
"if the\n/ timer could not be started.",
py::call_guard())
.def("start", py::overload_cast(&espp::Timer::start),
py::arg("delay"),
"/ @brief Start the timer with a delay.\n/ @details Starts the timer with a delay. If "
"the timer is already running,\n/ this will cancel the timer and start it "
"again with the new\n/ delay. If the timer is not running, this will start the "
"timer\n/ with the delay. Overwrites any previous delay that might have\n/ "
" been set.\n/ @param delay The delay before the first execution of the timer "
"callback.\n/ @return True if the timer was started or restarted, False if the timer\n/ "
" could not be started.",
py::call_guard())
.def("stop", &espp::Timer::stop,
"/ @brief Stop the timer, same as cancel().\n/ @details Stops the timer, same as "
"cancel().",
py::call_guard())
.def("cancel", &espp::Timer::cancel,
"/ @brief Cancel the timer.\n/ @details Cancels the timer.",
py::call_guard())
.def("set_period", &espp::Timer::set_period, py::arg("period"),
"/ @brief Set the period of the timer.\n/ @details Sets the period of the timer.\n/ "
"@param period The period of the timer.\n/ @note If the period is 0, the timer will run "
"once.\n/ @note If the period is negative, the period will not be set / updated.\n/ "
"@note If the timer is running, the period will be updated after the\n/ current "
"period has elapsed.")
.def("is_running", &espp::Timer::is_running,
"/ @brief Check if the timer is running.\n/ @details Checks if the timer is running.\n/ "
"@return True if the timer is running, False otherwise.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassTrajectoryPlanner = py::class_(
m, "TrajectoryPlanner", py::dynamic_attr(),
"*\n * @brief Converts normalized joystick velocity commands into smooth,\n * "
"dynamically feasible chassis motion commands (v, w).\n *\n * The planner is drive-system "
"independent -- it does not know about wheel\n * geometry or kinematics. It only enforces "
"velocity, acceleration, and\n * jerk limits on chassis-level commands. The downstream "
"kinematics layer\n * converts (v_ref, w_ref) into individual motor commands.\n *\n * ### "
"Algorithm\n * The jerk-limited mode uses a discrete optimal-control approach: at each\n * "
"step the planner computes the minimum velocity-change distance needed to\n * decelerate "
"the current acceleration to zero, then decides whether to\n * accelerate, maintain, or "
"decelerate to land exactly on the target without\n * overshoot -- equivalent to a "
"time-optimal S-curve under jerk and\n * acceleration constraints.\n *\n * ### Profiles\n "
"* Motion limits are grouped into two MotionProfile objects inside Config:\n * - "
"**driving_profile** -- used whenever the target is non-zero.\n * - **stopping_profile** -- "
"used when the target is (0, 0). Setting jerk to 0\n * gives a trapezoidal stop; higher "
"acceleration gives faster, firmer braking.\n *\n * A MotionProfile selects its mode "
"automatically:\n * - **Trapezoidal**: `max_linear_jerk == 0 && max_angular_jerk == 0`\n * "
"- **S-curve**: either jerk field is non-zero\n *\n * ### Timing\n * Two independent "
"timers run internally:\n * - **planning timer** -- calls `update()` at `planning_period` "
"(default 20 ms / 50 Hz).\n * Recommended range: 5-200 ms on microcontrollers.\n * - "
"**callback task** -- fires `output_callback` on every `update()` that produces a\n * new "
"output value (CV-notified by the planning timer); no separate period needed.\n *\n * This "
"class is thread-safe: set_target(), get_target(), output(), stop(),\n * and reset() may be "
"called from different threads concurrently.\n *\n * \\section trajectory_planner_ex0 "
"Quick-Start: Full Public API\n * \\snippet trajectory_planner_example.cpp "
"trajectory_planner quickstart\n * \\section trajectory_planner_ex1 S-Curve Driving / "
"Trapezoidal Stop\n * \\snippet trajectory_planner_example.cpp trajectory_planner example\n "
"* \\section trajectory_planner_ex2 High-Speed S-Curve with Centripetal Limiting\n * "
"\\snippet trajectory_planner_example.cpp trajectory_planner jerk example\n * \\section "
"trajectory_planner_ex3 Constraint Validation\n * \\snippet trajectory_planner_example.cpp "
"trajectory_planner validation\n");
{ // inner classes & enums of TrajectoryPlanner
auto pyClassTrajectoryPlanner_ClassMotionCommand =
py::class_(
pyClassTrajectoryPlanner, "MotionCommand", py::dynamic_attr(),
"*\n * @brief Chassis motion command produced by the planner.\n")
.def(py::init([](float linear_velocity = 0.0f, float angular_velocity = 0.0f) {
auto r_ctor_ = std::make_unique();
r_ctor_->linear_velocity = linear_velocity;
r_ctor_->angular_velocity = angular_velocity;
return r_ctor_;
}),
py::arg("linear_velocity") = 0.0f, py::arg("angular_velocity") = 0.0f)
.def_readwrite("linear_velocity",
&espp::TrajectoryPlanner::MotionCommand::linear_velocity,
"*< Linear velocity reference (m/s).")
.def_readwrite("angular_velocity",
&espp::TrajectoryPlanner::MotionCommand::angular_velocity,
"*< Angular velocity reference (rad/s).");
auto pyClassTrajectoryPlanner_ClassMotionProfile =
py::class_(
pyClassTrajectoryPlanner, "MotionProfile", py::dynamic_attr(),
"*\n * @brief Acceleration and jerk limits for one phase of motion.\n *\n * Set "
"max_linear_jerk / max_angular_jerk to 0 for a trapezoidal (ramp)\n * profile, or to "
"a positive value for an S-curve profile.\n *\n * @note Using a trapezoidal "
"stopping profile (jerk = 0) is recommended to\n * avoid S-curve overshoot "
"past zero when the planner decelerates from\n * a jerk-limited driving "
"phase.\n")
.def(py::init([](float max_linear_acceleration = 0.0f,
float max_angular_acceleration = 0.0f, float max_linear_jerk = 0.0f,
float max_angular_jerk = 0.0f) {
auto r_ctor_ = std::make_unique();
r_ctor_->max_linear_acceleration = max_linear_acceleration;
r_ctor_->max_angular_acceleration = max_angular_acceleration;
r_ctor_->max_linear_jerk = max_linear_jerk;
r_ctor_->max_angular_jerk = max_angular_jerk;
return r_ctor_;
}),
py::arg("max_linear_acceleration") = 0.0f,
py::arg("max_angular_acceleration") = 0.0f, py::arg("max_linear_jerk") = 0.0f,
py::arg("max_angular_jerk") = 0.0f)
.def_readwrite("max_linear_acceleration",
&espp::TrajectoryPlanner::MotionProfile::max_linear_acceleration,
"*< Linear acceleration limit (m/s^2).")
.def_readwrite("max_angular_acceleration",
&espp::TrajectoryPlanner::MotionProfile::max_angular_acceleration,
"*< Angular acceleration limit (rad/s^2).")
.def_readwrite("max_linear_jerk",
&espp::TrajectoryPlanner::MotionProfile::max_linear_jerk,
"*< Linear jerk limit (m/s^3). 0 = trapezoidal.")
.def_readwrite("max_angular_jerk",
&espp::TrajectoryPlanner::MotionProfile::max_angular_jerk,
"*< Angular jerk limit (rad/s^3). 0 = trapezoidal.");
auto pyClassTrajectoryPlanner_ClassConfig =
py::class_(
pyClassTrajectoryPlanner, "Config", py::dynamic_attr(),
"*\n * @brief Configuration for the TrajectoryPlanner.\n")
.def(
py::init(
[](float max_linear_velocity = 1.0f, float max_angular_velocity = 3.14159f,
espp::TrajectoryPlanner::MotionProfile driving_profile =
espp::TrajectoryPlanner::MotionProfile(),
espp::TrajectoryPlanner::MotionProfile stopping_profile =
espp::TrajectoryPlanner::MotionProfile(),
bool enforce_motion_envelope = false,
float max_centripetal_acceleration = 0.1f,
espp::TrajectoryPlanner::output_callback_t output_callback = nullptr,
std::chrono::duration planning_period = std::chrono::milliseconds(20),
espp::Task::BaseConfig planning_task_config = {.name = "TP_planning",
.stack_size_bytes = 4096,
.priority = 0,
.core_id = -1},
espp::Task::BaseConfig callback_task_config = {.name = "TP_cb",
.stack_size_bytes = 8192,
.priority = 0,
.core_id = -1},
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN) {
auto r_ctor_ = std::make_unique();
r_ctor_->max_linear_velocity = max_linear_velocity;
r_ctor_->max_angular_velocity = max_angular_velocity;
r_ctor_->driving_profile = driving_profile;
r_ctor_->stopping_profile = stopping_profile;
r_ctor_->enforce_motion_envelope = enforce_motion_envelope;
r_ctor_->max_centripetal_acceleration = max_centripetal_acceleration;
r_ctor_->output_callback = output_callback;
r_ctor_->planning_period = planning_period;
r_ctor_->planning_task_config = planning_task_config;
r_ctor_->callback_task_config = callback_task_config;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("max_linear_velocity") = 1.0f, py::arg("max_angular_velocity") = 3.14159f,
py::arg("driving_profile") = espp::TrajectoryPlanner::MotionProfile(),
py::arg("stopping_profile") = espp::TrajectoryPlanner::MotionProfile(),
py::arg("enforce_motion_envelope") = false,
py::arg("max_centripetal_acceleration") = 0.1f,
py::arg("output_callback") = py::none(),
py::arg("planning_period") = std::chrono::milliseconds(20),
py::arg("planning_task_config") = espp::Task::BaseConfig{.name = "TP_planning",
.stack_size_bytes = 4096,
.priority = 0,
.core_id = -1},
py::arg("callback_task_config") =
espp::Task::BaseConfig{
.name = "TP_cb", .stack_size_bytes = 8192, .priority = 0, .core_id = -1},
py::arg("log_level") = espp::Logger::Verbosity::WARN)
.def_readwrite("max_linear_velocity",
&espp::TrajectoryPlanner::Config::max_linear_velocity,
"*< Maximum linear velocity magnitude (m/s).")
.def_readwrite("max_angular_velocity",
&espp::TrajectoryPlanner::Config::max_angular_velocity,
"*< Maximum angular velocity magnitude (rad/s).")
.def_readwrite("driving_profile", &espp::TrajectoryPlanner::Config::driving_profile,
"*< Accel/jerk limits used when target != (0, 0).")
.def_readwrite(
"stopping_profile", &espp::TrajectoryPlanner::Config::stopping_profile,
"*< Accel/jerk limits used when target\n "
"== (0, 0). Set jerk to 0 here for a clean trapezoidal stop\n "
" with no overshoot. Higher acceleration than the\n "
" driving profile gives faster, firmer braking.")
.def_readwrite(
"enforce_motion_envelope",
&espp::TrajectoryPlanner::Config::enforce_motion_envelope,
"*< When True, enforces (v/vmax)^2+(w/wmax)^2= 1).\n/\n/ @note This class does "
"not manage RTP headers (sequence numbers, timestamps,\n/ SSRC). The caller wraps "
"each returned chunk into an RtpPacket.\n/\n/ \\section h264_packetizer_ex1 Example\n/ "
"\\snippet rtsp_example.cpp h264_packetizer_test");
{ // inner classes & enums of H264Packetizer
auto pyClassH264Packetizer_ClassConfig =
py::class_(pyClassH264Packetizer, "Config",
py::dynamic_attr(),
"/ Configuration for the H264Packetizer.")
.def(py::init(
[](size_t max_payload_size = {1400}, int payload_type = {96},
std::string profile_level_id = std::string(), int packetization_mode = {1},
std::vector sps = std::vector(),
std::vector pps = std::vector(),
espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->max_payload_size = max_payload_size;
r_ctor_->payload_type = payload_type;
r_ctor_->profile_level_id = profile_level_id;
r_ctor_->packetization_mode = packetization_mode;
r_ctor_->sps = sps;
r_ctor_->pps = pps;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("max_payload_size") = size_t{1400}, py::arg("payload_type") = int{96},
py::arg("profile_level_id") = std::string(),
py::arg("packetization_mode") = int{1}, py::arg("sps") = std::vector(),
py::arg("pps") = std::vector(),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("max_payload_size", &espp::H264Packetizer::Config::max_payload_size,
"/< Maximum payload bytes per RTP packet")
.def_readwrite("payload_type", &espp::H264Packetizer::Config::payload_type,
"/< Dynamic RTP payload type (typically 96-127).")
.def_readwrite("profile_level_id", &espp::H264Packetizer::Config::profile_level_id,
"/< H.264 profile-level-id hex string, e.g. \"42C01E\".")
.def_readwrite("packetization_mode", &espp::H264Packetizer::Config::packetization_mode,
"/< 0 = single NAL only, 1 = non-interleaved (FU-A allowed).")
.def_readwrite("sps", &espp::H264Packetizer::Config::sps,
"/< Sequence Parameter Set raw bytes (without start code).")
.def_readwrite("pps", &espp::H264Packetizer::Config::pps,
"/< Picture Parameter Set raw bytes (without start code).")
.def_readwrite("log_level", &espp::H264Packetizer::Config::log_level,
"/< Log verbosity level");
} // end of inner classes & enums of H264Packetizer
pyClassH264Packetizer.def(py::init())
.def("packetize", &espp::H264Packetizer::packetize, py::arg("frame_data"),
"/ Packetize a complete H.264 access unit (Annex B format).\n/\n/ The input may contain "
"multiple NAL units separated by 3-byte or 4-byte\n/ start codes. Each NAL is "
"individually packetized (single NAL or FU-A).\n/ The marker bit is set on the last "
"chunk of the last NAL unit in the\n/ access unit.\n/\n/ @param frame_data Raw Annex B "
"byte-stream of one access unit.\n/ @return Vector of RTP payload chunks ready for "
"transmission.")
.def("packetize_nal", &espp::H264Packetizer::packetize_nal, py::arg("nal_data"),
py::arg("is_last_nal") = true,
"/ Packetize a single pre-parsed NAL unit (no start code prefix).\n/\n/ @param nal_data "
"The raw NAL unit bytes (including NAL header byte).\n/ @param is_last_nal If True, the "
"marker bit is set on the last chunk.\n/ @return Vector of RTP payload chunks for this "
"NAL.")
.def("set_sps_pps", &espp::H264Packetizer::set_sps_pps, py::arg("sps"), py::arg("pps"),
"/ Update the SPS and PPS used for SDP generation.\n/ @param sps Sequence Parameter Set "
"raw bytes.\n/ @param pps Picture Parameter Set raw bytes.")
.def("get_payload_type", &espp::H264Packetizer::get_payload_type,
"/ Get the RTP payload type.\n/ @return The dynamic payload type configured for H.264.")
.def("get_clock_rate", &espp::H264Packetizer::get_clock_rate,
"/ Get the RTP clock rate for H.264 video.\n/ @return 90000 (fixed for H.264).")
.def("get_sdp_media_attributes", &espp::H264Packetizer::get_sdp_media_attributes,
"/ Get the SDP attribute lines for H.264.\n/ @return SDP a= lines (rtpmap and fmtp) "
"without trailing CRLF.")
.def("get_sdp_media_line", &espp::H264Packetizer::get_sdp_media_line,
"/ Get the SDP m= media line for H.264.\n/ @return SDP m= line without trailing CRLF.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassMjpegDepacketizer = py::class_(
m, "MjpegDepacketizer", py::dynamic_attr(),
"/ MJPEG depacketizer that reassembles JPEG frames from RTP packets.\n/\n/ This class "
"receives individual RTP packets containing RFC 2435 MJPEG\n/ payloads, reassembles the scan "
"data fragments, reconstructs the JPEG\n/ header from the MJPEG header fields, and delivers "
"complete JPEG frames\n/ through callbacks.");
{ // inner classes & enums of MjpegDepacketizer
auto pyClassMjpegDepacketizer_ClassConfig =
py::class_(pyClassMjpegDepacketizer, "Config",
py::dynamic_attr(),
"/ Configuration for the MJPEG depacketizer.")
.def(
py::init([](espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("log_level", &espp::MjpegDepacketizer::Config::log_level,
"/< Log verbosity level");
} // end of inner classes & enums of MjpegDepacketizer
pyClassMjpegDepacketizer.def(py::init())
.def("set_jpeg_frame_callback", &espp::MjpegDepacketizer::set_jpeg_frame_callback,
py::arg("cb"),
"/ Set callback for receiving complete JPEG frames.\n/ @param cb Callback receiving a "
"shared pointer to the completed JpegFrame.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassMjpegPacketizer = py::class_(
m, "MjpegPacketizer", py::dynamic_attr(),
"/ MJPEG packetizer that fragments JPEG frames into RFC 2435 RTP payloads.\n/\n/ This class "
"takes complete JPEG frames and produces RTP payload chunks\n/ suitable for MJPEG streaming. "
"Each chunk contains an RFC 2435 MJPEG\n/ header, and the first chunk additionally includes "
"quantization tables.");
{ // inner classes & enums of MjpegPacketizer
auto pyClassMjpegPacketizer_ClassConfig =
py::class_(pyClassMjpegPacketizer, "Config",
py::dynamic_attr(),
"/ Configuration for the MJPEG packetizer.")
.def(
py::init([](size_t max_payload_size = {1400},
espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) {
auto r_ctor_ = std::make_unique();
r_ctor_->max_payload_size = max_payload_size;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("max_payload_size") = size_t{1400},
py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN})
.def_readwrite("max_payload_size", &espp::MjpegPacketizer::Config::max_payload_size,
"/< Maximum payload bytes per RTP packet")
.def_readwrite("log_level", &espp::MjpegPacketizer::Config::log_level,
"/< Log verbosity level");
} // end of inner classes & enums of MjpegPacketizer
pyClassMjpegPacketizer.def(py::init())
.def("get_payload_type", &espp::MjpegPacketizer::get_payload_type,
"/ Get the RTP payload type for MJPEG.\n/ @return 26 (static JPEG payload type).")
.def("get_clock_rate", &espp::MjpegPacketizer::get_clock_rate,
"/ Get the RTP clock rate for MJPEG.\n/ @return 90000 Hz.")
.def("get_sdp_media_attributes", &espp::MjpegPacketizer::get_sdp_media_attributes,
"/ Get the SDP media attributes for MJPEG.\n/ @return SDP rtpmap attribute string.")
.def("get_sdp_media_line", &espp::MjpegPacketizer::get_sdp_media_line,
"/ Get the SDP media line for MJPEG.\n/ @return SDP media description line.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtpJpegPacket =
py::class_(
m, "RtpJpegPacket", py::dynamic_attr(),
"/ RTP packet for JPEG video.\n/ The RTP payload for JPEG is defined in RFC 2435.")
.def(py::init(), py::arg("data"),
"/ Construct an RTP packet from a buffer.\n/ @param data The buffer containing the "
"RTP packet.")
.def(
py::init(),
py::arg("type_specific"), py::arg("frag_type"), py::arg("q"), py::arg("width"),
py::arg("height"), py::arg("q0"), py::arg("q1"), py::arg("scan_data"),
"/ Construct an RTP packet from fields\n/ @details This will construct a packet with "
"quantization tables, so it\n/ can only be used for the first packet in a "
"frame.\n/ @param type_specific The type-specific field.\n/ @param frag_type The "
"fragment type field.\n/ @param q The q field.\n/ @param width The width field.\n/ "
"@param height The height field.\n/ @param q0 The first quantization table.\n/ "
"@param q1 The second quantization table.\n/ @param scan_data The scan data.")
.def(py::init(),
py::arg("type_specific"), py::arg("offset"), py::arg("frag_type"), py::arg("q"),
py::arg("width"), py::arg("height"), py::arg("scan_data"),
"/ Construct an RTP packet from fields\n/ @details This will construct a packet "
"without quantization tables, so it\n/ cannot be used for the first packet "
"in a frame.\n/ @param type_specific The type-specific field.\n/ @param offset The "
"offset field.\n/ @param frag_type The fragment type field.\n/ @param q The q "
"field.\n/ @param width The width field.\n/ @param height The height field.\n/ "
"@param scan_data The scan data.")
.def("get_type_specific", &espp::RtpJpegPacket::get_type_specific,
"/ Get the type-specific field.\n/ @return The type-specific field.")
.def("get_offset", &espp::RtpJpegPacket::get_offset,
"/ Get the offset field.\n/ @return The offset field.")
.def("get_q", &espp::RtpJpegPacket::get_q,
"/ Get the fragment type field.\n/ @return The fragment type field.")
.def("get_width", &espp::RtpJpegPacket::get_width,
"/ Get the fragment type field.\n/ @return The fragment type field.")
.def("get_height", &espp::RtpJpegPacket::get_height,
"/ Get the fragment type field.\n/ @return The fragment type field.")
.def("get_mjpeg_header", &espp::RtpJpegPacket::get_mjpeg_header,
"/ Get the mjepg header.\n/ @return The mjepg header.")
.def("has_q_tables", &espp::RtpJpegPacket::has_q_tables,
"/ Get whether the packet contains quantization tables.\n/ @note The quantization "
"tables are optional. If they are present, the\n/ number of quantization tables is "
"always 2.\n/ @note This check is based on the value of the q field. If the q "
"field\n/ is 128-256, the packet contains quantization tables.\n/ @return "
"Whether the packet contains quantization tables.")
.def("get_num_q_tables", &espp::RtpJpegPacket::get_num_q_tables,
"/ Get the number of quantization tables.\n/ @note The quantization tables are "
"optional. If they are present, the\n/ number of quantization tables is always "
"2.\n/ @note Only the first packet in a frame contains quantization tables.\n/ "
"@return The number of quantization tables.")
.def("get_q_table", &espp::RtpJpegPacket::get_q_table, py::arg("index"),
"/ Get the quantization table at the specified index.\n/ @param index The index of "
"the quantization table.\n/ @return The quantization table at the specified index.")
.def("set_q_table", &espp::RtpJpegPacket::set_q_table, py::arg("index"),
py::arg("q_table"),
"/ Set the quantization table at the specified index.\n/ @param index The index of "
"the quantization table.\n/ @param q_table The quantization table to set.\n/ @note "
"This will not change the size of the packet. If the index is out of\n/ "
"bounds, the quantization table will not be set.")
.def("get_jpeg_data", &espp::RtpJpegPacket::get_jpeg_data,
"/ Get the JPEG data.\n/ The jpeg data is the payload minus the mjpeg header and "
"quantization\n/ tables.\n/ @return The JPEG data.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassJpegFrame =
py::class_(
m, "JpegFrame", py::dynamic_attr(),
"/ A class that represents a complete JPEG frame.\n/\n/ This class is used to collect "
"the JPEG scans that are received in RTP\n/ packets and to serialize them into a "
"complete JPEG frame.")
.def(py::init(), py::arg("packet"),
"/ Construct a JpegFrame from a RtpJpegPacket.\n/\n/ This constructor will parse "
"the header of the packet and add the JPEG\n/ data to the frame.\n/\n/ @param "
"packet The packet to parse.")
.def(py::init(), py::arg("data"),
"/ Construct a JpegFrame from a vector of jpeg data.\n/ @param data The vector "
"containing the jpeg data.\n/ @note The vector must contain the complete JPEG data, "
"including the JPEG\n/ header and EOI marker.")
.def(py::init(), py::arg("data"),
"/ Construct a JpegFrame from a span of jpeg data.\n/ @param data The span "
"containing the jpeg data.\n/ @note The span must contain the complete JPEG data, "
"including the JPEG\n/ header and EOI marker.")
.def(py::init(), py::arg("data"), py::arg("size"),
"/ Construct a JpegFrame from buffer of jpeg data\n/ @param data The buffer "
"containing the jpeg data.\n/ @param size The size of the buffer.")
.def("get_header", &espp::JpegFrame::get_header,
"/ Get a reference to the header.\n/ @return A reference to the header.")
.def("get_width", &espp::JpegFrame::get_width,
"/ Get the width of the frame.\n/ @return The width of the frame.")
.def("get_height", &espp::JpegFrame::get_height,
"/ Get the height of the frame.\n/ @return The height of the frame.")
.def("is_complete", &espp::JpegFrame::is_complete,
"/ Check if the frame is complete.\n/ @return True if the frame is complete, False "
"otherwise.")
.def("append", &espp::JpegFrame::append, py::arg("packet"),
"/ Append a RtpJpegPacket to the frame.\n/ This will add the JPEG data to the "
"frame.\n/ @param packet The packet containing the scan to append.")
.def("add_scan",
py::overload_cast(&espp::JpegFrame::add_scan),
py::arg("packet"),
"/ Append a JPEG scan to the frame.\n/ This will add the JPEG data to the frame.\n/ "
"@note If the packet contains the EOI marker, the frame will be\n/ finalized, "
"and no further scans can be added.\n/ @param packet The packet containing the scan "
"to append.")
.def("get_data", &espp::JpegFrame::get_data,
"/ Get the serialized data.\n/ This will return the serialized data.\n/ @return The "
"serialized data.")
.def("get_scan_data", &espp::JpegFrame::get_scan_data,
"/ Get the scan data.\n/ This will return the scan data.\n/ @return The scan data.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassJpegHeader =
py::class_(
m, "JpegHeader", py::dynamic_attr(),
"/ A class to generate a JPEG header for a given image size and quantization tables.\n/ "
"The header is generated once and then cached for future use.\n/ The header is generated "
"according to the JPEG standard and is compatible with\n/ the ESP32 camera driver.")
.def(py::init(),
py::arg("width"), py::arg("height"), py::arg("q0_table"), py::arg("q1_table"),
"/ Create a JPEG header for a given image size and quantization tables.\n/ @param "
"width The image width in pixels.\n/ @param height The image height in pixels.\n/ "
"@param q0_table The quantization table for the Y channel.\n/ @param q1_table The "
"quantization table for the Cb and Cr channels.")
.def(py::init(), py::arg("data"),
"/ Create a JPEG header from a given JPEG header data.")
.def("get_width", &espp::JpegHeader::get_width,
"/ Get the image width.\n/ @return The image width in pixels.")
.def("get_height", &espp::JpegHeader::get_height,
"/ Get the image height.\n/ @return The image height in pixels.")
.def("size", &espp::JpegHeader::size,
"/ Get the size of the JPEG header data.\n/ @return The size of the JPEG header "
"data in bytes.\n/ @note This is the size of the serialized JPEG header, not the "
"image size.")
.def("is_valid", &espp::JpegHeader::is_valid,
"/ Returns whether this header parsed or serialized successfully.")
.def("get_data", &espp::JpegHeader::get_data,
"/ Get the JPEG header data.\n/ @return The JPEG header data.")
.def("get_quantization_table", &espp::JpegHeader::get_quantization_table,
py::arg("index"),
"/ Get the Quantization table at the index.\n/ @param index The index of the "
"quantization table.\n/ @return The quantization table.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtcpPacket =
py::class_(
m, "RtcpPacket", py::dynamic_attr(),
"/ @brief A class to represent a RTCP packet\n/ @details This class is used to represent "
"a RTCP packet.\n/ It is used as a base class for all RTCP packet types.\n/ "
"@note At the moment, this class is not used.")
.def(py::init(), "/ @brief Constructor, default")
.def("get_data", &espp::RtcpPacket::get_data,
"/ @brief Get the buffer of the packet\n/ @return The buffer of the packet");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtpPacket =
py::class_(
m, "RtpPacket", py::dynamic_attr(),
"/ RtpPacket is a class to parse RTP packet.\n/ It can be used to parse and serialize "
"RTP packets.\n/ The RTP header fields are stored in the class and can be modified.\n/ "
"The payload is stored in the packet_ vector and can be modified.")
.def(py::init(), "/ Construct an empty RtpPacket.\n/ The packet_ vector is empty and "
"the header fields are set to 0.")
.def(py::init(), py::arg("payload_size"),
"/ Construct an RtpPacket with a payload of size payload_size.\n/ The packet_ "
"vector is resized to RTP_HEADER_SIZE + payload_size.")
.def(py::init(), py::arg("data"),
"/ Construct an RtpPacket from a span of bytes.\n/ Stores the bytes in the packet_ "
"vector and parses the header.\n/ @param data The span of bytes to parse.")
.def("get_version", &espp::RtpPacket::get_version,
"/ Get the RTP version.\n/ @return The RTP version.")
.def("get_padding", &espp::RtpPacket::get_padding,
"/ Get the padding flag.\n/ @return The padding flag.")
.def("get_extension", &espp::RtpPacket::get_extension,
"/ Get the extension flag.\n/ @return The extension flag.")
.def("get_csrc_count", &espp::RtpPacket::get_csrc_count,
"/ Get the CSRC count.\n/ @return The CSRC count.")
.def("get_marker", &espp::RtpPacket::get_marker,
"/ Get the marker flag.\n/ @return The marker flag.")
.def("get_payload_type", &espp::RtpPacket::get_payload_type,
"/ Get the payload type.\n/ @return The payload type.")
.def("get_sequence_number", &espp::RtpPacket::get_sequence_number,
"/ Get the sequence number.\n/ @return The sequence number.")
.def("get_timestamp", &espp::RtpPacket::get_timestamp,
"/ Get the timestamp.\n/ @return The timestamp.")
.def("get_ssrc", &espp::RtpPacket::get_ssrc, "/ Get the SSRC.\n/ @return The SSRC.")
.def("set_version", &espp::RtpPacket::set_version, py::arg("version"),
"/ Set the RTP version.\n/ @param version The RTP version to set.")
.def("set_padding", &espp::RtpPacket::set_padding, py::arg("padding"),
"/ Set the padding flag.\n/ @param padding The padding flag to set.")
.def("set_extension", &espp::RtpPacket::set_extension, py::arg("extension"),
"/ Set the extension flag.\n/ @param extension The extension flag to set.")
.def("set_csrc_count", &espp::RtpPacket::set_csrc_count, py::arg("csrc_count"),
"/ Set the CSRC count.\n/ @param csrc_count The CSRC count to set.")
.def("set_marker", &espp::RtpPacket::set_marker, py::arg("marker"),
"/ Set the marker flag.\n/ @param marker The marker flag to set.")
.def("set_payload_type", &espp::RtpPacket::set_payload_type, py::arg("payload_type"),
"/ Set the payload type.\n/ @param payload_type The payload type to set.")
.def("set_sequence_number", &espp::RtpPacket::set_sequence_number,
py::arg("sequence_number"),
"/ Set the sequence number.\n/ @param sequence_number The sequence number to set.")
.def("set_timestamp", &espp::RtpPacket::set_timestamp, py::arg("timestamp"),
"/ Set the timestamp.\n/ @param timestamp The timestamp to set.")
.def("set_ssrc", &espp::RtpPacket::set_ssrc, py::arg("ssrc"),
"/ Set the SSRC.\n/ @param ssrc The SSRC to set.")
.def("serialize", &espp::RtpPacket::serialize,
"/ Serialize the RTP header.\n/ @note This method should be called after modifying "
"the RTP header fields.\n/ @note This method does not serialize the payload. To set "
"the payload, use\n/ set_payload().\n/ To get the payload, use "
"get_payload().")
.def("get_data", &espp::RtpPacket::get_data,
"/ Get a span view of the whole packet.\n/ @note The span is valid as long as the "
"packet_ vector is not modified.\n/ @note If you manually build the packet_ vector, "
"you should make sure that you\n/ call serialize() before calling this "
"method.\n/ @return A span of the whole packet.")
.def("get_rtp_header_size", &espp::RtpPacket::get_rtp_header_size,
"/ Get the size of the RTP header.\n/ @return The size of the RTP header.")
.def("get_rtp_header", &espp::RtpPacket::get_rtp_header,
"/ Get a span of bytes of the RTP header.\n/ @return A span of bytes of the RTP "
"header.")
.def("get_packet", &espp::RtpPacket::get_packet,
"/ Get a reference to the packet_ vector.\n/ @return A reference to the packet_ "
"vector.")
.def("get_payload", &espp::RtpPacket::get_payload,
"/ Get a span of bytes of the payload.\n/ @return A span of bytes of the payload.")
.def("set_payload", &espp::RtpPacket::set_payload, py::arg("payload"),
"/ Set the payload.\n/ @param payload The payload to set.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtspClient = py::class_(
m, "RtspClient", py::dynamic_attr(),
"/ A class for interacting with an RTSP server using RTP and RTCP over UDP\n/\n/ This class "
"is used to connect to an RTSP server and receive JPEG frames\n/ over RTP. It uses the TCP "
"socket to send RTSP requests and receive RTSP\n/ responses. It uses the UDP socket to "
"receive RTP and RTCP packets.\n/\n/ The RTSP client is designed to be used with the RTSP "
"server in the\n/ [camera-streamer]https://github.com/esp-cpp/camera-streamer) project, but "
"it\n/ should work with any RTSP server that sends JPEG frames over RTP.\n/\n/ \\section "
"rtsp_client_ex1 RtspClient Example\n/ \\snippet rtsp_example.cpp rtsp_client_example");
{ // inner classes & enums of RtspClient
auto pyClassRtspClient_ClassTrackInfo =
py::class_(pyClassRtspClient, "TrackInfo", py::dynamic_attr(),
"")
.def(py::init([](int track_id = {0}, int payload_type = {0}, int clock_rate = {0},
int channels = {1}, std::string media_type = std::string(),
std::string encoding_name = std::string(),
std::string control_path = std::string()) {
auto r_ctor_ = std::make_unique();
r_ctor_->track_id = track_id;
r_ctor_->payload_type = payload_type;
r_ctor_->clock_rate = clock_rate;
r_ctor_->channels = channels;
r_ctor_->media_type = media_type;
r_ctor_->encoding_name = encoding_name;
r_ctor_->control_path = control_path;
return r_ctor_;
}),
py::arg("track_id") = int{0}, py::arg("payload_type") = int{0},
py::arg("clock_rate") = int{0}, py::arg("channels") = int{1},
py::arg("media_type") = std::string(), py::arg("encoding_name") = std::string(),
py::arg("control_path") = std::string())
.def_readwrite("track_id", &espp::RtspClient::TrackInfo::track_id, "")
.def_readwrite("payload_type", &espp::RtspClient::TrackInfo::payload_type, "")
.def_readwrite("clock_rate", &espp::RtspClient::TrackInfo::clock_rate, "")
.def_readwrite("channels", &espp::RtspClient::TrackInfo::channels, "")
.def_readwrite("media_type", &espp::RtspClient::TrackInfo::media_type, "")
.def_readwrite("encoding_name", &espp::RtspClient::TrackInfo::encoding_name, "")
.def_readwrite("control_path", &espp::RtspClient::TrackInfo::control_path, "");
auto pyClassRtspClient_ClassConfig =
py::class_(pyClassRtspClient, "Config", py::dynamic_attr(),
"/ Configuration for the RTSP client")
.def(py::init(
[](std::string server_address = std::string(), int rtsp_port = {8554},
std::string path = {"/mjpeg/1"},
espp::RtspClient::frame_callback_t on_frame = {nullptr},
espp::RtspClient::jpeg_frame_callback_t on_jpeg_frame = {nullptr},
espp::RtspClient::disconnect_callback_t on_connection_lost = {nullptr},
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::INFO) {
auto r_ctor_ = std::make_unique();
r_ctor_->server_address = server_address;
r_ctor_->rtsp_port = rtsp_port;
r_ctor_->path = path;
r_ctor_->on_frame = on_frame;
r_ctor_->on_jpeg_frame = on_jpeg_frame;
r_ctor_->on_connection_lost = on_connection_lost;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("server_address") = std::string(), py::arg("rtsp_port") = int{8554},
py::arg("path") = std::string{"/mjpeg/1"},
py::arg("on_frame") = espp::RtspClient::frame_callback_t{nullptr},
py::arg("on_jpeg_frame") = espp::RtspClient::jpeg_frame_callback_t{nullptr},
py::arg("on_connection_lost") = espp::RtspClient::disconnect_callback_t{nullptr},
py::arg("log_level") = espp::Logger::Verbosity::INFO)
.def_readwrite("server_address", &espp::RtspClient::Config::server_address,
"/< The server IP Address to connect to")
.def_readwrite("rtsp_port", &espp::RtspClient::Config::rtsp_port,
"/< The port of the RTSP server")
.def_readwrite("path", &espp::RtspClient::Config::path,
"/< The path to the RTSP stream on the server. Will be appended")
.def_readwrite("on_frame", &espp::RtspClient::Config::on_frame,
"/ Generic frame callback for any codec (track_id, raw frame data)")
.def_readwrite("on_jpeg_frame", &espp::RtspClient::Config::on_jpeg_frame,
"/ JPEG-specific frame callback (backward compatible).\n/ If set and no "
"depacketizer is registered for PT 26, an MjpegDepacketizer\n/ is "
"automatically created.")
.def_readwrite(
"on_connection_lost", &espp::RtspClient::Config::on_connection_lost,
"/ Called once if the client loses the server after playback starts.\n/ This "
"callback is intended for applications that want to stop playback\n/ and re-enter "
"service discovery or reconnect logic automatically.")
.def_readwrite("log_level", &espp::RtspClient::Config::log_level,
"/< The verbosity of the logger");
} // end of inner classes & enums of RtspClient
pyClassRtspClient.def(py::init())
.def("send_request", &espp::RtspClient::send_request, py::arg("method"), py::arg("path"),
py::arg("extra_headers"), py::arg("ec"),
"/ Send an RTSP request to the server\n/ \note This is a blocking call\n/ \note This "
"will parse the response and set the session ID if it is\n/ present in the "
"response. If the response is not a 200 OK, then\n/ an error code will be set and "
"the response will be returned.\n/ If the response is a 200 OK, then the response "
"will be returned\n/ and the error code will be set to success.\n/ \\param method "
"The method to use for connecting.\n/ Options are \"OPTIONS\", \"DESCRIBE\", "
"\"SETUP\", \"PLAY\", and \"TEARDOWN\"\n/ \\param path The path to the RTSP stream on "
"the server.\n/ \\param extra_headers Any extra headers to send with the request. "
"These\n/ will be added to the request after the CSeq and Session headers. The\n/ "
" key is the header name and the value is the header value. For example,\n/ "
"{\"Accept\": \"application/sdp\"} will add \"Accept: application/sdp\" to the\n/ "
"request. The \"User-Agent\" header will be added automatically. The\n/ \"CSeq\" "
"and \"Session\" headers will be added automatically.\n/ The \"Accept\" header "
"will be added automatically. The \"Transport\"\n/ header will be added "
"automatically for the \"SETUP\" method. Defaults to\n/ an empty map.\n/ \\param "
"ec The error code to set if an error occurs\n/ \\return The response from the server",
py::call_guard())
.def("connect", &espp::RtspClient::connect, py::arg("ec"),
"/ Connect to the RTSP server\n/ Connects to the RTSP server and sends the OPTIONS "
"request.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("disconnect", &espp::RtspClient::disconnect, py::arg("ec"),
"/ Disconnect from the RTSP server\n/ Disconnects from the RTSP server and sends the "
"TEARDOWN request.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("describe", &espp::RtspClient::describe, py::arg("ec"),
"/ Describe the RTSP stream\n/ Sends the DESCRIBE request to the RTSP server and parses "
"the response.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("setup", py::overload_cast(&espp::RtspClient::setup), py::arg("ec"),
"/ Setup the RTSP stream\n/ \note Starts the RTP and RTCP threads.\n/ Sends the SETUP "
"request to the RTSP server and parses the response.\n/ \note The default ports are "
"5000 and 5001 for RTP and RTCP respectively.\n/ \note The default receive timeout is 5 "
"seconds.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("setup",
py::overload_cast(&espp::RtspClient::setup),
py::arg("rtp_port"), py::arg("rtcp_port"), py::arg("receive_timeout"), py::arg("ec"),
"/ Setup the RTSP stream\n/ Sends the SETUP request to the RTSP server and parses the "
"response.\n/ \note Starts the RTP and RTCP threads.\n/ \\param rtp_port The RTP client "
"port\n/ \\param rtcp_port The RTCP client port\n/ \\param receive_timeout The timeout "
"for receiving RTP and RTCP packets\n/ \\param ec The error code to set if an error "
"occurs",
py::call_guard())
.def("add_depacketizer", &espp::RtspClient::add_depacketizer, py::arg("payload_type"),
py::arg("depacketizer"),
"/ Register a depacketizer for a specific RTP payload type.\n/ When RTP packets with "
"this payload type are received, they are\n/ dispatched to the registered "
"depacketizer.\n/ @param payload_type The RTP payload type (e.g., 26 for MJPEG, 96 for "
"H264)\n/ @param depacketizer The depacketizer to handle packets of this type")
.def("play", &espp::RtspClient::play, py::arg("ec"),
"/ Play the RTSP stream\n/ Sends the PLAY request to the RTSP server and parses the "
"response.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("pause", &espp::RtspClient::pause, py::arg("ec"),
"/ Pause the RTSP stream\n/ Sends the PAUSE request to the RTSP server and parses the "
"response.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("teardown", &espp::RtspClient::teardown, py::arg("ec"),
"/ Teardown the RTSP stream\n/ Sends the TEARDOWN request to the RTSP server and parses "
"the response.\n/ \\param ec The error code to set if an error occurs",
py::call_guard())
.def("tracks", &espp::RtspClient::tracks,
"/ Get the parsed SDP track descriptions from the most recent DESCRIBE call.\n/ "
"\\return The ordered set of discovered media tracks.");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtspServer = py::class_(
m, "RtspServer", py::dynamic_attr(),
"/ Class for streaming MJPEG data from a camera using RTSP + RTP\n/ Starts a TCP socket to "
"listen for RTSP connections, and then spawns off a\n/ new RTSP session for each "
"connection.\n/ @see RtspSession\n/ @note This class does not currently send RTCP "
"packets\n/\n/ \\section rtsp_server_ex1 RtspServer example\n/ \\snippet rtsp_example.cpp "
"rtsp_server_example");
{ // inner classes & enums of RtspServer
auto pyClassRtspServer_ClassConfig =
py::class_(pyClassRtspServer, "Config", py::dynamic_attr(),
"/ @brief Configuration for the RTSP server")
.def(py::init(
[](std::string server_address = std::string(), int port = int(),
std::string path = std::string(), size_t max_data_size = 1000,
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN,
size_t accept_task_stack_size_bytes =
espp::RtspServer::Config::default_accept_task_stack_size_bytes,
size_t session_task_stack_size_bytes =
espp::RtspServer::Config::default_session_task_stack_size_bytes,
size_t control_task_stack_size_bytes =
espp::RtspSession::Config::default_control_task_stack_size_bytes) {
auto r_ctor_ = std::make_unique();
r_ctor_->server_address = server_address;
r_ctor_->port = port;
r_ctor_->path = path;
r_ctor_->max_data_size = max_data_size;
r_ctor_->log_level = log_level;
r_ctor_->accept_task_stack_size_bytes = accept_task_stack_size_bytes;
r_ctor_->session_task_stack_size_bytes = session_task_stack_size_bytes;
r_ctor_->control_task_stack_size_bytes = control_task_stack_size_bytes;
return r_ctor_;
}),
py::arg("server_address") = std::string(), py::arg("port") = int(),
py::arg("path") = std::string(), py::arg("max_data_size") = 1000,
py::arg("log_level") = espp::Logger::Verbosity::WARN,
py::arg("accept_task_stack_size_bytes") =
espp::RtspServer::Config::default_accept_task_stack_size_bytes,
py::arg("session_task_stack_size_bytes") =
espp::RtspServer::Config::default_session_task_stack_size_bytes,
py::arg("control_task_stack_size_bytes") =
espp::RtspSession::Config::default_control_task_stack_size_bytes)
.def_readwrite("server_address", &espp::RtspServer::Config::server_address,
"/< The ip address of the server")
.def_readwrite("port", &espp::RtspServer::Config::port, "/< The port to listen on")
.def_readwrite("path", &espp::RtspServer::Config::path,
"/< The path to the RTSP stream")
.def_readwrite("max_data_size", &espp::RtspServer::Config::max_data_size,
"/< The maximum size of RTP packet data for the MJPEG stream. Frames "
"will be broken")
.def_readwrite("log_level", &espp::RtspServer::Config::log_level,
"/< The log level for the RTSP server")
.def_readwrite("accept_task_stack_size_bytes",
&espp::RtspServer::Config::accept_task_stack_size_bytes,
"/< RTSP accept-task stack size, in bytes")
.def_readwrite("session_task_stack_size_bytes",
&espp::RtspServer::Config::session_task_stack_size_bytes,
"/< RTSP session-dispatch task stack size, in bytes")
.def_readwrite("control_task_stack_size_bytes",
&espp::RtspServer::Config::control_task_stack_size_bytes,
"/< Per-session RTSP");
auto pyClassRtspServer_ClassTrackConfig =
py::class_(
pyClassRtspServer, "TrackConfig", py::dynamic_attr(),
"/ Configuration for a media track to be registered with the server")
.def(py::init([](int track_id = {0}, std::shared_ptr packetizer =
std::shared_ptr()) {
auto r_ctor_ = std::make_unique();
r_ctor_->track_id = track_id;
r_ctor_->packetizer = packetizer;
return r_ctor_;
}),
py::arg("track_id") = int{0},
py::arg("packetizer") = std::shared_ptr())
.def_readwrite("track_id", &espp::RtspServer::TrackConfig::track_id,
"/< Track identifier")
.def_readwrite("packetizer", &espp::RtspServer::TrackConfig::packetizer,
"/< Codec-specific packetizer");
} // end of inner classes & enums of RtspServer
pyClassRtspServer.def(py::init())
.def("set_session_log_level", &espp::RtspServer::set_session_log_level, py::arg("log_level"),
"/ @brief Sets the log level for the RTSP sessions created by this server\n/ @note This "
"does not affect the log level of the RTSP server itself\n/ @note This does not change "
"the log level of any sessions that have\n/ already been created\n/ @param "
"log_level The log level to set")
.def("start", &espp::RtspServer::start, py::arg("accept_timeout") = std::chrono::seconds(5),
"/ @brief Start the RTSP server\n/ Starts the accept task, session task, and binds the "
"RTSP socket\n/ @param accept_timeout The timeout for accepting new connections\n/ "
"@return True if the server was started successfully, False otherwise",
py::call_guard())
.def("stop", &espp::RtspServer::stop,
"/ @brief Stop the FTP server\n/ Stops the accept task, session task, and closes the "
"RTSP socket",
py::call_guard())
.def("add_track", &espp::RtspServer::add_track, py::arg("config"),
"/ @brief Register a media track with the server.\n/ Each track has its own packetizer, "
"SSRC, and sequence number.\n/ @param config Track configuration including the "
"packetizer.")
.def("has_active_sessions", &espp::RtspServer::has_active_sessions,
"/ @brief Returns True when at least one session is actively playing.\n/ @return True "
"if an active RTSP session is ready to receive RTP packets.")
.def("get_capture_cooldown", &espp::RtspServer::get_capture_cooldown,
"/ @brief Returns how long capture should wait before queueing another frame.\n/ "
"@return Remaining RTP backpressure cooldown, or zero if sending may resume.")
.def("get_recommended_capture_period", &espp::RtspServer::get_recommended_capture_period,
"/ @brief Returns the minimum recommended period between captured frames.\n/ @return "
"Recommended capture period based on recent RTP backpressure history.")
.def("send_frame",
py::overload_cast(&espp::RtspServer::send_frame),
py::arg("track_id"), py::arg("frame_data"),
"/ @brief Send a frame on a specific track.\n/ The track's packetizer splits the frame "
"into RTP payload chunks,\n/ which are then wrapped with RTP headers and queued for "
"delivery.\n/ @note Overwrites any existing pending packets for this track.\n/ @param "
"track_id The track to send on.\n/ @param frame_data Raw encoded frame data.",
py::call_guard())
.def("send_frame", py::overload_cast(&espp::RtspServer::send_frame),
py::arg("frame"),
"/ @brief Send a JPEG frame over the RTSP connection (backward compatible).\n/ If no "
"tracks have been added, lazily creates a default MJPEG track on\n/ track 0. Uses the "
"legacy RtpJpegPacket packetization to preserve the\n/ exact wire format for existing "
"MJPEG users.\n/ @note Overwrites any existing frame that has not been sent.\n/ @param "
"frame The frame to send.",
py::call_guard())
.def("send_frame", py::overload_cast(&espp::RtspServer::send_frame),
py::arg("frame_data"),
"/ @brief Send raw JPEG bytes over the default MJPEG track.\n/ Uses the legacy MJPEG "
"RTP packetization path without copying the frame\n/ into an intermediate JpegFrame "
"object.\n/ @note Overwrites any existing frame that has not been sent.\n/ @param "
"frame_data Complete JPEG bytes, including header and EOI marker.",
py::call_guard());
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassRtspSession = py::class_(
m, "RtspSession", py::dynamic_attr(),
"/ Class that reepresents an RTSP session, which is uniquely identified by a\n/ session id "
"and sends frame data over RTP and RTCP to the client");
{ // inner classes & enums of RtspSession
auto pyClassRtspSession_ClassTrack =
py::class_(pyClassRtspSession, "Track", py::dynamic_attr(),
"/ Represents one media track within an RTSP session")
.def_readwrite("track_id", &espp::RtspSession::Track::track_id,
"/< Track identifier (matches trackID=N in SDP)")
.def_readwrite("control_path", &espp::RtspSession::Track::control_path,
"/< Control path suffix (e.g., \"trackID=0\")")
.def_readonly("rtp_socket", &espp::RtspSession::Track::rtp_socket,
"/< RTP socket for this track")
.def_readonly("rtcp_socket", &espp::RtspSession::Track::rtcp_socket,
"/< RTCP socket for this track")
.def_readwrite("client_rtp_port", &espp::RtspSession::Track::client_rtp_port,
"/< Client's RTP port")
.def_readwrite("client_rtcp_port", &espp::RtspSession::Track::client_rtcp_port,
"/< Client's RTCP port")
.def_readwrite("setup_complete", &espp::RtspSession::Track::setup_complete,
"/< Whether SETUP has been completed for this track")
.def(py::init());
auto pyClassRtspSession_ClassConfig =
py::class_(pyClassRtspSession, "Config", py::dynamic_attr(),
"/ Configuration for the RTSP session")
.def(py::init(
[](std::string server_address = std::string(),
std::string rtsp_path = std::string(),
std::chrono::duration receive_timeout = std::chrono::seconds(5),
size_t control_task_stack_size_bytes =
espp::RtspSession::Config::default_control_task_stack_size_bytes,
espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN) {
auto r_ctor_ = std::make_unique();
r_ctor_->server_address = server_address;
r_ctor_->rtsp_path = rtsp_path;
r_ctor_->receive_timeout = receive_timeout;
r_ctor_->control_task_stack_size_bytes = control_task_stack_size_bytes;
r_ctor_->log_level = log_level;
return r_ctor_;
}),
py::arg("server_address") = std::string(), py::arg("rtsp_path") = std::string(),
py::arg("receive_timeout") = std::chrono::seconds(5),
py::arg("control_task_stack_size_bytes") =
espp::RtspSession::Config::default_control_task_stack_size_bytes,
py::arg("log_level") = espp::Logger::Verbosity::WARN)
.def_readwrite("server_address", &espp::RtspSession::Config::server_address,
"/< The address of the server")
.def_readwrite("rtsp_path", &espp::RtspSession::Config::rtsp_path,
"/< The RTSP path of the session")
.def_readwrite("receive_timeout", &espp::RtspSession::Config::receive_timeout,
"/< The timeout for receiving data. Should be > 0.")
.def_readwrite("control_task_stack_size_bytes",
&espp::RtspSession::Config::control_task_stack_size_bytes,
"/< RTSP control-task stack size, in bytes")
.def_readwrite("sdp_generator", &espp::RtspSession::Config::sdp_generator, "")
.def_readwrite("log_level", &espp::RtspSession::Config::log_level,
"/< The log level of the session");
} // end of inner classes & enums of RtspSession
pyClassRtspSession
.def(py::init())
.def("get_session_id", &espp::RtspSession::get_session_id,
"/ @brief Get the session id\n/ @return The session id")
.def("is_closed", &espp::RtspSession::is_closed,
"/ @brief Check if the session is closed\n/ @return True if the session is closed, "
"False otherwise")
.def("is_connected", &espp::RtspSession::is_connected,
"/ Get whether the session is connected\n/ @return True if the session is connected, "
"False otherwise")
.def("is_active", &espp::RtspSession::is_active,
"/ Get whether the session is active\n/ @return True if the session is active, False "
"otherwise")
.def("play", &espp::RtspSession::play,
"/ Mark the session as active\n/ This will cause the server to start sending frames to "
"the client")
.def("pause", &espp::RtspSession::pause,
"/ Pause the session\n/ This will cause the server to stop sending frames to the "
"client\n/ @note This does not stop the session, it just pauses it\n/ @note This is "
"useful for when the client is buffering")
.def("teardown", &espp::RtspSession::teardown,
"/ Teardown the session\n/ This will cause the server to stop sending frames to the "
"client\n/ and close the connection")
.def("send_rtp_packet",
py::overload_cast(&espp::RtspSession::send_rtp_packet),
py::arg("track_id"), py::arg("packet"),
"/ Send an RTP packet on a specific track\n/ @param track_id The track to send on\n/ "
"@param packet The RTP packet to send\n/ @return True if the packet was sent "
"successfully, False otherwise",
py::call_guard())
.def("send_rtp_packet",
py::overload_cast(&espp::RtspSession::send_rtp_packet),
py::arg("track_id"), py::arg("packet_data"),
"/ Send a serialized RTP packet on a specific track.\n/ @param track_id The track to "
"send on\n/ @param packet_data Serialized RTP packet bytes\n/ @return True if the "
"packet was sent successfully, False otherwise",
py::call_guard())
.def("send_rtp_packet",
py::overload_cast(&espp::RtspSession::send_rtp_packet),
py::arg("packet"),
"/ Send an RTP packet to the client (backward compat - sends on default track 0)\n/ "
"@param packet The RTP packet to send\n/ @return True if the packet was sent "
"successfully, False otherwise",
py::call_guard())
.def("send_rtp_packet",
py::overload_cast(&espp::RtspSession::send_rtp_packet),
py::arg("packet_data"),
"/ Send a serialized RTP packet to the client (default track 0).\n/ @param packet_data "
"Serialized RTP packet bytes\n/ @return True if the packet was sent successfully, False "
"otherwise",
py::call_guard())
.def("send_rtcp_packet",
py::overload_cast(&espp::RtspSession::send_rtcp_packet),
py::arg("track_id"), py::arg("packet"),
"/ Send an RTCP packet on a specific track\n/ @param track_id The track to send on\n/ "
"@param packet The RTCP packet to send\n/ @return True if the packet was sent "
"successfully, False otherwise",
py::call_guard())
.def("send_rtcp_packet",
py::overload_cast(&espp::RtspSession::send_rtcp_packet),
py::arg("packet"),
"/ Send an RTCP packet to the client (backward compat - sends on default track 0)\n/ "
"@param packet The RTCP packet to send\n/ @return True if the packet was sent "
"successfully, False otherwise",
py::call_guard());
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassLowpassFilter = py::class_(
m, "LowpassFilter", py::dynamic_attr(),
"*\n * @brief Lowpass infinite impulse response (IIR) filter.\n");
{ // inner classes & enums of LowpassFilter
auto pyClassLowpassFilter_ClassConfig =
py::class_(
pyClassLowpassFilter, "Config", py::dynamic_attr(),
"*\n * @brief Configuration for the lowpass filter.\n")
.def(py::init(
[](float normalized_cutoff_frequency = float(), float q_factor = float()) {
auto r_ctor_ = std::make_unique();
r_ctor_->normalized_cutoff_frequency = normalized_cutoff_frequency;
r_ctor_->q_factor = q_factor;
return r_ctor_;
}),
py::arg("normalized_cutoff_frequency") = float(), py::arg("q_factor") = float())
.def_readwrite(
"normalized_cutoff_frequency",
&espp::LowpassFilter::Config::normalized_cutoff_frequency,
"*< Filter cutoff frequency in the range [0.0, 0.5] (normalizd\n "
" to sample frequency, = 2 * f_cutoff / f_sample).")
.def_readwrite(
"q_factor", &espp::LowpassFilter::Config::q_factor,
"*< Quality (Q) factor of the filter. The higher the Q the better the filter.");
} // end of inner classes & enums of LowpassFilter
pyClassLowpassFilter.def(py::init())
.def("configure", &espp::LowpassFilter::configure, py::arg("config"),
"*\n * @brief Set the filter coefficients based on the config.\n * @param config "
"Configuration struct.\n")
.def("update",
py::overload_cast(&espp::LowpassFilter::update),
py::arg("input"), py::arg("output"), py::arg("length"),
"*\n * @brief Filter the input samples, updating internal state, and writing the\n "
"* filtered values to the data pointed to by output.\n * @param input Pointer "
"to (floating point) array of new samples of the input data\n * @param output Pointer "
"to (floating point) array which will be filled with\n * the filtered input.\n "
" * @param length Number of samples, should be >= length of input & output memory.\n "
"* @note On ESP32, the input and output arrays must have\n * "
"__attribute__((aligned(16))) to ensure proper alignment for the ESP32\n * DSP "
"functions.\n")
.def("update", py::overload_cast(&espp::LowpassFilter::update), py::arg("input"),
"*\n * @brief Filter the signal sampled by input, updating internal state, and\n * "
" returning the filtered output.\n * @param input New sample of the input "
"data.\n * @return Filtered output based on input and history.\n")
.def("__call__", &espp::LowpassFilter::operator(), py::arg("input"),
"*\n * @brief Filter the signal sampled by input, updating internal state, and\n * "
" returning the filtered output.\n * @param input New sample of the input "
"data.\n * @return Filtered output based on input and history.\n")
.def("reset", &espp::LowpassFilter::reset,
"*\n * @brief Reset the filter state to zero.\n");
//////////////////// ////////////////////
//////////////////// ////////////////////
auto pyClassSimpleLowpassFilter = py::class_(
m, "SimpleLowpassFilter", py::dynamic_attr(),
"*\n * @brief Simple lowpass filter using a time constant and a stored value.\n");
{ // inner classes & enums of SimpleLowpassFilter
auto pyClassSimpleLowpassFilter_ClassConfig =
py::class_(
pyClassSimpleLowpassFilter, "Config", py::dynamic_attr(),
"*\n * @brief Configuration for the lowpass filter.\n")
.def(py::init([](float time_constant = 0.0f) {
auto r_ctor_ = std::make_unique();
r_ctor_->time_constant = time_constant;
return r_ctor_;
}),
py::arg("time_constant") = 0.0f)
.def_readwrite("time_constant", &espp::SimpleLowpassFilter::Config::time_constant,
"*< Time constant of the filter.");
} // end of inner classes & enums of SimpleLowpassFilter
pyClassSimpleLowpassFilter.def(py::init())
.def("set_time_constant", &espp::SimpleLowpassFilter::set_time_constant,
py::arg("time_constant"),
"*\n * @brief Set the time constant of the filter.\n * @param time_constant Time "
"constant of the filter.\n")
.def("get_time_constant", &espp::SimpleLowpassFilter::get_time_constant,
"*\n * @brief Get the time constant of the filter.\n * @return Time constant of the "
"filter.\n")
.def("update", &espp::SimpleLowpassFilter::update, py::arg("input"),
"*\n * @brief Filter the signal sampled by input, updating internal state, and\n * "
" returning the filtered output.\n * @param input New sample of the input "
"data.\n * @return Filtered output based on input, time, and history.\n")
.def("__call__", &espp::SimpleLowpassFilter::operator(), py::arg("input"),
"*\n * @brief Filter the signal sampled by input, updating internal state, and\n * "
" returning the filtered output.\n * @param input New sample of the input "
"data.\n * @return Filtered output based on input, time, and history.\n")
.def("reset", &espp::SimpleLowpassFilter::reset,
"*\n * @brief Reset the filter to its initial state.\n");
//////////////////// ////////////////////
// // Autogenerated code end
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! AUTOGENERATED CODE END !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
}