Changeset 33662 for trunk/Ohana/src/opihi/dvo/mextract.c

Timestamp:

Apr 1, 2012, 3:24:39 PM (14 years ago)

Author:

eugene

Message:

support for parallel dvo in mextract, avextract, avmerge, gstar, gcat; remove some if-def-ed out code already moved to libdvo; list -vectors and -buffers options (store names of vectors and buffers in lists); list -copy give error if missing source; catlist, hosts & remote functions to support parallel dvos; skyregion -save option; new skycoverage modes (-min-ubercal; -min-dmag-sys; -min-mcal; -max-mcal); gstar output formatting cleanups; add more average info to gstar output; functions for spectral similarity analysis; coords returns nans for projections off the sphere; fix cumulative function; add name:type option to opihi/read function (eg read a:float 2 b:time 3 c:int 4); line -frac option; vtype function to get vector types; threshold function; write -fits option; code for unfinished mtype function (matching vtype)

File:

• trunk/Ohana/src/opihi/dvo/mextract.c (modified) (10 diffs)

trunk/Ohana/src/opihi/dvo/mextract.c

@@ -25 +25 @@
   selection = NULL;
 
+  // fprintf (stderr, "start...");
   if ((N = get_argument (argc, argv, "-h"))) goto help;
   if ((N = get_argument (argc, argv, "--help"))) goto help;
@@ -34 +35 @@
   }
 
+  int PARALLEL = FALSE;
+  if ((N = get_argument (argc, argv, "-parallel"))) {
+    remove_argument (N, &argc, argv);
+    PARALLEL = TRUE;
+  }
+
+  // this is used to NOT save the results in the results file
+  // use this option when mextract is used in a script which does its
+  // own job of packaging the results
+  int SKIP_RESULTS = FALSE;
+  if ((N = get_argument (argc, argv, "-skip-results"))) {
+    remove_argument (N, &argc, argv);
+    SKIP_RESULTS = TRUE;
+  }
+
+  // init here so free in 'escape' block does not crash
   dvo_catalog_init (&catalog, TRUE);
 
@@ -40 +57 @@
   Nsecfilt = GetPhotcodeNsecfilt ();
   
+  // parse skyregion options.  NOTE: this is stripped off in parallel operation and always
+  // defined for the client via the -skyregion option.  The dvo_client parses this
+  // argument in the main program, before it is passed to the command (like mextract)
+  if ((selection = SetRegionSelection (&argc, argv)) == NULL) {
+    print_error();
+    goto escape;
+  }
+
   // init locally static variables (time refs)
   dbExtractMeasuresInit();
 
-  // parse skyregion options
-  if ((selection = SetRegionSelection (&argc, argv)) == NULL) goto escape;
-
-  // command-line is of the form: avextract field,field, field [where (field op value)...]
+  // command-line is of the form: mextract field,field, field [where (field op value)...]
 
   // parse the fields to be extracted and returned
@@ -84 +106 @@
   if ((skylist = SelectRegions (selection)) == NULL) goto escape;
 
+  // this does all the work of re-packaging the command, calling it on the remote machines, then loading in the results
+  if (PARALLEL && !HOST_ID) {
+    int status = HostTableParallelOps (argc, argv, RESULT_FILE, 0, VERBOSE);
+
+    dbFreeFields (fields, Nfields);
+    dbFreeStack (stack, Nstack);
+    free (stack);
+    FreeSkyRegionSelection (selection);
+    dvo_catalog_free (&catalog);
+
+    return status;
+  }
+
   // load image data if needed (for fields listed below)
   loadImages = FALSE;
   mosaicMode = FALSE;
   for (i = 0; !loadImages && (i < Nfields); i++) {
-    if (fields[i].ID == MEAS_XCCD) loadImages = TRUE;   // Are these still needed? Xccd and Yccd are in measurement
-    if (fields[i].ID == MEAS_YCCD) loadImages = TRUE;   // and dbExtractMeasures doesn't look at the image for these
-    if (fields[i].ID == MEAS_XMOSAIC) loadImages = mosaicMode = TRUE;
-    if (fields[i].ID == MEAS_YMOSAIC) loadImages = mosaicMode = TRUE;
+    if (!MEASURE_HAS_XCCD) {
+      // I'm keeping this code because it gives a way of handling dvo dbs that don't have
+      // measure.xccd if we need it
+      if (fields[i].ID == MEAS_XCCD) loadImages = TRUE;
+      if (fields[i].ID == MEAS_YCCD) loadImages = TRUE;
+    }
+    if (fields[i].ID == MEAS_XMOSAIC)   loadImages = mosaicMode = TRUE;
+    if (fields[i].ID == MEAS_YMOSAIC)   loadImages = mosaicMode = TRUE;
     if (fields[i].ID == MEAS_EXTERN_ID) loadImages = mosaicMode = TRUE;
+    if (fields[i].ID == MEAS_FLAT)      loadImages = mosaicMode = TRUE;
   }
   if (loadImages && !SetImageSelection (mosaicMode, selection)) goto escape;
