Files
Jungfraujoch/broker/JFJochBrokerParser.cpp
T
leonarski_fandClaude Opus 4.8 4cda46c6b0 Wire BraggIntegrationEngine into the pipeline; deterministic prediction; integration_model API
Replace the free functions BraggIntegrate2D/ProfileIntegrate2D with the
BraggIntegrationEngine (CPU/GPU) as the live integrator.

- IndexAndRefine no longer holds the integrator: ProcessImage takes a
  per-worker BraggIntegrateFn callback (ProcessImage is called concurrently by
  the shared IndexAndRefine, so the stateful engine must not be a member).
- WithoutFPGA/jfjoch_process: owns a GPU engine when a GPU is present, else CPU,
  and passes the GPU-resident preprocessed buffer so integration runs on-device.
- AfterFPGA: forces CPU and integrates straight off the assembled CompressedImage
  via a templated per-pixel sampler - only the reflection-disk pixels are read,
  no whole-image copy (the FPGA host runs up to 36 GB/s). Sampler maps type
  min/max to INT32_MIN/INT32_MAX on read; special/saturation only, no +/-1 band.
- Remove BraggIntegrate2D/ProfileIntegrate2D and their test; keep IntegratorMode.

Prediction: buffer up to 20000 candidates but return the 10000 closest to the
Ewald sphere (deterministic partial_sort on |dist_ewald|, hkl tiebreak) instead
of the GPU atomic-fill order. Serialized output stays <=10000, so the frame
transport headroom and its CBOR guard are unchanged.

integration_model exposed via OpenAPI (bragg_integration_settings schema,
/config/bragg_integration PUT/GET, added to jfjoch_settings and jfjoch_statistics)
and the frontend (BraggIntegrationSettings dropdown). Regenerated C++/TS clients
and redoc.

Validated old-vs-new on all 18 /data/rotation_test crystals: indexing rate and
space group bit-identical; ISa/CC identical on 16/18 (one improved, EcwtAL500
ISa 0.0->6.7); new CompressedImage-vs-buffer and GPU-vs-CPU parity tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-03 14:35:20 +02:00

