mx/Jungfraujoch
3
2 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 83 Wiki Activity
Files
4bf41983ed13bd71ed304029a7dbc837249b6bb2
Jungfraujoch/broker/JFJochBrokerHttp.cpp
T
leonarski_fandClaude Opus 5.5 4bf41983ed rugnux: the model is parsed by the caller and handed in as RugnuxModel 
The analysis read the --model file itself, four to five times per run
(the model kind, every validation, molecular replacement, the stills
reference intensities) and located the PAE sidecar from the file name.
The library is to do no file I/O, so the model now arrives in memory:

- RugnuxModel (image_analysis/structure_refinement/RugnuxModel.h): the
  gemmi structure as read, whether it is ModelCIF, the PAE matrix where a
  prediction came with one, and the name the report gives it.
- ParseRugnuxModel / ReadRugnuxModelFile (writer/RugnuxModelFile.h, in
  JFJochRugnuxFiles): bytes in, gzip by magic, PDB/mmCIF by content,
  _ma_qa_metric detection as before; the file reader takes the PAE JSON
  from the same place PaePathFor named.
- ProcessConfig::model replaces model_path; PrepareSearchModel,
  RunMolecularReplacement, ValidateAgainstModel and
  ModelReferenceIntensities take the model; result fields name it by
  model_name.
- rugnux reads the model before the run starts and stops with an error
  where the file holds no model (it used to run and report the model as
  not validated). The viewer reads it when the job is set up.
- The broker parses an uploaded model at once and answers 400 where it is
  not one; it keeps the parsed model in memory, with no temporary
  directory or file. It takes no PAE.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 16:11:58 +02:00

