From 8e134c405d7cfff42eecfb8b4845650170cff5f5 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Thu, 8 Oct 2026 15:52:22 +0200 Subject: [PATCH] tools: run the broker on a stored dataset, without a detector For demonstrating and testing a collection followed while it is written: - jfjoch_stream2_replay replays a stored dataset as a DECTRIS detector's stream2 (ZeroMQ PUSH on the stream2 port): start message, every image at a given rate, end message. - fake_simplon.py stands in for the detector's SIMPLON control interface: answers the configuration and status requests the broker makes, accepts every command, reads back idle. It needs port 80. Not installed. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01SVmAWnzCmRKAXVUCdc4iNi --- tools/CMakeLists.txt | 5 ++ tools/fake_simplon.py | 89 ++++++++++++++++++++++++++++++ tools/jfjoch_stream2_replay.cpp | 96 +++++++++++++++++++++++++++++++++ 3 files changed, 190 insertions(+) create mode 100644 tools/fake_simplon.py create mode 100644 tools/jfjoch_stream2_replay.cpp diff --git a/tools/CMakeLists.txt b/tools/CMakeLists.txt index 310097f76..b6754ec31 100644 --- a/tools/CMakeLists.txt +++ b/tools/CMakeLists.txt @@ -32,6 +32,11 @@ IF(NOT JFJOCH_PORTABLE_ONLY) TARGET_LINK_LIBRARIES(jfjoch_hdf5_test JFJochWriter JFJochReceiver JFJochCommon) INSTALL(TARGETS jfjoch_hdf5_test RUNTIME COMPONENT jfjoch) + # Not installed: replays a stored dataset as a DECTRIS detector's image stream, to run the broker + # without a detector (demonstrations and tests). + ADD_EXECUTABLE(jfjoch_stream2_replay jfjoch_stream2_replay.cpp) + TARGET_LINK_LIBRARIES(jfjoch_stream2_replay JFJochReader JFJochZMQ CBORStream2FrameSerialize JFJochCommon) + ADD_EXECUTABLE(jfjoch_pcie_status jfjoch_pcie_status.cpp) TARGET_LINK_LIBRARIES(jfjoch_pcie_status JFJochDevice JFJochCommon) INSTALL(TARGETS jfjoch_pcie_status RUNTIME COMPONENT jfjoch) diff --git a/tools/fake_simplon.py b/tools/fake_simplon.py new file mode 100644 index 000000000..758fcc5dc --- /dev/null +++ b/tools/fake_simplon.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +# SPDX-License-Identifier: GPL-3.0-only +"""A stand-in for a DECTRIS detector's SIMPLON control interface, for running jfjoch_broker without a +detector (demonstrations and tests). It answers the configuration and status requests the broker makes, +accepts every command, and always reads back as idle. The images come from tools/jfjoch_stream2_replay, +which plays the detector's stream2. + + fake_simplon.py --width 2068 --height 2164 --pixel-size 75e-6 [--port 80] + +The broker reaches SIMPLON on port 80, so this has to be able to bind it (e.g. run it as root in a +container on the host network). +""" + +import argparse +import base64 +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--width", type=int, required=True) + p.add_argument("--height", type=int, required=True) + p.add_argument("--pixel-size", type=float, required=True, help="metres") + p.add_argument("--sensor-thickness", type=float, default=320e-6, help="metres") + p.add_argument("--sensor-material", default="Si") + p.add_argument("--bit-depth", type=int, default=32) + p.add_argument("--description", default="Replayed detector") + p.add_argument("--port", type=int, default=80) + a = p.parse_args() + + mask = base64.b64encode(bytes(4 * a.width * a.height)).decode() + config = { + "description": {"value": a.description}, + "detector_number": {"value": "replay-1"}, + "x_pixel_size": {"value": a.pixel_size}, + "y_pixel_size": {"value": a.pixel_size}, + "sensor_material": {"value": a.sensor_material}, + "sensor_thickness": {"value": a.sensor_thickness}, + "bit_depth_image": {"value": a.bit_depth}, + "x_pixels_in_detector": {"value": a.width}, + "y_pixels_in_detector": {"value": a.height}, + "frame_time": {"value": 0.01, "min": 0.0001}, + "count_time": {"value": 0.01, "min": 0.0001}, + "detector_readout_time": {"value": 0.0}, + "threshold_energy": {"value": 4000.0, "min": 1000.0}, + "photon_energy": {"value": 8000.0}, + "nimages": {"value": 1}, + "ntrigger": {"value": 1}, + "pixel_mask": {"value": {"shape": [a.height, a.width], "type": "uint32", "data": mask}}, + } + status = {"state": {"value": "idle"}, "humidity": {"value": 0.0}, "temperature": {"value": 25.0}} + + class Handler(BaseHTTPRequestHandler): + def reply(self, body): + data = json.dumps(body).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self): + parts = self.path.strip("/").split("/") # module/api/version/task/key + key = parts[-1] + if len(parts) >= 5 and parts[3] == "status": + self.reply(status.get(key, {"value": 0})) + else: + self.reply(config.get(key, {"value": 0})) + + def do_POST(self): + parts = self.path.strip("/").split("/") + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) if length else b"" + if len(parts) >= 5 and parts[3] == "config" and body: + config.setdefault(parts[-1], {})["value"] = json.loads(body)["value"] + self.reply([]) + + do_PUT = do_POST + + def log_message(self, fmt, *args): + pass + + ThreadingHTTPServer(("0.0.0.0", a.port), Handler).serve_forever() + + +if __name__ == "__main__": + main() diff --git a/tools/jfjoch_stream2_replay.cpp b/tools/jfjoch_stream2_replay.cpp new file mode 100644 index 000000000..0cd97277b --- /dev/null +++ b/tools/jfjoch_stream2_replay.cpp @@ -0,0 +1,96 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +// Replays a stored dataset as the image stream of a DECTRIS detector: binds a ZeroMQ PUSH socket where +// the broker's DECTRIS receiver connects (tcp://*:31001) and sends the start message, every image at +// the given rate, and the end message. With a stand-in for the detector's control interface, the broker +// then runs a real collection - real geometry, real diffraction - with no detector. For demonstrations +// and tests only. + +#include +#include +#include +#include + +#include "../common/Logger.h" +#include "../common/ZMQWrappers.h" +#include "../frame_serialize/CBORStream2Serializer.h" +#include "../reader/JFJochHDF5Reader.h" + +void print_usage() { + std::cout << "Usage: jfjoch_stream2_replay [-r ] [-n ] [-a
] " << std::endl; + std::cout << " -r images sent per second (default 100)" << std::endl; + std::cout << " -n images to send (default all)" << std::endl; + std::cout << " -a ZeroMQ address to bind (default tcp://0.0.0.0:31001, the DECTRIS stream2 port)" << std::endl; +} + +int main(int argc, char **argv) { + Logger logger("jfjoch_stream2_replay"); + RegisterHDF5Filter(); + + double rate = 100.0; + std::optional images; + std::string address = "tcp://0.0.0.0:31001"; + + int opt; + while ((opt = getopt(argc, argv, "r:n:a:")) != -1) { + switch (opt) { + case 'r': rate = atof(optarg); break; + case 'n': images = atoll(optarg); break; + case 'a': address = optarg; break; + default: print_usage(); return EXIT_FAILURE; + } + } + if (optind + 1 != argc || rate <= 0.0) { + print_usage(); + return EXIT_FAILURE; + } + + JFJochHDF5Reader reader; + reader.ReadFile(argv[optind]); + const auto dataset = reader.GetDataset(); + const int64_t n = std::min(images.value_or(reader.GetNumberOfImages()), reader.GetNumberOfImages()); + + // What a DECTRIS detector states about itself in its start message; the broker takes the + // geometry of the collection from its own /start. + DiffractionExperiment x(dataset->experiment); + x.ImagesPerTrigger(n).NumTriggers(1); + StartMessage start; + x.FillMessage(start); + const auto stored = reader.GetStoredPixelFormat(); + start.bit_depth_image = stored.bit_depth; + start.pixel_signed = stored.is_signed; + start.number_of_images = n; + + ZMQSocket socket(ZMQSocketType::Push); + socket.SendWaterMark(1000); + socket.Bind(address); + logger.Info("Bound {}; sending {} images at {} Hz", address, n, rate); + + std::vector buffer(256 * 1024 * 1024); + CBORStream2Serializer serializer(buffer.data(), buffer.size()); + serializer.SerializeSequenceStart(start); + socket.Send(buffer.data(), serializer.GetBufferSize()); + + const auto t0 = std::chrono::steady_clock::now(); + JFJochReaderRawImage raw; + for (int64_t i = 0; i < n; i++) { + std::this_thread::sleep_until(t0 + std::chrono::duration(static_cast(i) / rate)); + reader.ReadRawImage(i, raw); + DataMessage message{}; + message.image = raw.image; + message.number = i; + serializer.SerializeImage(message); + socket.Send(buffer.data(), serializer.GetBufferSize()); + } + + EndMessage end{}; + end.max_image_number = n; + serializer.SerializeSequenceEnd(end); + socket.Send(buffer.data(), serializer.GetBufferSize()); + logger.Info("Sent {} images in {:.1f} s", n, + std::chrono::duration(std::chrono::steady_clock::now() - t0).count()); + // Let the end message leave before the socket closes. + std::this_thread::sleep_for(std::chrono::seconds(2)); + return EXIT_SUCCESS; +}