indexamajig: Add --wait-for-file

6 files changed, 100 insertions, 19 deletions

diff --git a/doc/man/indexamajig.1 b/doc/man/indexamajig.1
index ab581120..1394b4a4 100644
--- a/doc/man/indexamajig.1
+++ b/doc/man/indexamajig.1
@@ -215,6 +215,11 @@ Display timing data for performance monitoring.
 .PD
 Put the temporary folder under \fIpath\fR.
 
+.PD 0
+.IP \fB--wait-for-file=\fIn\fR
+.PD
+Wait at most \fIn\fR seconds for each image file in the input list to be created before trying to process it.  This is useful for some automated processing pipelines.  It obviously only really works for single-frame files.  If a file exists but is not readable when this option is set non-zero, a second attempt will be made after ten seconds.  This is to allow to incompletely written files.  A value of -1 means to wait forever.  The default value is \fB--wait-for-file=0\fR.
+
 .SH PEAK SEARCH OPTIONS
 .PD 0
 .IP \fB--peaks=\fR\fImethod\fR
diff --git a/src/indexamajig.c b/src/indexamajig.c
index c0334d29..f9b2c94d 100644
--- a/src/indexamajig.c
+++ b/src/indexamajig.c
@@ -85,6 +85,7 @@ static void show_help(const char *s)
 "     --highres=<n>         Absolute resolution cutoff in Angstroms\n"
 "     --profile             Show timing data for performance monitoring\n"
 "     --temp-dir=<path>     Put the temporary folder under <path>\n"
+"     --wait-for-file=<n>   Time to wait for each file before processing\n"
 "\nPeak search options:\n\n"
 "     --peaks=<method>      Peak search method (zaef,peakfinder8,peakfinder9,hdf5,cxi)\n"
 "                            Default: zaef\n"
@@ -382,6 +383,7 @@ int main(int argc, char *argv[])
 	iargs.copyme = new_imagefile_field_list();
 	iargs.min_peaks = 0;
 	iargs.overpredict = 0;
+	iargs.wait_for_file = 0;
 	if ( iargs.copyme == NULL ) {
 		ERROR("Couldn't allocate HDF5 field list.\n");
 		return 1;
@@ -528,6 +530,7 @@ int main(int argc, char *argv[])
 		{"xgandalf-max-lattice-vector-length",       1, NULL, 356},
 		{"xgandalf-max-lvl",                         1, NULL, 356},
 	        {"spectrum-file",                            1, NULL, 357},
+		{"wait-for-file",            1, NULL,        358},
 
 		{0, 0, NULL, 0}
 	};
@@ -922,6 +925,14 @@ int main(int argc, char *argv[])
 			spectrum_fn = strdup(optarg);
 			break;
 
+			case 358:
+			if (sscanf(optarg, "%d", &iargs.wait_for_file) != 1)
+			{
+				ERROR("Invalid value for --wait-for-file\n");
+				return 1;
+			}
+			break;
+
 			case 0 :
 			break;
 
diff --git a/src/process_image.c b/src/process_image.c
index 2fbf0e4a..31abcb21 100644
--- a/src/process_image.c
+++ b/src/process_image.c
@@ -37,6 +37,7 @@
 #include <gsl/gsl_statistics_double.h>
 #include <gsl/gsl_sort.h>
 #include <unistd.h>
+#include <sys/stat.h>
 
 #include "utils.h"
 #include "hdf5-file.h"
