Add opencmw source block and rda3 client

.gitattributes

@@ -0,0 +1 @@
+*.jar filter=lfs diff=lfs merge=lfs -text

.restyled.yaml

@@ -19,6 +19,9 @@ ignore_labels:
 
 restylers_version: stable
 
+also_exclude:
+  - "**/*.jar"
+
 restylers:
   - clang-format:
       image:

CMakeLists.txt

@@ -13,6 +13,12 @@ if(NOT DEFINED GR_DIGITIZERS_TOPLEVEL_PROJECT)
   endif()
 endif()
 
+if(NOT EMSCRIPTEN)
+  option(ENABLE_PICOSCOPE "Enable PicoScope support" ON)
+  option(ENABLE_TIMING "Enable TimingReceiver support" ON)
+  option(ENABLE_OPENCMW "Enable OpenCmw support" ON)
+endif()
+
 if(CMAKE_CXX_COMPILER_ID MATCHES ".*Clang") # set default C++ STL to Clang's
                                             # libc++ when using Clang
   add_compile_options(-stdlib=libc++ -fcolor-diagnostics)
@@ -32,12 +38,25 @@ FetchContent_Declare(
   GIT_REPOSITORY https://github.com/fair-acc/gnuradio4.git
   GIT_TAG d14bb560909aaa151b13a535ad4c7f98ad3e9959 # main as of 2026-02-24
   EXCLUDE_FROM_ALL)
+list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST gnuradio4)
+
+if(ENABLE_GR_DIGITIZERS_TESTING)
+  FetchContent_Declare(
+    ut
+    GIT_REPOSITORY https://github.com/boost-ext/ut.git
+    GIT_TAG v2.3.1 # latest tag as of 2025-04-02
+    EXCLUDE_FROM_ALL)
+  list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST ut)
+endif()
 
-FetchContent_Declare(
-  ut
-  GIT_REPOSITORY https://github.com/boost-ext/ut.git
-  GIT_TAG v2.3.1 # latest tag as of 2025-04-02
-  EXCLUDE_FROM_ALL)
+if(ENABLE_OPENCMW)
+  FetchContent_Declare(
+    opencmw-cpp
+    GIT_REPOSITORY https://github.com/fair-acc/opencmw-cpp.git
+    GIT_TAG e30b78193052bf2ea39160a8ec913044552fe103 # latest as of 2026-03-19
+    EXCLUDE_FROM_ALL)
+  list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST "opencmw-cpp")
+endif()
 
 if(CMAKE_CXX_COMPILER_ID MATCHES "(Clang|GNU)")
   if(ADDRESS_SANITIZER)
@@ -68,7 +87,7 @@ include(cmake/CompilerWarnings.cmake)
 set_project_warnings(gr-digitizers-options)
 
 if(EMSCRIPTEN)
-  FetchContent_MakeAvailable(gnuradio4 ut)
+  FetchContent_MakeAvailable(${GR_DIGITIZER_FETCH_CONTENT_LIST})
   set(CMAKE_EXECUTABLE_SUFFIX ".js")
   target_compile_options(gr-digitizers-options INTERFACE -fwasm-exceptions
                                                          -pthread)
@@ -87,14 +106,16 @@ else()
       imgui
       GIT_REPOSITORY https://github.com/ocornut/imgui.git
       GIT_TAG v1.91.9b-docking)
+    list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST imgui)
 
     # Enables 32 bit vertex indices for ImGui
     add_compile_definitions("ImDrawIdx=unsigned int")
     FetchContent_Declare(
       implot
       GIT_REPOSITORY https://github.com/epezent/implot.git
       GIT_TAG v0.16)
-    FetchContent_MakeAvailable(imgui implot gnuradio4 ut)
+    list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST implot)
+    FetchContent_MakeAvailable(${GR_DIGITIZER_FETCH_CONTENT_LIST})
 
     find_package(SDL3 REQUIRED)
     find_package(OpenGL REQUIRED COMPONENTS OpenGL)