@@ -115 +155 @@
   interrupt = FALSE;
 
+  // fprintf (stderr, "done setup...");
+
   for (i = 0; (i < skylist[0].Nregions) && !interrupt; i++) {
+
+    // does this host ID match the desired location for the table?
+    if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue;
+
     /* lock, load, unlock catalog */
-    catalog.filename = skylist[0].filename[i];
+    char hostfile[1024];
+    snprintf (hostfile, 1024, "%s/%s.cpt", HOSTDIR, skylist[0].regions[i]->name);
+    catalog.filename = HOST_ID ? hostfile : skylist[0].filename[i];
     catalog.catflags = LOAD_AVES | LOAD_MEAS | LOAD_SECF;
     catalog.Nsecfilt = Nsecfilt;
@@ -126 +174 @@
     if (!dvo_catalog_open (&catalog, NULL, FALSE, "r")) {
       gprint (GP_ERR, "ERROR: failure to open catalog file %s\n", catalog.filename);
-      exit (2);
+      return FALSE;
     }
     dvo_catalog_unlock (&catalog);
@@ -132 +180 @@
     /* XXX need to call dvo_catalog_chipcoords here passing the loaded images */
 
+    // fprintf (stderr, "done read...");
+
     for (j = 0; (j < catalog.Naverage) && !interrupt; j++) {
       m = catalog.average[j].measureOffset;
-      dbExtractMeasuresInitAve (); // reset counters for saved fields 
+      if (m > catalog.Nmeasure) {
+        gprint (GP_ERR, "ERROR: inconsistent average->measure offset.  Unsorted database?\n");
+        goto escape;
+      }
+
+      dbExtractMeasuresInitAve (); // reset counters for saved fields (costs very little
 
       for (k = 0; (k < catalog.average[j].Nmeasure); k++, m++) {
+        if (catalog.measure[m].averef != j) {
+          gprint (GP_ERR, "ERROR: inconsistent measure->average link.  Unsorted database?\n");
+          goto escape;
+        }
 
         // extract the relevant values for this measurement
-        dbExtractMeasuresInitMeas (); // reset counters for saved fields 
+        dbExtractMeasuresInitMeas (); // reset counters for saved fields  (costs very little
         for (n = 0; n < Nfields; n++) {
           values[n] = dbExtractMeasures (&catalog.average[j], &catalog.secfilt[j*Nsecfilt], &catalog.measure[m], &fields[n]);
@@ -173 +232 @@
   interrupt = FALSE;
 
+  // fprintf (stderr, "done load...");
+
   for (n = 0; n < Nreturn; n++) {
-    ResetVector (vec[n], fields[n].type, MAX(1,Npts));
+    ResetVector (vec[n], fields[n].type, Npts);
+  }
+
+  // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere)
+  if (RESULT_FILE && !SKIP_RESULTS) {
+    int status = WriteVectorTableFITS (RESULT_FILE, "RESULT", vec, Nreturn, FALSE, NULL);
+    if (!status) goto escape;
   }
 
@@ -185 +252 @@
   SkyListFree (skylist);
   FreeSkyRegionSelection (selection);
+
+  // fprintf (stderr, "done extr...\n");
   return (TRUE);
 
@@ -307 +376 @@
     gprint (GP_ERR, "  imageID : ID of source image (32 bit)\n");
     gprint (GP_ERR, "  externID : externID of source image (32 bit)\n");
+
+    gprint (GP_ERR, "  Mcal_offset : difference wrt nominal zero point (clouds are positive)\n");
+    gprint (GP_ERR, "  flat : flat-field correction (measure.Mcal - image.Mcal)\n");
     return (FALSE);
   }