mx/Jungfraujoch
3
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 81 Wiki Activity

tools: run the broker on a stored dataset, without a detector

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SVmAWnzCmRKAXVUCdc4iNi
This commit is contained in:
leonarski_f
2026-10-08 15:52:22 +02:00
co-authored by Claude Opus 5.5
parent 21f30549a7
commit 8e134c405d
3 changed files with 190 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/CMakeLists.txt
+5
View File
@@ -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)
tools/fake_simplon.py
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
# 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()
tools/jfjoch_stream2_replay.cpp
+96
View File
@@ -0,0 +1,96 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// 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 <chrono>
#include <iostream>
#include <thread>
#include <getopt.h>
#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 <images per s>] [-n <images>] [-a <address>] <input master>" << 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<int64_t> 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<int64_t>(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<uint8_t> 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<double>(static_cast<double>(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<double>(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;
}