<SWAUTO-4024> Integrate ethernet packet in the ASAM CMP data sink

Aleksandr Kuzin · Aleksandr Kuzin · commit c32caec9ac42 · 2025-08-20T09:39:41.000Z
diff --git a/asam_cmp_capture_module/src/capture_fb.cpp b/asam_cmp_capture_module/src/capture_fb.cpp
@@ -11,9 +11,9 @@ BEGIN_NAMESPACE_ASAM_CMP_CAPTURE_MODULE
 
 CaptureFb::CaptureFb(const ContextPtr& ctx, const ComponentPtr& parent, const StringPtr& localId, const CaptureFbInit& init)
     : asam_cmp_common_lib::CaptureCommonFb(ctx, parent, localId)
+    , allowJumboFrames(false)
     , ethernetWrapper(init.ethernetWrapper)
     , selectedEthernetDeviceName(init.selectedDeviceName)
-    , allowJumboFrames(false)
 {
     initProperties();
     initEncoders();
diff --git a/asam_cmp_capture_module/src/interface_fb.cpp b/asam_cmp_capture_module/src/interface_fb.cpp
@@ -15,9 +15,9 @@ InterfaceFb::InterfaceFb(const ContextPtr& ctx,
                          const asam_cmp_common_lib::InterfaceCommonInit& init,
                          const InterfaceFbInit& internalInit)
     : InterfaceCommonFb(ctx, parent, localId, init)
+    , statusSync(internalInit.statusSync)
     , encoders(internalInit.encoders)
     , deviceStatus(internalInit.deviceStatus)
-    , statusSync(internalInit.statusSync)
     , vendorDataAsString("")
     , ethernetWrapper(internalInit.ethernetWrapper)
     , allowJumboFrames(internalInit.allowJumboFrames)
diff --git a/asam_cmp_capture_module/src/stream_fb.cpp b/asam_cmp_capture_module/src/stream_fb.cpp
@@ -30,15 +30,15 @@ StreamFb::StreamFb(const ContextPtr& ctx,
                    const asam_cmp_common_lib::StreamCommonInit& init,
                    const StreamInit& internalInit)
     : asam_cmp_common_lib::StreamCommonFb(ctx, parent, localId, init)
+    , interfaceId(internalInit.interfaceId)
     , streamIdsList(internalInit.streamIdsList)
     , statusSync(internalInit.statusSync)
-    , interfaceId(internalInit.interfaceId)
-    , ethernetWrapper(internalInit.ethernetWrapper)
-    , dataContext(createEncoderDataContext())
-    , allowJumboFrames(internalInit.allowJumboFrames)
     , encoders(internalInit.encoderBank)
     , parentInterfaceUpdater(internalInit.parentInterfaceUpdater)
     , isConfigured(false)
+    , ethernetWrapper(internalInit.ethernetWrapper)
+    , allowJumboFrames(internalInit.allowJumboFrames)
+    , dataContext(createEncoderDataContext())
 {
     createInputPort();
     initStatuses();
diff --git a/asam_cmp_common_lib/include/asam_cmp_common_lib/stream_common_fb_impl.h b/asam_cmp_common_lib/include/asam_cmp_common_lib/stream_common_fb_impl.h
@@ -70,9 +70,9 @@ StreamCommonFbImpl<Interfaces...>::StreamCommonFbImpl(const ContextPtr& ctx,
                                                       const StringPtr& localId,
                                                       const StreamCommonInit& init)
     : FunctionBlockImpl<IFunctionBlock, IStreamCommon, Interfaces...>(CreateType(), ctx, parent, localId)
-    , streamIdManager(init.streamIdManager)
     , streamId(init.id)
     , payloadType(init.payloadType)
+    , streamIdManager(init.streamIdManager)
 {
     initProperties();
 }
diff --git a/asam_cmp_common_lib/src/ethernet_pcpp_impl.cpp b/asam_cmp_common_lib/src/ethernet_pcpp_impl.cpp
@@ -3,6 +3,7 @@
 #include <SystemUtils.h>
 #include <EthLayer.h>
 #include <PayloadLayer.h>
+#include <Packet.h>
 
 BEGIN_NAMESPACE_ASAM_CMP_COMMON
 
diff --git a/asam_cmp_common_lib/src/interface_common_fb.cpp b/asam_cmp_common_lib/src/interface_common_fb.cpp
@@ -42,7 +42,7 @@ void InterfaceCommonFb::initProperties()
         [this](PropertyObjectPtr& obj, PropertyValueEventArgsPtr& args) { propertyChangedIfNotUpdating(); };
 
     propName = "PayloadType";
-    ListPtr<StringPtr> payloadTypes{"Undefined", "CAN", "CAN FD", "Analog"};
+    ListPtr<StringPtr> payloadTypes{"Undefined", "CAN", "CAN FD", "Analog", "Ethernet"};
     prop = SelectionPropertyBuilder(propName, payloadTypes, 0).build();
     objPtr.addProperty(prop);
     objPtr.getOnPropertyValueWrite(propName) +=
@@ -172,7 +172,7 @@ FunctionBlockPtr InterfaceCommonFb::onAddFunctionBlock(const StringPtr& typeId,
 }
 
 const static std::unordered_map<uint32_t, Int> payloadTypeToIndexMap = {
-    {PayloadType::invalid, 0}, {PayloadType::can, 1}, {PayloadType::canFd, 2}, {PayloadType::analog, 3}};
+    {PayloadType::invalid, 0}, {PayloadType::can, 1}, {PayloadType::canFd, 2}, {PayloadType::analog, 3}, {PayloadType::ethernet, 4}};
 
 const std::vector<uint32_t> indexToPayloadTypeMap = []() {
     std::vector<uint32_t> vec;
diff --git a/asam_cmp_data_sink/include/asam_cmp_data_sink/status_handler.h b/asam_cmp_data_sink/include/asam_cmp_data_sink/status_handler.h
@@ -29,8 +29,8 @@ class StatusMt final
 {
 public:
     StatusMt(const ASAM::CMP::Status& st, std::mutex& mt)
-        : statusRef(st)
-        , mtRef(mt)
+        : mtRef(mt)
+        , statusRef(st)
     {
     }
 
diff --git a/asam_cmp_data_sink/include/asam_cmp_data_sink/stream_fb.h b/asam_cmp_data_sink/include/asam_cmp_data_sink/stream_fb.h
@@ -17,6 +17,7 @@
 #pragma once
 #include <asam_cmp/analog_payload.h>
 #include <asam_cmp/can_payload.h>
+#include <asam_cmp/ethernet_payload.h>
 #include <asam_cmp/packet.h>
 #include <opendaq/packet_factory.h>
 
@@ -36,11 +37,22 @@ struct CANData
 };
 #pragma pack(pop)
 
+#pragma pack(push, 1)
+struct EthernetData
+{
+    uint16_t flags;
+    uint16_t reserved;
+    uint16_t length;
+};
+#pragma pack(pop)
+
+
 class StreamFb final : public asam_cmp_common_lib::StreamCommonFbImpl<IAsamCmpPacketsSubscriber>
 {
 private:
     using Packet = ASAM::CMP::Packet;
     using CanPayload = ASAM::CMP::CanPayload;
+    using EthernetPayload = ASAM::CMP::EthernetPayload;
     using AnalogPayload = ASAM::CMP::AnalogPayload;
 
 public:
@@ -67,11 +79,12 @@ class StreamFb final : public asam_cmp_common_lib::StreamCommonFbImpl<IAsamCmpPa
     void createSignals();
     void buildDataDescriptor();
     void buildCanDescriptor();
+    void buildEthernetDescriptor();
     void buildAnalogDescriptor(const AnalogPayload& payload);
     void buildAsyncDomainDescriptor();
     void buildSyncDomainDescriptor(const float sampleInterval);
-    void processAsyncData(const std::vector<std::shared_ptr<Packet>>& packets);
-    void fillCanData(CANData* const data, const std::shared_ptr<Packet>& packet);
+    void processCanData(const std::vector<std::shared_ptr<Packet>>& packets);
+    void processEthernetData(const std::vector<std::shared_ptr<Packet>>& packets);
     void processSyncData(const std::shared_ptr<Packet>& packet);
     bool domainChanged(const AnalogPayload& payload);
     bool dataChanged(const AnalogPayload& payload);
diff --git a/asam_cmp_data_sink/src/capture_fb.cpp b/asam_cmp_data_sink/src/capture_fb.cpp
@@ -24,9 +24,9 @@ CaptureFb::CaptureFb(const ContextPtr& ctx,
                      CapturePacketsPublisher& capturePacketsPublisher,
                      ASAM::CMP::DeviceStatus&& deviceStatus)
     : CaptureCommonFbImpl(ctx, parent, localId)
+    , deviceStatus(std::move(deviceStatus))
     , dataPacketsPublisher(dataPacketsPublisher)
     , capturePacketsPublisher(capturePacketsPublisher)
-    , deviceStatus(std::move(deviceStatus))
 {
     initDeviceInfoProperties(true);
     setProperties();
diff --git a/asam_cmp_data_sink/src/data_sink_module_fb.cpp b/asam_cmp_data_sink/src/data_sink_module_fb.cpp
@@ -8,6 +8,7 @@
 
 #include <SystemUtils.h>
 #include <asam_cmp_common_lib/ethernet_pcpp_impl.h>
+#include <Packet.h>
 
 #include <iostream>
 
diff --git a/asam_cmp_data_sink/src/interface_fb.cpp b/asam_cmp_data_sink/src/interface_fb.cpp
@@ -12,8 +12,8 @@ InterfaceFb::InterfaceFb(const ContextPtr& ctx,
                          const uint16_t& deviceId,
                          DataPacketsPublisher& publisher)
     : InterfaceCommonFb(ctx, parent, localId, init)
-    , deviceId(deviceId)
     , publisher(publisher)
+    , deviceId(deviceId)
 {
 }
 
@@ -26,8 +26,8 @@ InterfaceFb::InterfaceFb(const ContextPtr& ctx,
                          ASAM::CMP::InterfaceStatus&& ifStatus)
     : InterfaceCommonFb(ctx, parent, localId, init)
     , interfaceStatus(std::move(ifStatus))
-    , deviceId(deviceId)
     , publisher(publisher)
+    , deviceId(deviceId)
 {
     createFbs();
 }
diff --git a/asam_cmp_data_sink/src/stream_fb.cpp b/asam_cmp_data_sink/src/stream_fb.cpp
@@ -2,6 +2,7 @@
 
 #include <asam_cmp_common_lib/unit_converter.h>
 #include <asam_cmp_data_sink/stream_fb.h>
+#include <opendaq/binary_data_packet_factory.h>
 
 
 #include <chrono>
@@ -63,14 +64,19 @@ void StreamFb::receive(const std::vector<std::shared_ptr<Packet>>& packets)
     if (packets.front()->getPayload().getType() != payloadType)
         return;
 
-    if (payloadType == PayloadType::analog)
-    {
-        for (auto& packet : packets)
-            processSyncData(packet);
-    }
-    else
+    switch (payloadType.getType())
     {
-        processAsyncData(packets);
+        case PayloadType::analog:
+            for (auto& packet : packets)
+                processSyncData(packet);
+            break;
+        case PayloadType::can:
+        case PayloadType::canFd:
+            processCanData(packets);
+            break;
+        case PayloadType::ethernet:
+            processEthernetData(packets);
+            break;
     }
 }
 
@@ -104,6 +110,9 @@ void StreamFb::buildDataDescriptor()
             break;
         case PayloadType::analog:
             break;
+        case PayloadType::ethernet:
+            buildEthernetDescriptor();
+            break;
     }
 }
 
@@ -128,6 +137,19 @@ void StreamFb::buildCanDescriptor()
     dataSignal.setDescriptor(canMsgDescriptor);
 }
 
