#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, std::span>( &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, std::span>( &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 no\nrate limit. @note Only " "calls that have _rate_limited suffixed will be rate limited.") .def_readwrite("level", &espp::Logger::Config::level, "*< The verbosity level for the logger."); } // end of inner classes & enums of Logger pyClassLogger.def(py::init()) .def("get_verbosity", &espp::Logger::get_verbosity, "*\n * @brief Get the current verbosity for the logger.\n * @return The current " "verbosity level.\n * \\sa Logger::Verbosity\n *\n") .def("set_verbosity", &espp::Logger::set_verbosity, py::arg("level"), "*\n * @brief Change the verbosity for the logger. \\sa Logger::Verbosity\n * " "@param level new verbosity level\n") .def("set_tag", &espp::Logger::set_tag, py::arg("tag"), "*\n * @brief Change the tag for the logger.\n * @param tag The new tag.\n") .def("get_tag", &espp::Logger::get_tag, "*\n * @brief Get the current tag for the logger.\n * @return A const reference to " "the current tag.\n") .def("set_include_time", &espp::Logger::set_include_time, py::arg("include_time"), "*\n * @brief Whether to include the time in the log.\n * @param include_time " "Whether to include the time in the log.\n * @note The time is in seconds since boot " "and is represented as a floating\n * point number with precision to the " "millisecond.\n") .def("set_rate_limit", &espp::Logger::set_rate_limit, py::arg("rate_limit"), "*\n * @brief Change the rate limit for the logger.\n * @param rate_limit The new " "rate limit.\n * @note Only calls that have _rate_limited suffixed will be rate " "limited.\n") .def("get_rate_limit", &espp::Logger::get_rate_limit, "*\n * @brief Get the current rate limit for the logger.\n * @return The current " "rate limit.\n") .def_static( "get_time", &espp::Logger::get_time, "*\n * Get the current time in seconds since the start of the logging system.\n * " " @return time in seconds since the start of the logging system.\n"); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassBezier_espp_Vector2f = py::class_>( m, "Bezier_espp_Vector2f", py::dynamic_attr(), "*\n * @brief Implements rational / weighted and unweighted cubic bezier curves\n * " "between control points.\n * @note See https://pomax.github.io/bezierinfo/ for information " "on bezier\n * curves.\n * @note Template class which can be used individually on " "floating point\n * values directly or on containers such as Vector2d.\n * " "@tparam T The type of the control points, e.g. float or Vector2d.\n * @note The " "bezier curve is defined by 4 control points, P0, P1, P2, P3.\n * The curve is defined " "by the equation:\n * \\f$B(t) = (1-t)^3 * P0 + 3 * (1-t)^2 * t * P1 + 3 * (1-t) * t^2 " "* P2 + t^3 * P3\\f$\n * where t is the evaluation parameter, [0, 1].\n *\n * @note The " "weighted bezier curve is defined by 4 control points, P0, P1, P2, P3\n * and 4 " "weights, W0, W1, W2, W3.\n * The curve is defined by the equation:\n * \\f$B(t) = " "(W0 * (1-t)^3 * P0 + W1 * 3 * (1-t)^2 * t * P1 + W2 * 3 * (1-t) * t^2 * P2 + W3 *\n * t^3 * " "P3) / (W0 + W1 + W2 + W3)\\f$ where t is the evaluation parameter, [0, 1].\n *\n * " "\\section bezier_ex1 Example\n * \\snippet math_example.cpp bezier example\n"); { // inner classes & enums of Bezier_espp_Vector2f auto pyClassBezier_ClassConfig = py::class_::Config>( pyClassBezier_espp_Vector2f, "Config", py::dynamic_attr(), "*\n * @brief Unweighted cubic bezier configuration for 4 control points.\n") .def(py::init<>()) // implicit default constructor .def_readwrite("control_points", &espp::Bezier::Config::control_points, "/< Array of 4 control points"); auto pyClassBezier_ClassWeightedConfig = py::class_::WeightedConfig>( pyClassBezier_espp_Vector2f, "WeightedConfig", py::dynamic_attr(), "*\n * @brief Weighted cubic bezier configuration for 4 control points with\n * " " individual weights.\n") .def(py::init<>()) // implicit default constructor .def_readwrite("control_points", &espp::Bezier::WeightedConfig::control_points, "/< Array of 4 control points") .def_readwrite("weights", &espp::Bezier::WeightedConfig::weights, "/< Array of 4 weights, default is array of 1.0"); } // end of inner classes & enums of Bezier_espp_Vector2f pyClassBezier_espp_Vector2f.def(py::init::Config &>()) .def(py::init::WeightedConfig &>()) .def("__call__", &espp::Bezier::operator(), py::arg("t"), "*\n * @brief Evaluate the bezier at \\p t.\n * @note Convienience wrapper around " "the at() method.\n * @param t The evaluation parameter, [0, 1].\n * @return The " "bezier evaluated at \\p t.\n"); //////////////////// //////////////////// //////////////////// //////////////////// m.def("deg_to_rad", espp::deg_to_rad, py::arg("degrees"), "*\n * @brief Convert degrees to radians\n * @param degrees Angle in degrees\n * @return " "Angle in radians\n"); m.def("rad_to_deg", espp::rad_to_deg, py::arg("radians"), "*\n * @brief Convert radians to degrees\n * @param radians Angle in radians\n * @return " "Angle in degrees\n"); m.def("square", espp::square, py::arg("f"), "*\n * @brief Simple square of the input.\n * @param f Value to square.\n * @return The " "square of f (f*f).\n"); m.def("cube", espp::cube, py::arg("f"), "*\n * @brief Simple cube of the input.\n * @param f Value to cube.\n * @return The cube " "of f (f*f*f).\n"); m.def("fast_inv_sqrt", espp::fast_inv_sqrt, py::arg("value"), "*\n * @brief Fast inverse square root approximation.\n * @note Using " "https://reprap.org/forum/read.php?147,219210 and\n * " "https://en.wikipedia.org/wiki/Fast_inverse_square_root\n * @param value Value to take the " "inverse square root of.\n * @return Approximation of the inverse square root of value.\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("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_::Config>( 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::Config>(); 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_::Config>( 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::Config>(); 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) <= 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__", [](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( "__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) <= 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__", [](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( "__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 &, const espp::TcpSocket::TransmitConfig &>( &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 &, const espp::TcpSocket::TransmitConfig &>( &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, const espp::TcpSocket::TransmitConfig &>( &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 &, size_t>(&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 &, const espp::UdpSocket::SendConfig &>( &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, const espp::UdpSocket::SendConfig &>( &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 on\n " " output to prevent infeasible combined commands.") .def_readwrite( "max_centripetal_acceleration", &espp::TrajectoryPlanner::Config::max_centripetal_acceleration, "*< Maximum centripetal acceleration |v*w| (m/s^2).\n " " 0 disables the limit. Both v and w are scaled\n " " proportionally when the limit is exceeded.") .def_readwrite( "output_callback", &espp::TrajectoryPlanner::Config::output_callback, "/**< Optional callback invoked after each update()\nwith the latest MotionCommand " "output.\nLeave as None to disable. */\n --- Periodic task configuration ---") .def_readwrite("planning_period", &espp::TrajectoryPlanner::Config::planning_period, "*< Planner\n " "update rate (default 50 Hz). Recommended: 5-200\n " " ms on microcontrollers.") .def_readwrite("planning_task_config", &espp::TrajectoryPlanner::Config::planning_task_config, "*< Underlying task config for the timer.") .def_readwrite("callback_task_config", &espp::TrajectoryPlanner::Config::callback_task_config, "*< Underlying task config for the callback timer.") .def_readwrite("log_level", &espp::TrajectoryPlanner::Config::log_level, "*< Logger verbosity."); } // end of inner classes & enums of TrajectoryPlanner pyClassTrajectoryPlanner.def(py::init()) .def("set_config", &espp::TrajectoryPlanner::set_config, py::arg("config"), py::arg("reset_state") = true, "*\n * @brief Update the planner configuration.\n * @param config New configuration " "parameters.\n * @param reset_state If True (default), resets velocity/acceleration " "state to zero.\n * @return True if the configuration was successfully applied.\n") .def("get_config", &espp::TrajectoryPlanner::get_config, "*\n * @brief Get the current configuration.\n * @return Const reference to the " "active Config.\n") .def("set_target", &espp::TrajectoryPlanner::set_target, py::arg("linear"), py::arg("angular"), "*\n * @brief Set the desired chassis velocity target using normalized joystick " "inputs.\n *\n * Both inputs are in the range [-1, +1] and are scaled internally:\n " " * @code\n * v_target = linear * max_linear_velocity\n * w_target = " "angular * max_angular_velocity\n * @endcode\n * Values outside [-1, +1] are " "clamped before scaling.\n *\n * @param linear Normalized linear velocity command " "[-1, +1].\n * +1 = full forward, -1 = full reverse.\n * @param " "angular Normalized angular velocity command [-1, +1].\n * +1 = full " "left turn, -1 = full right turn.\n") .def("get_target", &espp::TrajectoryPlanner::get_target, "*\n * @brief Get the current normalized velocity target.\n * @note If " "enforce_motion_envelope is enabled the stored target may differ\n * from the " "value passed to set_target() (it is projected onto the unit circle).\n * @return " "Pair of {linear, angular} in [-1, +1].\n") .def("output", &espp::TrajectoryPlanner::output, "*\n * @brief Get the current smoothed motion command.\n * @return MotionCommand " "containing the trajectory-limited (v_ref, w_ref).\n") .def("stop", &espp::TrajectoryPlanner::stop, "*\n * @brief Command the planner to decelerate to a full stop.\n *\n * " "Equivalent to set_target(0, 0). The planner ramps down respecting all\n * configured " "limits rather than cutting output immediately.\n") .def("reset", &espp::TrajectoryPlanner::reset, "*\n * @brief Reset velocity, acceleration state, and target to zero immediately.\n " "*\n * The next call to output() will return (0, 0). Use after an emergency stop\n " "* or before re-initialising with a new configuration.\n") .def("is_running", &espp::TrajectoryPlanner::is_running, "*\n * @brief Check whether the periodic update task is currently running.\n * " "@return True if the task is running.\n"); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassThreadPool = py::class_( m, "ThreadPool", py::dynamic_attr(), "*\n * @brief A thread pool that dispatches submitted jobs to a fixed set of worker " "threads.\n *\n * Workers are implemented as espp::Task instances. Jobs are queued and\n * " "consumed in FIFO order. The queue can be optionally bounded; when full,\n * new submissions " "are either rejected immediately or blocked until space\n * becomes available, depending on " "the configuration.\n *\n * \\section thread_pool_ex1 Lifecycle: start / stop / is_running / " "worker_count\n * \\snippet thread_pool_example.cpp lifecycle example\n * \\section " "thread_pool_ex2 Submit Jobs\n * \\snippet thread_pool_example.cpp submit example\n * " "\\section thread_pool_ex3 try_submit - Non-Blocking Rejection When Full\n * \\snippet " "thread_pool_example.cpp try_submit example\n * \\section thread_pool_ex4 Blocking Submit " "When Full\n * \\snippet thread_pool_example.cpp blocking submit example\n * \\section " "thread_pool_ex5 Submit Rejected After stop()\n * \\snippet thread_pool_example.cpp submit " "after stop example\n * \\section thread_pool_ex6 Concurrent start / stop\n * \\snippet " "thread_pool_example.cpp concurrent lifecycle example\n * \\section thread_pool_ex7 " "Concurrent submit and try_submit\n * \\snippet thread_pool_example.cpp concurrent submit " "example\n * \\section thread_pool_ex8 Chained Pools\n * \\snippet thread_pool_example.cpp " "chained pools example\n * \\section thread_pool_ex9 Self-Submit\n * \\snippet " "thread_pool_example.cpp self-submit example\n"); { // inner classes & enums of ThreadPool auto pyClassThreadPool_ClassStats = py::class_(pyClassThreadPool, "Stats", py::dynamic_attr(), "/ @brief Snapshot of pool activity counters.") .def(py::init<>([](std::uint64_t submitted = 0, std::uint64_t executed = 0, std::uint64_t rejected = 0) { auto r_ctor_ = std::make_unique(); r_ctor_->submitted = submitted; r_ctor_->executed = executed; r_ctor_->rejected = rejected; return r_ctor_; }), py::arg("submitted") = 0, py::arg("executed") = 0, py::arg("rejected") = 0) .def_readwrite("submitted", &espp::ThreadPool::Stats::submitted, "/< Total jobs accepted into the queue.") .def_readwrite("executed", &espp::ThreadPool::Stats::executed, "/< Total jobs successfully executed.") .def_readwrite("rejected", &espp::ThreadPool::Stats::rejected, "/< Total jobs rejected (invalid job, stopped/stopping, or queue") .def_readwrite("band_submitted", &espp::ThreadPool::Stats::band_submitted, "/< Jobs accepted per band (by the band they were submitted to).") .def_readwrite("band_executed", &espp::ThreadPool::Stats::band_executed, "/< Jobs executed per band (by the band they were popped from, i.e. " "after any aging promotions).") .def_readwrite("band_aged", &espp::ThreadPool::Stats::band_aged, "/< Aging promotions OUT of each band (an entry moved from band i to " "band i-1).") .def_readwrite("band_rejected", &espp::ThreadPool::Stats::band_rejected, "/< Jobs rejected per band (by the band they were submitted to)."); auto pyClassThreadPool_ClassConfig = py::class_( pyClassThreadPool, "Config", py::dynamic_attr(), "/ @brief Configuration parameters for constructing a ThreadPool.") .def(py::init<>( [](std::size_t worker_count = 1, std::size_t max_queue_size = 0, bool auto_start = true, bool block_on_submit_when_full = false, espp::Task::BaseConfig worker_task_config = { ///< Base configuration applied to every worker task. .name = "thread_pool_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1, }, espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN, std::chrono::milliseconds aging_threshold = std::chrono::milliseconds{100}, std::array band_worker_counts = {}, std::array band_task_priorities = {10, 7, 5, 1}, bool band_workers_realtime = false) { auto r_ctor_ = std::make_unique(); r_ctor_->worker_count = worker_count; r_ctor_->max_queue_size = max_queue_size; r_ctor_->auto_start = auto_start; r_ctor_->block_on_submit_when_full = block_on_submit_when_full; r_ctor_->worker_task_config = worker_task_config; r_ctor_->log_level = log_level; r_ctor_->aging_threshold = aging_threshold; r_ctor_->band_worker_counts = band_worker_counts; r_ctor_->band_task_priorities = band_task_priorities; r_ctor_->band_workers_realtime = band_workers_realtime; return r_ctor_; }), py::arg("worker_count") = 1, py::arg("max_queue_size") = 0, py::arg("auto_start") = true, py::arg("block_on_submit_when_full") = false, py::arg("worker_task_config") = espp::Task::BaseConfig{ ///< Base configuration applied to every worker task. .name = "thread_pool_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1, }, py::arg("log_level") = espp::Logger::Verbosity::WARN, py::arg("aging_threshold") = std::chrono::milliseconds{100}, py::arg("band_worker_counts") = std::array{}, py::arg("band_task_priorities") = std::array{10, 7, 5, 1}, py::arg("band_workers_realtime") = false) .def_readwrite("worker_count", &espp::ThreadPool::Config::worker_count, "/< Number of worker threads to spawn.") .def_readwrite("max_queue_size", &espp::ThreadPool::Config::max_queue_size, "/< Maximum pending jobs (0 = unbounded).") .def_readwrite("auto_start", &espp::ThreadPool::Config::auto_start, "/< Start workers immediately on construction.") .def_readwrite( "block_on_submit_when_full", &espp::ThreadPool::Config::block_on_submit_when_full, "/< If True, submit() blocks when the queue is full instead of rejecting.") .def_readwrite("worker_task_config", &espp::ThreadPool::Config::worker_task_config, "") .def_readwrite("log_level", &espp::ThreadPool::Config::log_level, "/< Logger verbosity level.") .def_readwrite("aging_threshold", &espp::ThreadPool::Config::aging_threshold, "/< Starvation guard: a queued job whose wait exceeds this is promoted " "up one band. 0 disables aging (strict band priority).") .def_readwrite("band_worker_counts", &espp::ThreadPool::Config::band_worker_counts, "/< Opt-in per-band worker counts (index = QosBand); all zero = " "disabled (identical workers service all bands).") .def_readwrite("band_task_priorities", &espp::ThreadPool::Config::band_task_priorities, "/< Task priorities for per-band workers (only used when " "band_worker_counts is set).") .def_readwrite("band_workers_realtime", &espp::ThreadPool::Config::band_workers_realtime, "/< Opt-in for OS real-time scheduling of per-band workers on host " "platforms (SCHED_FIFO; see Task.BaseConfig.host_realtime)."); } // end of inner classes & enums of ThreadPool pyClassThreadPool.def(py::init()) .def("start", &espp::ThreadPool::start, "/ @brief Start all worker threads.\n/ @return True if all workers were successfully " "started, False otherwise.\n/ @note No-op if the pool is already running and return " "True immediately.\n/ @note If any workers could not be started, the pool will roll " "back to the stopped state.", py::call_guard()) .def("stop", &espp::ThreadPool::stop, "/ @brief Stop all worker threads and reject further submissions.\n/ @note Blocks until " "every worker has exited; queued jobs may not be executed.", py::call_guard()) .def("is_running", &espp::ThreadPool::is_running, "/ @brief Query whether the pool is currently running.\n/ @return True if workers are " "active, False otherwise.") .def("submit", static_cast( &espp::ThreadPool::submit), py::arg("job"), "/ @brief Submit a job at QosBand.Normal, optionally blocking when the queue is " "full.\n/\n/ Blocks if Config::block_on_submit_when_full is True and the queue has\n/ " "reached its capacity limit. Otherwise behaves identically to try_submit().\n/ @param " "job Callable to enqueue; moved into the queue on acceptance.\n/ @return True if the " "job was accepted, False if it was rejected.", py::call_guard()) .def("submit", static_cast( &espp::ThreadPool::submit), py::arg("job"), py::arg("band"), "/ @brief Submit a job at the given priority band, optionally blocking when the queue " "is full.\n/ @param job Callable to enqueue; moved into the queue on acceptance.\n/ " "@param band Priority band to enqueue the job at.\n/ @return True if the job was " "accepted, False if it was rejected.", py::call_guard()) .def("try_submit", static_cast( &espp::ThreadPool::try_submit), py::arg("job"), "/ @brief Attempt to submit a job at QosBand.Normal without blocking.\n/\n/ Returns " "immediately with False when the queue is full.\n/ @param job Callable to enqueue; " "moved into the queue on acceptance.\n/ @return True if the job was accepted, False if " "it was rejected.") .def("try_submit", static_cast( &espp::ThreadPool::try_submit), py::arg("job"), py::arg("band"), "/ @brief Attempt to submit a job at the given priority band without blocking.\n/ " "@param job Callable to enqueue; moved into the queue on acceptance.\n/ @param band " "Priority band to enqueue the job at.\n/ @return True if the job was accepted, False " "if it was rejected.") .def("queue_size", &espp::ThreadPool::queue_size, "/ @brief Return the number of jobs currently waiting in the queue.\n/ @return Pending " "job count.") .def("worker_count", &espp::ThreadPool::worker_count, "/ @brief Return the number of worker threads in the pool.\n/ @return Worker thread " "count.") .def("stats", &espp::ThreadPool::stats, "/ @brief Return a snapshot of the pool's activity counters.\n/ @return Stats struct " "with submitted, executed, and rejected counts."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassJoystick = py::class_( m, "Joystick", py::dynamic_attr(), "*\n * @brief 2-axis Joystick with axis mapping / calibration.\n *\n * \\section " "joystick_ex1 Basic Circular and Rectangular Joystick Example\n * \\snippet " "joystick_example.cpp circular joystick example\n * \\section joystick_ex2 ADC Joystick " "Example\n * \\snippet joystick_example.cpp adc joystick example\n"); { // inner classes & enums of Joystick auto pyEnumType = py::enum_( pyClassJoystick, "Type", py::arithmetic(), "*\n * @brief Type of the joystick.\n * @note When using a " "Type::CIRCULAR joystick, it's recommended to set the\n * " "individual x/y calibration deadzones to be 0 and to only use the\n * " " deadzone_radius field to set the deadzone around the center.\n") .value("rectangular", espp::Joystick::Type::RECTANGULAR, "/< The default type of joystick. Uses the rangemappers for") .value("circular", espp::Joystick::Type::CIRCULAR, "/< The joystick is configured to have a circular output. This"); auto pyClassJoystick_ClassConfig = py::class_( pyClassJoystick, "Config", py::dynamic_attr(), "*\n * @brief Configuration structure for the joystick.\n") .def( py::init<>([](espp::FloatRangeMapper::Config x_calibration = espp::FloatRangeMapper::Config(), espp::FloatRangeMapper::Config y_calibration = espp::FloatRangeMapper::Config(), espp::Joystick::Type type = {espp::Joystick::Type::RECTANGULAR}, float center_deadzone_radius = {0}, float range_deadzone = {0}, espp::Joystick::get_values_fn get_values = {nullptr}, espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}) { auto r_ctor_ = std::make_unique(); r_ctor_->x_calibration = x_calibration; r_ctor_->y_calibration = y_calibration; r_ctor_->type = type; r_ctor_->center_deadzone_radius = center_deadzone_radius; r_ctor_->range_deadzone = range_deadzone; r_ctor_->get_values = get_values; r_ctor_->log_level = log_level; return r_ctor_; }), py::arg("x_calibration") = espp::FloatRangeMapper::Config(), py::arg("y_calibration") = espp::FloatRangeMapper::Config(), py::arg("type") = espp::Joystick::Type{espp::Joystick::Type::RECTANGULAR}, py::arg("center_deadzone_radius") = float{0}, py::arg("range_deadzone") = float{0}, py::arg("get_values") = espp::Joystick::get_values_fn{nullptr}, py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN}) .def_readwrite("x_calibration", &espp::Joystick::Config::x_calibration, "*< Configuration for the x axis.") .def_readwrite("y_calibration", &espp::Joystick::Config::y_calibration, "*< Configuration for the y axis.") .def_readwrite("type", &espp::Joystick::Config::type, "*< The type of the joystick. See\n " " Type enum for more information.") .def_readwrite( "center_deadzone_radius", &espp::Joystick::Config::center_deadzone_radius, "*< The radius of the unit circle's deadzone [0, 1.0] around the center, only " "used\n when the joystick is configured as Type::CIRCULAR.") .def_readwrite( "range_deadzone", &espp::Joystick::Config::range_deadzone, "*< The deadzone around the edge of the unit circle, only used when\n " " the joystick is configured as Type::CIRCULAR. This scales the output " "so\n that the output appears to have magnitude 1 " "(meaning it appears to be on\n the edge of the unit " "circle) when the joystick value magnitude is within\n " "the range [1-range_deadzone, 1].") .def_readwrite("get_values", &espp::Joystick::Config::get_values, "*< Function to retrieve the latest\n " " unmapped joystick values. Required if\n " " you want to use update(), unused if\n " " you call update(float raw_x, float\n " " raw_y).") .def_readwrite("log_level", &espp::Joystick::Config::log_level, "*< Verbosity for the Joystick logger_."); } // end of inner classes & enums of Joystick pyClassJoystick.def(py::init()) .def( "set_type", &espp::Joystick::set_type, py::arg("type"), py::arg("radius") = 0, py::arg("range_deadzone") = 0, "*\n * @brief Set the type of the joystick.\n * @param type The Type of the " "joystick.\n * @param radius Optional radius parameter used when \\p type is\n * " " Type::CIRCULAR. When the magnitude of the joystick's mapped\n * position " "vector is less than this value, the vector is set to\n * (0,0).\n * @param " "range_deadzone Optional deadzone around the edge of the unit circle\n * when " "\\p type is Type::CIRCULAR. This scales the output so that the\n * output " "appears to have magnitude 1 (meaning it appears to be on the\n * edge of the " "unit circle) if the magnitude of the mapped position\n * vector is greater " "than 1-range_deadzone. Example: if the range\n * deadzone is 0.1, then the " "output will be scaled so that the\n * magnitude of the output is 1 if the " "magnitude of the mapped\n * position vector is greater than 0.9.\n * @note " "If the Joystick is Type::CIRCULAR, the actual calibrations that are\n * saved " "into the joystick will have 0 deadzone around the center value\n * and range " "values, so that center and range deadzones are actually\n * applied on the " "vector value instead of on the individual axes\n * independently.\n * @sa " "set_center_deadzone_radius\n * @sa set_range_deadzone\n * @sa set_calibration\n") .def("type", &espp::Joystick::type, "*\n * @brief Get the type of the joystick.\n * @return The Type of the joystick.\n") .def("set_center_deadzone_radius", &espp::Joystick::set_center_deadzone_radius, py::arg("radius"), "*\n * @brief Sets the center deadzone radius.\n * @note Radius is only applied " "when \\p deadzone is Deadzone::CIRCULAR.\n * @param radius Optional radius parameter " "used when \\p deadzone is\n * Deadzone::CIRCULAR. When the magnitude of the " "joystick's mapped\n * position vector is less than this value, the vector is " "set to\n * (0,0).\n") .def("center_deadzone_radius", &espp::Joystick::center_deadzone_radius, "*\n * @brief Get the center deadzone radius.\n * @return The center deadzone " "radius.\n") .def("set_range_deadzone", &espp::Joystick::set_range_deadzone, py::arg("range_deadzone"), "*\n * @brief Sets the range deadzone.\n * @note Range deadzone is only applied " "when \\p deadzone is Deadzone::CIRCULAR.\n * @param range_deadzone Optional deadzone " "around the edge of the unit circle\n * when \\p deadzone is " "Deadzone::CIRCULAR. This scales the output so\n * that the output appears to " "have magnitude 1 (meaning it appears to\n * be on the edge of the unit " "circle) if the magnitude of the mapped\n * position vector is greater than " "1-range_deadzone. Example: if the\n * range deadzone is 0.1, then the output " "will be scaled so that the\n * magnitude of the output is 1 if the magnitude " "of the mapped position\n * vector is greater than 0.9.\n") .def("range_deadzone", &espp::Joystick::range_deadzone, "*\n * @brief Get the range deadzone.\n * @return The range deadzone.\n") .def("set_calibration", &espp::Joystick::set_calibration, py::arg("x_calibration"), py::arg("y_calibration"), py::arg("center_deadzone_radius") = 0, py::arg("range_deadzone") = 0, "*\n * @brief Update the x and y axis mapping.\n * @param x_calibration New x-axis " "range mapping configuration to use.\n * @param y_calibration New y-axis range " "mapping configuration to use.\n * @param center_deadzone_radius The radius of the " "unit circle's deadzone [0,\n * 1.0] around the center, only used when the " "joystick is configured\n * as Type::CIRCULAR.\n * @param range_deadzone " "Optional deadzone around the edge of the unit circle\n * when \\p type is " "Type::CIRCULAR. This scales the output so that the\n * output appears to " "have magnitude 1 (meaning it appears to be on the\n * edge of the unit " "circle) if the magnitude of the mapped position\n * vector is greater than " "1-range_deadzone. Example: if the range\n * deadzone is 0.1, then the output " "will be scaled so that the\n * magnitude of the output is 1 if the magnitude " "of the mapped\n * position vector is greater than 0.9.\n * @note If the " "Joystick is Type::CIRCULAR, the actual calibrations that are\n * saved into " "the joystick will have 0 deadzone around the center and range values,\n * so " "that center and range deadzones are actually applied on the vector value.\n * @sa " "set_center_deadzone_radius\n * @sa set_range_deadzone\n") .def( "update", [](espp::Joystick &self) { return self.update(); }, "*\n * @brief Read the raw values and use the calibration data to update the\n * " " position.\n * @note Requires that the get_values_ function is set.\n") .def("update", py::overload_cast(&espp::Joystick::update), py::arg("raw_x"), py::arg("raw_y"), "*\n * @brief Update the joystick's position using the provided raw x and y\n * " " values.\n * @param raw_x The raw x-axis value.\n * @param raw_y The raw y-axis " "value.\n * @note This function is useful when you have the raw values and don't " "want\n * to use the get_values_ function.\n") .def("x", &espp::Joystick::x, "*\n * @brief Get the most recently updated x axis calibrated position.\n * @return " "The most recent x-axis position (from when update() was last\n * called).\n") .def("y", &espp::Joystick::y, "*\n * @brief Get the most recently updated y axis calibrated position.\n * @return " "The most recent y-axis position (from when update() was last\n * called).\n") .def("position", &espp::Joystick::position, "*\n * @brief Get the most recently updated calibrated position.\n * @return The " "most recent position (from when update() was last called).\n * @note The returned " "reference is valid as long as the Joystick object is alive.\n") .def("raw", &espp::Joystick::raw, "*\n * @brief Get the most recently updated raw / uncalibrated readings. This\n * " " function is useful for externally performing a calibration routine\n * " "and creating updated calibration / mapper configuration\n * structures.\n * " "@return The most recent raw measurements (from when update() was last\n * " "called).\n * @note The returned reference is valid as long as the Joystick object is " "alive.\n"); //////////////////// //////////////////// //////////////////// //////////////////// auto pyEnumMediaType = py::enum_(m, "MediaType", py::arithmetic(), "/ Describes a media type for RTSP tracks.") .value("video", espp::MediaType::VIDEO, "/< Video media (MJPEG, H264, etc.)") .value("audio", espp::MediaType::AUDIO, "/< Audio media (PCM, Opus, AAC, etc.)"); auto pyClassRtpPayloadChunk = py::class_( m, "RtpPayloadChunk", py::dynamic_attr(), "/ Represents one RTP payload chunk ready to be wrapped in an RtpPacket.\n/ Packetizers " "produce these; the server wraps them with RTP headers.") .def(py::init<>( [](std::vector data = std::vector(), bool marker = {false}) { auto r_ctor_ = std::make_unique(); r_ctor_->data = data; r_ctor_->marker = marker; return r_ctor_; }), py::arg("data") = std::vector(), py::arg("marker") = bool{false}) .def_readwrite("data", &espp::RtpPayloadChunk::data, "/< The payload data for this chunk") .def_readwrite("marker", &espp::RtpPayloadChunk::marker, "/< Set on last chunk of a frame/access unit"); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassRtpDepacketizer = py::class_>( m, "RtpDepacketizer", py::dynamic_attr(), "/ Abstract base class for reassembling media frames from incoming RTP packets.\n/ Concrete " "depacketizers (e.g. MJPEG, H.264) override process_packet() to\n/ accumulate payload data " "and invoke the frame callback when a complete frame\n/ has been assembled."); { // inner classes & enums of RtpDepacketizer auto pyClassRtpDepacketizer_ClassConfig = py::class_(pyClassRtpDepacketizer, "Config", py::dynamic_attr(), "/ Configuration for RtpDepacketizer.") .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::RtpDepacketizer::Config::log_level, "/< Log verbosity level"); } // end of inner classes & enums of RtpDepacketizer pyClassRtpDepacketizer .def( "process_packet", &espp::RtpDepacketizer::process_packet, py::arg("packet"), "/ Process an incoming RTP packet, accumulating payload data.\n/ When a complete frame " "is assembled the frame callback is invoked.\n/ @param packet The RTP packet to process.") .def("set_frame_callback", &espp::RtpDepacketizer::set_frame_callback, py::arg("cb"), "/ Set the callback for completed frames.\n/ @param cb The callback to invoke when a " "full frame is ready."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassRtpPacketizer = py::class_>( m, "RtpPacketizer", py::dynamic_attr(), "/ Abstract base class for splitting media frames into RTP payload chunks.\n/ Concrete " "packetizers (e.g. MJPEG, H.264) override the pure-virtual methods\n/ to produce " "codec-specific payloads. The RTSP server wraps each returned\n/ RtpPayloadChunk with an RTP " "header before sending."); { // inner classes & enums of RtpPacketizer auto pyClassRtpPacketizer_ClassConfig = py::class_(pyClassRtpPacketizer, "Config", py::dynamic_attr(), "/ Configuration for RtpPacketizer.") .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::RtpPacketizer::Config::max_payload_size, "/< Maximum payload bytes per RTP packet") .def_readwrite("log_level", &espp::RtpPacketizer::Config::log_level, "/< Log verbosity level"); } // end of inner classes & enums of RtpPacketizer pyClassRtpPacketizer .def("packetize", &espp::RtpPacketizer::packetize, py::arg("frame_data"), "/ Packetize a complete media frame into RTP payload chunks.\n/ @param frame_data The " "raw frame bytes to packetize.\n/ @return A vector of RtpPayloadChunk ready to be " "wrapped in RTP packets.") .def("get_payload_type", &espp::RtpPacketizer::get_payload_type, "/ Get the RTP payload type number for this codec.\n/ @return The RTP payload type " "(e.g. 26 for MJPEG, 96 for dynamic).") .def("get_clock_rate", &espp::RtpPacketizer::get_clock_rate, "/ Get the RTP clock rate for timestamp calculation.\n/ @return The clock rate in Hz " "(e.g. 90000 for video, 8000 for audio).") .def("get_sdp_media_attributes", &espp::RtpPacketizer::get_sdp_media_attributes, "/ Generate the SDP media-level attributes for this codec.\n/ @return A string " "containing SDP a= lines (without trailing CRLF).") .def("get_sdp_media_line", &espp::RtpPacketizer::get_sdp_media_line, "/ Generate the SDP m= line for this codec.\n/ @return A string containing the SDP m= " "line (without trailing CRLF)."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassGenericDepacketizer = py::class_>( m, "GenericDepacketizer", py::dynamic_attr(), "/ A generic RTP depacketizer that reassembles media frames from incoming RTP\n/ packets. It " "accumulates payload data until a packet with the marker bit set\n/ is received, then " "delivers the complete frame via the frame callback. If a\n/ packet arrives with a different " "RTP timestamp than the current accumulation\n/ buffer, the old buffer is discarded and a " "new one is started.\n/\n/ This is suitable for audio codecs (PCM, G.711, Opus, etc.) or any " "payload\n/ format that uses simple marker-based framing.\n/\n/ \\section " "generic_depacketizer_ex1 Example\n/ \\snippet rtsp_example.cpp generic_depacketizer_test"); { // inner classes & enums of GenericDepacketizer auto pyClassGenericDepacketizer_ClassConfig = py::class_(pyClassGenericDepacketizer, "Config", py::dynamic_attr(), "/ Configuration for GenericDepacketizer.") .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::GenericDepacketizer::Config::log_level, "/< Log verbosity level"); } // end of inner classes & enums of GenericDepacketizer pyClassGenericDepacketizer.def(py::init()) .def("process_packet", &espp::GenericDepacketizer::process_packet, py::arg("packet"), "/ Process an incoming RTP packet.\n/ Payload data is accumulated until a packet with " "the marker bit set is\n/ received. At that point the assembled frame is delivered via " "the frame\n/ callback and the buffer is reset.\n/ @param packet The RTP packet to " "process."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassGenericPacketizer = py::class_>( m, "GenericPacketizer", py::dynamic_attr(), "/ A generic RTP packetizer suitable for audio codecs (PCM, G.711, Opus, etc.)\n/ or any " "pre-formatted data that simply needs MTU-based chunking. It splits\n/ frame data into " "chunks of at most max_payload_size bytes and marks the last\n/ chunk with the RTP marker " "bit.\n/\n/ \\section generic_packetizer_ex1 Example\n/ \\snippet rtsp_example.cpp " "generic_packetizer_test"); { // inner classes & enums of GenericPacketizer auto pyClassGenericPacketizer_ClassConfig = py::class_(pyClassGenericPacketizer, "Config", py::dynamic_attr(), "/ Configuration for GenericPacketizer.") .def( py::init<>([](size_t max_payload_size = {1400}, int payload_type = {96}, uint32_t clock_rate = {48000}, std::string encoding_name = {"L16"}, int channels = {1}, std::string fmtp = std::string(), espp::MediaType media_type = {espp::MediaType::AUDIO}, 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_->clock_rate = clock_rate; r_ctor_->encoding_name = encoding_name; r_ctor_->channels = channels; r_ctor_->fmtp = fmtp; r_ctor_->media_type = media_type; 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("clock_rate") = uint32_t{48000}, py::arg("encoding_name") = std::string{"L16"}, py::arg("channels") = int{1}, py::arg("fmtp") = std::string(), py::arg("media_type") = espp::MediaType{espp::MediaType::AUDIO}, py::arg("log_level") = espp::Logger::Verbosity{espp::Logger::Verbosity::WARN}) .def_readwrite("max_payload_size", &espp::GenericPacketizer::Config::max_payload_size, "/< Maximum payload bytes per RTP packet") .def_readwrite("payload_type", &espp::GenericPacketizer::Config::payload_type, "/< RTP payload type number") .def_readwrite("clock_rate", &espp::GenericPacketizer::Config::clock_rate, "/< Clock rate in Hz for RTP timestamps") .def_readwrite("encoding_name", &espp::GenericPacketizer::Config::encoding_name, "/< Encoding name for SDP rtpmap line") .def_readwrite("channels", &espp::GenericPacketizer::Config::channels, "/< Number of audio channels") .def_readwrite("fmtp", &espp::GenericPacketizer::Config::fmtp, "/< Optional format parameters for SDP fmtp line") .def_readwrite("media_type", &espp::GenericPacketizer::Config::media_type, "/< Media type for the SDP m= line") .def_readwrite("log_level", &espp::GenericPacketizer::Config::log_level, "/< Log verbosity level"); } // end of inner classes & enums of GenericPacketizer pyClassGenericPacketizer.def(py::init()) .def("packetize", &espp::GenericPacketizer::packetize, py::arg("frame_data"), "/ Split frame data into RTP payload chunks of at most max_payload_size.\n/ The last " "(or only) chunk has its marker flag set.\n/ @param frame_data The raw frame bytes to " "packetize.\n/ @return A vector of RtpPayloadChunk ready to be wrapped in RTP packets.") .def("get_payload_type", &espp::GenericPacketizer::get_payload_type, "/ Get the RTP payload type number.\n/ @return The configured RTP payload type.") .def("get_clock_rate", &espp::GenericPacketizer::get_clock_rate, "/ Get the RTP clock rate.\n/ @return The configured clock rate in Hz.") .def("get_sdp_media_attributes", &espp::GenericPacketizer::get_sdp_media_attributes, "/ Generate the SDP media-level attribute lines for this codec.\n/ Produces an a=rtpmap " "line and, if fmtp is non-empty, an a=fmtp line.\n/ @return A string containing the SDP " "a= lines.") .def("get_sdp_media_line", &espp::GenericPacketizer::get_sdp_media_line, "/ Generate the SDP m= line for this codec.\n/ @return A string such as \"m=audio 0 " "RTP/AVP 96\"."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassH264Depacketizer = py::class_>( m, "H264Depacketizer", py::dynamic_attr(), "/ @brief RTP depacketizer for H.264 video per RFC 6184.\n/\n/ Reassembles H.264 access " "units from incoming RTP packets. Supports:\n/ - Single NAL unit packets (NAL type " "1-23)\n/ - STAP-A aggregation packets (NAL type 24)\n/ - FU-A " "fragmentation packets (NAL type 28)\n/\n/ When the RTP marker bit is set, the accumulated " "NAL units are delivered\n/ as one Annex B byte-stream (each NAL prefixed with 0x00 0x00 " "0x00 0x01)\n/ via the frame callback set with set_frame_callback().\n/\n/ \\section " "h264_depacketizer_ex1 Example\n/ \\snippet rtsp_example.cpp h264_depacketizer_test"); { // inner classes & enums of H264Depacketizer auto pyClassH264Depacketizer_ClassConfig = py::class_(pyClassH264Depacketizer, "Config", py::dynamic_attr(), "/ Configuration for the H264Depacketizer.") .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::H264Depacketizer::Config::log_level, "/< Log verbosity level"); } // end of inner classes & enums of H264Depacketizer pyClassH264Depacketizer.def(py::init()) .def("process_packet", &espp::H264Depacketizer::process_packet, py::arg("packet"), "/ Process an incoming RTP packet containing H.264 payload.\n/\n/ Handles single NAL, " "STAP-A, and FU-A packet types. NAL units are\n/ buffered until the RTP marker bit " "indicates the end of an access unit,\n/ at which point the complete Annex B frame is " "delivered via the callback.\n/\n/ @param packet The RTP packet to process."); //////////////////// //////////////////// //////////////////// //////////////////// auto pyClassH264Packetizer = py::class_>( m, "H264Packetizer", py::dynamic_attr(), "/ @brief RTP packetizer for H.264 video per RFC 6184.\n/\n/ Accepts H.264 access units " "in Annex B byte-stream format (NAL units\n/ separated by 0x00000001 or 0x000001 start " "codes) and produces a sequence\n/ of RTP payload chunks suitable for " "transmission.\n/\n/ Supports two NAL-unit packetization strategies:\n/ - Single " "NAL unit mode - NAL fits within max_payload_size.\n/ - FU-A fragmentation " "- NAL exceeds max_payload_size (packetization_mode >= 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, std::span, std::span>(), 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, std::span>(), 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 &, std::error_code &>(&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, const espp::RtspSession::Config &>()) .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 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! }