@@ -137,11 +158,6 @@ if(CMAKE_CXX_COMPILER MATCHES "/em\\+\\+(-[a-zA-Z0-9.])?$") # if this hasn't
   set(EMSCRIPTEN true)
 endif()
 
-if(NOT EMSCRIPTEN)
-  option(ENABLE_PICOSCOPE "Enable PicoScope support" ON)
-  option(ENABLE_TIMING "Enable TimingReceiver support" ON)
-endif()
-
 find_package(PkgConfig REQUIRED)
 
 if(ENABLE_PICOSCOPE)

blocklib/CMakeLists.txt

@@ -5,3 +5,7 @@ endif()
 if(ENABLE_TIMING)
   add_subdirectory(timing)
 endif()
+
+if(ENABLE_OPENCMW)
+  add_subdirectory(opencmw)
+endif()

blocklib/opencmw/CMakeLists.txt

@@ -0,0 +1,16 @@
+add_library(fair-opencmw INTERFACE)
+target_include_directories(
+  fair-opencmw INTERFACE $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+                         $<INSTALL_INTERFACE:include/>)
+
+target_link_libraries(fair-opencmw INTERFACE gr-digitizers-options client)
+set_target_properties(
+  fair-opencmw
+  PROPERTIES
+    PUBLIC_HEADER
+    "include/fair/opencmw/cmwlight/DirectoryLightClient.hpp;include/fair/opencmw/cmwlight/CmwLightClient.hpp;include/fair/opencmw/OpenCmwSource.hpp"
+)
+
+if(ENABLE_GR_DIGITIZERS_TESTING)
+  add_subdirectory(test)
+endif()

blocklib/opencmw/include/fair/opencmw/OpenCmwSource.hpp

