Dev/rx disable datastream port (#1448)

- changed - command from `datastream` to `udp_datastream` to be more specific - Detector API: - setNumberofUDPInterfaces(pos) =>setNumberofUDPInterfaces() # no position anymore - getDataStream(portPosition, pos) =>getUDPDataStream(portPosition, pos) - setDataStream(portPosition, pos) => setUDPDataStream(portPosition) # no position anymore - modified the dataprocessing thread: - if gui or call back, start zmq processing (connecting sockets) before starting receiver (disabled ports sends dummy that the client zmq sockets are not ready for) - if only progress, starting progress processing thread after startReceiver - - added - Detector API: - getRxDisabledUDPPortIndices() - getPortPositionList() - allow disabling/enabling udp interface in receiver for Jungfrua/Moench when both udp interfaces is enabled (This is to go fast, where only one half of the module matters) - write ports type and disabled ports to master file (tests for it as well) - tests for python commands as well - datastream command to deprecated commands - refactored tests to make the new parameters fit * first draft of disabling data port at the receiver side. Todo: master file and stream for gui to use * updated help in command * formatting * md5 unchanged * fix test, check port position for detector type * wip, udp port enable metadata in client, to:receiver, zmq streaming * wip, master attributes * wip, imple=> save only disabled ports with port index, 1 interface, is empty * works in file. still needs refactoring * updated writer versions * udp ports type pass, disabled with api yet * fixed disabled ports meta data as well * fixed in developer via other PR * renamed.more intuitive * fix in gui that wont wait for disabled ports * added tests for master file * extra tests * move the master fiel tests into its own tests with the disable marker for github tests * refactoring * python test * more intuitive and doc * release notes * example for python * back to before * minor documentation fix and sending nports unnecessarily for getRxUDPPortDisableMetadata * doc * review fixes, 1. using std::array instead of vector, 2. moving udp_datastream command testing with different values into master attributes testing file because we test the master file * minor * review: update help * progress printed before assigned * try catch inside hdf5 * auto generated code for CLI and formatting * missing packets should not be checked at github workflow level * formatting * improved python doc for udp_datastream help * indent doc python done
2026-08-07 00:22:25 +02:00 · 2026-07-13 14:58:25 +02:00
parent 9cbf588217
commit c3ad3d2e73
48 changed files with 1269 additions and 142431 deletions
@@ -1126,10 +1126,17 @@ int DetectorImpl::acquire() {

        // start receiver
        if (receiver) {
-            Parallel(&Module::startReceiver, {});
-        }

-        startProcessingThread(receiver);
+            // to catch dummy zmq packets for disabled ports,
+            // start processing thread before startReceiver
+            if (dataReady != nullptr)
+                startRxZmqProcessingThread();
+
+            Parallel(&Module::startReceiver, {});
+
+            if (dataReady == nullptr)
+                startRxProgressThread();
+        }

        // start and read all
        try {
@@ -1314,53 +1321,46 @@ void DetectorImpl::printProgress(double progress) {
    std::cout << '\r' << std::flush;
 }

-void DetectorImpl::startProcessingThread(bool receiver) {
+void DetectorImpl::startRxZmqProcessingThread() {
    dataProcessingThread =
-        std::thread(&DetectorImpl::processData, this, receiver);
+        std::thread(&DetectorImpl::readFrameFromReceiver, this);
 }

-void DetectorImpl::processData(bool receiver) {
-    if (receiver) {
-        if (dataReady != nullptr) {
-            readFrameFromReceiver();
-        }
-        // only update progress
-        else {
-            LOG(logINFO) << "Type 'q' and hit enter to stop acquisition";
-            double progress = 0;
-            printProgress(progress);
+void DetectorImpl::startRxProgressThread() {
+    dataProcessingThread = std::thread(&DetectorImpl::printRxProgress, this);
+}

-            while (true) {
-                // to exit acquire by typing q
-                if (kbhit() != 0) {
-                    if (fgetc(stdin) == 'q') {
-                        LOG(logINFO)
-                            << "Caught the command to stop acquisition";
-                        stopDetector({});
-                    }
-                }
-                // get and print progress
-                double temp =
-                    (double)Parallel(&Module::getReceiverProgress, {0})
-                        .squash();
-                if (temp != progress) {
-                    printProgress(progress);
-                    progress = temp;
-                }
+void DetectorImpl::printRxProgress() {
+    LOG(logINFO) << "Type 'q' and hit enter to stop acquisition";
+    double progress = 0;
+    printProgress(progress);

-                // exiting loop
-                if (getJoinThreadFlag()) {
-                    // print progress one final time before exiting
-                    progress =
-                        (double)Parallel(&Module::getReceiverProgress, {0})
-                            .squash();
-                    printProgress(progress);
-                    break;
-                }
-                // otherwise error when connecting to the receiver too fast
-                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    while (true) {
+        // to exit acquire by typing q
+        if (kbhit() != 0) {
+            if (fgetc(stdin) == 'q') {
+                LOG(logINFO) << "Caught the command to stop acquisition";
+                stopDetector({});
            }
        }
+        // get and print progress
+        double temp =
+            (double)Parallel(&Module::getReceiverProgress, {0}).squash();
+        if (temp != progress) {
+            progress = temp;
+            printProgress(progress);
+        }
+
+        // exiting loop
+        if (getJoinThreadFlag()) {
+            // print progress one final time before exiting
+            progress =
+                (double)Parallel(&Module::getReceiverProgress, {0}).squash();
+            printProgress(progress);
+            break;
+        }
+        // otherwise error when connecting to the receiver too fast
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
 }

@@ -1644,6 +1644,77 @@ void DetectorImpl::verifyUniqueHost(
    }
 }

+void DetectorImpl::assertTwoUDPDataInterfaces(const std::string &cmd) const {
+    // assert globally
+    auto numInterfaces =
+        Parallel(&Module::getNumberofUDPInterfacesFromShm, {})
+            .tsquash("Inconsistent number of UDP interfaces among modules");
+    if (numInterfaces != 2) {
+        throw RuntimeError(
+            "Cannot " + cmd +
+            ". Change number of udp interfaces to 2 (cmd = numinterfaces).");
+    }
+}
+
+Result<bool> DetectorImpl::getUDPDataStream(const defs::portPosition port,
+                                            Positions pos) const {
+    assertTwoUDPDataInterfaces("get enable/disable UDP ports");
+    return Parallel(&Module::getUDPDataStream, pos, port);
+}
+
+void DetectorImpl::setUDPDataStream(const defs::portPosition port,
+                                    const bool enable, Positions pos) {
+    assertTwoUDPDataInterfaces("set enable/disable UDP ports");
+    Parallel(&Module::setUDPDataStream, pos, port, enable);
+    updateRxUDPDatastreamMetadata();
+}
+
+void DetectorImpl::updateRxUDPDatastreamMetadata() {
+    assertTwoUDPDataInterfaces(
+        "update Disbaled UDP ports metadata in receiver");
+
+    std::vector<int> disable;
+    auto portList = getPortPositionList();
+    if (portList.size() != 2) {
+        throw RuntimeError("Invalid port size. Expected 2.");
+    }
+    // bottom and left is port 0
+    auto port0 = Parallel(&Module::getUDPDataStream, {}, portList[0]);
+    auto port1 = Parallel(&Module::getUDPDataStream, {}, portList[1]);
+
+    // if any of them are disabled
+    if (port0.any(false) || port1.any(false)) {
+        // for each module: if disabled, push port index
+        for (size_t i = 0; i != port0.size(); ++i) {
+            if (!port0[i]) {
+                disable.push_back(i * 2);
+            }
+            if (!port1[i]) {
+                disable.push_back(i * 2 + 1);
+            }
+        }
+    }
+
+    modules[0]->updateRxUDPPortDisableMetadata(disable);
+}
+
+std::vector<int> DetectorImpl::getRxDisabledUDPPortIndices() const {
+    assertTwoUDPDataInterfaces("get Disbaled UDP ports metadata from receiver");
+    return modules[0]->getRxUDPPortDisableMetadata();
+}
+
+std::vector<defs::portPosition> DetectorImpl::getPortPositionList() const {
+    switch (shm()->detType) {
+    case defs::JUNGFRAU:
+    case defs::MOENCH:
+        return std::vector<defs::portPosition>{defs::BOTTOM, defs::TOP};
+    case defs::EIGER:
+        return std::vector<defs::portPosition>{defs::LEFT, defs::RIGHT};
+    default:
+        throw RuntimeError("port Position does not exist for this detector");
+    }
+}
+
 std::vector<defs::ROI> DetectorImpl::getRxROI(int module_id) const {
    if (shm()->detType == CHIPTESTBOARD ||
        shm()->detType == defs::XILINX_CHIPTESTBOARD) {