Cast node

.github/workflows/main.workflow.yml

@@ -67,6 +67,8 @@ jobs:
         exclude:
           - os: windows-latest
             cmake: '3.10.x'
+          - os: macos-latest # Skip the old cmake on latest macos - doesn't handle ARM64 aarch correctly
+            cmake: '3.10.x'
 
     steps:
     - name: Cache .hunter folder
@@ -333,4 +335,4 @@ jobs:
         upload_url: ${{ steps.createRelease.outputs.upload_url }}
         asset_path: ${{ github.workspace }}/depthai-core-${{ steps.tag.outputs.version }}-win32-no-opencv.zip
         asset_name: depthai-core-${{ steps.tag.outputs.version }}-win32-no-opencv.zip
-        asset_content_type: application/octet-stream
+        asset_content_type: application/octet-stream

CMakeLists.txt

@@ -231,6 +231,7 @@ add_library(${TARGET_CORE_NAME}
     src/pipeline/node/DetectionParser.cpp
     src/pipeline/node/UVC.cpp
     src/pipeline/node/PointCloud.cpp
+    src/pipeline/node/Cast.cpp
     src/pipeline/datatype/Buffer.cpp
     src/pipeline/datatype/ImgFrame.cpp
     src/pipeline/datatype/EncodedFrame.cpp

cmake/Depthai/DepthaiBootloaderConfig.cmake

@@ -3,5 +3,5 @@ set(DEPTHAI_BOOTLOADER_MATURITY "release")
 # set(DEPTHAI_BOOTLOADER_MATURITY "snapshot")
 
 # "version if applicable"
-set(DEPTHAI_BOOTLOADER_VERSION "0.0.27")
-# set(DEPTHAI_BOOTLOADER_VERSION "0.0.24+57c26493754e2f00e57f6594b0b1a317f762d5f2")
+set(DEPTHAI_BOOTLOADER_VERSION "0.0.28")
+# set(DEPTHAI_BOOTLOADER_VERSION "0.0.27+5fb331f993adceeeda72202c233a9e3939ab3dab")

cmake/Depthai/DepthaiDeviceSideConfig.cmake

@@ -2,7 +2,7 @@
 set(DEPTHAI_DEVICE_SIDE_MATURITY "snapshot")
 
 # "full commit hash of device side binary"
-set(DEPTHAI_DEVICE_SIDE_COMMIT "aace8f1c8098546e743f69fc9b3b856d43aed457")
+set(DEPTHAI_DEVICE_SIDE_COMMIT "045361e9d6601e4c20d016182b93dce4159ec38b")
 
 # "version if applicable"
 set(DEPTHAI_DEVICE_SIDE_VERSION "")

include/depthai/device/DeviceBase.hpp

@@ -446,6 +446,16 @@ class DeviceBase {
      */
     int getXLinkChunkSize();
 
+    /**
+     * Sets the maximum transmission rate for the XLink connection on device side,
+     * using a simple token bucket algorithm. Useful for bandwidth throttling
+     *
+     * @param maxRateBytesPerSecond Rate limit in Bytes/second
+     * @param burstSize Size in Bytes for how much to attempt to send once, 0 = auto
+     * @param waitUs Time in microseconds to wait for replenishing tokens, 0 = auto
+     */
+    void setXLinkRateLimit(int maxRateBytesPerSecond, int burstSize = 0, int waitUs = 0);
+
     /**
      * Get the Device Info object o the device which is currently running
      *

include/depthai/pipeline/node/Cast.hpp

@@ -0,0 +1,67 @@
+#pragma once
+
+#include <depthai/pipeline/Node.hpp>
+
+#include "depthai/pipeline/datatype/Tracklets.hpp"
+
+// standard
+#include <fstream>
+
+// shared
+#include <depthai-shared/properties/CastProperties.hpp>
+
+namespace dai {
+namespace node {
+
+/**
+ * @brief Cast node.
+ */
+class Cast : public NodeCRTP<Node, Cast, CastProperties> {
+   public:
+    constexpr static const char* NAME = "Cast";
+
+    Cast(const std::shared_ptr<PipelineImpl>& par, int64_t nodeId);
+    Cast(const std::shared_ptr<PipelineImpl>& par, int64_t nodeId, std::unique_ptr<Properties> props);
+
+    /**
+     * Input NNData or ImgFrame message.
+     */
+    Input input{*this, "input", Input::Type::SReceiver, false, 4, true, {{DatatypeEnum::ImgFrame, false}, {DatatypeEnum::NNData, false}}};
+
+    /**
+     * Output ImgFrame message.
+     */
+    Output output{*this, "output", Output::Type::MSender, {{DatatypeEnum::ImgFrame, false}}};
+
+    /**
+     * Passthrough input message.
+     */
+    Output passthroughInput{*this, "passthroughInput", Output::Type::MSender, {{DatatypeEnum::ImgFrame, false}, {DatatypeEnum::NNData, false}}};
+
+    /**
+     * Set number of frames in pool
+     * @param numFramesPool Number of frames in pool
+     */
+    Cast& setNumFramesPool(int numFramesPool);
+
+    /**
+     * Set output frame type
+     * @param outputType Output frame type
+     */
+    Cast& setOutputFrameType(dai::RawImgFrame::Type outputType);
+
+    /**
+     * Set scale
+     * @param scale Scale
+     */
+    Cast& setScale(float scale);
+
+    /**
+     * Set offset
+     * @param offset Offset
+     */
+    Cast& setOffset(float offset);
+};
+
+}  // namespace node
+}  // namespace dai

