mx/Jungfraujoch
0
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 37 Wiki Activity

JFJochReceiver: Transfer all radial integration profiles in one ProtoBuf

Browse Source
This commit is contained in:
leonarski_f
2023-05-09 22:17:50 +02:00
parent 5b49971dd7
commit e5739f5408
17 changed files with 324 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
broker/JFJochBroker.cpp
View File

@@ -54,6 +54,11 @@ grpc::Status JFJochBroker::GetPlots(grpc::ServerContext *context, const JFJochPr
GRPC_RUN( *response = state_machine.GetPlots(*request) );
}
grpc::Status JFJochBroker::GetRadialIntegrationProfiles(grpc::ServerContext *context, const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::RadialIntegrationProfiles *response) {
GRPC_RUN( *response = state_machine.GetRadialIntegrationProfiles() );
}
grpc::Status JFJochBroker::GetPreview(grpc::ServerContext *context, const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::PreviewFrame *response) {
GRPC_RUN( *response = services.GetPreviewFrame() );

3
broker/JFJochBroker.h
View File

@@ -44,6 +44,9 @@ public:
grpc::Status GetPlots(grpc::ServerContext *context, const JFJochProtoBuf::PlotRequest *request,
JFJochProtoBuf::Plot *response) override;
grpc::Status GetRadialIntegrationProfiles(grpc::ServerContext *context, const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::RadialIntegrationProfiles *response) override;
grpc::Status GetPreview(grpc::ServerContext *context, const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::PreviewFrame *response) override;

8
broker/JFJochServices.cpp
View File

@@ -151,6 +151,14 @@ JFJochProtoBuf::Plot JFJochServices::GetPlots(const JFJochProtoBuf::PlotRequest
}
}
JFJochProtoBuf::RadialIntegrationProfiles JFJochServices::GetRadialIntegrationProfiles() {
try {
return receiver.GetRadialIntegrationProfiles();
} catch (...) {
return JFJochProtoBuf::RadialIntegrationProfiles();
}
}
void JFJochServices::SetDataProcessingSettings(const JFJochProtoBuf::DataProcessingSettings &settings) {
receiver.SetDataProcessingSettings(settings);
}

1
broker/JFJochServices.h
View File

@@ -32,6 +32,7 @@ public:
JFJochProtoBuf::ReceiverStatus GetReceiverStatus();
JFJochProtoBuf::PreviewFrame GetPreviewFrame();
JFJochProtoBuf::Plot GetPlots(const JFJochProtoBuf::PlotRequest &request);
JFJochProtoBuf::RadialIntegrationProfiles GetRadialIntegrationProfiles();
void SetDataProcessingSettings(const JFJochProtoBuf::DataProcessingSettings &settings);
JFJochServices& Receiver(const std::string &addr);

4
broker/JFJochStateMachine.cpp
View File

@@ -501,6 +501,10 @@ JFJochProtoBuf::Plot JFJochStateMachine::GetPlots(const JFJochProtoBuf::PlotRequ
return services.GetPlots(request);
}
JFJochProtoBuf::RadialIntegrationProfiles JFJochStateMachine::GetRadialIntegrationProfiles() const {
return services.GetRadialIntegrationProfiles();
}
void JFJochStateMachine::SetDataProcessingSettings(const JFJochProtoBuf::DataProcessingSettings &settings) {
std::unique_lock<std::mutex> ul(data_processing_settings_mutex);
DiffractionExperiment::CheckDataProcessingSettings(settings);

1
broker/JFJochStateMachine.h
View File

@@ -85,6 +85,7 @@ public:
JFJochProtoBuf::BrokerStatus GetStatus() const;
JFJochProtoBuf::Plot GetPlots(const JFJochProtoBuf::PlotRequest &request) const;
JFJochProtoBuf::RadialIntegrationProfiles GetRadialIntegrationProfiles() const;
JFJochProtoBuf::Image GetNeXusMask() const;
JFJochProtoBuf::Image GetPedestalG0() const;

15
grpc/JFJochReceiverClient.cpp
View File

@@ -143,4 +143,19 @@ JFJochProtoBuf::Plot JFJochReceiverClient::GetPlots(const JFJochProtoBuf::PlotRe
// TODO: Write some dummy plots
}
return ret;
}
JFJochProtoBuf::RadialIntegrationProfiles JFJochReceiverClient::GetRadialIntegrationProfiles() {
JFJochProtoBuf::RadialIntegrationProfiles ret;
if (_stub) {
grpc::ClientContext context;
JFJochProtoBuf::Empty empty;
auto status = _stub->GetRadialIntegrationProfiles(&context, empty, &ret);
if (!status.ok()) throw JFJochException(JFJochExceptionCategory::gRPCError,
"JFJochReceiver: " + status.error_message());
} else {
// TODO: Write some dummy plots
}
return ret;
}