@@ -0,0 +1,164 @@
+#ifndef GR_DIGITIZERS_OPENCMWSOURCE_HPP
+#define GR_DIGITIZERS_OPENCMWSOURCE_HPP
+
+#include <Client.hpp>
+#include <ClientContext.hpp>
+#include <RestClient.hpp>
+#include <URI.hpp>
+#include <fair/opencmw/cmwlight/CmwLightClient.hpp>
+#include <gnuradio-4.0/Block.hpp>
+#include <gnuradio-4.0/BlockRegistry.hpp>
+
+namespace fair::opencmw {
+static ::opencmw::client::ClientContext initGlobalClientContext(::opencmw::zmq::Context& zctx) {
+    using namespace ::opencmw::client;
+    std::string nameserver = "http://localhost:7500";
+    if (const auto ptr = std::getenv("CMW_NAMESERVER"); ptr != nullptr) {
+        nameserver = std::string{ptr};
+    }
+    std::vector<std::unique_ptr<ClientBase>> clients{};
+    clients.emplace_back(std::make_unique<RestClient>());
+    clients.emplace_back(std::make_unique<MDClientCtx>(zctx, 100ms, "OpenCmwMajordomoClient"));
+    clients.emplace_back(std::make_unique<cmwlight::CmwLightClientCtx>(zctx, nameserver, 100ms, "OpenCmwRda3Client"));
+    return ClientContext{std::move(clients)};
+}
+
+static ::opencmw::client::ClientContext& getGloablClientContext() {
+    static ::opencmw::zmq::Context          zctx{};
+    static ::opencmw::client::ClientContext clientContext = initGlobalClientContext(zctx);
+    return clientContext;
+}
+
+struct OpenCmwSource : gr::Block<OpenCmwSource> {
+    using FloatLimits = gr::Limits<std::numeric_limits<float>::lowest(), std::numeric_limits<float>::max()>;
+    gr::PortOut<gr::pmt::Value> out;
+
+    gr::Annotated<std::string, "url", gr::Visible, gr::Doc<"URL of the resource to subscribe to">>                     url;
+    gr::Annotated<std::string, "signal name", gr::Visible, gr::Doc<"Identifier for the signal">>                       signal_name;
+    gr::Annotated<std::string, "signal quantity", gr::Visible, gr::Doc<"Physical quantity represented by the signal">> signal_quantity;
+    gr::Annotated<std::string, "signal unit", gr::Visible, gr::Doc<"Unit of measurement for the signal values">>       signal_unit;
+    gr::Annotated<float, "signal min", gr::Doc<"Minimum expected value for the signal">, FloatLimits>                  signal_min        = std::numeric_limits<float>::lowest();
+    gr::Annotated<float, "signal max", gr::Doc<"Maximum expected value for the signal">, FloatLimits>                  signal_max        = std::numeric_limits<float>::max();
+    gr::Annotated<bool, "verbose console", gr::Doc<"For debugging">>                                                   verbose_console   = false;
+    gr::Annotated<float, "reconnect timeout", gr::Doc<"reconnect timeout in sec">>                                     reconnect_timeout = 5.f;
+    gr::Annotated<std::string, "nameserver", gr::Doc<"Nameserver to use for device lookup">>                           nameserver        = "http://localhost:7500";
+
+    GR_MAKE_REFLECTABLE(OpenCmwSource, out, url, signal_name, signal_unit, signal_quantity, signal_min, signal_max, verbose_console, reconnect_timeout);
+
+    std::mutex                 mutex;
+    std::queue<gr::pmt::Value> updates;
+
+    void start() {
+        if (verbose_console) {
+            std::println("OpenCmwSource: subscribe to {}", url);
+        }
+        auto& clientContext = getGloablClientContext();
+        clientContext.subscribe(::opencmw::URI(url), [this](const ::opencmw::mdp::Message& response) {
+            if (verbose_console) {
+                std::println("OpenCmwSource: received update");
+            }
+            auto res    = gr::property_map{};
+            auto header = gr::property_map{};
+            header.try_emplace("url", std::pmr::string{url});
+            if (response.error.empty()) {
+                using AcquisitionMap = std::map<std::string, std::variant<bool, float, double, int8_t, int16_t, int32_t, int64_t, std::string, std::vector<bool>, std::vector<float>, std::vector<double>, std::vector<int8_t>, std::vector<int16_t>, std::vector<int32_t>, std::vector<int64_t>, std::vector<std::string>>>;
+                ::opencmw::IoBuffer             buffer(response.data);
+                AcquisitionMap                  acqMap{};
+                ::opencmw::DeserialiserInfo     info;
+                ::opencmw::FieldDescriptionLong field;
+                ::opencmw::FieldHeaderReader<::opencmw::CmwLight>::get<::opencmw::ProtocolCheck::LENIENT>(buffer, info, field);
+                ::opencmw::IoSerialiser<::opencmw::CmwLight, AcquisitionMap>::deserialise(buffer, field, acqMap);
+                if (verbose_console) {
+                    std::println("Deserialised subscription reply:\n  fields: {}\n  fieldTypes: {}\n  errors: {}\n  mapFields: {}", info.additionalFields | std::views::keys, info.additionalFields | std::views::values, info.exceptions | std::views::transform([](auto& e) { return e.what(); }), acqMap | std::views::keys);
+                }
+                auto val = gr::property_map{};
+                for (auto& [key, value] : acqMap) {
+                    std::visit([&](auto&& v) { val.insert(std::make_pair(key, gr::pmt::Value(v))); }, value);
+                }
+                res.try_emplace("data", val);
+                auto errors = gr::pmt::Value{info.exceptions | std::views::transform([](auto& e) { return e.what(); }) | std::ranges::to<std::vector<std::string>>()};
+                res.try_emplace("errors", errors);
+            } else {
+                if (verbose_console) {
+                    std::println("error in subscription: {}", response.error);
+                }
+                auto errors = gr::pmt::Value{response.error};
+                res.try_emplace("errors", errors);
+            }
+            res.try_emplace("header", header);
+            std::lock_guard guard{mutex};
+            updates.push(std::move(res));
+            this->progress->incrementAndGet();
+        });
+    }
+
+    void stop() {
+        auto& clientContext = getGloablClientContext();
+        if (verbose_console) {
+            std::println("OpenCmwSource: unsubscribe from {}", url);
+        }
+        clientContext.unsubscribe(::opencmw::URI(url));
+    }
+
+    auto processBulk(gr::OutputSpanLike auto& output) noexcept {
+        size_t          written = 0;
+        std::lock_guard guard{mutex};
+        while (written < output.size() && !updates.empty()) {
+            if (verbose_console) {
+                std::println("OpenCmwSource: publish update(s): {}", updates.front());
+            }
+            *(output.begin() + written) = updates.front();
+            ++written;
+            updates.pop();
+        }
+        output.publish(written);
+        // todo: propagate metadata
+        return written == 0UZ ? gr::work::Status::INSUFFICIENT_INPUT_ITEMS : gr::work::Status::OK;
+    }
+};
+
+template<typename T>
+struct ExtractFromMap : gr::Block<ExtractFromMap<T>> {
+    gr::Annotated<std::string, "fieldname", gr::Visible, gr::Doc<"Field to extract and publish from the map">> fieldname;
+
+    gr::PortIn<gr::pmt::Value> in;
+    gr::PortOut<T>             out;
+
+    GR_MAKE_REFLECTABLE(ExtractFromMap, in, out, fieldname);
+
+    [[nodiscard]] constexpr gr::work::Status processBulk(gr::InputSpanLike auto& input, gr::OutputSpanLike auto& output) noexcept {
+        std::size_t nIn  = 0;
+        std::size_t nOut = 0;
+        for (const auto& map : input) {
+            if (const gr::pmt::Value::Map* dataPtr = map.template get_if<gr::pmt::Value::Map>(); dataPtr != nullptr && dataPtr->contains("data")) {
+                if (const auto* mapPtr = dataPtr->at("data").get_if<gr::pmt::Value::Map>(); mapPtr != nullptr && mapPtr->contains(fieldname)) {
+                    if (const T* valuePtr = mapPtr->at(std::pmr::string{fieldname.value}).get_if<T>(); valuePtr != nullptr) {
+                        if (nOut >= output.size()) {
+                            break;
+                        }
+                        output[nOut] = *valuePtr;
+                        ++nOut;
+                    } else if (const gr::Tensor<T>* tensorPtr = mapPtr->at(std::pmr::string{fieldname.value}).get_if<gr::Tensor<T>>(); tensorPtr != nullptr) {
+                        if (nOut + tensorPtr->size() >= output.size()) {
+                            break;
+                        }
+                        std::copy(tensorPtr->begin(), tensorPtr->end(), output.begin() + nOut);
+                        nOut += tensorPtr->size();
+                    } else {
+                        std::println("field {1} exists, but has unsupported field type: {2}, supported: {0} or Tensor<{0}>", gr::meta::type_name<T>(), fieldname.value, magic_enum::enum_name<>(magic_enum::enum_cast<gr::pmt::Value::ValueType>(mapPtr->at(std::pmr::string{fieldname.value})._value_type).value_or(gr::pmt::Value::ValueType::Monostate)));
+                    }
+                }
+            }
+            ++nIn;
+        }
+        output.publish(nOut);
+        input.consume(nIn);
+        return nOut > 0 ? gr::work::Status::OK : (nIn > 0 ? gr::work::Status::INSUFFICIENT_OUTPUT_ITEMS : gr::work::Status::INSUFFICIENT_INPUT_ITEMS);
+    }
+};
+
+} // namespace fair::opencmw
+
+inline auto registerRemoteStreamSource = ::gr::registerBlock<fair::opencmw::OpenCmwSource>(::gr::globalBlockRegistry());
+
+#endif // GR_DIGITIZERS_OPENCMWSOURCE_HPP