src/device/DeviceBase.cpp

@@ -1219,6 +1219,10 @@ int DeviceBase::getXLinkChunkSize() {
     return pimpl->rpcClient->call("getXLinkChunkSize").as<int>();
 }
 
+void DeviceBase::setXLinkRateLimit(int maxRateBytesPerSecond, int burstSize, int waitUs) {
+    pimpl->rpcClient->call("setXLinkRateLimit", maxRateBytesPerSecond, burstSize, waitUs);
+}
+
 DeviceInfo DeviceBase::getDeviceInfo() const {
     return deviceInfo;
 }

src/device/DeviceBootloader.cpp

@@ -417,6 +417,19 @@ void DeviceBootloader::init(bool embeddedMvcmd, const dai::Path& pathToMvcmd, tl
             // Retrieve bootloader version
             version = requestVersion();
             flashedVersion = version;
+
+            auto recommendedMinVersion = DeviceBootloader::Version(0, 0, 28);
+            if(version < recommendedMinVersion) {
+                logger::warn(
+                    "[{}] [{}] Flashed bootloader version {}, less than {} is susceptible to bootup/restart failure. Upgrading is advised, flashing "
+                    "main/factory (not user) bootloader. Available: {}",
+                    deviceInfo.mxid,
+                    deviceInfo.name,
+                    version.toString(),
+                    recommendedMinVersion.toString(),
+                    getEmbeddedBootloaderVersion().toString());
+            }
+
             if(version >= Version(0, 0, 12)) {
                 // If version is adequate, do an in memory boot.
 

src/pipeline/datatype/StreamMessageParser.cpp

@@ -8,6 +8,8 @@
 #include <XLink/XLinkPublicDefines.h>
 #include <spdlog/spdlog.h>
 
+#include "utility/Logging.hpp"
+
 // project
 #include "depthai/pipeline/datatype/ADatatype.hpp"
 #include "depthai/pipeline/datatype/AprilTagConfig.hpp"
@@ -67,6 +69,8 @@
 
 namespace dai {
 
+static constexpr std::array<uint8_t, 16> endOfPacketMarker = {0xAB, 0xCD, 0xEF, 0x01, 0x23, 0x45, 0x67, 0x89, 0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0};
+
 // Reads int from little endian format
 inline int readIntLE(uint8_t* data) {
     return data[0] + data[1] * 256 + data[2] * 256 * 256 + data[3] * 256 * 256 * 256;
@@ -85,28 +89,40 @@ inline std::shared_ptr<T> parseDatatype(std::uint8_t* metadata, size_t size, std
 }
 
 static std::tuple<DatatypeEnum, size_t, size_t> parseHeader(streamPacketDesc_t* const packet) {
-    if(packet->length < 8) {
-        throw std::runtime_error("Bad packet, couldn't parse (not enough data)");
+    if(packet->length < 24) {
+        throw std::runtime_error(fmt::format("Bad packet, couldn't parse (not enough data), total size {}", packet->length));
+    }
+    const std::uint32_t packetLength = packet->length - endOfPacketMarker.size();
+    const int serializedObjectSize = readIntLE(packet->data + packetLength - 4);
+    const auto objectType = static_cast<DatatypeEnum>(readIntLE(packet->data + packetLength - 8));
+
+    uint8_t* marker = packet->data + packetLength;
+    if(memcmp(marker, endOfPacketMarker.data(), endOfPacketMarker.size()) != 0) {
+        std::string hex;
+        for(std::uint32_t i = 0; i < endOfPacketMarker.size(); i++) {
+            hex += fmt::format("{:02X}", marker[i]);
+        }
+        logger::warn("StreamMessageParser end-of-packet marker mismatch, got: " + hex);
     }
-    const int serializedObjectSize = readIntLE(packet->data + packet->length - 4);
-    const auto objectType = static_cast<DatatypeEnum>(readIntLE(packet->data + packet->length - 8));
+
+    const auto info = fmt::format(", total size {}, type {}, metadata size {}", packet->length, objectType, serializedObjectSize);
 
     if(serializedObjectSize < 0) {
-        throw std::runtime_error("Bad packet, couldn't parse (metadata size negative)");
-    } else if(serializedObjectSize > static_cast<int>(packet->length)) {
-        throw std::runtime_error("Bad packet, couldn't parse (metadata size larger than packet length)");
+        throw std::runtime_error("Bad packet, couldn't parse (metadata size negative)" + info);
+    } else if(serializedObjectSize > static_cast<int>(packetLength)) {
+        throw std::runtime_error("Bad packet, couldn't parse (metadata size larger than packet length)" + info);
     }
-    if(static_cast<int>(packet->length) - 8 - serializedObjectSize < 0) {
-        throw std::runtime_error("Bad packet, couldn't parse (data too small)");
+    if(static_cast<int>(packetLength) - 8 - serializedObjectSize < 0) {
+        throw std::runtime_error("Bad packet, couldn't parse (data too small)" + info);
     }
-    const std::uint32_t bufferLength = packet->length - 8 - serializedObjectSize;
-    if(bufferLength > packet->length) {
-        throw std::runtime_error("Bad packet, couldn't parse (data too large)");
+    const std::uint32_t bufferLength = packetLength - 8 - serializedObjectSize;
+    if(bufferLength > packetLength) {
+        throw std::runtime_error("Bad packet, couldn't parse (data too large)" + info);
     }
     auto* const metadataStart = packet->data + bufferLength;
 
-    if(metadataStart < packet->data || metadataStart >= packet->data + packet->length) {
-        throw std::runtime_error("Bad packet, couldn't parse (metadata out of bounds)");
+    if(metadataStart < packet->data || metadataStart >= packet->data + packetLength) {
+        throw std::runtime_error("Bad packet, couldn't parse (metadata out of bounds)" + info);
     }
 
     return {objectType, serializedObjectSize, bufferLength};
@@ -217,7 +233,8 @@ std::shared_ptr<RawBuffer> StreamMessageParser::parseMessage(streamPacketDesc_t*
             break;
     }
 
-    throw std::runtime_error("Bad packet, couldn't parse");
+    throw std::runtime_error(
+        fmt::format("Bad packet, couldn't parse, total size {}, type {}, metadata size {}", packet->length, objectType, serializedObjectSize));
 }
 
 std::shared_ptr<ADatatype> StreamMessageParser::parseMessageToADatatype(streamPacketDesc_t* const packet, DatatypeEnum& objectType) {
@@ -324,8 +341,10 @@ std::shared_ptr<ADatatype> StreamMessageParser::parseMessageToADatatype(streamPa
             break;
     }
 
-    throw std::runtime_error("Bad packet, couldn't parse (invalid message type)");
+    throw std::runtime_error(fmt::format(
+        "Bad packet, couldn't parse (invalid message type), total size {}, type {}, metadata size {}", packet->length, objectType, serializedObjectSize));
 }
+
 std::shared_ptr<ADatatype> StreamMessageParser::parseMessageToADatatype(streamPacketDesc_t* const packet) {
     DatatypeEnum objectType;
     return parseMessageToADatatype(packet, objectType);
@@ -337,6 +356,7 @@ std::vector<std::uint8_t> StreamMessageParser::serializeMessage(const RawBuffer&
     // 2. serialize and append metadata
     // 3. append datatype enum (4B LE)
     // 4. append size (4B LE) of serialized metadata
+    // 5. append 16-byte marker/canary
 
     DatatypeEnum datatype;
     std::vector<std::uint8_t> metadata;
@@ -350,11 +370,12 @@ std::vector<std::uint8_t> StreamMessageParser::serializeMessage(const RawBuffer&
     for(int i = 0; i < 4; i++) leMetadataSize[i] = (metadataSize >> i * 8) & 0xFF;
 
     std::vector<std::uint8_t> ser;
-    ser.reserve(data.data.size() + metadata.size() + leDatatype.size() + leMetadataSize.size());
+    ser.reserve(data.data.size() + metadata.size() + leDatatype.size() + leMetadataSize.size() + endOfPacketMarker.size());
     ser.insert(ser.end(), data.data.begin(), data.data.end());
     ser.insert(ser.end(), metadata.begin(), metadata.end());
     ser.insert(ser.end(), leDatatype.begin(), leDatatype.end());
     ser.insert(ser.end(), leMetadataSize.begin(), leMetadataSize.end());
+    ser.insert(ser.end(), endOfPacketMarker.begin(), endOfPacketMarker.end());
 
     return ser;
 }

src/pipeline/node/Cast.cpp

@@ -0,0 +1,36 @@
+#include "depthai/pipeline/node/Cast.hpp"
+
+#include "spdlog/fmt/fmt.h"
+
+namespace dai {
+namespace node {
+
+Cast::Cast(const std::shared_ptr<PipelineImpl>& par, int64_t nodeId) : Cast(par, nodeId, std::make_unique<Cast::Properties>()) {}
+Cast::Cast(const std::shared_ptr<PipelineImpl>& par, int64_t nodeId, std::unique_ptr<Properties> props)
+    : NodeCRTP<Node, Cast, CastProperties>(par, nodeId, std::move(props)) {
+    setInputRefs({&input});
+    setOutputRefs({&output, &passthroughInput});
+}
+
+Cast& Cast::setNumFramesPool(int numFramesPool) {
+    properties.numFramesPool = numFramesPool;
+    return *this;
+}
+
+Cast& Cast::setOutputFrameType(dai::RawImgFrame::Type outputType) {
+    properties.outputType = outputType;
+    return *this;
+}
+
+Cast& Cast::setScale(float scale) {
+    properties.scale = scale;
+    return *this;
+}
+
+Cast& Cast::setOffset(float offset) {
+    properties.offset = offset;
+    return *this;
+}
+
+}  // namespace node
+}  // namespace dai

tests/src/stream_message_parser_test.cpp

@@ -6,6 +6,8 @@
 
 // TODO(themarpe) - fuzz me instead
 
+constexpr auto MARKER_SIZE = 16;
+
 TEST_CASE("Correct message") {
     dai::ImgFrame frm;
     auto ser = dai::StreamMessageParser::serializeMessage(frm);
@@ -20,12 +22,29 @@ TEST_CASE("Correct message") {
     REQUIRE(ser == ser2);
 }
 
+TEST_CASE("Correct message, but padding corrupted, a warning should be printed") {
+    dai::ImgFrame frm;
+    auto ser = dai::StreamMessageParser::serializeMessage(frm);
+    ser[ser.size() - 1] = 0x55;
+
+    streamPacketDesc_t packet;
+    packet.data = ser.data();
+    packet.length = ser.size();
+
+    auto des = dai::StreamMessageParser::parseMessageToADatatype(&packet);
+    auto ser2 = dai::StreamMessageParser::serializeMessage(des);
+    // Just for the binary compare to match, cause the same corruption in re-serialized msg
+    ser2[ser.size() - 1] = 0x55;
+
+    REQUIRE(ser == ser2);
+}
+
 TEST_CASE("Incorrect message bad size") {
     dai::ImgFrame frm;
     auto ser = dai::StreamMessageParser::serializeMessage(frm);
 
     // wreak havoc on serialized data
-    ser[ser.size() - 1] = 100;
+    ser[ser.size() - 1 - MARKER_SIZE] = 100;
 
     streamPacketDesc_t packet;
     packet.data = ser.data();
@@ -39,7 +58,7 @@ TEST_CASE("Incorrect message negative size") {
     auto ser = dai::StreamMessageParser::serializeMessage(frm);
 
     // wreak havoc on serialized data
-    ser[ser.size() - 1] = 200;
+    ser[ser.size() - 1 - MARKER_SIZE] = 200;
 
     streamPacketDesc_t packet;
     packet.data = ser.data();
@@ -87,7 +106,7 @@ TEST_CASE("Raw - Incorrect message bad size") {
     auto ser = dai::StreamMessageParser::serializeMessage(frm);
 
     // wreak havoc on serialized data
-    ser[ser.size() - 1] = 100;
+    ser[ser.size() - 1 - MARKER_SIZE] = 100;
 
     streamPacketDesc_t packet;
     packet.data = ser.data();
@@ -101,7 +120,7 @@ TEST_CASE("Raw - Incorrect message negative size") {
     auto ser = dai::StreamMessageParser::serializeMessage(frm);
 
     // wreak havoc on serialized data
-    ser[ser.size() - 1] = 200;
+    ser[ser.size() - 1 - MARKER_SIZE] = 200;
 
     streamPacketDesc_t packet;
     packet.data = ser.data();
@@ -128,4 +147,4 @@ TEST_CASE("Raw - Incorrect message too small size 2") {
     packet.length = ser.size();
 
     REQUIRE_THROWS(dai::StreamMessageParser::parseMessage(&packet));
-}
+}