1165 lines
49 KiB
C++
Raw Blame History

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
// Using OpenAPI licensed with Apache License 2.0
#include "JFJochBrokerHttp.h"
#include <set>
#include <cstring>
#include <fstream>
#include <sstream>
#include <vector>
#include "Helpers.h"
#include "../common/GitInfo.h"
#include "../preview/JFJochTIFF.h"
#include "OpenAPIConvert.h"
#include "../image_pusher/LiveCollection.h"
#include "../writer/RugnuxModelFile.h"
#include <base64/Base64.h>
#include "gen/model/Error_message.h"
using namespace org::openapitools::server::model;
namespace {
template<class T>
bool fromStringValue(const std::string &s, T &out) {
std::istringstream is(s);
is >> out;
return !is.fail() && is.eof();
}
template<>
bool fromStringValue<std::string>(const std::string &s, std::string &out) {
out = s;
return true;
}
template<>
bool fromStringValue<bool>(const std::string &s, bool &out) {
if (s == "true" || s == "1") {
out = true;
return true;
}
if (s == "false" || s == "0") {
out = false;
return true;
}
return false;
}
template<class T>
std::optional<T> parse_query_value(const httplib::Request &req, const char *name) {
if (!req.has_param(name))
return std::nullopt;
T v{};
if (fromStringValue(req.get_param_value(name), v))
return v;
return std::nullopt;
}
std::optional<std::string> parse_query_string(const httplib::Request &req, const char *name) {
if (!req.has_param(name))
return std::nullopt;
return req.get_param_value(name);
}
std::string mime_from_path(const std::string &path) {
if (path.ends_with(".js")) return "text/javascript";
if (path.ends_with(".css")) return "text/css";
if (path.ends_with(".html")) return "text/html";
if (path.ends_with(".json")) return "application/json";
if (path.ends_with(".cbor")) return "application/cbor";
if (path.ends_with(".jpeg") || path.ends_with(".jpg")) return "image/jpeg";
if (path.ends_with(".tiff") || path.ends_with(".tif")) return "image/tiff";
if (path.ends_with(".png")) return "image/png";
if (path.ends_with(".svg")) return "image/svg+xml";
if (path.ends_with(".bin")) return "application/octet-stream";
return "application/octet-stream";
}
inline bool read_file_to_string(const std::string &path, std::string &out) {
std::ifstream f(path, std::ios::binary);
if (!f)
return false;
std::ostringstream ss;
ss << f.rdbuf();
out = ss.str();
return true;
}
}
std::string JFJochBrokerHttp::error_message_json(const std::string &msg, const std::string &reason) {
Error_message m;
m.setMsg(msg);
m.setReason(reason);
nlohmann::json j;
to_json(j, m);
return j.dump();
}
JFJochBrokerHttp::JFJochBrokerHttp(const DiffractionExperiment &experiment,
const SpotFindingSettings &spot_finding_settings)
: state_machine(experiment, services, logger, spot_finding_settings) {
}
void JFJochBrokerHttp::AddDetectorSetup(const DetectorSetup &setup) {
state_machine.AddDetectorSetup(setup);
logger.Info("Added detector {}", setup.GetDescription());
}
JFJochServices &JFJochBrokerHttp::Services() {
return services;
}
JFJochBrokerHttp &JFJochBrokerHttp::FrontendDirectory(const std::string &directory) {
frontend_directory = directory;
return *this;
}
JFJochBrokerHttp::HttpError JFJochBrokerHttp::handleParsingException(const std::exception &ex) const noexcept {
try {
throw;
} catch (const nlohmann::detail::exception &e) {
return {400, e.what(), "text/plain"};
} catch (const org::openapitools::server::helpers::ValidationException &e) {
return {400, e.what(), "text/plain"};
} catch (const std::exception &e) {
return {500, error_message_json(e.what(), "Other"), "application/json"};
}
}
JFJochBrokerHttp::HttpError JFJochBrokerHttp::handleOperationException(const std::exception &ex) const noexcept {
try {
throw;
} catch (const WrongDAQStateException &) {
return {500, error_message_json(ex.what(), "WrongDAQState"), "application/json"};
} catch (const std::exception &) {
return {500, error_message_json(ex.what(), "Other"), "application/json"};
}
}
// The read endpoints that expose the current dataset (its name and unit cell, and its images).
// While the dataset was started with tokens, they answer 401 to a request without one; the same
// set carries the bearerAuth security scheme in jfjoch_api.yaml. Everything else - the control
// plane, instrument configuration, broker status - stays open.
static const std::set<std::string> PROTECTED_PATHS{
"/statistics/data_collection",
"/result/scan",
"/image_buffer/start.cbor",
"/image_buffer/image.cbor",
"/image_buffer/image.jpeg",
"/image_buffer/image.tiff",
"/preview/plot",
"/preview/plot.bin",
"/live/events",
"/config/rugnux",
"/config/rugnux/model",
"/rugnux/status"
};
std::optional<std::string> JFJochBrokerHttp::AuthorizationHeader(const httplib::Request &req) {
if (req.has_header("Authorization"))
return req.get_header_value("Authorization");
return std::nullopt;
}
bool JFJochBrokerHttp::Authorized(const httplib::Request &req, httplib::Response &res) const {
if (state_machine.AcceptToken(AuthorizationHeader(req)))
return true;
res.status = 401;
res.set_header("WWW-Authenticate", "Bearer");
res.set_content("Bearer token required for the current dataset", "text/plain");
return false;
}
void JFJochBrokerHttp::attach(httplib::Server &server) {
server.set_pre_routing_handler([this](const httplib::Request &req, httplib::Response &res) {
if (PROTECTED_PATHS.count(req.path) && !Authorized(req, res))
return httplib::Server::HandlerResponse::Handled;
return httplib::Server::HandlerResponse::Unhandled;
});
// Last resort for an exception that got past a route's own handler - httplib would otherwise
// answer with a bodyless 500, which no generated client can deserialize into an error_message.
server.set_exception_handler([this](const httplib::Request &, httplib::Response &res,
const std::exception_ptr &ep) {
try {
std::rethrow_exception(ep);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
} catch (...) {
send_error(res, {500, error_message_json("Unknown exception", "Other"), "application/json"});
}
});
register_routes(server);
}
void JFJochBrokerHttp::register_routes(httplib::Server &server) {
auto bind_noarg = [this](auto method) {
return [this, method](const httplib::Request &, httplib::Response &res) {
try {
(this->*method)(res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
};
};
auto bind_json = [this](auto method, auto model_tag) {
return [this, method](const httplib::Request &req, httplib::Response &res) {
using Model = decltype(model_tag);
Model v;
try {
nlohmann::json::parse(req.body).get_to(v);
v.validate();
} catch (const std::exception &e) {
// Malformed or invalid request body.
send_error(res, handleParsingException(e));
return;
}
try {
(this->*method)(v, res);
} catch (const std::exception &e) {
// Operation failure (e.g. WrongDAQState, a failed synchronous start): return the
// structured Error_message JSON with a reason field, like the no-arg endpoints, so
// clients can distinguish the cause instead of receiving opaque plain text.
send_error(res, handleOperationException(e));
}
};
};
auto bind_req = [this](auto method) {
return [this, method](const httplib::Request &req, httplib::Response &res) {
try {
(this->*method)(req, res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
};
};
server.Post("/cancel", bind_noarg(&JFJochBrokerHttp::cancel_post));
server.Get("/config/azim_int", bind_noarg(&JFJochBrokerHttp::config_azim_int_get));
server.Put("/config/azim_int", bind_json(&JFJochBrokerHttp::config_azim_int_put, Azim_int_settings{}));
server.Get("/config/dark_mask", bind_noarg(&JFJochBrokerHttp::config_dark_mask_get));
server.Put("/config/dark_mask", bind_json(&JFJochBrokerHttp::config_dark_mask_put, Dark_mask_settings{}));
server.Get("/config/detector", bind_noarg(&JFJochBrokerHttp::config_detector_get));
server.Put("/config/detector", bind_json(&JFJochBrokerHttp::config_detector_put, Detector_settings{}));
server.Get("/config/file_writer", bind_noarg(&JFJochBrokerHttp::config_file_writer_get));
server.Put("/config/file_writer", bind_json(&JFJochBrokerHttp::config_file_writer_put, File_writer_settings{}));
server.Post("/config/image_format/conversion", bind_noarg(&JFJochBrokerHttp::config_image_format_conversion_post));
server.Get("/config/image_format", bind_noarg(&JFJochBrokerHttp::config_image_format_get));
server.Put("/config/image_format", bind_json(&JFJochBrokerHttp::config_image_format_put, Image_format_settings{}));
server.Post("/config/image_format/raw", bind_noarg(&JFJochBrokerHttp::config_image_format_raw_post));
server.Get("/config/indexing", bind_noarg(&JFJochBrokerHttp::config_indexing_get));
server.Put("/config/indexing", bind_json(&JFJochBrokerHttp::config_indexing_put, Indexing_settings{}));
server.Get("/config/rugnux", bind_noarg(&JFJochBrokerHttp::config_rugnux_get));
server.Put("/config/rugnux", bind_json(&JFJochBrokerHttp::config_rugnux_put, Rugnux_settings{}));
server.Put("/config/rugnux/model", bind_req(&JFJochBrokerHttp::config_rugnux_model_put));
server.Delete("/config/rugnux/model", bind_noarg(&JFJochBrokerHttp::config_rugnux_model_delete));
server.Get("/rugnux/status", bind_noarg(&JFJochBrokerHttp::rugnux_status_get));
server.Get("/config/bragg_integration", bind_noarg(&JFJochBrokerHttp::config_bragg_integration_get));
server.Put("/config/bragg_integration", bind_json(&JFJochBrokerHttp::config_bragg_integration_put, Bragg_integration_settings{}));
server.Get("/config/instrument", bind_noarg(&JFJochBrokerHttp::config_instrument_get));
server.Put("/config/instrument", bind_json(&JFJochBrokerHttp::config_instrument_put, Instrument_metadata{}));
server.Put("/config/internal_generator_image", [this](const httplib::Request &req, httplib::Response &res) {
try {
config_internal_generator_image_put(parse_query_value<int64_t>(req, "id"), req, res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Put("/config/internal_generator_image.tiff", [this](const httplib::Request &req, httplib::Response &res) {
try {
config_internal_generator_image_tiff_put(parse_query_value<int64_t>(req, "id"), req, res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/config/mask", bind_noarg(&JFJochBrokerHttp::config_mask_get));
server.Get("/config/mask.tiff", bind_noarg(&JFJochBrokerHttp::config_mask_tiff_get));
server.Get("/config/roi", bind_noarg(&JFJochBrokerHttp::config_roi_get));
server.Put("/config/roi", bind_json(&JFJochBrokerHttp::config_roi_put, Roi_definitions{}));
server.Get("/config/select_detector", bind_noarg(&JFJochBrokerHttp::config_select_detector_get));
server.Put("/config/select_detector",
bind_json(&JFJochBrokerHttp::config_select_detector_put, Detector_selection{}));
server.Get("/config/spot_finding", bind_noarg(&JFJochBrokerHttp::config_spot_finding_get));
server.Put("/config/spot_finding", bind_json(&JFJochBrokerHttp::config_spot_finding_put, Spot_finding_settings{}));
server.Get("/config/user_mask", bind_noarg(&JFJochBrokerHttp::config_user_mask_get));
server.Put("/config/user_mask", bind_req(&JFJochBrokerHttp::config_user_mask_put));
server.Get("/config/user_mask.tiff", bind_noarg(&JFJochBrokerHttp::config_user_mask_tiff_get));
server.Put("/config/user_mask.tiff", bind_req(&JFJochBrokerHttp::config_user_mask_tiff_put));
server.Get("/config/zeromq_metadata", bind_noarg(&JFJochBrokerHttp::config_zeromq_metadata_get));
server.Put("/config/zeromq_metadata",
bind_json(&JFJochBrokerHttp::config_zeromq_metadata_put, Zeromq_metadata_settings{}));
server.Get("/config/zeromq_preview", bind_noarg(&JFJochBrokerHttp::config_zeromq_preview_get));
server.Put("/config/zeromq_preview",
bind_json(&JFJochBrokerHttp::config_zeromq_preview_put, Zeromq_preview_settings{}));
server.Post("/deactivate", bind_noarg(&JFJochBrokerHttp::deactivate_post));
server.Get("/detector/status", bind_noarg(&JFJochBrokerHttp::detector_status_get));
server.Get("/fpga_status", bind_noarg(&JFJochBrokerHttp::fpga_status_get));
server.Post("/image_buffer/clear", bind_noarg(&JFJochBrokerHttp::image_buffer_clear_post));
server.Get("/image_buffer/image.cbor", [this](const httplib::Request &req, httplib::Response &res) {
try {
image_buffer_image_cbor_get(parse_query_value<int64_t>(req, "id"), res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/image_buffer/image.jpeg", [this](const httplib::Request &req, httplib::Response &res) {
try {
image_buffer_image_jpeg_get(
parse_query_value<int64_t>(req, "id"),
parse_query_value<bool>(req, "show_user_mask"),
parse_query_value<bool>(req, "show_roi"),
parse_query_value<bool>(req, "show_spots"),
parse_query_value<bool>(req, "show_predictions"),
parse_query_value<bool>(req, "show_beam_center"),
parse_query_value<float>(req, "saturation"),
parse_query_value<int64_t>(req, "jpeg_quality"),
parse_query_value<float>(req, "show_res_ring"),
parse_query_string(req, "color"),
parse_query_value<bool>(req, "show_res_est"),
res
);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/image_buffer/image.tiff", [this](const httplib::Request &req, httplib::Response &res) {
try {
image_buffer_image_tiff_get(parse_query_value<int64_t>(req, "id"), res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/image_buffer/start.cbor", bind_noarg(&JFJochBrokerHttp::image_buffer_start_cbor_get));
server.Get("/live/events", bind_noarg(&JFJochBrokerHttp::live_events_get));
server.Get("/image_buffer/status", bind_noarg(&JFJochBrokerHttp::image_buffer_status_get));
server.Get("/image_pusher/status", bind_noarg(&JFJochBrokerHttp::image_pusher_status_get));
server.Post("/initialize", bind_noarg(&JFJochBrokerHttp::initialize_post));
server.Post("/pedestal", bind_noarg(&JFJochBrokerHttp::pedestal_post));
server.Get("/preview/pedestal.tiff", [this](const httplib::Request &req, httplib::Response &res) {
try {
preview_pedestal_tiff_get(parse_query_value<int32_t>(req, "gain_level"),
parse_query_value<int32_t>(req, "sc"),
res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/preview/plot.bin", [this](const httplib::Request &req, httplib::Response &res) {
try {
preview_plot_bin_get(parse_query_string(req, "type"),
parse_query_string(req, "roi"),
res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/preview/plot", [this](const httplib::Request &req, httplib::Response &res) {
try {
preview_plot_get(parse_query_string(req, "type"),
parse_query_value<int32_t>(req, "binning"),
parse_query_value<float>(req, "fill"),
parse_query_value<bool>(req, "experimental_coord"),
parse_query_string(req, "azint_unit"),
res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/result/scan", bind_noarg(&JFJochBrokerHttp::result_scan_get));
server.Post("/start", bind_json(&JFJochBrokerHttp::start_post, Dataset_settings{}));
server.Get("/statistics/calibration", bind_noarg(&JFJochBrokerHttp::statistics_calibration_get));
server.Get("/statistics/data_collection", bind_noarg(&JFJochBrokerHttp::statistics_data_collection_get));
server.Get("/statistics", [this](const httplib::Request &req, httplib::Response &res) {
try {
statistics_get(req, res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/status", bind_noarg(&JFJochBrokerHttp::status_get));
server.Post("/trigger", bind_noarg(&JFJochBrokerHttp::trigger_post));
server.Get("/version", bind_noarg(&JFJochBrokerHttp::version_get));
server.Post("/wait_till_done", [this](const httplib::Request &req, httplib::Response &res) {
try {
wait_till_done_post(parse_query_value<int32_t>(req, "timeout"), res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Post("/wait_until_running", [this](const httplib::Request &req, httplib::Response &res) {
try {
wait_until_running_post(parse_query_value<int32_t>(req, "timeout"), res);
} catch (const std::exception &e) {
send_error(res, handleOperationException(e));
}
});
server.Get("/xfel/event_code", bind_noarg(&JFJochBrokerHttp::xfel_event_code_get));
server.Get("/xfel/pulse_id", bind_noarg(&JFJochBrokerHttp::xfel_pulse_id_get));
}
void JFJochBrokerHttp::cancel_post(httplib::Response &response) {
state_machine.Cancel();
response.status = 200;
}
void JFJochBrokerHttp::deactivate_post(httplib::Response &response) {
state_machine.Deactivate();
response.status = 200;
}
void JFJochBrokerHttp::initialize_post(httplib::Response &response) {
state_machine.Initialize();
response.status = 200;
}
void JFJochBrokerHttp::start_post(const Dataset_settings &datasetSettings, httplib::Response &response) {
nlohmann::json j = datasetSettings;
j.erase("tokens"); // secrets never reach the log
logger.Info("Start {}", j.dump());
std::vector<std::string> tokens;
if (datasetSettings.tokensIsSet())
tokens = datasetSettings.getTokens();
state_machine.Start(Convert(datasetSettings), datasetSettings.isAsyncStart(), std::move(tokens));
response.status = 200;
}
void JFJochBrokerHttp::status_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetStatus()), response);
}
void JFJochBrokerHttp::wait_until_running_post(const std::optional<int32_t> &timeout, httplib::Response &response) {
BrokerStatus status;
if (!timeout)
status = state_machine.WaitTillNotBusy(std::chrono::minutes(1));
else if ((timeout.value() > 3600) || (timeout.value() < 0)) {
response.status = 400;
response.set_content("timeout must be in range 0..3600", "text/plain");
return;
} else
// A zero timeout still goes through the wait function rather than GetStatus(): the predicate
// is evaluated before the wait expires, so the answer is the same, but a start that failed
// asynchronously is reported instead of being read back as a plain state.
status = state_machine.WaitTillNotBusy(std::chrono::seconds(timeout.value()));
logger.Info("Wait until running");
switch (status.state) {
case JFJochState::Measuring:
response.status = 200;
break;
case JFJochState::Inactive:
response.status = 502;
break;
case JFJochState::Error:
throw WrongDAQStateException(status.message.value_or("Unknown error"));
case JFJochState::Idle:
case JFJochState::Busy:
case JFJochState::Calibration:
response.status = 504;
break;
}
}
void JFJochBrokerHttp::wait_till_done_post(const std::optional<int32_t> &timeout, httplib::Response &response) {
BrokerStatus status;
if (!timeout)
status = state_machine.WaitTillMeasurementDone(std::chrono::minutes(1));
else if ((timeout.value() > 3600) || (timeout.value() < 0)) {
response.status = 400;
response.set_content("timeout must be in range 0..3600", "text/plain");
return;
} else
status = state_machine.WaitTillMeasurementDone(std::chrono::seconds(timeout.value()));
logger.Info("Wait till done");
switch (status.state) {
case JFJochState::Idle:
// An operation that finished badly leaves the state Idle - the detector is still
// usable - and says so only through the message severity. Reporting that as a plain
// 200 makes a run that lost packets, or one whose writer could not write, look to a
// script exactly like a good one. A cancelled collection does not land here: the
// cancel is reported with Info severity, and MeasurementThread checks it before it
// looks at the completeness of the data, so the incomplete run a cancel leaves behind
// is never reported as an error. A Warning - a receiver that could not keep up - is
// still a 200: the data is there, and the run is done.
if (status.message_severity == BrokerStatus::MessageSeverity::Error) {
send_error(response, {500, error_message_json(status.message.value_or("Unknown error"), "Other"),
"application/json"});
return;
}
response.status = 200;
break;
case JFJochState::Inactive:
response.status = 502;
break;
case JFJochState::Error:
throw WrongDAQStateException(status.message.value_or("Unknown error"));
case JFJochState::Measuring:
case JFJochState::Busy:
case JFJochState::Calibration:
response.status = 504;
break;
}
}
void JFJochBrokerHttp::trigger_post(httplib::Response &response) {
state_machine.Trigger();
response.status = 200;
}
void JFJochBrokerHttp::pedestal_post(httplib::Response &response) {
state_machine.Pedestal();
response.status = 200;
}
void JFJochBrokerHttp::config_detector_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetDetectorSettings()), response);
}
void JFJochBrokerHttp::config_detector_put(const Detector_settings &detectorSettings,
httplib::Response &response) {
state_machine.LoadDetectorSettings(Convert(detectorSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_azim_int_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetRadialIntegrationSettings()), response);
}
void JFJochBrokerHttp::config_azim_int_put(const Azim_int_settings &radIntSettings,
httplib::Response &response) {
state_machine.SetRadialIntegrationSettings(Convert(radIntSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_select_detector_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetDetectorsList()), response);
}
void JFJochBrokerHttp::config_select_detector_put(const Detector_selection &detectorSelection,
httplib::Response &response) {
state_machine.SelectDetector(detectorSelection.getId());
response.status = 200;
}
void JFJochBrokerHttp::config_spot_finding_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetSpotFindingSettings()), response);
}
void JFJochBrokerHttp::config_spot_finding_put(const Spot_finding_settings &spotFindingSettings,
httplib::Response &response) {
state_machine.SetSpotFindingSettings(Convert(spotFindingSettings));
response.status = 200;
}
void JFJochBrokerHttp::statistics_calibration_get(httplib::Response &response) {
nlohmann::json j;
for (const auto &d: Convert(state_machine.GetCalibrationStatistics()))
j.push_back(d);
logger.Info("Calibration statistics: {}", j.dump());
response.set_content(j.dump(), "application/json");
response.status = 200;
}
void JFJochBrokerHttp::statistics_data_collection_get(httplib::Response &response) {
auto stats = state_machine.GetMeasurementStatistics();
if (stats)
ProcessOutput(Convert(stats.value()), response);
else
response.status = 404;
}
void JFJochBrokerHttp::config_file_writer_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetFileWriterSettings()), response);
}
void JFJochBrokerHttp::config_file_writer_put(const File_writer_settings &fileWriterSettings,
httplib::Response &response) {
state_machine.LoadFileWriterSettings(Convert(fileWriterSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_image_format_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetImageFormatSettings()), response);
}
void JFJochBrokerHttp::config_image_format_put(const Image_format_settings &imageFormatSettings,
httplib::Response &response) {
state_machine.LoadImageFormatSettings(Convert(imageFormatSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_image_format_conversion_post(httplib::Response &response) {
state_machine.ConvImageFormatSettings();
response.status = 200;
}
void JFJochBrokerHttp::config_image_format_raw_post(httplib::Response &response) {
state_machine.RawImageFormatSettings();
response.status = 200;
}
void JFJochBrokerHttp::fpga_status_get(httplib::Response &response) {
nlohmann::json j;
for (const auto &d: Convert(state_machine.GetDeviceStatus()))
j.push_back(d);
response.set_content(j.dump(), "application/json");
response.status = 200;
}
void JFJochBrokerHttp::statistics_get(const httplib::Request &request, httplib::Response &response) {
Jfjoch_statistics statistics;
// The aggregate is what the web frontend polls for the whole instrument view, so a protected
// dataset only drops the measurement block here instead of refusing the request.
auto data_collection_statistics = state_machine.GetMeasurementStatistics();
if (data_collection_statistics && state_machine.AcceptToken(AuthorizationHeader(request)))
statistics.setMeasurement(Convert(data_collection_statistics.value()));
statistics.setFpga(Convert(state_machine.GetDeviceStatus()));
statistics.setCalibration(Convert(state_machine.GetCalibrationStatistics()));
statistics.setBroker(Convert(state_machine.GetStatus()));
auto det_status = state_machine.GetDetectorStatus();
if (det_status.has_value())
statistics.setDetector(Convert(det_status.value()));
statistics.setDetectorSettings(Convert(state_machine.GetDetectorSettings()));
statistics.setDetectorList(Convert(state_machine.GetDetectorsList()));
statistics.setDataProcessingSettings(Convert(state_machine.GetSpotFindingSettings()));
statistics.setInstrumentMetadata(Convert(state_machine.GetInstrumentMetadata()));
statistics.setImageFormatSettings(Convert(state_machine.GetImageFormatSettings()));
statistics.setPixelMask(Convert(state_machine.GetPixelMaskStatistics()));
statistics.setRoi(Convert(state_machine.GetROIDefintion()));
statistics.setFileWriterSettings(Convert(state_machine.GetFileWriterSettings()));
statistics.setAzInt(Convert(state_machine.GetRadialIntegrationSettings()));
statistics.setBuffer(Convert(state_machine.GetImageBufferStatus()));
statistics.setIndexing(Convert(state_machine.GetIndexingSettings()));
statistics.setBraggIntegration(Convert(state_machine.GetBraggIntegrationSettings()));
statistics.setDarkMask(Convert(state_machine.GetDarkMaskSettings()));
statistics.setImagePusher(Convert(state_machine.GetImagePusherStatus()));
auto zeromq_prev = state_machine.GetPreviewSocketSettings();
if (!zeromq_prev.address.empty())
statistics.setZeromqPreview(Convert(zeromq_prev));
auto zeromq_metadata = state_machine.GetMetadataSocketSettings();
if (!zeromq_metadata.address.empty())
statistics.setZeromqMetadata(Convert(zeromq_metadata));
nlohmann::json j = statistics;
response.set_content(j.dump(), "application/json");
response.status = 200;
}
void JFJochBrokerHttp::config_zeromq_preview_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetPreviewSocketSettings()), response);
}
void JFJochBrokerHttp::config_zeromq_preview_put(const Zeromq_preview_settings &zeromqPreviewSettings,
httplib::Response &response) {
state_machine.SetPreviewSocketSettings(Convert(zeromqPreviewSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_zeromq_metadata_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetMetadataSocketSettings()), response);
}
void JFJochBrokerHttp::config_zeromq_metadata_put(const Zeromq_metadata_settings &zeromqMetadataSettings,
httplib::Response &response) {
state_machine.SetMetadataSocketSettings(Convert(zeromqMetadataSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_internal_generator_image_put(const std::optional<int64_t> &id,
const httplib::Request &request,
httplib::Response &response) {
int64_t image_number = id.value_or(0);
if ((image_number < 0) || (image_number > 127)) {
response.status = 400;
response.set_content("id must be in range 0-127", "text/plain");
return;
}
state_machine.LoadInternalGeneratorImage(request.body.data(), request.body.size(), image_number);
logger.Info("Internal generator image #{} loaded", image_number);
response.status = 200;
}
void JFJochBrokerHttp::config_internal_generator_image_tiff_put(const std::optional<int64_t> &id,
const httplib::Request &request,
httplib::Response &response) {
int64_t image_number = id.value_or(0);
if ((image_number < 0) || (image_number > 127)) {
response.status = 400;
response.set_content("id must be in range 0-127", "text/plain");
return;
}
state_machine.LoadInternalGeneratorImageTIFF(request.body, image_number);
response.status = 200;
}
void JFJochBrokerHttp::image_buffer_clear_post(httplib::Response &response) {
state_machine.ClearImageBuffer();
response.status = 200;
}
void JFJochBrokerHttp::image_buffer_image_cbor_get(const std::optional<int64_t> &imageNumber,
httplib::Response &response) {
std::vector<uint8_t> tmp_vector;
state_machine.GetImageFromBuffer(tmp_vector, imageNumber.value_or(-1));
std::string s(reinterpret_cast<char *>(tmp_vector.data()), tmp_vector.size());
if (!s.empty())
response.set_content(s, "application/cbor");
else
response.status = 404;
}
void JFJochBrokerHttp::image_buffer_image_jpeg_get(const std::optional<int64_t> &id,
const std::optional<bool> &showUserMask,
const std::optional<bool> &showRoi,
const std::optional<bool> &showSpots,
const std::optional<bool> &showPredictions,
const std::optional<bool> &showBeamCenter,
const std::optional<float> &saturation,
const std::optional<int64_t> &jpegQuality,
const std::optional<float> &showResRing,
const std::optional<std::string> &color,
const std::optional<bool> &showResEst,
httplib::Response &response) {
int64_t image_id = id.value_or(ImageBuffer::MaxImage);
PreviewImageSettings settings{};
settings.show_user_mask = showUserMask.value_or(false);
settings.show_roi = showRoi.value_or(false);
settings.show_spots = showSpots.value_or(true);
settings.show_predictions = showPredictions.value_or(false);
settings.saturation_value = saturation;
settings.background_value = 0.0;
settings.jpeg_quality = jpegQuality.value_or(100);
if (showResEst.value_or(false)) {
settings.show_res_est = true;
} else {
settings.show_res_est = false;
settings.resolution_ring = showResRing;
}
settings.scale = ConvertColorScale(color);
settings.show_beam_center = showBeamCenter.value_or(true);
settings.format = PreviewImageFormat::JPEG;
std::string s = state_machine.GetPreviewJPEG(settings, image_id);
if (!s.empty())
response.set_content(s, "image/jpeg");
else
response.status = 404;
}
void JFJochBrokerHttp::image_buffer_image_tiff_get(const std::optional<int64_t> &id,
httplib::Response &response) {
int64_t image_id = id.value_or(ImageBuffer::MaxImage);
std::string s = state_machine.GetPreviewTIFF(image_id);
if (!s.empty())
response.set_content(s, "image/tiff");
else
response.status = 404;
}
void JFJochBrokerHttp::image_buffer_start_cbor_get(httplib::Response &response) {
std::vector<uint8_t> tmp_vector;
state_machine.GetStartMessageFromBuffer(tmp_vector);
std::string s(reinterpret_cast<char *>(tmp_vector.data()), tmp_vector.size());
if (!s.empty())
response.set_content(s, "application/cbor");
else
response.status = 404;
}
namespace {
std::string ServerSentEvent(const std::string &event, const nlohmann::json &data) {
return "event: " + event + "\ndata: " + data.dump() + "\n\n";
}
}
void JFJochBrokerHttp::live_events_get(httplib::Response &response) {
LiveCollection *live = services.GetLiveCollection();
if (live == nullptr) {
response.status = 409;
response.set_content("Following a collection needs the TCP or the HDF5 image pusher", "text/plain");
return;
}
const uint64_t generation = live->Get().generation;
if (generation == 0) {
response.status = 404;
response.set_content("No collection has started", "text/plain");
return;
}
if (!live->AcquireFollower()) {
response.status = 409;
response.set_content("Another reader is following the current collection", "text/plain");
return;
}
// The whole history of the collection first, then every change, until the collection ends or a new
// one begins. A comment every 10 s when nothing happens, so a reader can tell a quiet collection
// from a dead connection.
struct Cursor {
bool collection_sent = false;
size_t files_sent = 0;
bool done = false;
};
auto cursor = std::make_shared<Cursor>();
response.set_header("Cache-Control", "no-cache");
response.set_header("X-Accel-Buffering", "no");
response.set_chunked_content_provider(
"text/event-stream",
[live, generation, cursor](size_t, httplib::DataSink &sink) {
if (cursor->done) {
sink.done();
return true;
}
const auto s = live->Wait(generation, cursor->collection_sent, cursor->files_sent,
std::chrono::seconds(10));
std::string out;
// The start message first, once it is known: a collection is announced at /start, and a
// DECTRIS detector sends its start message only afterwards.
if (!cursor->collection_sent && s.started && s.generation == generation) {
const auto cbor = live->StartCBOR().value_or(std::vector<uint8_t>{});
out += ServerSentEvent("start", {
{"run_number", s.run_number}, {"run_name", s.run_name},
{"file_prefix", s.file_prefix}, {"number_of_images", s.number_of_images},
{"images_per_file", s.images_per_file},
{"cbor", macaron::Encode(std::string(cbor.begin(), cbor.end()))}});
cursor->collection_sent = true;
}
if (s.generation != generation) {
out += ServerSentEvent("superseded", {{"run_number", s.run_number}});
cursor->done = true;
} else {
for (; cursor->files_sent < s.files.size(); cursor->files_sent++) {
const auto &f = s.files[cursor->files_sent];
out += ServerSentEvent("file", {
{"file_number", f.file_number}, {"file", f.file},
{"first_image", f.first_image}, {"image_count", f.image_count}});
}
if (s.ended) {
nlohmann::json end = {{"images", s.end_images}};
if (!s.end_error.empty())
end["error"] = s.end_error;
out += ServerSentEvent("end", end);
cursor->done = true;
}
}
if (out.empty())
out = ": keepalive\n\n";
return sink.write(out.data(), out.size());
},
[live](bool) { live->ReleaseFollower(); });
}
void JFJochBrokerHttp::image_buffer_status_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetImageBufferStatus()), response);
}
void JFJochBrokerHttp::image_pusher_status_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetImagePusherStatus()), response);
}
void JFJochBrokerHttp::config_mask_tiff_get(httplib::Response &response) {
auto s = state_machine.GetFullPixelMaskTIFF();
response.set_content(s, "image/tiff");
response.status = 200;
}
void JFJochBrokerHttp::config_user_mask_tiff_get(httplib::Response &response) {
auto s = state_machine.GetUserPixelMaskTIFF();
response.set_content(s, "image/tiff");
response.status = 200;
}
void JFJochBrokerHttp::config_user_mask_tiff_put(const httplib::Request &request,
httplib::Response &response) {
std::vector<uint8_t> buffer;
state_machine.SetUserPixelMask(ReadTIFF(request.body, buffer));
response.status = 200;
}
void JFJochBrokerHttp::config_mask_get(httplib::Response &response) {
const auto v = state_machine.GetFullPixelMask();
response.set_content(reinterpret_cast<const char *>(v.data()),
v.size() * sizeof(uint32_t),
"application/octet-stream");
response.status = 200;
}
void JFJochBrokerHttp::config_user_mask_get(httplib::Response &response) {
const auto v = state_machine.GetUserPixelMask();
response.set_content(reinterpret_cast<const char *>(v.data()),
v.size() * sizeof(uint32_t),
"application/octet-stream");
response.status = 200;
}
void JFJochBrokerHttp::config_user_mask_put(const httplib::Request &request,
httplib::Response &response) {
if (request.body.empty()) {
response.status = 400;
response.set_content("Request body cannot be empty", "text/plain");
return;
}
if (request.body.size() % 4 != 0) {
response.status = 400;
response.set_content("Request has to be 32-bit", "text/plain");
return;
}
std::vector<uint32_t> v(request.body.size() / 4);
std::memcpy(v.data(), request.body.data(), request.body.size());
state_machine.SetUserPixelMask(v);
response.status = 200;
}
void JFJochBrokerHttp::version_get(httplib::Response &response) {
response.set_content(jfjoch_version(), "text/plain");
response.status = 200;
}
void JFJochBrokerHttp::config_instrument_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetInstrumentMetadata()), response);
}
void JFJochBrokerHttp::config_instrument_put(const Instrument_metadata &instrumentMetadata,
httplib::Response &response) {
state_machine.LoadInstrumentMetadata(Convert(instrumentMetadata));
response.status = 200;
}
void JFJochBrokerHttp::config_indexing_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetIndexingSettings()), response);
}
void JFJochBrokerHttp::config_indexing_put(const Indexing_settings &indexingSettings,
httplib::Response &response) {
state_machine.SetIndexingSettings(Convert(indexingSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_rugnux_get(httplib::Response &response) {
const RugnuxLiveSettings s = state_machine.GetRugnuxLive();
Rugnux_status ret;
ret.setEnable(s.enable);
ret.setReservedCpus(s.reserved_cpus);
ret.setDeadlineS(s.deadline_s);
ret.setModelLoaded(s.model != nullptr);
if (s.model)
ret.setModelSizeBytes(s.model_size_bytes);
ProcessOutput(ret, response);
}
void JFJochBrokerHttp::config_rugnux_put(const Rugnux_settings &rugnuxSettings,
httplib::Response &response) {
state_machine.SetRugnuxLive(rugnuxSettings.isEnable(), rugnuxSettings.getReservedCpus(),
rugnuxSettings.getDeadlineS());
response.status = 200;
}
void JFJochBrokerHttp::config_rugnux_model_put(const httplib::Request &request,
httplib::Response &response) {
if (request.body.empty()) {
response.status = 400;
response.set_content("Request body cannot be empty", "text/plain");
return;
}
if (request.body.size() > 64 * 1024 * 1024) {
response.status = 413;
response.set_content("Model cannot be larger than 64 MiB", "text/plain");
return;
}
std::shared_ptr<const RugnuxModel> model;
try {
model = std::make_shared<const RugnuxModel>(ParseRugnuxModel(request.body, "model"));
} catch (const std::exception &e) {
response.status = 400;
response.set_content(std::string("Not an atomic model: ") + e.what(), "text/plain");
return;
}
state_machine.SetRugnuxModel(std::move(model), static_cast<int64_t>(request.body.size()));
logger.Info("Rugnux model uploaded ({} bytes)", request.body.size());
response.status = 200;
}
void JFJochBrokerHttp::config_rugnux_model_delete(httplib::Response &response) {
state_machine.SetRugnuxModel(nullptr, 0);
logger.Info("Rugnux model removed");
response.status = 200;
}
void JFJochBrokerHttp::rugnux_status_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetRugnuxStatus()), response);
}
void JFJochBrokerHttp::config_bragg_integration_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetBraggIntegrationSettings()), response);
}
void JFJochBrokerHttp::config_bragg_integration_put(const Bragg_integration_settings &braggIntegrationSettings,
httplib::Response &response) {
state_machine.SetBraggIntegrationSettings(Convert(braggIntegrationSettings));
response.status = 200;
}
void JFJochBrokerHttp::result_scan_get(httplib::Response &response) {
auto ret = state_machine.GetScanResult();
if (ret.has_value())
ProcessOutput(Convert(ret.value()), response);
else
response.status = 404;
}
void JFJochBrokerHttp::config_dark_mask_put(const Dark_mask_settings &darkMaskSettings,
httplib::Response &response) {
state_machine.SetDarkMaskSettings(Convert(darkMaskSettings));
response.status = 200;
}
void JFJochBrokerHttp::config_dark_mask_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetDarkMaskSettings()), response);
}
void JFJochBrokerHttp::detector_status_get(httplib::Response &response) {
auto out = state_machine.GetDetectorStatus();
if (out)
ProcessOutput(Convert(out.value()), response);
else
response.status = 404;
}
void JFJochBrokerHttp::preview_pedestal_tiff_get(const std::optional<int32_t> &gainLevel,
const std::optional<int32_t> &sc,
httplib::Response &response) {
if (!gainLevel) {
response.status = 400;
response.set_content("gain_level is required", "text/plain");
return;
}
std::string s = state_machine.GetPedestalTIFF(gainLevel.value(), sc ? sc.value() : 0);
if (!s.empty())
response.set_content(s, "image/tiff");
else
response.status = 404;
}
void JFJochBrokerHttp::preview_plot_bin_get(const std::optional<std::string> &type,
const std::optional<std::string> &roi,
httplib::Response &response) {
std::vector<float> ret;
state_machine.GetPlotRaw(ret, ConvertPlotType(type), roi.value_or(""));
if (ret.empty()) {
response.status = 404;
return;
}
response.set_content(reinterpret_cast<const char *>(ret.data()),
ret.size() * sizeof(float),
"application/octet-stream");
response.status = 200;
}
void JFJochBrokerHttp::preview_plot_get(const std::optional<std::string> &type,
const std::optional<int32_t> &binning,
const std::optional<float> &fill,
const std::optional<bool> &experimentalCoord,
const std::optional<std::string> &azintUnit,
httplib::Response &response) {
PlotAzintUnit unit = PlotAzintUnit::Q_recipA;
if (azintUnit.has_value()) {
if (azintUnit == "Q_recipA" || azintUnit == "q_recipa")
unit = PlotAzintUnit::Q_recipA;
else if (azintUnit == "d_A" || azintUnit == "d_a")
unit = PlotAzintUnit::d_A;
else if (azintUnit == "two_theta_deg")
unit = PlotAzintUnit::TwoTheta_deg;
}
PlotRequest req{
.type = ConvertPlotType(type),
.binning = 1,
.experimental_coord = experimentalCoord.value_or(false),
.azint_unit = unit,
.fill_value = fill
};
if (binning) {
if (binning.value() < 0) {
response.status = 400;
response.set_content("Binning cannot be negative", "text/plain");
return;
}
req.binning = binning.value();
}
auto plot = state_machine.GetPlots(req);
ProcessOutput(Convert(plot), response);
}
void JFJochBrokerHttp::config_roi_get(httplib::Response &response) {
ProcessOutput(Convert(state_machine.GetROIDefintion()), response);
}
void JFJochBrokerHttp::config_roi_put(const Roi_definitions &roiDefinitions,
httplib::Response &response) {
state_machine.SetROIDefinition(Convert(roiDefinitions));
response.status = 200;
}
void JFJochBrokerHttp::xfel_event_code_get(httplib::Response &response) {
auto array = state_machine.GetXFELEventCode();
if (array.empty()) {
response.status = 404;
return;
}
nlohmann::json j = array;
response.set_content(j.dump(), "application/json");
response.status = 200;
}
void JFJochBrokerHttp::xfel_pulse_id_get(httplib::Response &response) {
auto array = state_machine.GetXFELPulseID();
if (array.empty()) {
response.status = 404;
return;
}
nlohmann::json j = array;
response.set_content(j.dump(), "application/json");
response.status = 200;
}
Reference in new issue View Git Blame Copy Permalink