1
grpc/JFJochReceiverClient.h
View File

@@ -23,6 +23,7 @@ public:
JFJochProtoBuf::PreviewFrame GetPreviewFrame();
JFJochProtoBuf::ReceiverNetworkConfig GetNetworkConfig();
JFJochProtoBuf::Plot GetPlots(const JFJochProtoBuf::PlotRequest& request);
JFJochProtoBuf::RadialIntegrationProfiles GetRadialIntegrationProfiles();
};

14
grpc/jfjoch.proto
View File

@@ -275,11 +275,9 @@ message PlotRequest {
optional uint64 file_number = 3;
}
message ReceiverDataProcessingPlots {
Plot bkg_estimate = 1;
Plot radial_int_profile = 2;
Plot spot_count = 3;
Plot indexing_rate = 4;
message RadialIntegrationProfiles {
repeated Plot file_plots = 1;
Plot total = 2;
}
// Writer
@@ -384,10 +382,6 @@ message FPGAStatus {
uint32 error_packet_len = 39;
}
message IndexerDataProcessingPlots {
Plot indexing_rate = 1;
}
message DataProcessingSettings {
float signal_to_noise_threshold = 1; // STRONG_PIXEL in XDS
int64 photon_count_threshold = 2; // Threshold in photon counts
@@ -505,6 +499,7 @@ service gRPC_JFJochReceiver {
rpc GetStatus (Empty) returns (ReceiverStatus) {}
rpc SetDataProcessingSettings(DataProcessingSettings) returns (Empty) {}
rpc GetDataProcessingPlots(PlotRequest) returns (Plot) {}
rpc GetRadialIntegrationProfiles(Empty) returns (RadialIntegrationProfiles) {}
rpc GetPreviewFrame (Empty) returns (PreviewFrame) {}
rpc GetNetworkConfig (Empty) returns (ReceiverNetworkConfig) {}
}
@@ -541,6 +536,7 @@ service gRPC_JFJochBroker {
rpc GetDataProcessingSettings (Empty) returns (DataProcessingSettings) {}
rpc PutDataProcessingSettings (DataProcessingSettings) returns (Empty) {}
rpc GetPlots (PlotRequest) returns (Plot) {}
rpc GetRadialIntegrationProfiles(Empty) returns (RadialIntegrationProfiles) {}
rpc GetPreview (Empty) returns (PreviewFrame) {}
rpc GetPedestalG0 (Empty) returns (Image) {}

11
python/jfjoch_grpc2http.py
View File

@@ -199,6 +199,17 @@ async def get_settings(request: Request):
raise HTTPException(status_code=400, detail=e.details())
@app.post("/data_processing/rad_int_profiles")
async def get_settings():
try:
stub = jfjoch_pb2_grpc.gRPC_JFJochBrokerStub(channel)
return MessageToDict(
stub.GetRadialIntegrationProfiles(jfjoch_pb2.Empty()), including_default_value_fields=True
)
except grpc.RpcError as e:
raise HTTPException(status_code=400, detail=e.details())
@app.get("/detector/measurement_statistics")
async def get_meas_stats():
try:

132
python/jfjoch_pb2.py
View File

File diff suppressed because one or more lines are too long

66
python/jfjoch_pb2_grpc.py
View File

@@ -49,6 +49,11 @@ class gRPC_JFJochReceiverStub(object):
request_serializer=jfjoch__pb2.PlotRequest.SerializeToString,
response_deserializer=jfjoch__pb2.Plot.FromString,
)
self.GetRadialIntegrationProfiles = channel.unary_unary(
'/JFJochProtoBuf.gRPC_JFJochReceiver/GetRadialIntegrationProfiles',
request_serializer=jfjoch__pb2.Empty.SerializeToString,
response_deserializer=jfjoch__pb2.RadialIntegrationProfiles.FromString,
)
self.GetPreviewFrame = channel.unary_unary(
'/JFJochProtoBuf.gRPC_JFJochReceiver/GetPreviewFrame',
request_serializer=jfjoch__pb2.Empty.SerializeToString,
@@ -106,6 +111,12 @@ class gRPC_JFJochReceiverServicer(object):
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetRadialIntegrationProfiles(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetPreviewFrame(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
@@ -156,6 +167,11 @@ def add_gRPC_JFJochReceiverServicer_to_server(servicer, server):
request_deserializer=jfjoch__pb2.PlotRequest.FromString,
response_serializer=jfjoch__pb2.Plot.SerializeToString,
),
'GetRadialIntegrationProfiles': grpc.unary_unary_rpc_method_handler(
servicer.GetRadialIntegrationProfiles,
request_deserializer=jfjoch__pb2.Empty.FromString,
response_serializer=jfjoch__pb2.RadialIntegrationProfiles.SerializeToString,
),
'GetPreviewFrame': grpc.unary_unary_rpc_method_handler(
servicer.GetPreviewFrame,
request_deserializer=jfjoch__pb2.Empty.FromString,
@@ -295,6 +311,23 @@ class gRPC_JFJochReceiver(object):
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetRadialIntegrationProfiles(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/JFJochProtoBuf.gRPC_JFJochReceiver/GetRadialIntegrationProfiles',
jfjoch__pb2.Empty.SerializeToString,
jfjoch__pb2.RadialIntegrationProfiles.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetPreviewFrame(request,
target,
@@ -767,6 +800,11 @@ class gRPC_JFJochBrokerStub(object):
request_serializer=jfjoch__pb2.PlotRequest.SerializeToString,
response_deserializer=jfjoch__pb2.Plot.FromString,
)
self.GetRadialIntegrationProfiles = channel.unary_unary(
'/JFJochProtoBuf.gRPC_JFJochBroker/GetRadialIntegrationProfiles',
request_serializer=jfjoch__pb2.Empty.SerializeToString,
response_deserializer=jfjoch__pb2.RadialIntegrationProfiles.FromString,
)
self.GetPreview = channel.unary_unary(
'/JFJochProtoBuf.gRPC_JFJochBroker/GetPreview',
request_serializer=jfjoch__pb2.Empty.SerializeToString,
@@ -897,6 +935,12 @@ class gRPC_JFJochBrokerServicer(object):
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetRadialIntegrationProfiles(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetPreview(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
@@ -1017,6 +1061,11 @@ def add_gRPC_JFJochBrokerServicer_to_server(servicer, server):
request_deserializer=jfjoch__pb2.PlotRequest.FromString,
response_serializer=jfjoch__pb2.Plot.SerializeToString,
),
'GetRadialIntegrationProfiles': grpc.unary_unary_rpc_method_handler(
servicer.GetRadialIntegrationProfiles,
request_deserializer=jfjoch__pb2.Empty.FromString,
response_serializer=jfjoch__pb2.RadialIntegrationProfiles.SerializeToString,
),
'GetPreview': grpc.unary_unary_rpc_method_handler(
servicer.GetPreview,
request_deserializer=jfjoch__pb2.Empty.FromString,
@@ -1317,6 +1366,23 @@ class gRPC_JFJochBroker(object):
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetRadialIntegrationProfiles(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/JFJochProtoBuf.gRPC_JFJochBroker/GetRadialIntegrationProfiles',
jfjoch__pb2.Empty.SerializeToString,
jfjoch__pb2.RadialIntegrationProfiles.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetPreview(request,
target,

12
receiver/JFJochReceiver.cpp
View File

@@ -683,6 +683,18 @@ JFJochProtoBuf::Plot JFJochReceiver::GetPlots(const JFJochProtoBuf::PlotRequest
return ret;
}
JFJochProtoBuf::RadialIntegrationProfiles JFJochReceiver::GetRadialIntegrationProfiles() {
JFJochProtoBuf::RadialIntegrationProfiles ret;
for (auto &i: rad_int_profile_per_file)
i->GetPlot(*ret.add_file_plots());
if (rad_int_profile)
rad_int_profile->GetPlot(*ret.mutable_total());
return ret;
}
void JFJochReceiver::PrepareConversionOnCPU() {
if (experiment.GetStorageCellNumber() != 1)
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,

1
receiver/JFJochReceiver.h
View File

@@ -130,6 +130,7 @@ public:
float GetAvailableSendBuffers() const;
JFJochProtoBuf::Plot GetPlots(const JFJochProtoBuf::PlotRequest& request);
JFJochProtoBuf::RadialIntegrationProfiles GetRadialIntegrationProfiles();
};
#endif //JUNGFRAUJOCH_JFJOCHRECEIVER_H

16
receiver/JFJochReceiverService.cpp
View File

@@ -138,9 +138,9 @@ grpc::Status JFJochReceiverService::GetStatus(grpc::ServerContext *context, cons
return grpc::Status::OK;
}
grpc::Status
JFJochReceiverService::GetDataProcessingPlots(grpc::ServerContext *context, const JFJochProtoBuf::PlotRequest *request,
JFJochProtoBuf::Plot *response) {
grpc::Status JFJochReceiverService::GetDataProcessingPlots(grpc::ServerContext *context,
const JFJochProtoBuf::PlotRequest *request,
JFJochProtoBuf::Plot *response) {
// Need to hold mutex, as receiver might not exist here, if state is idle
std::unique_lock<std::mutex> ul(state_mutex);
if (receiver)
@@ -148,6 +148,16 @@ JFJochReceiverService::GetDataProcessingPlots(grpc::ServerContext *context, cons
return grpc::Status::OK;
}
grpc::Status JFJochReceiverService::GetRadialIntegrationProfiles(grpc::ServerContext *context,
const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::RadialIntegrationProfiles *response) {
// Need to hold mutex, as receiver might not exist here, if state is idle
std::unique_lock<std::mutex> ul(state_mutex);
if (receiver)
*response = receiver->GetRadialIntegrationProfiles();
return grpc::Status::OK;
}
grpc::Status JFJochReceiverService::SetDataProcessingSettings(grpc::ServerContext *context,
const JFJochProtoBuf::DataProcessingSettings *request,
JFJochProtoBuf::Empty *response) {

2
receiver/JFJochReceiverService.h
View File

@@ -52,6 +52,8 @@ public:
JFJochProtoBuf::ReceiverNetworkConfig *response) override;
grpc::Status GetDataProcessingPlots(grpc::ServerContext *context, const JFJochProtoBuf::PlotRequest *request,
JFJochProtoBuf::Plot *response) override;
grpc::Status GetRadialIntegrationProfiles(grpc::ServerContext *context, const JFJochProtoBuf::Empty *request,
JFJochProtoBuf::RadialIntegrationProfiles *response) override;
};

111
tests/JFJochFullIntegrationTest.cpp
View File

@@ -1274,6 +1274,117 @@ TEST_CASE("JFJochIntegrationTest_ZMQ_lysozyme_spot_and_index", "[JFJochReceiver]
}
TEST_CASE("JFJochIntegrationTest_ZMQ_lysozyme_rad_int", "[JFJochReceiver]") {
Logger logger("JFJochIntegrationTest_ZMQ_lysozyme_rad_int");
RegisterHDF5Filter();
size_t nimages = 24;
int64_t ndatastream = 1;
int64_t nmodules = 8;
JFJochServices services(logger);
JFJochStateMachine state_machine(services, logger);
state_machine.AddDetectorSetup(DetectorGeometry(ndatastream * nmodules, 2, 8, 36));
state_machine.NotThreadSafe_Experiment().DataStreams(ndatastream);
state_machine.NotThreadSafe_Experiment().PedestalG0Frames(0).PedestalG1Frames(0).PedestalG2Frames(0)
.SpotFindingPeriod(10ms);
state_machine.NotThreadSafe_Experiment().LowQForRadialInt_recipA(0.5).HighQForRadialInt_recipA(3.5)
.QSpacingForRadialInt_recipA(1.0);
services.Writer("unix:writer_test", "inproc://#1").Receiver("unix:fpga_receiver_test");
logger.Verbose(true);
std::vector<uint16_t> image(RAW_MODULE_SIZE);
HDF5File data("../../tests/test_data/compression_benchmark.h5", false, false, false);
HDF5DataSet dataset(data, "/entry/data/data");
HDF5DataSpace file_space(dataset);
std::vector<int16_t> image_conv(nimages * file_space.GetDimensions()[1] * file_space.GetDimensions()[2]);
std::vector<hsize_t> start = {0, 0, 0};
std::vector<hsize_t> file_size = {nimages, file_space.GetDimensions()[1], file_space.GetDimensions()[2]};
dataset.ReadVector(image_conv, start, file_size);
std::vector<int16_t> image_raw_geom(
nimages * state_machine.NotThreadSafe_Experiment().GetModulesNum() * RAW_MODULE_SIZE * sizeof(int16_t));
for (int i = 0; i < nimages; i++)
ConvertedToRawGeometry(state_machine.NotThreadSafe_Experiment(),
image_raw_geom.data() +
i * RAW_MODULE_SIZE * state_machine.NotThreadSafe_Experiment().GetModulesNum(),
image_conv.data() + i * file_space.GetDimensions()[1] * file_space.GetDimensions()[2]);
std::vector<std::unique_ptr<AcquisitionDevice>> aq_devices;
auto *test = new MockAcquisitionDevice(0, 512);
for (int m = 0; m < state_machine.NotThreadSafe_Experiment().GetModulesNum(0); m++) {
for (int image_num = 1; image_num <= nimages; image_num++)
test->AddModule(image_num, m, (uint16_t *) (image_raw_geom.data()
+ (m + (image_num - 1) *
state_machine.NotThreadSafe_Experiment().GetModulesNum(
0))
* RAW_MODULE_SIZE));
}
test->Terminate();
aq_devices.emplace_back(test);
ZMQContext zmq_context;
std::vector<AcquisitionDevice *> tmp_devices;
for (const auto &i: aq_devices)
tmp_devices.emplace_back(i.get());
ZMQImagePusher pusher(zmq_context, {"inproc://#1"});
JFJochReceiverService fpga_receiver(tmp_devices, logger, pusher);
JFJochWriterService writer(zmq_context, logger);;
auto fpga_receiver_server = gRPCServer("unix:fpga_receiver_test", fpga_receiver);
auto writer_server = gRPCServer("unix:writer_test", writer);
REQUIRE_NOTHROW(state_machine.Initialize());
logger.Info("Initialized");
JFJochProtoBuf::DatasetSettings setup;
setup.set_ntrigger(1);
setup.set_detector_distance_mm(75);
setup.set_summation(1);
setup.set_file_prefix("spot_finding_test");
setup.set_images_per_trigger(nimages);
setup.set_photon_energy_kev(12.4);
setup.set_beam_x_pxl(1090);
setup.set_beam_y_pxl(1136);
setup.mutable_unit_cell()->set_a(36.9);
setup.mutable_unit_cell()->set_b(78.95);
setup.mutable_unit_cell()->set_c(78.95);
setup.mutable_unit_cell()->set_alpha(90.0);
setup.mutable_unit_cell()->set_beta(90.0);
setup.mutable_unit_cell()->set_gamma(90.0);
setup.set_data_file_count(4);
REQUIRE_NOTHROW(state_machine.Start(setup));
logger.Info("Started measurement");
REQUIRE_NOTHROW(state_machine.Stop());
logger.Info("Stopped measurement");
auto rad_int = state_machine.GetRadialIntegrationProfiles();
REQUIRE(rad_int.file_plots_size() == 4);
REQUIRE(rad_int.file_plots(0).x_size() == 3);
REQUIRE(rad_int.file_plots(0).y_size() == 3);
REQUIRE(rad_int.file_plots(3).x_size() == 3);
REQUIRE(rad_int.file_plots(3).y_size() == 3);
REQUIRE(rad_int.file_plots(3).x(0) == Approx(1.0));
REQUIRE(rad_int.total().x_size() == 3);
fpga_receiver_server->Shutdown();
writer_server->Shutdown();
}
TEST_CASE("JFJochIntegrationTest_ZMQ_lysozyme_spot_and_index_sum", "[JFJochReceiver]") {
Logger logger("JFJochIntegrationTest_ZMQ_lysozyme_spot_and_index_sum");