+void StreamFb::buildEthernetDescriptor()
+{
+    auto metadata = Dict<IString, IString>();
+    metadata["DataType"] = "Ethernet";
+    const auto ethernetMsgDescriptor =
+        DataDescriptorBuilder()
+        .setSampleType(SampleType::Binary)
+        .setMetadata(metadata)
+        .build();
+
+    dataSignal.setDescriptor(ethernetMsgDescriptor);
+}
+
 void StreamFb::buildAnalogDescriptor(const AnalogPayload& payload)
 {
     const auto inputDataType = payload.getSampleDt() == AnalogPayload::SampleDt::aInt16 ? SampleType::Int16 : SampleType::Int32;
@@ -182,7 +204,7 @@ void StreamFb::buildSyncDomainDescriptor(const float sampleInterval)
     analogHeader.setSampleInterval(sampleInterval);
 }
 
-void StreamFb::processAsyncData(const std::vector<std::shared_ptr<Packet>>& packets)
+void StreamFb::processCanData(const std::vector<std::shared_ptr<Packet>>& packets)
 {
     const uint64_t newSamples = packets.size();
     auto timestamp = packets.front()->getTimestamp();
@@ -195,13 +217,11 @@ void StreamFb::processAsyncData(const std::vector<std::shared_ptr<Packet>>& pack
 
     for (auto& packet : packets)
     {
-        switch (payloadType.getType())
-        {
-            case PayloadType::can:
-            case PayloadType::canFd:
-                fillCanData(buffer, packet);
-                break;
-        }
+        auto& payload = static_cast<const CanPayload&>(packet->getPayload());
+        buffer->arbId = payload.getId();
+        buffer->length = payload.getDataLength();
+        memcpy(buffer->data, payload.getData(), buffer->length);
+
         *domainBuffer++ = packet->getTimestamp();
         buffer++;
     }
@@ -210,13 +230,27 @@ void StreamFb::processAsyncData(const std::vector<std::shared_ptr<Packet>>& pack
     domainSignal.sendPacket(domainPacket);
 }
 
-void StreamFb::fillCanData(CANData* const data, const std::shared_ptr<Packet>& packet)
+void StreamFb::processEthernetData(const std::vector<std::shared_ptr<Packet>>& packets)
 {
-    auto& payload = static_cast<const CanPayload&>(packet->getPayload());
+    static constexpr size_t analogMessagePayloadSize{6};
+    for (auto& packet : packets)
+    {
+        auto timestamp = packet->getTimestamp();
+
+        const auto domainPacket = DataPacket(domainSignal.getDescriptor(), 1, timestamp);
+        auto domainBuffer = static_cast<uint64_t*>(domainPacket.getRawData());
 
-    data->arbId = payload.getId();
-    data->length = payload.getDataLength();
-    memcpy(data->data, payload.getData(), data->length);
+        auto& payload = static_cast<const EthernetPayload&>(packet->getPayload());
+        auto payloadLen = payload.getLength() - analogMessagePayloadSize;
+
+        const auto dataPacket = BinaryDataPacket(domainPacket, dataSignal.getDescriptor(), payloadLen);
+        auto buffer = static_cast<uint8_t*>(dataPacket.getRawData());
+        memcpy(buffer, payload.getData(), payloadLen);
+        *domainBuffer = packet->getTimestamp();
+
+        dataSignal.sendPacket(dataPacket);
+        domainSignal.sendPacket(domainPacket);
+    }
 }
 
 void StreamFb::processSyncData(const std::shared_ptr<Packet>& packet)
diff --git a/asam_cmp_data_sink/tests/test_data_packets_publisher.cpp b/asam_cmp_data_sink/tests/test_data_packets_publisher.cpp
@@ -2,6 +2,7 @@
 #include <gmock/gmock.h>
 
 #include <asam_cmp_data_sink/data_packets_publisher.h>
+#include <Packet.h>
 
 using namespace daq;
 
diff --git a/asam_cmp_data_sink/tests/test_data_sink_module_fb.cpp b/asam_cmp_data_sink/tests/test_data_sink_module_fb.cpp
@@ -11,6 +11,8 @@
 #include <asam_cmp_data_sink/data_sink_module_fb.h>
 #include <asam_cmp_data_sink/module_dll.h>
 
+#include <Packet.h>
+
 using namespace daq;
 using daq::asam_cmp_common_lib::PcppPacketReceivedCallbackType;
 
diff --git a/asam_cmp_data_sink/tests/test_status_fb.cpp b/asam_cmp_data_sink/tests/test_status_fb.cpp
@@ -9,6 +9,8 @@
 #include <asam_cmp_data_sink/status_handler.h>
 #include <asam_cmp_data_sink/status_fb_impl.h>
 
+#include <Packet.h>
+
 using namespace daq;
 
 using ASAM::CMP::CaptureModulePayload;
diff --git a/asam_cmp_data_sink/tests/test_stream_fb.cpp b/asam_cmp_data_sink/tests/test_stream_fb.cpp
diff --git a/external/AsamCmpLib/CMakeLists.txt b/external/AsamCmpLib/CMakeLists.txt
diff --git a/external/PcapPlusPlus/CMakeLists.txt b/external/PcapPlusPlus/CMakeLists.txt

Original file line number	Diff line number	Diff line change
`@@ -70,9 +70,9 @@ StreamCommonFbImpl<Interfaces...>::StreamCommonFbImpl(const ContextPtr& ctx,`
`70`	`70`	`const StringPtr& localId,`
`71`	`71`	`const StreamCommonInit& init)`
`72`	`72`	`: FunctionBlockImpl<IFunctionBlock, IStreamCommon, Interfaces...>(CreateType(), ctx, parent, localId)`
`73`		`- , streamIdManager(init.streamIdManager)`
`74`	`73`	`, streamId(init.id)`
`75`	`74`	`, payloadType(init.payloadType)`
	`75`	`+ , streamIdManager(init.streamIdManager)`
`76`	`76`	`{`
`77`	`77`	`initProperties();`
`78`	`78`	`}`
Original file line number	Diff line number	Diff line change
`@@ -29,8 +29,8 @@ class StatusMt final`
`29`	`29`	`{`
`30`	`30`	`public:`
`31`	`31`	`StatusMt(const ASAM::CMP::Status& st, std::mutex& mt)`
`32`		`- : statusRef(st)`
`33`		`- , mtRef(mt)`
	`32`	`+ : mtRef(mt)`
	`33`	`+ , statusRef(st)`
`34`	`34`	`{`
`35`	`35`	`}`
`36`	`36`