receiver: the indexer belongs to the two receivers that index
IndexAndRefine and the indexer pool move from the base JFJochReceiver into JFJochReceiverLite and JFJochReceiverFPGA. JFJochReceiverStream no longer takes the pool, and the rugnux receivers no longer pass it nullptr. The consensus cell and the indexer-error report of the end message are added by both through one base helper, AddIndexingToEndMessage. No change for Lite and FPGA; the other receivers carried an indexer that never ran and reported no cell from it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK
This commit is contained in:
1 parent
7080109a4a
commit
97cfae1fd8
11 files changed
+47
-34
No files matched your search
+16
-16
@@ -4,6 +4,7 @@
|
||||
#include "JFJochReceiver.h"
|
||||
#include "../common/time_utc.h"
|
||||
#include "../frame_serialize/CBORStream2Serializer.h"
|
||||
#include "../image_analysis/IndexAndRefine.h"
|
||||
|
||||
JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
|
||||
ImageBuffer &in_image_buffer,
|
||||
@@ -14,8 +15,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
|
||||
Logger &logger,
|
||||
const PixelMask &in_pixel_mask,
|
||||
ZMQPreviewSocket *in_zmq_preview_socket,
|
||||
ZMQMetadataSocket *in_zmq_metadata_socket,
|
||||
IndexerThreadPool *indexing_thread_pool)
|
||||
ZMQMetadataSocket *in_zmq_metadata_socket)
|
||||
: logger(logger),
|
||||
experiment(in_experiment),
|
||||
spot_finding_settings(spot_finding_settings),
|
||||
@@ -27,9 +27,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
|
||||
plots(in_plots),
|
||||
scan_result(in_experiment),
|
||||
serialmx_filter(in_experiment),
|
||||
pixel_mask(in_pixel_mask),
|
||||
// retain_outcomes=false: the receivers index only, there is no scaling or merging at the end
|
||||
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true) {
|
||||
pixel_mask(in_pixel_mask) {
|
||||
logger.Debug("Initializing receiver");
|
||||
// Ensure there is nothing running for now
|
||||
if (!image_buffer.Finalize(std::chrono::seconds(1)))
|
||||
@@ -265,17 +263,6 @@ void JFJochReceiver::SendEndMessage() {
|
||||
|
||||
message.az_int_result["dataset"] = plots.GetAzIntProfile();
|
||||
|
||||
message.unit_cell = indexer.GetConsensusUnitCell();
|
||||
// The images are written regardless, but the indexing rate is then not a statement about the crystal
|
||||
if (const int64_t error_count = indexer.GetIndexerErrorCount(); error_count > 0) {
|
||||
try {
|
||||
indexer.RethrowIndexerError();
|
||||
} catch (const std::exception &e) {
|
||||
logger.Error("Indexing failed {} times in this collection, first with: {}. Those attempts are "
|
||||
"left out of the indexing rate", error_count, e.what());
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < adu_histogram_module.size(); i++)
|
||||
message.adu_histogram["module" + std::to_string(i)] = adu_histogram_module[i]->GetHistogram();
|
||||
|
||||
@@ -304,6 +291,19 @@ void JFJochReceiver::SendEndMessage() {
|
||||
zmq_preview_socket->EndDataCollection(message);
|
||||
}
|
||||
|
||||
void JFJochReceiver::AddIndexingToEndMessage(EndMessage &message, const IndexAndRefine &indexer) {
|
||||
message.unit_cell = indexer.GetConsensusUnitCell();
|
||||
// The images are written regardless, but the indexing rate is then not a statement about the crystal
|
||||
if (const int64_t error_count = indexer.GetIndexerErrorCount(); error_count > 0) {
|
||||
try {
|
||||
indexer.RethrowIndexerError();
|
||||
} catch (const std::exception &e) {
|
||||
logger.Error("Indexing failed {} times in this collection, first with: {}. Those attempts are "
|
||||
"left out of the indexing rate", error_count, e.what());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
JFJochReceiverOutput JFJochReceiver::GetFinalStatistics() const {
|
||||
JFJochReceiverOutput ret;
|
||||
ret.efficiency = GetEfficiency();
|
||||
|
||||
@@ -18,8 +18,8 @@
|
||||
#include "LossyFilter.h"
|
||||
#include "../common/MovingAverage.h"
|
||||
#include "../common/ScanResultGenerator.h"
|
||||
#include "../image_analysis/indexing/IndexerThreadPool.h"
|
||||
#include "../image_analysis/IndexAndRefine.h"
|
||||
|
||||
class IndexAndRefine;
|
||||
|
||||
class JFJochReceiver {
|
||||
protected:
|
||||
@@ -82,8 +82,6 @@ protected:
|
||||
std::optional<uint64_t> max_delay;
|
||||
std::mutex max_delay_mutex;
|
||||
|
||||
IndexAndRefine indexer;
|
||||
|
||||
std::string writer_error;
|
||||
|
||||
void UpdateMaxDelay(uint64_t delay);
|
||||
@@ -107,6 +105,8 @@ protected:
|
||||
|
||||
// The last word on the end message: a receiver with an analysis of its own fills in what it found
|
||||
virtual void AddToEndMessage(EndMessage &message) {}
|
||||
// For the receivers that index: the consensus cell, and how often the indexer failed
|
||||
void AddIndexingToEndMessage(EndMessage &message, const IndexAndRefine &indexer);
|
||||
virtual uint64_t GetImagesDropped() const { return 0; }
|
||||
virtual void Cancel(const JFJochException &e);
|
||||
public:
|
||||
@@ -119,8 +119,7 @@ public:
|
||||
Logger &logger,
|
||||
const PixelMask &pixel_mask,
|
||||
ZMQPreviewSocket *zmq_preview_socket = nullptr,
|
||||
ZMQMetadataSocket *zmq_metadata_socket = nullptr,
|
||||
IndexerThreadPool *indexing_thread_pool = nullptr);
|
||||
ZMQMetadataSocket *zmq_metadata_socket = nullptr);
|
||||
virtual ~JFJochReceiver();
|
||||
|
||||
virtual void StopReceiver() = 0;
|
||||
|
||||
@@ -29,9 +29,10 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
|
||||
in_logger,
|
||||
in_pixel_mask,
|
||||
in_zmq_preview_socket,
|
||||
in_zmq_metadata_socket,
|
||||
indexing_thread_pool),
|
||||
in_zmq_metadata_socket),
|
||||
calibration(nullptr),
|
||||
// retain_outcomes=false: the receiver indexes only, there is no scaling or merging at the end
|
||||
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true),
|
||||
acquisition_device(in_aq_device),
|
||||
ndatastreams(experiment.GetDataStreamsNum()),
|
||||
frame_transformation_nthreads(
|
||||
@@ -498,6 +499,10 @@ float JFJochReceiverFPGA::GetProgress() const {
|
||||
return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames);
|
||||
}
|
||||
|
||||
void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) {
|
||||
AddIndexingToEndMessage(message, indexer);
|
||||
}
|
||||
|
||||
void JFJochReceiverFPGA::FinalizeMeasurement() {
|
||||
for (auto &future: frame_transformation_futures)
|
||||
future.get();
|
||||
|
||||
@@ -39,6 +39,8 @@
|
||||
class JFJochReceiverFPGA : public JFJochReceiver {
|
||||
const JFCalibration *calibration;
|
||||
|
||||
IndexAndRefine indexer;
|
||||
|
||||
std::vector<uint16_t> roi_map;
|
||||
|
||||
std::vector<std::future<void>> frame_transformation_futures;
|
||||
@@ -70,6 +72,7 @@ class JFJochReceiverFPGA : public JFJochReceiver {
|
||||
uint32_t threadid,
|
||||
ModuleSummation &summation);
|
||||
void Cancel(const JFJochException &e) override;
|
||||
void AddToEndMessage(EndMessage &message) override;
|
||||
void FinalizeMeasurement();
|
||||
|
||||
void SendCalibration();
|
||||
|
||||
@@ -52,8 +52,9 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen
|
||||
in_plots,
|
||||
in_image_buffer,
|
||||
in_zmq_preview_socket,
|
||||
in_zmq_metadata_socket,
|
||||
indexing_thread_pool),
|
||||
in_zmq_metadata_socket),
|
||||
// retain_outcomes=false: the receiver indexes only, there is no scaling or merging at the end
|
||||
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true),
|
||||
analysis(nthreads) {
|
||||
StartThreads();
|
||||
}
|
||||
@@ -89,6 +90,11 @@ void JFJochReceiverLite::ImageThread(uint32_t id) {
|
||||
logger.Debug("Thread {} finished", id);
|
||||
}
|
||||
|
||||
void JFJochReceiverLite::AddToEndMessage(EndMessage &message) {
|
||||
JFJochReceiverStream::AddToEndMessage(message);
|
||||
AddIndexingToEndMessage(message, indexer);
|
||||
}
|
||||
|
||||
void JFJochReceiverLite::OnImage(uint32_t id, DataMessage &data_msg, std::chrono::steady_clock::time_point received) {
|
||||
auto start = std::chrono::high_resolution_clock::now();
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
#include "../image_analysis/MXAnalysisWithoutFPGA.h"
|
||||
|
||||
class JFJochReceiverLite : public JFJochReceiverStream {
|
||||
IndexAndRefine indexer;
|
||||
std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> analysis; // per image thread
|
||||
|
||||
// Each thread builds a GPU analysis of its own
|
||||
@@ -17,6 +18,7 @@ class JFJochReceiverLite : public JFJochReceiverStream {
|
||||
|
||||
void ImageThread(uint32_t id) override;
|
||||
void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
|
||||
void AddToEndMessage(EndMessage &message) override;
|
||||
public:
|
||||
JFJochReceiverLite(const DiffractionExperiment& experiment,
|
||||
const PixelMask &pixel_mask,
|
||||
|
||||
@@ -100,7 +100,7 @@ JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExpe
|
||||
ZMQMetadataSocket *in_zmq_metadata_socket)
|
||||
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger,
|
||||
in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
|
||||
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr),
|
||||
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
|
||||
reader(in_image_buffer),
|
||||
rugnux_nthreads(in_rugnux_nthreads),
|
||||
host_budget(RugnuxHostBudget(in_image_buffer)),
|
||||
|
||||
@@ -26,7 +26,7 @@ JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperime
|
||||
IndexerThreadPool *in_indexing_thread_pool)
|
||||
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger,
|
||||
in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
|
||||
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr),
|
||||
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
|
||||
model(std::move(in_model)),
|
||||
deadline_s(in_deadline_s),
|
||||
live_status(in_live_status),
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
#include "FPGAImagePuller.h"
|
||||
#include "../preview/PreviewImage.h"
|
||||
#include "../image_puller/ZMQImagePuller.h"
|
||||
#include "../image_analysis/indexing/IndexerThreadPool.h"
|
||||
#include "../acquisition_device/AcquisitionDeviceGroup.h"
|
||||
|
||||
#include "../preview/ZMQMetadataSocket.h"
|
||||
|
||||
@@ -44,8 +44,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
|
||||
JFJochReceiverPlots &in_plots,
|
||||
ImageBuffer &in_image_buffer,
|
||||
ZMQPreviewSocket *in_zmq_preview_socket,
|
||||
ZMQMetadataSocket *in_zmq_metadata_socket,
|
||||
IndexerThreadPool *indexing_thread_pool)
|
||||
ZMQMetadataSocket *in_zmq_metadata_socket)
|
||||
: JFJochReceiver(in_experiment,
|
||||
in_image_buffer,
|
||||
in_image_pusher,
|
||||
@@ -55,8 +54,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
|
||||
in_logger,
|
||||
in_pixel_mask,
|
||||
in_zmq_preview_socket,
|
||||
in_zmq_metadata_socket,
|
||||
indexing_thread_pool),
|
||||
in_zmq_metadata_socket),
|
||||
image_puller(in_image_puller),
|
||||
nthreads(in_nthreads),
|
||||
measurement_started(1),
|
||||
|
||||
@@ -68,8 +68,7 @@ public:
|
||||
JFJochReceiverPlots &plots,
|
||||
ImageBuffer &image_buffer,
|
||||
ZMQPreviewSocket *zmq_preview_socket = nullptr,
|
||||
ZMQMetadataSocket *zmq_metadata_socket = nullptr,
|
||||
IndexerThreadPool *indexing_thread_pool = nullptr);
|
||||
ZMQMetadataSocket *zmq_metadata_socket = nullptr);
|
||||
|
||||
void StopReceiver() override;
|
||||
void Cancel(bool silent) override;
|
||||
|
||||
Reference in new issue
Block a user