Merge pull request #19 from tlm-solutions/marenz.borzoi_integration

Refactor packet parsing and integrate borzoi sender

CMakeLists.txt

@@ -7,28 +7,34 @@ option(NIX_BUILD "Is CMake called by a nix build?" OFF)
 
 add_executable(tetra-decoder
                src/main.cpp
-               src/reporter.cpp
-        src/decoder.cpp
-        src/bit_stream_decoder.cpp
-        src/iq_stream_decoder.cpp
-        src/prometheus.cpp
-        src/l2/access_assignment_channel.cpp
-        src/l2/broadcast_synchronization_channel.cpp
-        src/l2/logical_link_control.cpp
-        src/l2/lower_mac.cpp
-        src/l2/slot.cpp
-        src/l2/timebase_counter.cpp
-        src/l2/upper_mac.cpp
-        src/l2/upper_mac_packet.cpp
-        src/l2/upper_mac_packet_builder.cpp
-        src/l2/upper_mac_packet_formatter.cpp
-        src/l3/mobile_link_entity.cpp
-        src/l3/mobile_management.cpp
-        src/l3/circuit_mode_control_entity.cpp
-        src/l3/short_data_service.cpp
-        src/utils/address.cpp
-        src/utils/bit_vector.cpp
-        src/utils/viter_bi_codec.cpp)
+               src/decoder.cpp
+               src/bit_stream_decoder.cpp
+               src/iq_stream_decoder.cpp
+               src/prometheus.cpp
+               src/borzoi/borzoi_converter.cpp
+               src/borzoi/borzoi_sender.cpp
+               src/l2/access_assignment_channel.cpp
+               src/l2/broadcast_synchronization_channel.cpp        
+               src/l2/logical_link_control_formatter.cpp
+               src/l2/logical_link_control_packet.cpp
+               src/l2/lower_mac.cpp
+               src/l2/slot.cpp
+               src/l2/timebase_counter.cpp
+               src/l2/upper_mac.cpp
+               src/l2/upper_mac_packet.cpp
+               src/l2/upper_mac_packet_builder.cpp
+               src/l2/upper_mac_packet_formatter.cpp
+               src/l3/circuit_mode_control_entity_formatter.cpp
+               src/l3/circuit_mode_control_entity_packet.cpp
+               src/l3/mobile_link_entity_formatter.cpp
+               src/l3/mobile_link_entity_packet.cpp
+               src/l3/mobile_management_formatter.cpp
+               src/l3/mobile_management_packet.cpp
+               src/l3/short_data_service_formatter.cpp
+               src/l3/short_data_service_packet.cpp
+               src/utils/address.cpp
+               src/utils/bit_vector.cpp
+               src/utils/viter_bi_codec.cpp)
 
 add_executable(tetra-puncturing
                src/examples/tetra_puncturing.cpp)
@@ -50,14 +56,15 @@ find_package(ZLIB REQUIRED)
 find_package(fmt REQUIRED)
 find_package(nlohmann_json REQUIRED)
 find_package(prometheus-cpp CONFIG REQUIRED)
+find_package(cpr REQUIRED)
 
 include_directories(${CMAKE_SOURCE_DIR}/include)
 
 if (NOT NIX_BUILD)
 	target_link_libraries(tetra-decoder cxxopts::cxxopts)
 endif()
 
-target_link_libraries(tetra-decoder ZLIB::ZLIB fmt::fmt nlohmann_json::nlohmann_json viterbi prometheus-cpp::pull)
+target_link_libraries(tetra-decoder ZLIB::ZLIB fmt::fmt nlohmann_json::nlohmann_json viterbi prometheus-cpp::pull cpr::cpr)
 target_link_libraries(tetra-viterbi viterbi)
 
 install(TARGETS tetra-decoder DESTINATION bin)

derivation.nix

@@ -7,6 +7,7 @@
 , nlohmann_json
 , prometheus-cpp
 , curlFull