275 lines
12 KiB
C++

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#include "JFJochBrokerParser.h"
#include "../common/NetworkAddressConvert.h"
#include "../image_pusher/ZMQStream2Pusher.h"
#include "../image_pusher/CBORFilePusher.h"
#include "../image_pusher/HDF5FilePusher.h"
#include "OpenAPIConvert.h"
#include "Detector_type.h"
#include "../image_pusher/NonePusher.h"
#include "../image_pusher/TCPStreamPusher.h"
DetectorGeometryModular ParseStandardDetectorGeometry(const org::openapitools::server::model::Detector &j) {
auto s = j.getStandardGeometry();
return DetectorGeometryModular(s.getNmodules(), s.getModulesInRow(), s.getGapX(), s.getGapY(), j.isMirrorY());
}
DetectorModuleGeometry::Direction Convert(const org::openapitools::server::model::Detector_module_direction& d) {
switch (d.getValue()) {
case org::openapitools::server::model::Detector_module_direction::eDetector_module_direction::XP:
return DetectorModuleGeometry::Direction::Xpos;
case org::openapitools::server::model::Detector_module_direction::eDetector_module_direction::XN:
return DetectorModuleGeometry::Direction::Xneg;
case org::openapitools::server::model::Detector_module_direction::eDetector_module_direction::YP:
return DetectorModuleGeometry::Direction::Ypos;
case org::openapitools::server::model::Detector_module_direction::eDetector_module_direction::YN:
return DetectorModuleGeometry::Direction::Yneg;
default:
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "invalid detector direction");
}
}
DetectorType Convert(const org::openapitools::server::model::Detector_type &d) {
switch (d.getValue()) {
case org::openapitools::server::model::Detector_type::eDetector_type::EIGER:
return DetectorType::EIGER;
case org::openapitools::server::model::Detector_type::eDetector_type::JUNGFRAU:
return DetectorType::JUNGFRAU;
case org::openapitools::server::model::Detector_type::eDetector_type::DECTRIS:
return DetectorType::DECTRIS;
default:
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "invalid detector type");
}
}
DetectorGeometryModular ParseCustomDetectorGeometry(const org::openapitools::server::model::Detector &j) {
std::vector<DetectorModuleGeometry> modules;
for (const auto &iter: j.getCustomGeometry()) {
auto fast = Convert(iter.getFastAxis());
auto slow = Convert(iter.getSlowAxis());
modules.emplace_back(iter.getX0(), iter.getY0(), fast, slow);
}
return DetectorGeometryModular(modules, j.isMirrorY());
}
DetectorGeometryModular ParseDetectorGeometry(const org::openapitools::server::model::Detector &d) {
if (d.standardGeometryIsSet() && d.customGeometryIsSet())
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Need to set EITHER standard or custom geometry");
if (d.standardGeometryIsSet())
return ParseStandardDetectorGeometry(d);
else if (d.customGeometryIsSet())
return ParseCustomDetectorGeometry(d);
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Need to set EITHER standard or custom geometry");
}
DetectorSetup ParseDetectorSetup(const org::openapitools::server::model::Detector &d) {
DetectorType detector_type = Convert(d.getType());
if (detector_type == DetectorType::DECTRIS) {
std::string hostname;
if (d.getHostname().size() > 1)
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
"DECTRIS detector requires single hostname (or none)");
else if (d.getHostname().size() == 1)
hostname = d.getHostname()[0];
DetectorSetup setup = DetDECTRIS(1,1, d.getDescription(), hostname);
if (d.roiModeIsSet())
setup.DECTRISROI(d.getRoiMode());
return setup;
}
DetectorGeometryModular geom = ParseDetectorGeometry(d);
DetectorSetup setup(geom, detector_type, d.getDescription(), d.getHostname());
auto calib = d.getCalibrationFile();
auto trim_energies = d.getTrimEnergiesEV();
if (!calib.empty()) {
switch (detector_type) {
case DetectorType::EIGER:
setup.SetTrimFiles(calib);
if (trim_energies.empty())
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
"Trimming energies not provided");
setup.TrimEnergies_eV(trim_energies);
break;
case DetectorType::JUNGFRAU:
setup.LoadGain(calib);
break;
default:
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
"Detector type not supported");
}
}
switch (detector_type) {
case DetectorType::EIGER:
case DetectorType::JUNGFRAU:
setup.PixelSize_um(75.0f);
break;
default:
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
"Detector type not supported");
}
if (d.highVoltageVIsSet())
setup.HighVoltage(d.getHighVoltageV());
setup.UDPInterfaceCount(d.getUdpInterfaceCount())
.SerialNumber(d.getSerialNumber())
.ModuleSync(d.isModuleSync());
// Only override the sensor from the request when it explicitly sets these. The API model defaults
// them (320 um / Si) with IsSet=false, so an unconditional set would clobber the detector-reported
// value (DECTRIS SIMPLON read) or a detector-specific default with 320 um (highVoltageV above is
// guarded the same way).
if (d.sensorThicknessUmIsSet())
setup.SensorThickness_um(d.getSensorThicknessUm());
if (d.sensorMaterialIsSet())
setup.SensorMaterial(d.getSensorMaterial());
if (d.readoutTimeNsIsSet())
setup.ReadOutTime(std::chrono::nanoseconds(d.getReadoutTimeNs()));
if (d.baseDataIpv4AddressIsSet())
setup.BaseIPv4Addr(d.getBaseDataIpv4Address());
if (d.txDelayIsSet())
setup.TxDelay(d.getTxDelay());
if (d.minCountTimeNsIsSet())
setup.MinCountTime(std::chrono::nanoseconds(d.getMinCountTimeNs()));
if (d.minFrameTimeNsIsSet())
setup.MinFrameTime(std::chrono::nanoseconds(d.getMinFrameTimeNs()));
if (d.defaultSettingsIsSet())
setup.DefaultSettings(Convert(d.getDefaultSettings()));
if (d.tempThresoldDegCIsSet())
setup.TempThreshold_degC(d.getTempThresoldDegC());
return setup;
}
void ParseFacilityConfiguration(const org::openapitools::server::model::Jfjoch_settings &j, DiffractionExperiment &experiment) {
if (j.instrumentIsSet())
experiment.ImportInstrumentMetadata(Convert(j.getInstrument()));
if (j.fileWriterIsSet())
experiment.ImportFileWriterSettings(Convert(j.getFileWriter()));
if (j.detectorSettingsIsSet())
experiment.ImportDetectorSettings(Convert(j.getDetectorSettings()));
if (j.azimIntIsSet())
experiment.ImportAzimuthalIntegrationSettings(Convert(j.getAzimInt()));
if (j.imageFormatIsSet())
experiment.ImportImageFormatSettings(Convert(j.getImageFormat()));
if (j.indexingIsSet())
experiment.ImportIndexingSettings(Convert(j.getIndexing()));
if (j.braggIntegrationIsSet())
experiment.ImportBraggIntegrationSettings(Convert(j.getBraggIntegration()));
if (j.darkMaskIsSet())
experiment.ImportDarkMaskSettings(Convert(j.getDarkMask()));
}
std::unique_ptr<ImagePusher> ParseZMQImagePusher(const org::openapitools::server::model::Jfjoch_settings &j) {
if (!j.zeromqIsSet())
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "ZeroMQ settings must be provided");
std::optional<int32_t> send_buffer_size;
if (j.getZeromq().sendBufferSizeIsSet())
send_buffer_size = j.getZeromq().getSendBufferSize();
std::optional<int32_t> send_watermark;
if (j.getZeromq().sendWatermarkIsSet())
send_watermark = j.getZeromq().getSendWatermark();
auto tmp = std::make_unique<ZMQStream2Pusher>(j.getZeromq().getImageSocket(),
send_watermark,
send_buffer_size);
if (j.getZeromq().writerNotificationSocketIsSet())
tmp->WriterNotificationSocket(j.getZeromq().getWriterNotificationSocket());
return std::move(tmp);
}
std::unique_ptr<ImagePusher> ParseTCPImagePusher(const org::openapitools::server::model::Jfjoch_settings &j) {
if (!j.tcpIsSet())
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "TCP/IP Socket settings must be provided");
std::optional<int32_t> send_buffer_size;
if (j.getTcp().sendBufferSizeIsSet())
send_buffer_size = j.getTcp().getSendBufferSize();
auto tmp = std::make_unique<TCPStreamPusher>(j.getTcp().getImageSocket(), j.getTcp().getNwriters(), send_buffer_size);
// Optional liveness/backpressure tuning; unset -> keep the pusher's built-in defaults.
if (j.getTcp().peerLivenessTimeoutMsIsSet() && j.getTcp().getPeerLivenessTimeoutMs() > 0)
tmp->SetPeerLivenessTimeout(std::chrono::milliseconds(j.getTcp().getPeerLivenessTimeoutMs()));
if (j.getTcp().maxBackpressureTimeoutMsIsSet() && j.getTcp().getMaxBackpressureTimeoutMs() > 0)
tmp->SetMaxBackpressureTimeout(std::chrono::milliseconds(j.getTcp().getMaxBackpressureTimeoutMs()));
return std::move(tmp);
}
std::unique_ptr<ImagePusher> ParseImagePusher(const org::openapitools::server::model::Jfjoch_settings &j) {
switch (j.getImagePusher().getValue()) {
case org::openapitools::server::model::Image_pusher_type::eImage_pusher_type::ZEROMQ:
return ParseZMQImagePusher(j);
case org::openapitools::server::model::Image_pusher_type::eImage_pusher_type::TCP:
return ParseTCPImagePusher(j);
case org::openapitools::server::model::Image_pusher_type::eImage_pusher_type::HDF5:
return std::make_unique<HDF5FilePusher>();
case org::openapitools::server::model::Image_pusher_type::eImage_pusher_type::NONE:
return std::make_unique<NonePusher>();
case org::openapitools::server::model::Image_pusher_type::eImage_pusher_type::CBOR:
return std::make_unique<CBORFilePusher>();
default:
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Invalid value");
}
}
void ParseAcquisitionDeviceGroup(const org::openapitools::server::model::Jfjoch_settings &input,
AcquisitionDeviceGroup &aq_devices) {
if (!input.pcieIsSet())
aq_devices.AddHLSDevice(256);
else for (auto &p: input.getPcie()) {
std::optional<uint32_t> ipv4_addr = {};
if (p.ipv4IsSet())
ipv4_addr = IPv4AddressFromStr(p.getIpv4());
aq_devices.AddPCIeDevice(p.getBlk(), ipv4_addr);
}
}
void ParseReceiverSettings(const org::openapitools::server::model::Jfjoch_settings &input, JFJochReceiverService &service) {
// Using default in case
service.NumThreads(input.getReceiverThreads());
if (input.zeromqPreviewIsSet()) {
service.PreviewSocket(input.getZeromqPreview().getSocketAddress());
service.PreviewSocketSettings(Convert(input.getZeromqPreview()));
}
if (input.zeromqMetadataIsSet()) {
service.MetadataSocket(input.getZeromqMetadata().getSocketAddress());
service.MetadataSocketSettings(Convert(input.getZeromqMetadata()));
}
}
SpotFindingSettings ParseSpotFindingSettings(const org::openapitools::server::model::Jfjoch_settings &input) {
if (input.spotFindingIsSet())
return Convert(input.getSpotFinding());
return SpotFindingSettings();
}