@@ -102,7 +103,6 @@ void process_image(const struct index_args *iargs, struct pattern_args *pargs,
                    int serial, struct sb_shm *sb_shared, TimeAccounts *taccs,
                    char *last_task)
 {
-	int check;
 	struct imagefile *imfile;
 	struct image image;
 	int i;
@@ -111,6 +111,9 @@ void process_image(const struct index_args *iargs, struct pattern_args *pargs,
 	char *rn;
 	float **prefilter;
 	int any_crystals;
+	int wait_message_done = 0;
+	int read_retry_done = 0;
+	signed int file_wait_time = iargs->wait_for_file;
 
 	image.features = NULL;
 	image.copyme = iargs->copyme;
@@ -124,22 +127,81 @@ void process_image(const struct index_args *iargs, struct pattern_args *pargs,
 	image.serial = serial;
 	image.indexed_by = INDEXING_NONE;
 
+	time_accounts_set(taccs, TACC_WAITFILE);
+	set_last_task(last_task, "wait for file");
+
+	do {
+
+		struct stat statbuf;
+
+		sb_shared->pings[cookie]++;
+		r = stat(image.filename, &statbuf);
+		if ( r ) {
+
+			if ( (iargs->wait_for_file != 0)
+			  && (file_wait_time != 0) )
+			{
+
+				if ( !wait_message_done ) {
+					STATUS("Waiting for '%s'\n",
+					       image.filename);
+					wait_message_done = 1;
+				}
+
+				sleep(1);
+				if ( iargs->wait_for_file != -1 ) {
+					file_wait_time--;
+				}
+				continue;
+
+			}
+
+			ERROR("File %s not found\n", image.filename);
+			return;
+		}
+
+	} while ( r );
+
 	time_accounts_set(taccs, TACC_HDF5OPEN);
 	set_last_task(last_task, "open file");
 	sb_shared->pings[cookie]++;
-	imfile = imagefile_open(image.filename);
-	if ( imfile == NULL ) {
-		ERROR("Couldn't open file: %s\n", image.filename);
-		return;
-	}
 
-	time_accounts_set(taccs, TACC_HDF5READ);
-	set_last_task(last_task, "read file");
-	sb_shared->pings[cookie]++;
-	check = imagefile_read(imfile, &image, image.event);
-	if ( check ) {
-		return;
-	}
+	do {
+		imfile = imagefile_open(image.filename);
+		if ( imfile == NULL ) {
+			if ( iargs->wait_for_file && !read_retry_done ) {
+				read_retry_done = 1;
+				r = 1;
+				STATUS("File '%s' exists but could not be opened."
+				       "  Trying again after 10 seconds.\n",
+				       image.filename);
+				sleep(10);
+				continue;
+			}
+			ERROR("Couldn't open file: %s\n", image.filename);
+			return;
+		}
+
+		time_accounts_set(taccs, TACC_HDF5READ);
+		set_last_task(last_task, "read file");
+		sb_shared->pings[cookie]++;
+
+		r = imagefile_read(imfile, &image, image.event);
+		if ( r ) {
+			if ( iargs->wait_for_file && !read_retry_done ) {
+				read_retry_done = 1;
+				imagefile_close(imfile);
+				STATUS("File '%s' exists but could not be read."
+				       "  Trying again after 10 seconds.\n",
+				       image.filename);
+				sleep(10);
+				continue;
+			}
+			ERROR("Couldn't open file: %s\n", image.filename);
+			return;
+		}
+
+	} while ( r );
 
 	/* Take snapshot of image before applying horrible noise filters */
 	time_accounts_set(taccs, TACC_FILTER);
diff --git a/src/process_image.h b/src/process_image.h
index d4b1ce59..90925592 100644
--- a/src/process_image.h
+++ b/src/process_image.h
@@ -113,6 +113,7 @@ struct index_args
 	struct xgandalf_options xgandalf_opts;
 	struct felix_options felix_opts;
 	struct spectrum *spectrum;
+	signed int wait_for_file; /* -1 means wait forever */
 };
 
 
diff --git a/src/time-accounts.c b/src/time-accounts.c
index 15a984cd..2c1b5c7c 100644
--- a/src/time-accounts.c
+++ b/src/time-accounts.c
@@ -3,11 +3,11 @@
  *
  * Simple profiling according to wall clock time
  *
- * Copyright © 2016 Deutsches Elektronen-Synchrotron DESY,
- *                  a research centre of the Helmholtz Association.
+ * Copyright © 2016-2018 Deutsches Elektronen-Synchrotron DESY,
+ *                       a research centre of the Helmholtz Association.
  *
  * Authors:
- *  2016 Thomas White <taw@physics.org>
+ *  2016-2018 Thomas White <taw@physics.org>
  *
  * This file is part of CrystFEL.
  *
@@ -173,6 +173,7 @@ static const char *taccname(enum timeaccount acc)
 		case TACC_ENDCHECK : return "Checking end";
 		case TACC_WAKEUP : return "Waking up workers";
 		case TACC_WAITPID : return "Waiting on workers";
+		case TACC_WAITFILE : return "Waiting for image file";
 		case TACC_HDF5OPEN : return "Opening image file";
 		case TACC_HDF5READ : return "Reading image file";
 		case TACC_FILTER : return "Image filters";
diff --git a/src/time-accounts.h b/src/time-accounts.h
index 92b459e1..43cc56ea 100644
--- a/src/time-accounts.h
+++ b/src/time-accounts.h
@@ -3,11 +3,11 @@
  *
  * Simple profiling according to wall clock time
  *
- * Copyright © 2016 Deutsches Elektronen-Synchrotron DESY,
- *                  a research centre of the Helmholtz Association.
+ * Copyright © 2016-2018 Deutsches Elektronen-Synchrotron DESY,
+ *                       a research centre of the Helmholtz Association.
  *
  * Authors:
- *  2016 Thomas White <taw@physics.org>
+ *  2016-2018 Thomas White <taw@physics.org>
  *
  * This file is part of CrystFEL.
  *
@@ -44,6 +44,7 @@ enum timeaccount
 	TACC_ENDCHECK,
 	TACC_WAKEUP,
 	TACC_WAITPID,
+	TACC_WAITFILE,
 	TACC_HDF5OPEN,
 	TACC_HDF5READ,
 	TACC_FILTER,