+, libcpr
 }:
 clangStdenv.mkDerivation {
   name = "tetra-decoder";
@@ -21,6 +22,7 @@ clangStdenv.mkDerivation {
     nlohmann_json
     curlFull
     prometheus-cpp
+    libcpr
   ];
 
   cmakeFlags = [ "-DNIX_BUILD=ON" ];

include/borzoi/borzoi_converter.hpp

@@ -0,0 +1,17 @@
+/*
+ * Copyright (C) 2024 Transit Live Mapping Solutions
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include "l2/slot.hpp"
+#include "l3/short_data_service_packet.hpp"
+
+struct BorzoiConverter {
+    static auto to_json(ShortDataServicePacket* packet) -> nlohmann::json;
+    static auto to_json(const Slots& slots) -> nlohmann::json;
+};

include/borzoi/borzoi_sender.hpp

@@ -0,0 +1,56 @@
+/*
+ * Copyright (C) 2022-2024 Transit Live Mapping Solutions
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include "l2/logical_link_control_packet.hpp"
+#include "l2/slot.hpp"
+#include "thread_safe_fifo.hpp"
+#include <atomic>
+#include <cpr/cpr.h>
+#include <thread>
+#include <variant>
+
+class BorzoiSender {
+  public:
+    BorzoiSender() = delete;
+
+    /// This class sends the HTTP Post requests to borzoi. https://github.com/tlm-solutions/borzoi
+    /// \param queue the queue holds either the parsed packets (std::unique_ptr<LogicalLinkControlPacket>) or Slots that
+    /// failed to decode
+    /// \param termination_flag this flag is set when the sender should terminate after finishing all work
+    /// \param borzoi_url the URL of borzoi
+    /// \param borzoi_uuid the station UUID of this instance of tetra-decoder sending to borzoi
+    BorzoiSender(ThreadSafeFifo<std::variant<std::unique_ptr<LogicalLinkControlPacket>, Slots>>& queue,
+                 std::atomic_bool& termination_flag, const std::string& borzoi_url, std::string borzoi_uuid);
+
+    ~BorzoiSender();
+
+  private:
+    /// The thread function for continously process incomming parsed packets or failed slots.
+    auto worker() -> void;
+
+    void send_packet(const std::unique_ptr<LogicalLinkControlPacket>& packet);
+    void send_failed_slots(const Slots& slots);
+
+    /// The input queue
+    ThreadSafeFifo<std::variant<std::unique_ptr<LogicalLinkControlPacket>, Slots>>& queue_;
+
+    /// The flag that is set when terminating the program
+    std::atomic_bool& termination_flag_;
+
+    /// The urls of borzoi
+    cpr::Url borzoi_url_sds_;
+    cpr::Url borzoi_url_failed_slots_;
+
+    /// The Station UUID of borzoi
+    std::string borzoi_uuid_;
+
+    /// The worker thread
+    std::thread worker_thread_;
+};

include/decoder.hpp

@@ -9,16 +9,17 @@
 
 #pragma once
 
+#include "bit_stream_decoder.hpp"
+#include "borzoi/borzoi_sender.hpp"
+#include "iq_stream_decoder.hpp"
+#include "l2/lower_mac.hpp"
 #include "l2/upper_mac.hpp"
+#include "thread_safe_fifo.hpp"
+#include <atomic>
 #include <memory>
 #include <optional>
 #include <string>
 
-#include <bit_stream_decoder.hpp>
-#include <iq_stream_decoder.hpp>
-#include <l2/lower_mac.hpp>
-#include <reporter.hpp>
-
 /**
  * Tetra downlink decoder for PI/4-DQPSK modulation
  *
@@ -39,21 +40,36 @@
  */
 class Decoder {
   public:
-    Decoder(unsigned int receive_port, unsigned int send_port, bool packed, std::optional<std::string> input_file,
-            std::optional<std::string> output_file, bool iq_or_bit_stream,
+    Decoder(unsigned int receive_port, const std::string& borzoi_url, const std::string& borzoi_uuid, bool packed,
+            std::optional<std::string> input_file, std::optional<std::string> output_file, bool iq_or_bit_stream,
             std::optional<unsigned int> uplink_scrambling_code,
             const std::shared_ptr<PrometheusExporter>& prometheus_exporter);
     ~Decoder();
 
     void main_loop();
 
   private:
+    /// This flag is set when the program should termiate. It is pass down to the next stage in the chain when
+    /// processing is done in the current stage.
+    std::atomic_bool termination_flag_ = false;
+
+    /// This flag is passed for the StreamingOrderedOutputThreadPoolExecutor to the upper mac.
+    std::atomic_bool upper_mac_termination_flag_ = false;
+
+    /// This flag is passed from the upper mac to the borzoi sender.
+    std::atomic_bool borzoi_sender_termination_flag_ = false;
+    /// This queue is used to pass data from the upper mac to the borzoi sender.
+    ThreadSafeFifo<std::variant<std::unique_ptr<LogicalLinkControlPacket>, Slots>> bozoi_queue_;
+
     /// The worker queue for the lower mac
     std::shared_ptr<StreamingOrderedOutputThreadPoolExecutor<LowerMac::return_type>> lower_mac_work_queue_;
 
     /// The reference to the upper mac thread class
     std::unique_ptr<UpperMac> upper_mac_;
 
+    /// The reference to the borzoi sender thread class
+    std::unique_ptr<BorzoiSender> borzoi_sender_;
+
     std::shared_ptr<BitStreamDecoder> bit_stream_decoder_;
     std::unique_ptr<IQStreamDecoder> iq_stream_decoder_;
 

include/iq_stream_decoder.hpp

@@ -9,15 +9,13 @@
 
 #pragma once
 
+#include "bit_stream_decoder.hpp"
+#include "fixed_queue.hpp"
+#include "l2/lower_mac.hpp"
+#include "streaming_ordered_output_thread_pool_executor.hpp"
 #include <complex>
 #include <memory>
 
-#include <bit_stream_decoder.hpp>
-#include <fixed_queue.hpp>
-#include <l2/lower_mac.hpp>
-#include <signal_handler.hpp>
-#include <streaming_ordered_output_thread_pool_executor.hpp>
-
 /**
  * Tetra downlink decoder for PI/4-DQPSK modulation
  *

include/l2/logical_link_control_packet.hpp

@@ -0,0 +1,73 @@
+/*
+ * Copyright (C) 2024 Transit Live Mapping Solutions
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include "l2/upper_mac_packet.hpp"
+#include "utils/bit_vector.hpp"
+#include <optional>
+
+/// The type of the basic link packet.
+enum class BasicLinkType {
+    kBlAdataWithoutFcs,
+    kBlDataWithoutFcs,
+    kBlUdataWithoutFcs,
+    kBlAckWithoutFcs,
+    kBlAdataWithFcs,
+    kBlDataWithFcs,
+    kBlUdataWithFcs,
+    kBlAckWithFcs,
+};
+
+constexpr auto to_string(BasicLinkType type) noexcept -> const char* {
+    switch (type) {
+    case BasicLinkType::kBlAdataWithoutFcs:
+        return "BL-ADATA without FCS";
+    case BasicLinkType::kBlDataWithoutFcs:
+        return "BL-DATA without FCS";
+    case BasicLinkType::kBlUdataWithoutFcs:
+        return "BL-UDATA without FCS";
+    case BasicLinkType::kBlAckWithoutFcs:
+        return "BL-ACK without FCS";
+    case BasicLinkType::kBlAdataWithFcs:
+        return "BL-ADATA with FCS";
+    case BasicLinkType::kBlDataWithFcs:
+        return "BL-DATA with FCS";
+    case BasicLinkType::kBlUdataWithFcs:
+        return "BL-UDATA with FCS";
+    case BasicLinkType::kBlAckWithFcs:
+        return "BL-ACK with FCS";
+    }
+};
+
+struct BasicLinkInformation {
+    BasicLinkType basic_link_type_;
+    std::optional<unsigned _BitInt(1)> n_r_;
+    std::optional<unsigned _BitInt(1)> n_s_;
+    std::optional<bool> fcs_good_;
+
+    BasicLinkInformation() = delete;
+
+    /// construct a BasicLinkInformation from a BitVector
+    explicit BasicLinkInformation(BitVector& data);
+};
+
+auto operator<<(std::ostream& stream, const BasicLinkInformation& bli) -> std::ostream&;
+
+/// The packet that is parsed in the logical link control layer. Currently we only implement basic link.
+struct LogicalLinkControlPacket : public UpperMacCPlaneSignallingPacket {
+    std::optional<BasicLinkInformation> basic_link_information_;
+    /// The data that is passed from the Logical Link Control layer to the Mobile Link Entity
+    BitVector tl_sdu_;
+
+    LogicalLinkControlPacket() = delete;
+
+    explicit LogicalLinkControlPacket(const UpperMacCPlaneSignallingPacket& packet);
+};
+
+auto operator<<(std::ostream& stream, const LogicalLinkControlPacket& llc) -> std::ostream&;