From 94ea1c4468fed5ee4be48e3334fe75541ecaba02 Mon Sep 17 00:00:00 2001 From: Thomas White Date: Tue, 8 Jan 2019 15:44:52 +0100 Subject: ZMQ connection and setup stuff --- src/im-sandbox.c | 138 ++++++++++++------ src/im-sandbox.h | 7 +- src/im-zmq.c | 402 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/im-zmq.h | 57 ++++++++ src/indexamajig.c | 20 ++- src/process_image.c | 6 +- src/process_image.h | 7 +- src/zmq.c | 313 ---------------------------------------- src/zmq.h | 48 ------- 9 files changed, 582 insertions(+), 416 deletions(-) create mode 100644 src/im-zmq.c create mode 100644 src/im-zmq.h delete mode 100644 src/zmq.c delete mode 100644 src/zmq.h (limited to 'src') diff --git a/src/im-sandbox.c b/src/im-sandbox.c index e8fbe763..37574310 100644 --- a/src/im-sandbox.c +++ b/src/im-sandbox.c @@ -3,17 +3,18 @@ * * Sandbox for indexing * - * Copyright © 2012-2018 Deutsches Elektronen-Synchrotron DESY, + * Copyright © 2012-2019 Deutsches Elektronen-Synchrotron DESY, * a research centre of the Helmholtz Association. * Copyright © 2012 Richard Kirian * Copyright © 2012 Lorenzo Galli * * Authors: - * 2010-2018 Thomas White + * 2010-2019 Thomas White * 2014 Valerio Mariani * 2011 Richard Kirian * 2012 Lorenzo Galli * 2012 Chunhong Yoon + * 2017 Stijn de Graaf * * This file is part of CrystFEL. * @@ -64,6 +65,7 @@ #include "im-sandbox.h" #include "process_image.h" #include "time-accounts.h" +#include "im-zmq.h" struct sandbox @@ -92,6 +94,10 @@ struct sandbox const char *tmpdir; + /* ZMQ mode */ + int zmq; + const char *zmq_address; + /* Final output */ Stream *stream; }; @@ -364,7 +370,19 @@ static void run_work(const struct index_args *iargs, Stream *st, int cookie, const char *tmpdir, struct sandbox *sb) { int allDone = 0; - TimeAccounts *taccs = time_accounts_init(); + TimeAccounts *taccs; + struct im_zmq *zmqstuff; + + /* Connect via ZMQ */ + if ( sb->zmq ) { + zmqstuff = im_zmq_connect(sb->zmq_address); + if ( zmqstuff == NULL ) { + ERROR("ZMQ setup failed.\n"); + return; + } + } + + taccs = time_accounts_init(); while ( !allDone ) { @@ -375,58 +393,66 @@ static void run_work(const struct index_args *iargs, Stream *st, struct event *ev; int r; - /* Wait until an event is ready */ - time_accounts_set(taccs, TACC_EVENTWAIT); - set_last_task(sb->shared->last_task[cookie], "wait_event"); - if ( sem_wait(sb->queue_sem) != 0 ) { - ERROR("Failed to wait on queue semaphore: %s\n", - strerror(errno)); - } + if ( !sb->zmq ) { - /* Get the event from the queue */ - set_last_task(sb->shared->last_task[cookie], "read_queue"); - pthread_mutex_lock(&sb->shared->queue_lock); - if ( (sb->shared->n_events==0) && (sb->shared->no_more) ) { - /* Queue is empty and no more coming, so exit */ - pthread_mutex_unlock(&sb->shared->queue_lock); - allDone = 1; - continue; - } - if ( sb->shared->n_events == 0 ) { - ERROR("Got the semaphore, but no events in queue!\n"); - ERROR("no_more = %i\n", sb->shared->no_more); + /* Wait until an event is ready */ + time_accounts_set(taccs, TACC_EVENTWAIT); + set_last_task(sb->shared->last_task[cookie], "wait_event"); + if ( sem_wait(sb->queue_sem) != 0 ) { + ERROR("Failed to wait on queue semaphore: %s\n", + strerror(errno)); + } + + /* Get the event from the queue */ + set_last_task(sb->shared->last_task[cookie], "read_queue"); + pthread_mutex_lock(&sb->shared->queue_lock); + if ( (sb->shared->n_events==0) && (sb->shared->no_more) ) { + /* Queue is empty and no more coming, so exit */ + pthread_mutex_unlock(&sb->shared->queue_lock); + allDone = 1; + continue; + } + if ( sb->shared->n_events == 0 ) { + ERROR("Got the semaphore, but no events in queue!\n"); + ERROR("no_more = %i\n", sb->shared->no_more); + pthread_mutex_unlock(&sb->shared->queue_lock); + allDone = 1; + continue; + } + r = sscanf(sb->shared->queue[0], "%s %s %i", + filename, event_str, &ser); + if ( r != 3 ) { + STATUS("Invalid event string '%s'\n", + sb->shared->queue[0]); + } + memcpy(sb->shared->last_ev[cookie], sb->shared->queue[0], + MAX_EV_LEN); + shuffle_events(sb->shared); pthread_mutex_unlock(&sb->shared->queue_lock); - allDone = 1; - continue; - } - r = sscanf(sb->shared->queue[0], "%s %s %i", - filename, event_str, &ser); - if ( r != 3 ) { - STATUS("Invalid event string '%s'\n", - sb->shared->queue[0]); - } - memcpy(sb->shared->last_ev[cookie], sb->shared->queue[0], - MAX_EV_LEN); - shuffle_events(sb->shared); - pthread_mutex_unlock(&sb->shared->queue_lock); - if ( r != 3 ) continue; + if ( r != 3 ) continue; - pargs.filename_p_e = initialize_filename_plus_event(); - pargs.filename_p_e->filename = strdup(filename); + pargs.filename_p_e = initialize_filename_plus_event(); + pargs.filename_p_e->filename = strdup(filename); - if ( strcmp(event_str, "(none)") != 0 ) { + if ( strcmp(event_str, "(none)") != 0 ) { + + ev = get_event_from_event_string(event_str); + if ( ev == NULL ) { + ERROR("Bad event string '%s'\n", event_str); + continue; + } + pargs.filename_p_e->ev = ev; + + } else { + + pargs.filename_p_e->ev = NULL; - ev = get_event_from_event_string(event_str); - if ( ev == NULL ) { - ERROR("Bad event string '%s'\n", event_str); - continue; } - pargs.filename_p_e->ev = ev; } else { - pargs.filename_p_e->ev = NULL; + pargs.msgpack_obj = im_zmq_fetch(zmqstuff); } @@ -434,10 +460,16 @@ static void run_work(const struct index_args *iargs, Stream *st, process_image(iargs, &pargs, st, cookie, tmpdir, ser, sb->shared, taccs, sb->shared->last_task[cookie]); - free_filename_plus_event(pargs.filename_p_e); + if ( !sb->zmq ) { + free_filename_plus_event(pargs.filename_p_e); + } else { + im_zmq_clean(zmqstuff); + } } + im_zmq_shutdown(zmqstuff); + time_accounts_set(taccs, TACC_FINALCLEANUP); cleanup_indexing(iargs->ipriv); free_detector_geometry(iargs->det); @@ -804,6 +836,11 @@ static int setup_shm(struct sandbox *sb) /* Assumes the caller is already holding queue_lock! */ static int fill_queue(struct get_pattern_ctx *gpctx, struct sandbox *sb) { + if ( sb->zmq ) { + /* Do nothing */ + return 0; + } + while ( sb->shared->n_events < QUEUE_SIZE ) { struct filename_plus_event *ne; @@ -998,7 +1035,8 @@ char *create_tempdir(const char *temp_location) * If the return value is zero, something is probably wrong. */ int create_sandbox(struct index_args *iargs, int n_proc, char *prefix, int config_basename, FILE *fh, - Stream *stream, const char *tmpdir, int serial_start) + Stream *stream, const char *tmpdir, int serial_start, + const char *zmq_address) { int i; struct sandbox *sb; @@ -1027,6 +1065,12 @@ int create_sandbox(struct index_args *iargs, int n_proc, char *prefix, sb->iargs = iargs; sb->serial = serial_start; sb->tmpdir = tmpdir; + if ( zmq_address != NULL ) { + sb->zmq = 1; + sb->zmq_address = zmq_address; + } else { + sb->zmq = 0; + } sb->fds = NULL; sb->fhs = NULL; diff --git a/src/im-sandbox.h b/src/im-sandbox.h index ee2de993..9da11526 100644 --- a/src/im-sandbox.h +++ b/src/im-sandbox.h @@ -3,13 +3,13 @@ * * Sandbox for indexing * - * Copyright © 2012-2018 Deutsches Elektronen-Synchrotron DESY, + * Copyright © 2012-2019 Deutsches Elektronen-Synchrotron DESY, * a research centre of the Helmholtz Association. * Copyright © 2012 Richard Kirian * Copyright © 2012 Lorenzo Galli * * Authors: - * 2010-2018 Thomas White + * 2010-2019 Thomas White * 2011 Richard Kirian * 2012 Lorenzo Galli * 2012 Chunhong Yoon @@ -83,6 +83,7 @@ extern void set_last_task(char *lt, const char *task); extern int create_sandbox(struct index_args *iargs, int n_proc, char *prefix, int config_basename, FILE *fh, Stream *stream, - const char *tempdir, int serial_start); + const char *tempdir, int serial_start, + const char *zmq_address); #endif /* IM_SANDBOX_H */ diff --git a/src/im-zmq.c b/src/im-zmq.c new file mode 100644 index 00000000..c299b980 --- /dev/null +++ b/src/im-zmq.c @@ -0,0 +1,402 @@ +/* + * zmq.c + * + * ZMQ data interface + * + * Copyright © 2017-2018 Deutsches Elektronen-Synchrotron DESY, + * a research centre of the Helmholtz Association. + * + * Authors: + * 2018 Thomas White + * 2014 Valerio Mariani + * 2017 Stijna de Graaf + * + * This file is part of CrystFEL. + * + * CrystFEL is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * CrystFEL is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with CrystFEL. If not, see . + * + */ + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "events.h" +#include "image.h" +#include "hdf5-file.h" +#include "utils.h" + + +struct im_zmq +{ + void *ctx; + void *socket; + zmq_msg_t msg; + msgpack_unpacked unpacked; + int unpacked_set; +}; + + +struct im_zmq *im_zmq_connect(const char *zmq_address) +{ + struct im_zmq *z; + + z = malloc(sizeof(struct im_zmq)); + if ( z == NULL ) return NULL; + + z->unpacked_set = 0; + + z->ctx = zmq_ctx_new(); + if ( z->ctx == NULL ) return NULL; + + z->socket = zmq_socket(z->ctx, ZMQ_REQ); + if ( z->socket == NULL ) return NULL; + + STATUS("Connecting to ZMQ at '%s'\n", zmq_address); + if ( zmq_connect(z->socket, zmq_address) == -1 ) { + ERROR("ZMQ connection failed: %s\n", zmq_strerror(errno)); + return NULL; + } + STATUS("ZMQ connected.\n"); + + return z; +} + + +msgpack_object *im_zmq_fetch(struct im_zmq *z) +{ + int msg_size; + int r; + + if ( zmq_send(z->socket, "m", 1, 0) == -1 ) { + ERROR("ZMQ message send failed: %s\n", zmq_strerror(errno)); + return NULL; + } + + zmq_msg_init(&z->msg); + msg_size = zmq_msg_recv(&z->msg, z->socket, 0); + if ( msg_size == -1 ) { + ERROR("ZMQ recieve failed: %s\n", zmq_strerror(errno)); + zmq_msg_close(&z->msg); + return NULL; + } + + msgpack_unpacked_init(&z->unpacked); + r = msgpack_unpack_next(&z->unpacked, zmq_msg_data(&z->msg), + msg_size, NULL); + if ( r != MSGPACK_UNPACK_SUCCESS ) { + ERROR("Msgpack unpack failed: %i\n", r); + zmq_msg_close(&z->msg); + return NULL; + } + z->unpacked_set = 1; + + return &z->unpacked.data; +} + + +/* Clean structures ready for next frame */ +void im_zmq_clean(struct im_zmq *z) +{ + if ( z->unpacked_set ) { + msgpack_unpacked_destroy(&z->unpacked); + zmq_msg_close(&z->msg); + z->unpacked_set = 0; + } +} + + +void im_zmq_shutdown(struct im_zmq *z) +{ + if ( z == NULL ) return; + + zmq_msg_close(&z->msg); + zmq_close(z->socket); + zmq_ctx_destroy(z->ctx); +} + +/** + * get_peaks_onda: + * @obj: A %msgpack_object containing data in OnDA format + * @image: An %image structure + * @half_pixel_shift: Non-zero if 0.5 should be added to all peak coordinates + * + * Get peaks from msgpack_object. The data should be in a map, with the value + * given by "peak_list" as an array of arrays. The first of these should contain + * the list of fs positions of the peaks, the second the ss positions, and the + * third the intensities of the peaks. + * + * http://c.msgpack.org/c/ provides documentation on msgpack objects + * + * CrystFEL considers all peak locations to be distances from the corner of the + * detector panel, in pixel units, consistent with its description of detector + * geometry (see 'man crystfel_geometry'). The software which generates the + * CXI files, including Cheetah, may instead consider the peak locations to be + * pixel indices in the data array. In this case, the peak coordinates should + * have 0.5 added to them. This will be done if @half_pixel_shift is non-zero. + * + * Returns: non-zero on error, zero otherwise. + * + */ +int get_peaks_onda(msgpack_object *obj, struct image *image, + int half_pixel_shift) +{ + + int num_peaks; + int pk; + double peak_offset = half_pixel_shift ? 0.5 : 0.0; + + int entry; + char *key_str; + msgpack_object map_val, peak_list; + + /* iterate over key-value pairs in msgpack_object + * object has structure: + * {"peak_list": [[peak_x], [peak_y], [peak_i]],"key2":val2,...} + */ + for ( entry = 0; entry < obj->via.map.size; entry++ ) { + key_str = (char *)obj->via.map.ptr[entry].key.via.str.ptr; + /* check if key matches "peak_list" */ + if (strncmp(key_str, "peak_list", 9) == 0) { + map_val = obj->via.map.ptr[entry].val; + /* length of peak_x array gives number of peaks */ + num_peaks = map_val.via.array.ptr[0].via.array.size; + peak_list = map_val; + } + } + + if ( image->features != NULL ) { + image_feature_list_free(image->features); + } + image->features = image_feature_list_new(); + image->num_peaks = num_peaks; + + for ( pk = 0; pkdet, fs, ss); + if ( p == NULL ) continue; + if ( p->no_index ) continue; + + /* Convert coordinates to panel-relative */ + fs = fs - p->orig_min_fs; + ss = ss - p->orig_min_ss; + + image_add_feature(image->features, fs, ss, p, image, val, NULL); + } + + return 0; +} + + +static void onda_fill_in_clen(struct detector *det) +{ + int i = 0; + + for ( i=0; in_panels; i++) { + + struct panel *p = &det->panels[i]; + + if ( p->clen_from != NULL ) { + + ERROR("Can't get clen from OnDA yet.\n"); + } + + adjust_centering_for_rail(p); + + } +} + + +/* Equivalent to fill_in_beam_parameters but without reference to imagefiles */ +static void onda_fill_in_beam_parameters(struct beam_params *beam, + struct image *image) +{ + double eV; + + if (beam->photon_energy_from == NULL ) { + + /* Explicit value given */ + eV = beam->photon_energy; + + } else { + + ERROR("Can't get photon energy from OnDA yet.\n"); + eV = 0.0; + + } + + image->lambda = ph_en_to_lambda(eV_to_J(eV))*beam->photon_energy_scale; +} + + +/* Unpacks the raw panel data from a msgpack_object, appliespanel geometry, + * and stores the resulting data in an image struct. Object has structure + * { + * "corr_data": + * { + * "data": binary_data, + * "shape": [data_height, data_width], + * ... + * ... + * }, + * "key2": val2, + * ... + * ... + * } + */ +int obj_read(msgpack_object *obj, struct image *image) +{ + + uint16_t *flags = NULL; + float *sat = NULL; + int pi; + int entry, sub_entry; + int data_width, data_height; + double *data; + char *key_str; + msgpack_object map_val; + + // Iterate over key-value pairs in msgpack_object + for ( entry=0; entryvia.map.size; entry++ ) { + key_str = (char *)obj->via.map.ptr[entry].key.via.str.ptr; + // Check for key is "corr_data" + if ( strncmp(key_str, "corr_data", 9) == 0 ) { + map_val = obj->via.map.ptr[entry].val; + // Iterate over key-value pairs in inner map + for ( sub_entry=0; sub_entrydet == NULL ) { + ERROR("Geometry not available.\n"); + return 1; + } + + image->dp = malloc(image->det->n_panels*sizeof(float *)); + image->bad = malloc(image->det->n_panels*sizeof(int *)); + image->sat = malloc(image->det->n_panels*sizeof(float *)); + if ( (image->dp == NULL) || (image->bad == NULL) || (image->sat == NULL) ) { + ERROR("Failed to allocate data arrays.\n"); + return 1; + } + + for ( pi=0; pidet->n_panels; pi++ ) { + + struct panel *p; + int fs, ss; + + p = &image->det->panels[pi]; + image->dp[pi] = malloc(p->w*p->h*sizeof(float)); + image->bad[pi] = malloc(p->w*p->h*sizeof(int)); + image->sat[pi] = malloc(p->w*p->h*sizeof(float)); + if ( (image->dp[pi] == NULL) || (image->bad[pi] == NULL) || (image->sat[pi] == NULL) ) + { + ERROR("Failed to allocate panel\n"); + return 1; + } + + if ( (p->orig_min_fs + p->w > data_width) + || (p->orig_min_ss + p->h > data_height) ) + { + ERROR("Panel %s is outside range of data provided\n", + p->name); + return 1; + } + + for ( ss=0; ssh; ss++) { + for ( fs=0; fsw; fs++) { + + int idx; + int cfs, css; + int bad = 0; + + cfs = fs+p->orig_min_fs; + css = ss+p->orig_min_ss; + idx = cfs + css*data_width; + + image->dp[pi][fs+p->w*ss] = data[idx]; + + if ( sat != NULL ) { + image->sat[pi][fs+p->w*ss] = sat[idx]; + } else { + image->sat[pi][fs+p->w*ss] = INFINITY; + } + + if ( p->no_index ) bad = 1; + + if ( in_bad_region(image->det, p, cfs, css) ) { + bad = 1; + } + + if ( flags != NULL ) { + + int f; + + f = flags[idx]; + + if ( (f & image->det->mask_good) + != image->det->mask_good ) bad = 1; + + if ( f & image->det->mask_bad ) bad = 1; + + } + image->bad[pi][fs+p->w*ss] = bad; + } + } + + } + + // might need to do some freeing of memory for msgpack object here + + if ( image->beam != NULL ) { + onda_fill_in_beam_parameters(image->beam, image); + if ( image->lambda > 1000 ) { + ERROR("Warning: Missing or nonsensical wavelength " + "(%e m).\n", + image->lambda); + } + } + onda_fill_in_clen(image->det); + fill_in_adu(image); + + return 0; +} diff --git a/src/im-zmq.h b/src/im-zmq.h new file mode 100644 index 00000000..e6abe562 --- /dev/null +++ b/src/im-zmq.h @@ -0,0 +1,57 @@ +/* + * zmq.h + * + * ZMQ data interface + * + * Copyright © 2017-2019 Deutsches Elektronen-Synchrotron DESY, + * a research centre of the Helmholtz Association. + * + * Authors: + * 2018-2019 Thomas White + * 2014 Valerio Mariani + * 2017 Stijn de Graaf + * + * This file is part of CrystFEL. + * + * CrystFEL is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * CrystFEL is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with CrystFEL. If not, see . + * + */ + + +#ifndef CRYSTFEL_ZMQ_H +#define CRYSTFEL_ZMQ_H + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include + +#include "image.h" + +extern struct im_zmq *im_zmq_connect(const char *zmq_address); + +extern msgpack_object *im_zmq_fetch(struct im_zmq *z); + +extern void im_zmq_clean(struct im_zmq *z); + +extern void im_zmq_shutdown(struct im_zmq *z); + +extern int get_peaks_onda(msgpack_object *obj, struct image *image, + int half_pixel_shift); + +extern int obj_read(msgpack_object *obj, struct image *image); + + +#endif /* CRYSTFEL_ZMQ_H */ diff --git a/src/indexamajig.c b/src/indexamajig.c index 2d584726..356d0c94 100644 --- a/src/indexamajig.c +++ b/src/indexamajig.c @@ -86,6 +86,7 @@ static void show_help(const char *s) " --profile Show timing data for performance monitoring\n" " --temp-dir= Put the temporary folder under \n" " --wait-for-file= Time to wait for each file before processing\n" +" --zmq-msgpack Receive data in MessagePack format over ZMQ\n" "\nPeak search options:\n\n" " --peaks= Peak search method (zaef,peakfinder8,peakfinder9,hdf5,cxi)\n" " Default: zaef\n" @@ -340,6 +341,8 @@ int main(int argc, char *argv[]) int if_retry = 1; int serial_start = 1; char *spectrum_fn = NULL; + int zmq = 0; + char *zmq_address = NULL; /* Defaults */ iargs.cell = NULL; @@ -457,6 +460,7 @@ int main(int argc, char *argv[]) {"no-multi", 0, &if_multi, 0}, {"multi", 0, &if_multi, 1}, {"overpredict", 0, &iargs.overpredict, 1}, + {"zmq-msgpack", 0, &zmq, 1}, /* Long-only options which don't actually do anything */ {"no-sat-corr", 0, &iargs.satcorr, 0}, @@ -1297,8 +1301,22 @@ int main(int argc, char *argv[]) gsl_set_error_handler_off(); + if ( zmq ) { + char line[1024]; + char *rval; + rval = fgets(line, 1024, fh); + if ( rval == NULL ) { + ERROR("Failed to read ZMQ server/port from input.\n"); + return 1; + } + chomp(line); + zmq_address = strdup(line); + /* In future, read multiple addresses and hand them out + * evenly to workers */ + } + r = create_sandbox(&iargs, n_proc, prefix, config_basename, fh, - st, tmpdir, serial_start); + st, tmpdir, serial_start, zmq_address); free_imagefile_field_list(iargs.copyme); cell_free(iargs.cell); diff --git a/src/process_image.c b/src/process_image.c index 2f2eb698..b3b3b1da 100644 --- a/src/process_image.c +++ b/src/process_image.c @@ -3,12 +3,13 @@ * * The processing pipeline for one image * - * Copyright © 2012-2018 Deutsches Elektronen-Synchrotron DESY, + * Copyright © 2012-2019 Deutsches Elektronen-Synchrotron DESY, * a research centre of the Helmholtz Association. * * Authors: - * 2010-2017 Thomas White + * 2010-2019 Thomas White * 2014-2017 Valerio Mariani + * 2017 Stijn de Graaf * * This file is part of CrystFEL. * @@ -38,6 +39,7 @@ #include #include #include +#include #include "utils.h" #include "hdf5-file.h" diff --git a/src/process_image.h b/src/process_image.h index 2a43d11d..b61fe83f 100644 --- a/src/process_image.h +++ b/src/process_image.h @@ -3,11 +3,11 @@ * * The processing pipeline for one image * - * Copyright © 2012-2018 Deutsches Elektronen-Synchrotron DESY, + * Copyright © 2012-2019 Deutsches Elektronen-Synchrotron DESY, * a research centre of the Helmholtz Association. * * Authors: - * 2010-2016 Thomas White + * 2010-2019 Thomas White * 2014-2017 Valerio Mariani * 2017-2018 Yaroslav Gevorkov * @@ -37,6 +37,8 @@ struct index_args; +#include + #include "integration.h" #include "im-sandbox.h" #include "time-accounts.h" @@ -122,6 +124,7 @@ struct pattern_args { /* "Input" */ struct filename_plus_event *filename_p_e; + msgpack_object *msgpack_obj; }; diff --git a/src/zmq.c b/src/zmq.c deleted file mode 100644 index 41bd2eec..00000000 --- a/src/zmq.c +++ /dev/null @@ -1,313 +0,0 @@ -/* - * zmq.c - * - * ZMQ data interface - * - * Copyright © 2017-2018 Deutsches Elektronen-Synchrotron DESY, - * a research centre of the Helmholtz Association. - * - * Authors: - * 2018 Thomas White - * 2014 Valerio Mariani - * 2017 Stijna de Graaf - * - * This file is part of CrystFEL. - * - * CrystFEL is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * CrystFEL is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with CrystFEL. If not, see . - * - */ - -#ifdef HAVE_CONFIG_H -#include -#endif - -#include -#include -#include -#include -#include -#include -#include - -#include "events.h" -#include "image.h" -#include "hdf5-file.h" -#include "utils.h" - - -/** - * get_peaks_onda: - * @obj: A %msgpack_object containing data in OnDA format - * @image: An %image structure - * @half_pixel_shift: Non-zero if 0.5 should be added to all peak coordinates - * - * Get peaks from msgpack_object. The data should be in a map, with the value - * given by "peak_list" as an array of arrays. The first of these should contain - * the list of fs positions of the peaks, the second the ss positions, and the - * third the intensities of the peaks. - * - * http://c.msgpack.org/c/ provides documentation on msgpack objects - * - * CrystFEL considers all peak locations to be distances from the corner of the - * detector panel, in pixel units, consistent with its description of detector - * geometry (see 'man crystfel_geometry'). The software which generates the - * CXI files, including Cheetah, may instead consider the peak locations to be - * pixel indices in the data array. In this case, the peak coordinates should - * have 0.5 added to them. This will be done if @half_pixel_shift is non-zero. - * - * Returns: non-zero on error, zero otherwise. - * - */ -int get_peaks_onda(msgpack_object *obj, struct image *image, - int half_pixel_shift) -{ - - int num_peaks; - int pk; - double peak_offset = half_pixel_shift ? 0.5 : 0.0; - - int entry; - char *key_str; - msgpack_object map_val, peak_list; - - /* iterate over key-value pairs in msgpack_object - * object has structure: - * {"peak_list": [[peak_x], [peak_y], [peak_i]],"key2":val2,...} - */ - for ( entry = 0; entry < obj->via.map.size; entry++ ) { - key_str = (char *)obj->via.map.ptr[entry].key.via.str.ptr; - /* check if key matches "peak_list" */ - if (strncmp(key_str, "peak_list", 9) == 0) { - map_val = obj->via.map.ptr[entry].val; - /* length of peak_x array gives number of peaks */ - num_peaks = map_val.via.array.ptr[0].via.array.size; - peak_list = map_val; - } - } - - if ( image->features != NULL ) { - image_feature_list_free(image->features); - } - image->features = image_feature_list_new(); - image->num_peaks = num_peaks; - - for ( pk = 0; pkdet, fs, ss); - if ( p == NULL ) continue; - if ( p->no_index ) continue; - - /* Convert coordinates to panel-relative */ - fs = fs - p->orig_min_fs; - ss = ss - p->orig_min_ss; - - image_add_feature(image->features, fs, ss, p, image, val, NULL); - } - - return 0; -} - - -static void onda_fill_in_clen(struct detector *det) -{ - int i = 0; - - for ( i=0; in_panels; i++) { - - struct panel *p = &det->panels[i]; - - if ( p->clen_from != NULL ) { - - ERROR("Can't get clen from OnDA yet.\n"); - } - - adjust_centering_for_rail(p); - - } -} - - -/* Equivalent to fill_in_beam_parameters but without reference to imagefiles */ -static void onda_fill_in_beam_parameters(struct beam_params *beam, - struct image *image) -{ - double eV; - - if (beam->photon_energy_from == NULL ) { - - /* Explicit value given */ - eV = beam->photon_energy; - - } else { - - ERROR("Can't get photon energy from OnDA yet.\n"); - eV = 0.0; - - } - - image->lambda = ph_en_to_lambda(eV_to_J(eV))*beam->photon_energy_scale; -} - - -/* Unpacks the raw panel data from a msgpack_object, appliespanel geometry, - * and stores the resulting data in an image struct. Object has structure - * { - * "corr_data": - * { - * "data": binary_data, - * "shape": [data_height, data_width], - * ... - * ... - * }, - * "key2": val2, - * ... - * ... - * } - */ -int obj_read(msgpack_object *obj, struct image *image) -{ - - uint16_t *flags = NULL; - float *sat = NULL; - int pi; - int entry, sub_entry; - int data_width, data_height; - double *data; - char *key_str; - msgpack_object map_val; - - // Iterate over key-value pairs in msgpack_object - for ( entry=0; entryvia.map.size; entry++ ) { - key_str = (char *)obj->via.map.ptr[entry].key.via.str.ptr; - // Check for key is "corr_data" - if ( strncmp(key_str, "corr_data", 9) == 0 ) { - map_val = obj->via.map.ptr[entry].val; - // Iterate over key-value pairs in inner map - for ( sub_entry=0; sub_entrydet == NULL ) { - ERROR("Geometry not available.\n"); - return 1; - } - - image->dp = malloc(image->det->n_panels*sizeof(float *)); - image->bad = malloc(image->det->n_panels*sizeof(int *)); - image->sat = malloc(image->det->n_panels*sizeof(float *)); - if ( (image->dp == NULL) || (image->bad == NULL) || (image->sat == NULL) ) { - ERROR("Failed to allocate data arrays.\n"); - return 1; - } - - for ( pi=0; pidet->n_panels; pi++ ) { - - struct panel *p; - int fs, ss; - - p = &image->det->panels[pi]; - image->dp[pi] = malloc(p->w*p->h*sizeof(float)); - image->bad[pi] = malloc(p->w*p->h*sizeof(int)); - image->sat[pi] = malloc(p->w*p->h*sizeof(float)); - if ( (image->dp[pi] == NULL) || (image->bad[pi] == NULL) || (image->sat[pi] == NULL) ) - { - ERROR("Failed to allocate panel\n"); - return 1; - } - - if ( (p->orig_min_fs + p->w > data_width) - || (p->orig_min_ss + p->h > data_height) ) - { - ERROR("Panel %s is outside range of data provided\n", - p->name); - return 1; - } - - for ( ss=0; ssh; ss++) { - for ( fs=0; fsw; fs++) { - - int idx; - int cfs, css; - int bad = 0; - - cfs = fs+p->orig_min_fs; - css = ss+p->orig_min_ss; - idx = cfs + css*data_width; - - image->dp[pi][fs+p->w*ss] = data[idx]; - - if ( sat != NULL ) { - image->sat[pi][fs+p->w*ss] = sat[idx]; - } else { - image->sat[pi][fs+p->w*ss] = INFINITY; - } - - if ( p->no_index ) bad = 1; - - if ( in_bad_region(image->det, p, cfs, css) ) { - bad = 1; - } - - if ( flags != NULL ) { - - int f; - - f = flags[idx]; - - if ( (f & image->det->mask_good) - != image->det->mask_good ) bad = 1; - - if ( f & image->det->mask_bad ) bad = 1; - - } - image->bad[pi][fs+p->w*ss] = bad; - } - } - - } - - // might need to do some freeing of memory for msgpack object here - - if ( image->beam != NULL ) { - onda_fill_in_beam_parameters(image->beam, image); - if ( image->lambda > 1000 ) { - ERROR("Warning: Missing or nonsensical wavelength " - "(%e m).\n", - image->lambda); - } - } - onda_fill_in_clen(image->det); - fill_in_adu(image); - - return 0; -} diff --git a/src/zmq.h b/src/zmq.h deleted file mode 100644 index ee04f437..00000000 --- a/src/zmq.h +++ /dev/null @@ -1,48 +0,0 @@ -/* - * zmq.h - * - * ZMQ data interface - * - * Copyright © 2017-2018 Deutsches Elektronen-Synchrotron DESY, - * a research centre of the Helmholtz Association. - * - * Authors: - * 2018 Thomas White - * 2014 Valerio Mariani - * 2017 Stijna de Graaf - * - * This file is part of CrystFEL. - * - * CrystFEL is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * CrystFEL is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with CrystFEL. If not, see . - * - */ - - -#ifndef CRYSTFEL_ZMQ_H -#define CRYSTFEL_ZMQ_H - -#ifdef HAVE_CONFIG_H -#include -#endif - -#include - -#include "image.h" - -extern int get_peaks_onda(msgpack_object *obj, struct image *image, - int half_pixel_shift); - -extern int obj_read(msgpack_object *obj, struct image *image); - -#endif /* CRYSTFEL_ZMQ_H */ -- cgit v1.2.3