// Copyright (2019-2023) Paul Scherrer Institute #include "JFJochReceiverService.h" JFJochReceiverService::JFJochReceiverService(AcquisitionDeviceGroup &in_aq_devices, Logger &in_logger, ImagePusher &pusher, size_t send_buffer_size_MiB) : logger(in_logger), aq_devices(in_aq_devices), image_pusher(pusher), spot_finding_settings(DiffractionExperiment::DefaultDataProcessingSettings()), buffer(send_buffer_size_MiB * 1024 * 1024) {} JFJochReceiverService& JFJochReceiverService::NumThreads(int64_t input) { if (input <= 0) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Thread number must be above zero"); nthreads = input; return *this; } JFJochReceiverService &JFJochReceiverService::NUMAPolicy(const NUMAHWPolicy &policy) { numa_policy = policy; return *this; } JFJochReceiverService &JFJochReceiverService::NUMAPolicy(const std::string &policy) { numa_policy = NUMAHWPolicy(policy); return *this; } void JFJochReceiverService::FinalizeMeasurement() { receiver->StopReceiver(); { std::unique_lock ul(state_mutex); state = ReceiverState::Idle; measurement_done.notify_all(); } } std::optional JFJochReceiverService::GetStatus() { // Need to hold mutex, as receiver might not exist here, if state is idle std::unique_lock ul(state_mutex); if (state == ReceiverState::Running) return receiver->GetStatus(); else return {}; } void JFJochReceiverService::Start(const DiffractionExperiment &experiment, const JFCalibration *calibration) { std::unique_lock ul_state(state_mutex); // unique lock, as it will destroy and create receiver object if (state != ReceiverState::Idle) throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Receiver not idle, cannot start"); try { // Thanks to properties of unique_ptr, starting new measurement will call destructor of JFJochReceiver, which will // ensure that everything was rolled back receiver = std::make_unique(experiment, calibration, aq_devices, image_pusher, logger, nthreads, numa_policy, spot_finding_settings, buffer); try { // Don't want to stop receiver->SetSpotFindingSettings(spot_finding_settings); } catch (...) {} measurement = std::async(std::launch::async, &JFJochReceiverService::FinalizeMeasurement, this); state = ReceiverState::Running; } catch (const JFJochException &e) { logger.ErrorException(e); throw; } } void JFJochReceiverService::Cancel(bool silent) { std::unique_lock ul(state_mutex); if (state == ReceiverState::Running) receiver->Cancel(silent); } JFJochReceiverOutput JFJochReceiverService::Stop() { std::unique_lock ul(state_mutex); measurement_done.wait(ul, [this] { return (state != ReceiverState::Running);}); if (state != ReceiverState::Idle) throw JFJochException(JFJochExceptionCategory::WrongReceiverState, "Receiver in weird state"); try { if (measurement.valid()) measurement.get(); } catch (JFJochException &e) { logger.ErrorException(e); throw; } if (!receiver) { logger.Warning("Request to stop while receiver not running"); throw JFJochException(JFJochExceptionCategory::WrongReceiverState, "Receiver idle, cannot stop"); } return receiver->GetStatistics(); } void JFJochReceiverService::SetSpotFindingSettings(const SpotFindingSettings &settings) { try { std::unique_lock ul(state_mutex); DiffractionExperiment::CheckDataProcessingSettings(settings); spot_finding_settings = settings; if (state != ReceiverState::Idle) receiver->SetSpotFindingSettings(settings); } catch (std::exception &e) { logger.ErrorException(e); throw; } } MultiLinePlot JFJochReceiverService::GetDataProcessingPlot(const PlotRequest &request) { // Need to hold mutex, as receiver might not exist here, if state is idle std::unique_lock ul(state_mutex); if (receiver) return receiver->GetPlots(request); else return {}; } std::vector JFJochReceiverService::GetNetworkConfig() { return aq_devices.GetNetworkConfig(); } std::string JFJochReceiverService::GetTIFF(bool calibration) const { std::unique_lock ul(state_mutex); if (receiver) return receiver->GetTIFF(calibration); else return ""; } std::string JFJochReceiverService::GetJPEG(const PreviewJPEGSettings &settings) const { std::unique_lock ul(state_mutex); if (receiver) return receiver->GetJPEG(settings); else return ""; } void JFJochReceiverService::LoadInternalGeneratorImage(const DiffractionExperiment &experiment, const std::vector &image, uint64_t image_number) { std::vector raw_geom, eiger_geom; const uint16_t *frame; if (image.size() == RAW_MODULE_SIZE * experiment.GetModulesNum()) { frame = image.data(); } else if (image.size() == experiment.GetPixelsNum()) { raw_geom.resize(RAW_MODULE_SIZE * experiment.GetModulesNum()); ConvertedToRawGeometry(experiment, raw_geom.data(), image.data()); frame = raw_geom.data(); } else throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Size of input array with raw expected image is wrong"); for (int i = 0; i < experiment.GetDataStreamsNum(); i++) { uint32_t module0 = experiment.GetFirstModuleOfDataStream(i); switch (experiment.GetDetectorSetup().GetDetectorType()) { case DetectorType::EIGER: eiger_geom.resize(RAW_MODULE_SIZE); for (int m = 0; m < experiment.GetModulesNum(i); m++) { RawToEigerInput(eiger_geom.data(), frame + (module0 + m) * RAW_MODULE_SIZE); aq_devices[i].SetInternalGeneratorFrame(eiger_geom.data(), m + experiment.GetModulesNum(i) * image_number); } break; case DetectorType::JUNGFRAU: for (int m = 0; m < experiment.GetModulesNum(i); m++) aq_devices[i].SetInternalGeneratorFrame(frame + (module0 + m) * RAW_MODULE_SIZE, m + experiment.GetModulesNum(i) * image_number); break; default: throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Detector not supported"); } } } void JFJochReceiverService::GetXFELEventCode(std::vector &v) const { std::unique_lock ul(state_mutex); if (receiver) return receiver->GetXFELEventCode(v); } void JFJochReceiverService::GetXFELPulseID(std::vector &v) const { std::unique_lock ul(state_mutex); if (receiver) return receiver->GetXFELPulseID(v); }