wip, udp port enable metadata in client, to:receiver, zmq streaming

slsDetectorSoftware/include/sls/Detector.h

@@ -913,6 +913,8 @@ class Detector {
     void setDataStream(const defs::portPosition port, const bool enable,
                        Positions pos = {});
 
+    /* list of possible port positions */
+    std::vector<defs::portPosition> getPortPositionList() const;
     ///@}
 
     /** @name Receiver Configuration */

slsDetectorSoftware/src/Detector.cpp

@@ -1090,6 +1090,7 @@ void Detector::setNumberofUDPInterfaces(int n, Positions pos) {
     }
     // also called by vetostream (for gotthard2)
     setNumberofUDPInterfaces_(n, pos);
+    pimpl->updateRxUDPDatastreamMetadata();
 }
 
 void Detector::setNumberofUDPInterfaces_(int n, Positions pos) {
@@ -1325,6 +1326,11 @@ Result<bool> Detector::getDataStream(const defs::portPosition port,
 void Detector::setDataStream(const defs::portPosition port, const bool enable,
                              Positions pos) {
     pimpl->Parallel(&Module::setDataStream, pos, port, enable);
+    pimpl->updateRxUDPDatastreamMetadata();
+}
+
+std::vector<defs::portPosition> Detector::getPortPositionList() const {
+    return pimpl->getPortPositionList();
 }
 
 // Receiver

slsDetectorSoftware/src/DetectorImpl.cpp

@@ -1644,6 +1644,42 @@ void DetectorImpl::verifyUniqueHost(
     }
 }
 
+std::vector<defs::portPosition> DetectorImpl::getPortPositionList() const {
+    switch (shm()->detType) {
+    case defs::JUNGFRAU:
+    case defs::MOENCH:
+        return std::vector<defs::portPosition>{defs::TOP, defs::BOTTOM};
+    case defs::EIGER:
+        return std::vector<defs::portPosition>{defs::LEFT, defs::RIGHT};
+    default:
+        throw RuntimeError("port Position does not exist for this detector");
+    }
+}
+
+void DetectorImpl::updateRxUDPDatastreamMetadata() {
+    auto detType = shm()->detType;
+    if (detType != defs::EIGER && detType != defs::JUNGFRAU &&
+        detType != defs::MOENCH)
+        throw RuntimeError(
+            "Datastream enable not implemented for this detector");
+    auto portList = getPortPositionList();
+    if (portList.size() != 2) {
+        throw RuntimeError("Invalid port size. Expected 2.");
+    }
+    std::array<std::vector<bool>, 2> results;
+    size_t i = 0;
+    for (const auto &port : portList) {
+        results[i] = Parallel(&Module::getDataStream, {}, port);
+        if (static_cast<int>(results[i].size()) != size()) {
+            throw RuntimeError("udp datastream enable list does not match size "
+                               "of module list");
+        }
+        ++i;
+    }
+
+    modules[0]->updateRxUDPDatastreamMetadata(results);
+}
+
 std::vector<defs::ROI> DetectorImpl::getRxROI(int module_id) const {
     if (shm()->detType == CHIPTESTBOARD ||
         shm()->detType == defs::XILINX_CHIPTESTBOARD) {

slsDetectorSoftware/src/DetectorImpl.h

@@ -310,6 +310,9 @@ class DetectorImpl : public virtual slsDetectorDefs {
     std::vector<std::pair<std::string, uint16_t>>
     verifyUniqueRxHost(const std::vector<std::string> &names) const;
 
+    std::vector<defs::portPosition> getPortPositionList() const;
+    void updateRxUDPDatastreamMetadata();
+
     defs::xy getPortGeometry() const;
     std::vector<defs::ROI> getRxROI(int module_id = -1) const;
     void setRxROI(const std::vector<defs::ROI> &args);

slsDetectorSoftware/src/Module.cpp

@@ -1434,6 +1434,41 @@ void Module::setDataStream(const portPosition port, const bool enable) {
     }
 }
 
+void Module::updateRxUDPDatastreamMetadata(
+    const std::array<std::vector<bool>, 2> &res) {
+    if (!shm()->useReceiverFlag) {
+        return;
+    }
+    LOG(logDEBUG) << "Updating UDP data stream enable in receiver (metadata)";
+    auto client = ReceiverSocket(shm()->rxHostname, shm()->rxTCPPort);
+
+    client.Send(F_RECEIVER_UDP_DATASTREAM_METADATA);
+    client.setFnum(F_RECEIVER_UDP_DATASTREAM_METADATA);
+    // ensure both port list size match
+    if (res[0].size() != res[1].size()) {
+        throw RuntimeError(
+            "Size Mismatch of udp datastream enable list for ports");
+    }
+    // send udp data stream enable of all ports in all modules
+    auto nports = static_cast<int>(res.size());
+    auto nmods = static_cast<int>(res[0].size());
+    client.Send(nports);
+    client.Send(nmods);
+    for (const auto &r : res) {
+        // convert vector of bool to vector of int to send
+        std::vector<int> tmp;
+        tmp.reserve(r.size());
+        for (bool b : r) {
+            tmp.push_back(static_cast<int>(b));
+        }
+        client.Send(tmp);
+    }
+    if (client.Receive<int>() == FAIL) {
+        throw ReceiverError("Receiver " + std::to_string(moduleIndex) +
+                            " returned error: " + client.readErrorMessage());
+    }
+}
+
 // Receiver Config
 
 bool Module::getUseReceiverFlag() const { return shm()->useReceiverFlag; }

slsDetectorSoftware/src/Module.h

@@ -281,6 +281,8 @@ class Module : public virtual slsDetectorDefs {
     void setTransmissionDelayRight(int value);
     bool getDataStream(const portPosition port) const;
     void setDataStream(const portPosition port, const bool enable);
+    void
+    updateRxUDPDatastreamMetadata(const std::array<std::vector<bool>, 2> &res);
 
     /**************************************************
      *                                                *

slsReceiverSoftware/src/Implementation.cpp

@@ -1095,6 +1095,10 @@ void Implementation::setNumberofUDPInterfaces(const int n) {
         // number of portrois should be equal to number of interfaces
         ResetRois();
 
+        // reset udp data stream
+        udpDataStream[0] = true;
+        udpDataStream[1] = true;
+
         // create threads
         for (int i = 0; i < generalData->numUDPInterfaces; ++i) {
             // listener and dataprocessor threads

slsSupportLib/include/sls/sls_detector_funcs.h

@@ -424,6 +424,7 @@ enum detFuncs {
     F_RECEIVER_GET_ROI_METADATA,
     F_SET_RECEIVER_READOUT_SPEED,
     F_RECEIVER_GET_UDP_DATASTREAM,
+    F_RECEIVER_UDP_DATASTREAM_METADATA,
 
     NUM_REC_FUNCTIONS
 };
@@ -844,6 +845,8 @@ const char* getFunctionNameFromEnum(enum detFuncs func) {
     case F_RECEIVER_GET_ROI_METADATA:       return "F_RECEIVER_GET_ROI_METADATA";
     case F_SET_RECEIVER_READOUT_SPEED:      return "F_SET_RECEIVER_READOUT_SPEED";
     case F_RECEIVER_GET_UDP_DATASTREAM:     return "F_RECEIVER_GET_UDP_DATASTREAM";
+    case F_RECEIVER_UDP_DATASTREAM_METADATA:return "F_RECEIVER_UDP_DATASTREAM_METADATA";
+
 
     case NUM_REC_FUNCTIONS: 				return "NUM_REC_FUNCTIONS";
 	default:								return "Unknown Function";