Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/Makefile =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/Makefile (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/Makefile (revision 33543) @@ -23,4 +23,5 @@ $(SRC)/compare.$(ARCH).o \ $(SRC)/dvomisc.$(ARCH).o \ +$(SRC)/dvo_host_utils.$(ARCH).o \ $(SRC)/region_list.$(ARCH).o \ $(SRC)/find_matches.$(ARCH).o \ Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avextract.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avextract.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avextract.c (revision 33543) @@ -1,7 +1,3 @@ # include "dvoshell.h" - -int HostTableLaunchJobs (HostTable *table, char *basecmd); -Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec); -int HostTableParallelOps (int argc, char **argv, char *ResultFile); int avextract (int argc, char **argv) { @@ -74,5 +70,5 @@ if (!SetSkyRegions (selection)) goto escape; - int status = HostTableParallelOps (argc, argv, ResultFile); + int status = HostTableParallelOps (argc, argv, ResultFile, 0); return status; } @@ -145,5 +141,5 @@ // does this host ID match the desired location for the table? - if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; + if (PARALLEL && !HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; /* lock, load, unlock catalog */ Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avmatch.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avmatch.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/avmatch.c (revision 33543) @@ -14,7 +14,7 @@ off_t i, j, n, m, *index; - int N, Ncat, Npts, NPTS, last, Nfields, Nsecfilt; + int N, Ncat, Npts, NPTS, last, Nfields, Nsecfilt, Ninvec; int VERBOSE; - char name[1024]; + char name[1024], *found; void *Signal; float RADIUS; @@ -22,5 +22,5 @@ Catalog catalog; - Vector **vec, *RAvec, *DECvec; + Vector **vec, **invec, *RAvec, *DECvec; dbField *fields; dbValue *values; @@ -29,11 +29,12 @@ /* defaults */ vec = NULL; + invec = NULL; fields = NULL; values = NULL; skylist = NULL; + Ninvec = 0; if ((N = get_argument (argc, argv, "-h"))) goto help; if ((N = get_argument (argc, argv, "--help"))) goto help; - if (argc < 5) goto help; VERBOSE = FALSE; @@ -42,4 +43,29 @@ VERBOSE = TRUE; } + + int PARALLEL = FALSE; + if ((N = get_argument (argc, argv, "-parallel"))) { + remove_argument (N, &argc, argv); + PARALLEL = TRUE; + } + + // dump results directly to fits file (esp for parallel dvo) + char *ResultFile = NULL; + if ((N = get_argument (argc, argv, "-result"))) { + remove_argument (N, &argc, argv); + ResultFile = strcreate(argv[N]); + remove_argument (N, &argc, argv); + } + + // dump results directly to fits file (esp for parallel dvo) + char *CoordsFile = NULL; + if ((N = get_argument (argc, argv, "-coords"))) { + remove_argument (N, &argc, argv); + CoordsFile = strcreate(argv[N]); + remove_argument (N, &argc, argv); + } + + if (!CoordsFile && (argc < 5)) goto help; + if ( CoordsFile && (argc < 3)) goto help; dvo_catalog_init (&catalog, TRUE); @@ -49,12 +75,54 @@ Nsecfilt = GetPhotcodeNsecfilt (); + // 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) { + if (!CoordsFile) { + // get vectors corresponding to coordinates of interest + if ((RAvec = SelectVector (argv[1], ANYVECTOR, TRUE)) == NULL) goto help; + if ((DECvec = SelectVector (argv[2], ANYVECTOR, TRUE)) == NULL) goto help; + + ALLOCATE (vec, Vector *, 2); + vec[0] = RAvec; + vec[1] = DECvec; + + CoordsFile = abspath("coords.fits", 1024); + int status = WriteVectorTableFITS (CoordsFile, "COORDS", vec, 2, FALSE, NULL); + if (!status) goto escape; + } + + char *targv1 = argv[1]; + char *targv2 = argv[2]; + argv[1] = strcreate ("-coords"); + argv[2] = strcreate (CoordsFile); + free (CoordsFile); + + // I need to pass the RA & DEC vectors to the remote clients... + int status = HostTableParallelOps (argc, argv, ResultFile, RAvec->Nelements); + if (ResultFile) free (ResultFile); + if (vec) free (vec); + + free (argv[1]); + free (argv[2]); + argv[1] = targv1; + argv[2] = targv2; + + return status; + } + // get vectors corresponding to coordinates of interest - if ((RAvec = SelectVector (argv[1], ANYVECTOR, TRUE)) == NULL) goto help; - if ((DECvec = SelectVector (argv[2], ANYVECTOR, TRUE)) == NULL) goto help; - RADIUS = atof (argv[3]); - - // strip off RA, DEC, RADIUS arguments - remove_argument (1, &argc, argv); - remove_argument (1, &argc, argv); + if (CoordsFile) { + // read RAvec, DECvec from coords file (1st 2 fields?) + Ninvec = 0; + invec = ReadVectorTableFITS (CoordsFile, "COORDS", &Ninvec); + RAvec = invec[0]; + DECvec = invec[1]; + } else { + if ((RAvec = SelectVector (argv[1], ANYVECTOR, TRUE)) == NULL) goto help; + if ((DECvec = SelectVector (argv[2], ANYVECTOR, TRUE)) == NULL) goto help; + // strip off RA & DEC + remove_argument (1, &argc, argv); + remove_argument (1, &argc, argv); + } + RADIUS = atof (argv[1]); remove_argument (1, &argc, argv); @@ -92,4 +160,6 @@ } ALLOCATE (index, off_t, NPTS); + ALLOCATE (found, char, NPTS); + memset (found, 0, NPTS*sizeof(char)); // grab data from all selected sky regions @@ -97,6 +167,12 @@ interrupt = FALSE; for (i = 0; (i < skylist[0].Nregions) && !interrupt; i++) { + + // does this host ID match the desired location for the table? + if (PARALLEL && !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 = 0; @@ -124,5 +200,4 @@ if (Ncat == -2) continue; - // XXX set a 'found' vector to double check we catch everything? m = catalog.average[Ncat].measureOffset; @@ -141,4 +216,5 @@ } } + found[Npts] = TRUE; } dvo_catalog_free (&catalog); @@ -147,6 +223,61 @@ interrupt = FALSE; + // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere) + // only write the fields which were in a valid catalog + if (ResultFile) { + // extend the array by one to hold index array + REALLOCATE (vec, Vector *, Nfields + 1); + vec[Nfields] = InitVector(); + strcpy (vec[Nfields]->name, "index"); + ResetVector (vec[Nfields], OPIHI_INT, NPTS); + Vector *idxVec = vec[Nfields]; + + // only write out the rows which were found + Npts = 0; + for (i = 0; i < NPTS; i++) { + if (!found[i]) continue; + idxVec->elements.Int[Npts] = i; + Npts ++; + } + int Nfound = Npts; + idxVec->Nelements = Nfound; + + fprintf (stderr, "found %d of %d pts\n", Nfound, NPTS); + + for (i = 0; i < Nfields; i++) { + if (vec[i][0].type == OPIHI_FLT) { + opihi_flt *tmp = NULL; + ALLOCATE (tmp, opihi_flt, Nfound); + Npts = 0; + for (j = 0; j < NPTS; j++) { + if (!found[j]) continue; + tmp[Npts] = vec[i][0].elements.Flt[j]; + Npts++; + } + free (vec[i][0].elements.Flt); + vec[i][0].elements.Flt = tmp; + } else { + opihi_int *tmp = NULL; + ALLOCATE (tmp, opihi_int, Nfound); + Npts = 0; + for (j = 0; j < NPTS; j++) { + if (!found[j]) continue; + tmp[Npts] = vec[i][0].elements.Int[j]; + Npts++; + } + free (vec[i][0].elements.Int); + vec[i][0].elements.Int = tmp; + } + vec[i][0].Nelements = Nfound; + } + int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nfields + 1, FALSE, NULL); + free (vec[Nfields]->elements.Int); + free (vec[Nfields]); + if (!status) goto escape; + } + if (vec) free (vec); if (values) free (values); + if (invec) FreeVectorArray (invec, Ninvec); dbFreeFields (fields, Nfields); SkyListFree (skylist); @@ -156,4 +287,5 @@ if (vec) free (vec); if (values) free (values); + if (invec) FreeVectorArray (invec, Ninvec); dbFreeFields (fields, Nfields); SkyListFree (skylist); @@ -162,4 +294,5 @@ help: gprint (GP_ERR, "USAGE: avmatch (RA) (DEC) (RADIUS) field[,field,field...]\n"); + gprint (GP_ERR, " OR: avmatch -coords (filename.fits) (RADIUS) field[,field,field...]\n"); if ((argc > N + 1) && !strcasecmp (argv[N+1], "fields")) { Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/dvo_host_utils.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/dvo_host_utils.c (revision 33543) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/dvo_host_utils.c (revision 33543) @@ -0,0 +1,161 @@ +# include "dvoshell.h" + +# define DEBUG 0 +# define PARALLEL_MANUAL 0 +# define PARALLEL_SERIAL 0 +# define MAX_PATH_LENGTH 1024 + +int HostTableLaunchJobs (HostTable *table, char *basecmd) { + + int i; + for (i = 0; i < table->Nhosts; i++) { + + // ensure that the paths are absolute path names + char *tmppath = abspath (table->hosts[i].pathname, MAX_PATH_LENGTH); + free (table->hosts[i].pathname); + table->hosts[i].pathname = tmppath; + + char resultFile[MAX_PATH_LENGTH]; + snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname); + + char command[1024]; + snprintf (command, 1024, "dvo_client %s -result %s -hostID %d -hostdir %s", basecmd, resultFile, table->hosts[i].hostID, table->hosts[i].pathname); + + if (DEBUG || PARALLEL_MANUAL) fprintf (stderr, "command: %s\n", command); + + if (PARALLEL_MANUAL) { + continue; + } + + if (PARALLEL_SERIAL) { + int status = system (command); + if (status) { + fprintf (stderr, "ERROR running relphot_client\n"); + exit (2); + } + } else { + // launch the job on the remote machine (no handshake) + int errorInfo = 0; + int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE); + if (!pid) { + if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo); + exit (1); + } + table->hosts[i].pid = pid; // save for future reference + } + } + return TRUE; +} + +// bundle the arguments into a command and pass to dvo_client. this implementation +// expects there to be a result file from the clients, and to load this into vectors in +// the main shell. 'Nelements' is a temp hack : for most commands, the result vectors are +// concatenated, but for avmatch, the vectors are merged by index into a pre-known +// length. this is probably not a solution to a general problem.. +int HostTableParallelOps (int argc, char **argv, char *ResultFile, int Nelements) { + + int i; + + // load the list of hosts + SkyTable *sky = GetSkyTable(); + if (!sky) { + gprint (GP_ERR, "failed to load sky table for database\n"); + return FALSE; + } + + char *tmppath = GetCATDIR (); + if (!tmppath) { + gprint (GP_ERR, "failed to get CATDIR for database\n"); + return FALSE; + } + + char *CATDIR = abspath (tmppath, MAX_PATH_LENGTH); + if (!CATDIR) { + gprint (GP_ERR, "failed to make an absolute path from %s (too long)\n", tmppath); + return FALSE; + } + + HostTable *table = HostTableLoad (CATDIR, sky->hosts); + if (!table) { + gprint (GP_ERR, "ERROR: failure reading Host Table %s for database %s\n", sky->hosts, CATDIR); + return FALSE; + } + + // other things I need to append? + char *basecmd = paste_args (argc, argv); + + // determine the sky region + double Rmin, Rmax, Dmin, Dmax; + get_skyregion (&Rmin, &Rmax, &Dmin, &Dmax); + + // determine time reference and format + char *TimeRef = get_variable ("TIMEREF"); + if (!TimeRef) { + gprint (GP_ERR, "failed to find TIMEREF variable\n"); + return FALSE; + } + + char *TimeFormat = get_variable ("TIMEFORMAT"); + if (!TimeFormat) { + gprint (GP_ERR, "failed to find TIMEFORMAT variable\n"); + return FALSE; + } + + char tmp; + char *command = NULL; + int length = snprintf (&tmp, 0, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, Rmin, Rmax, Dmin, Dmax); + + ALLOCATE (command, char, length); + snprintf (command, length, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, Rmin, Rmax, Dmin, Dmax); + + // launch this command remotely + HostTableLaunchJobs (table, command); + free (command); + + if (PARALLEL_MANUAL) { + gprint (GP_ERR, "run the relphot_client commands above. when these are done, hit return\n"); + getchar(); + } + if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) { + HostTableWaitJobsGetIO (table, __FILE__, __LINE__); + } + + // load fields from file + int Nvec = 0; + Vector **vec = NULL; + for (i = 0; i < table->Nhosts; i++) { + if (table->hosts[i].status) continue; + + char resultFile[MAX_PATH_LENGTH]; + snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname); + + int Ninvec = 0; + Vector **invec = ReadVectorTableFITS (resultFile, "RESULT", &Ninvec); + + if (Nelements == 0) { + vec = MergeVectors (vec, &Nvec, invec, Ninvec); + if (vec != invec) { + FreeVectorArray (invec, Ninvec); + } + } else { + vec = MergeVectorsByIndex (vec, &Nvec, invec, Ninvec, Nelements); + FreeVectorArray (invec, Ninvec); + } + } + + // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere) + if (ResultFile) { + int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nvec, FALSE, NULL); + if (!status) { + gprint (GP_ERR, "failed to write result file %s\n", ResultFile); + return FALSE; + } + } + + for (i = 0; i < Nvec; i++) { + AssignVector (vec[i], vec[i]->name, ANYVECTOR, TRUE); + } + free (vec); + + return TRUE; +} Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/gstar.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/gstar.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/gstar.c (revision 33543) @@ -68,4 +68,10 @@ } + int PARALLEL = FALSE; + if ((N = get_argument (argc, argv, "-parallel"))) { + remove_argument (N, &argc, argv); + PARALLEL = TRUE; + } + if (argc != 4) { gprint (GP_ERR, "USAGE: gstar RA DEC Radius [-m]\n"); @@ -89,6 +95,25 @@ } + HostTable *table = NULL; + if (PARALLEL) { + char *CATDIR = GetCATDIR(); + table = HostTableLoad (CATDIR, sky->hosts); + if (!table) { + gprint (GP_ERR, "ERROR: failure reading Host Table %s for database %s\n", sky->hosts, CATDIR); + return FALSE; + } + + SkyRegion *region = skylist[0].regions[0]; + int hostID = (region->hostFlags & DATA_USE_BCK) ? region->backupID : region->hostID; + int index = table->index[hostID]; + + char hostfile[1024]; + snprintf (hostfile, 1024, "%s/%s.cpt", table->hosts[index].pathname, region->name); + catalog.filename = hostfile; + } else { + catalog.filename = skylist[0].filename[0]; + } + /* lock, load, unlock catalog */ - catalog.filename = skylist[0].filename[0]; catalog.catflags = GetMeasures ? LOAD_AVES | LOAD_MEAS | LOAD_SECF : LOAD_AVES | LOAD_SECF; catalog.Nsecfilt = 0; Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/mextract.c (revision 33543) @@ -1,12 +1,3 @@ # include "dvoshell.h" - -# define DEBUG 0 -# define PARALLEL_MANUAL 0 -# define PARALLEL_SERIAL 0 -# define MAX_PATH_LENGTH 1024 - -int HostTableLaunchJobs (HostTable *table, char *basecmd); -Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec); -int HostTableParallelOps (int argc, char **argv, char *ResultFile); int mextract (int argc, char **argv) { @@ -77,5 +68,5 @@ if (!SetSkyRegions (selection)) goto escape; - int status = HostTableParallelOps (argc, argv, ResultFile); + int status = HostTableParallelOps (argc, argv, ResultFile, 0); return status; } @@ -161,5 +152,5 @@ // does this host ID match the desired location for the table? - if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; + if (PARALLEL && !HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; /* lock, load, unlock catalog */ @@ -380,181 +371,2 @@ return (FALSE); } - -int HostTableLaunchJobs (HostTable *table, char *basecmd) { - - int i; - for (i = 0; i < table->Nhosts; i++) { - - // ensure that the paths are absolute path names - char *tmppath = abspath (table->hosts[i].pathname, MAX_PATH_LENGTH); - free (table->hosts[i].pathname); - table->hosts[i].pathname = tmppath; - - char resultFile[MAX_PATH_LENGTH]; - snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname); - - char command[1024]; - snprintf (command, 1024, "dvo_client %s -result %s -hostID %d -hostdir %s", basecmd, resultFile, table->hosts[i].hostID, table->hosts[i].pathname); - - if (DEBUG || PARALLEL_MANUAL) fprintf (stderr, "command: %s\n", command); - - if (PARALLEL_MANUAL) { - continue; - } - - if (PARALLEL_SERIAL) { - int status = system (command); - if (status) { - fprintf (stderr, "ERROR running relphot_client\n"); - exit (2); - } - } else { - // launch the job on the remote machine (no handshake) - int errorInfo = 0; - int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE); - if (!pid) { - if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo); - exit (1); - } - table->hosts[i].pid = pid; // save for future reference - } - } - return TRUE; -} - -// take two arrays of vectors and merge equal named vectors. -// for ease, require that the order of the names match & number of vectors match -Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec) { - - int i, j; - - if (vec == NULL) { - *Nvec = Ninvec; - return invec; - } - - myAssert (*Nvec == Ninvec, "programming error (1)"); - - for (i = 0; i < Ninvec; i++) { - myAssert (!strcmp(vec[i]->name, invec[i]->name), "programming error (2)"); - myAssert (vec[i]->type == invec[i]->type, "programming error (2)"); - - int N = vec[i]->Nelements; - if (vec[i]->type == OPIHI_FLT) { - REALLOCATE (vec[i]->elements.Flt, opihi_flt, vec[i]->Nelements + invec[i]->Nelements); - for (j = 0; j < invec[i]->Nelements; j++) { - vec[i]->elements.Flt[N+j] = invec[i]->elements.Flt[j]; - } - } else { - REALLOCATE (vec[i]->elements.Int, opihi_int, vec[i]->Nelements + invec[i]->Nelements); - for (j = 0; j < invec[i]->Nelements; j++) { - vec[i]->elements.Int[N+j] = invec[i]->elements.Int[j]; - } - } - vec[i]->Nelements += invec[i]->Nelements; - } - return vec; -} - -int HostTableParallelOps (int argc, char **argv, char *ResultFile) { - - int i; - - // load the list of hosts - SkyTable *sky = GetSkyTable(); - if (!sky) { - gprint (GP_ERR, "failed to load sky table for database\n"); - return FALSE; - } - - char *tmppath = GetCATDIR (); - if (!tmppath) { - gprint (GP_ERR, "failed to get CATDIR for database\n"); - return FALSE; - } - - char *CATDIR = abspath (tmppath, MAX_PATH_LENGTH); - if (!CATDIR) { - gprint (GP_ERR, "failed to make an absolute path from %s (too long)\n", tmppath); - return FALSE; - } - - HostTable *table = HostTableLoad (CATDIR, sky->hosts); - if (!table) { - gprint (GP_ERR, "ERROR: failure reading Host Table %s for database %s\n", sky->hosts, CATDIR); - return FALSE; - } - - // other things I need to append? - char *basecmd = paste_args (argc, argv); - - // determine the sky region - double Rmin, Rmax, Dmin, Dmax; - get_skyregion (&Rmin, &Rmax, &Dmin, &Dmax); - - // determine time reference and format - char *TimeRef = get_variable ("TIMEREF"); - if (!TimeRef) { - gprint (GP_ERR, "failed to find TIMEREF variable\n"); - return FALSE; - } - - char *TimeFormat = get_variable ("TIMEFORMAT"); - if (!TimeFormat) { - gprint (GP_ERR, "failed to find TIMEFORMAT variable\n"); - return FALSE; - } - - char tmp; - char *command = NULL; - int length = snprintf (&tmp, 0, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, Rmin, Rmax, Dmin, Dmax); - - ALLOCATE (command, char, length); - snprintf (command, length, "%s -D CATDIR %s -time %s %s -skyregion %f %f %f %f", basecmd, CATDIR, TimeRef, TimeFormat, Rmin, Rmax, Dmin, Dmax); - - // launch this command remotely - HostTableLaunchJobs (table, command); - free (command); - - if (PARALLEL_MANUAL) { - gprint (GP_ERR, "run the relphot_client commands above. when these are done, hit return\n"); - getchar(); - } - if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) { - HostTableWaitJobsGetIO (table, __FILE__, __LINE__); - } - - // load fields from file - int Nvec = 0; - Vector **vec = NULL; - for (i = 0; i < table->Nhosts; i++) { - if (table->hosts[i].status) continue; - - char resultFile[MAX_PATH_LENGTH]; - snprintf (resultFile, MAX_PATH_LENGTH, "%s/dvo.results.fits", table->hosts[i].pathname); - - int Ninvec = 0; - Vector **invec = ReadVectorTableFITS (resultFile, "RESULT", &Ninvec); - - vec = MergeVectors (vec, &Nvec, invec, Ninvec); - if (vec != invec) { - FreeVectorArray (invec, Ninvec); - } - } - - // write vectors to a table (this is used by parallel dvo operations, but can be used elsewhere) - if (ResultFile) { - int status = WriteVectorTableFITS (ResultFile, "RESULT", vec, Nvec, FALSE, NULL); - if (!status) { - gprint (GP_ERR, "failed to write result file %s\n", ResultFile); - return FALSE; - } - } - - for (i = 0; i < Nvec; i++) { - AssignVector (vec[i], vec[i]->name, ANYVECTOR, TRUE); - } - free (vec); - - return TRUE; -} Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/skyregion.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/skyregion.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/dvo/skyregion.c (revision 33543) @@ -107,4 +107,6 @@ ALLOCATE (new[0].regions, SkyRegion *, 1); ALLOCATE (new[0].filename, char *, 1); + new[0].Nregions = 0; + new[0].ownElements = FALSE; // this list is only holding a view to the elements // output list Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvomath.h =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvomath.h (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvomath.h (revision 33543) @@ -131,4 +131,6 @@ Vector *SelectVector PROTO((char *name, int mode, int verbose)); int AssignVector PROTO((Vector *vec, char *name, int mode, int verbose)); +Vector **MergeVectors PROTO((Vector **vec, int *Nvec, Vector **invec, int Ninvec)); +Vector **MergeVectorsByIndex PROTO((Vector **vec, int *Nvec, Vector **invec, int Ninvec, int Nelements)); /* vector IO functions */ Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvoshell.h =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvoshell.h (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/include/dvoshell.h (revision 33543) @@ -6,269 +6,4 @@ # ifndef DVOSHELL_H # define DVOSHELL_H - -#ifdef NOT_MOVED_TO_LIBDVO -typedef enum { - DB_STACK_NONE = 0, - DB_STACK_INT = 0x01, - DB_STACK_FIELD = 0x02, - DB_STACK_TEMP = 0x04, - DB_STACK_VALUE = 0x08, - DB_STACK_CLOSE_PAR = 0x10, - DB_STACK_OPEN_PAR, - DB_STACK_LOGIC, - DB_STACK_COMPARE, - DB_STACK_BITWISE, - DB_STACK_SUM, - DB_STACK_MULTIPLY, - DB_STACK_POWER, - DB_STACK_UNARY, -} dbStackTypes; - -/* magnitude types */ -enum {MAG_NONE, - MAG_INST, - MAG_CAT, - MAG_APER, - MAG_SYS, - MAG_REL, - MAG_CAL, - MAG_AVE, - MAG_REF, - MAG_ERR, - MAG_AVE_ERR, - MAG_PHOT_FLAGS, - MAG_CHISQ, - MAG_NCODE, - MAG_NPHOT, -}; - -/* measure fields */ -enum {MEAS_ZERO, - MEAS_GLON, - MEAS_GLAT, - MEAS_GLON_AVE, - MEAS_GLAT_AVE, - MEAS_ELON, - MEAS_ELAT, - MEAS_ELON_AVE, - MEAS_ELAT_AVE, - MEAS_RA, - MEAS_DEC, - MEAS_RA_AVE, - MEAS_DEC_AVE, - MEAS_RA_AVE_ERR, - MEAS_DEC_AVE_ERR, - MEAS_U_RA, - MEAS_U_DEC, - MEAS_U_RA_ERR, - MEAS_U_DEC_ERR, - MEAS_PAR, - MEAS_PAR_ERR, - MEAS_RA_OFFSET, - MEAS_DEC_OFFSET, - MEAS_RA_FIT_OFFSET, - MEAS_DEC_FIT_OFFSET, - MEAS_RA_OFFSET_ERR, - MEAS_DEC_OFFSET_ERR, - MEAS_CHISQ_POS, - MEAS_CHISQ_PM, - MEAS_CHISQ_PAR, - MEAS_TMEAN, - MEAS_TRANGE, - MEAS_NMEAS, - MEAS_NMISS, - MEAS_NPOS, - MEAS_OBJ_FLAGS, - MEAS_MAG, - MEAS_MINST, - MEAS_MCAT, - MEAS_MSYS, - MEAS_MREL, - MEAS_MCAL, - MEAS_EXPTIME, - MEAS_AIRMASS, - MEAS_ALT, - MEAS_AZ, - MEAS_PHOTCODE, - MEAS_PHOTCODE_EQUIV, - MEAS_TIME, - MEAS_FWHM, - MEAS_FWHM_MAJ, - MEAS_FWHM_MIN, - MEAS_THETA, - MEAS_POSANGLE, - MEAS_PLATESCALE, - MEAS_MXX, - MEAS_MXY, - MEAS_MYY, - MEAS_DOPHOT, - MEAS_DB_FLAGS, - MEAS_PHOT_FLAGS, - MEAS_XCCD, - MEAS_YCCD, - MEAS_XCCD_ERR, - MEAS_YCCD_ERR, - MEAS_POS_SYS_ERR, - MEAS_XMOSAIC, - MEAS_YMOSAIC, - MEAS_SKY, - MEAS_dSKY, - MEAS_DET_ID, - MEAS_OBJ_ID, - MEAS_CAT_ID, - MEAS_IMAGE_ID, - MEAS_PSF_QF, - MEAS_PSF_QF_PERFECT, - MEAS_PSF_CHISQ, - MEAS_PSF_NDOF, - MEAS_PSF_NPIX, - MEAS_CR_NSIGMA, - MEAS_EXT_NSIGMA, - MEAS_EXTERN_ID, -}; - -/* average fields */ -enum {AVE_ZERO, - AVE_RA, - AVE_DEC, - AVE_RA_ERR, - AVE_DEC_ERR, - AVE_GLON, - AVE_GLAT, - AVE_ELON, - AVE_ELAT, - AVE_U_RA, - AVE_U_DEC, - AVE_U_RA_ERR, - AVE_U_DEC_ERR, - AVE_PAR, - AVE_PAR_ERR, - AVE_CHISQ_POS, - AVE_CHISQ_PM, - AVE_CHISQ_PAR, - AVE_TMEAN, - AVE_TRANGE, - AVE_Xp, - AVE_NMEAS, - AVE_NMISS, - AVE_NPOS, - AVE_NPHOT, - AVE_NCODE, - AVE_MAG, - AVE_dMAG, - AVE_Xm, - AVE_OBJ_FLAGS, - AVE_TYPE, - AVE_TYPEFRAC, - AVE_OBJID, - AVE_CATID, - AVE_EXTID_HI, - AVE_EXTID_LO, -}; - -enum {IMAGE_ZERO, - IMAGE_RA, - IMAGE_DEC, - IMAGE_GLON, - IMAGE_GLAT, - IMAGE_ELON, - IMAGE_ELAT, - IMAGE_XM, - IMAGE_AIRMASS, - IMAGE_MCAL, - IMAGE_dMCAL, - IMAGE_PHOTCODE, - IMAGE_TIME, - IMAGE_FWHM, - IMAGE_FWHM_MEDIAN, - IMAGE_EXPTIME, - IMAGE_NSTAR, - IMAGE_NCAL, - IMAGE_SKY, - IMAGE_FLAGS, - IMAGE_CCDNUM, - IMAGE_NX_PIX, - IMAGE_NY_PIX, - IMAGE_THETA, - IMAGE_SKEW, - IMAGE_SCALE, - IMAGE_DSCALE, - IMAGE_APRESID, - IMAGE_DAPRESID, - IMAGE_SIDTIME, - IMAGE_LATITUDE, - IMAGE_DET_LIMIT, - IMAGE_SAT_LIMIT, - IMAGE_CERROR, - IMAGE_FWHM_MAJ, - IMAGE_FWHM_MIN, - IMAGE_FWHM_MAJ_MEDIAN, - IMAGE_FWHM_MIN_MEDIAN, - IMAGE_TRATE, - IMAGE_IMAGE_ID, - IMAGE_EXTERN_ID, - IMAGE_SOURCE_ID, - IMAGE_X_LL_CHIP, - IMAGE_X_LR_CHIP, - IMAGE_X_UL_CHIP, - IMAGE_X_UR_CHIP, - IMAGE_Y_LL_CHIP, - IMAGE_Y_LR_CHIP, - IMAGE_Y_UL_CHIP, - IMAGE_Y_UR_CHIP, - IMAGE_X_LL_FP, - IMAGE_X_LR_FP, - IMAGE_X_UL_FP, - IMAGE_X_UR_FP, - IMAGE_Y_LL_FP, - IMAGE_Y_LR_FP, - IMAGE_Y_UL_FP, - IMAGE_Y_UR_FP, - IMAGE_X_ERR_SYS, - IMAGE_Y_ERR_SYS, - IMAGE_MAG_ERR_SYS, - IMAGE_NFIT_PHOTOM, - IMAGE_NFIT_ASTROM, - IMAGE_NLINK_PHOTOM, - IMAGE_NLINK_ASTROM -}; - -enum {DVO_TABLE_AVERAGE, DVO_TABLE_MEASURE, DVO_TABLE_IMAGE}; -enum {DVO_DB_CMDLINE_ERROR, DVO_DB_CMDLINE_IS_END, DVO_DB_CMDLINE_IS_WHERE, DVO_DB_CMDLINE_IS_MATCH}; - -// options for selecting the ra,dec limits of the db selections -typedef struct { - char *name; - char *list; - int useDisplay; - int useSkyregion; -} SkyRegionSelection; - -// a single db field -typedef struct { - char *name; - int extract; - int table; - int ID; - int magMode; - char type; - PhotCode *photcode; -} dbField; - -// db boolean operations -typedef struct { - char *name; - char type; - int field; - opihi_flt FltValue; - opihi_int IntValue; -} dbStack; - -typedef struct { - opihi_flt Flt; - opihi_int Int; -} dbValue; - -#endif // notdef MOVED_TO_LIBDVO typedef struct { @@ -320,9 +55,4 @@ void FreeDVO PROTO((void)); int InitPhotcodes PROTO((void)); -#ifdef NOT_MOVED_TO_DVO -Image *LoadImages PROTO((off_t *Nimage)); -void FreeImages PROTO((Image *images)); -Image *MatchImage PROTO((unsigned int time, short int source, unsigned int imageID)); -#endif Coords *MatchMosaic PROTO((unsigned int time, short int source)); int Quality PROTO((Measure *measure, int IsDophot)); @@ -357,23 +87,5 @@ CMPstars *cmpReadText PROTO((FILE *f, off_t *nstars)); int RD_to_XYpic PROTO((double *x, double *y, double r, double d, Coords *coords, double Rmin, double Rmax, double Rmid, int *leftside)); -int wordhash (char *word); - -int HostTableLaunchJobs (HostTable *table, char *basecmd); - -#ifdef NOT_MOVED_TO_LIBDVO -// dvo DB field functions -dbField *dbCmdlineFields PROTO((int argc, char **argv, int table, int *last, int *nfields)); -int dbCmdlineConditions PROTO((int argc, char **argv, int first, int *nextField)); -dbStack *dbRPN PROTO((int argc, char **argv, int *nstack)); -int dbCheckStack PROTO((dbStack *stack, int Nstack, int table, dbField **inFields, int *Nfields)); -int dbBooleanCond PROTO((dbStack *inStack, int NinStack, dbValue *fields)); -void dbInitStack PROTO((dbStack *stack)); -void dbFreeStack PROTO((dbStack *stack, int Nstack)); -void dbFreeEntry PROTO((dbStack *stack)); -void dbFreeTempEntry PROTO((dbStack *stack)); - -dbStack *dbBinary PROTO((dbStack *V1, dbStack *V2, char *op, dbValue *fields)); -dbStack *dbUnary PROTO((dbStack *V1, char *op, dbValue *fields)); -#endif +int wordhash PROTO((char *word)); int GetMagMode PROTO((char *string)); @@ -385,31 +97,6 @@ dbValue dbExtractMeasures PROTO((Average *average, SecFilt *secfilt, Measure *measure, dbField *field)); dbValue dbExtractImages PROTO((Image *image, off_t Nimage, off_t N, dbField *field)); -#ifdef NOT_MOVED_TO_LIBDVO -dbValue dbExtractAverages PROTO((Average *average, SecFilt *secfilt, Measure *measure, dbField *field)); -void dbInitField PROTO((dbField *field)); -void dbFreeFields PROTO((dbField *fields, int Nfields)); -int dbAstroRegionLimits PROTO((dbStack **stack, int *nstack, SkyRegionSelection *selection, int table)); -int get_skyregion (double *Rs, double *Re, double *Ds, double *De); -int set_skyregion (double Rs, double Re, double Ds, double De); -void FreeImageSelection (void); -char *strfloat PROTO((float value)); - - -void FreeSkyRegionSelection (SkyRegionSelection *selection); - - -int dbExtractMeasuresInitTransform (CoordTransformSystem target); -int dbExtractMeasuresInitAve (void); -int dbExtractMeasuresInitMeas (void); -int dbExtractMeasuresInit (void); - -int dbExtractAveragesInitTransform (CoordTransformSystem target); -int dbExtractAveragesInit (void); - -int dbExtractImagesInitTransform (CoordTransformSystem target); -int dbExtractImagesInit (void); -int dbExtractImagesReset (void); -#endif // NOT_MOVED_TO_LIBDVO - +int HostTableLaunchJobs PROTO((HostTable *table, char *basecmd)); +int HostTableParallelOps PROTO((int argc, char **argv, char *ResultFile, int Nelements)); # endif Index: /branches/eam_branches/ipp-20111122/Ohana/src/opihi/lib.shell/VectorOps.c =================================================================== --- /branches/eam_branches/ipp-20111122/Ohana/src/opihi/lib.shell/VectorOps.c (revision 33542) +++ /branches/eam_branches/ipp-20111122/Ohana/src/opihi/lib.shell/VectorOps.c (revision 33543) @@ -338,2 +338,94 @@ return (Nvectors); } + +// Take two arrays of vectors and merge equal named vectors. +// Output is a single array in (vec), with vectors lengths of len(vec) + len(invec) +// For ease, require that the order of the names match & number of vectors match +Vector **MergeVectors (Vector **vec, int *Nvec, Vector **invec, int Ninvec) { + + int i, j; + + if (vec == NULL) { + *Nvec = Ninvec; + return invec; + } + + myAssert (*Nvec == Ninvec, "programming error (1)"); + + for (i = 0; i < Ninvec; i++) { + myAssert (!strcmp(vec[i]->name, invec[i]->name), "programming error (2)"); + myAssert (vec[i]->type == invec[i]->type, "programming error (2)"); + + int N = vec[i]->Nelements; + if (vec[i]->type == OPIHI_FLT) { + REALLOCATE (vec[i]->elements.Flt, opihi_flt, vec[i]->Nelements + invec[i]->Nelements); + for (j = 0; j < invec[i]->Nelements; j++) { + vec[i]->elements.Flt[N+j] = invec[i]->elements.Flt[j]; + } + } else { + REALLOCATE (vec[i]->elements.Int, opihi_int, vec[i]->Nelements + invec[i]->Nelements); + for (j = 0; j < invec[i]->Nelements; j++) { + vec[i]->elements.Int[N+j] = invec[i]->elements.Int[j]; + } + } + vec[i]->Nelements += invec[i]->Nelements; + } + return vec; +} + +// Take two arrays of vectors and merge equal named vectors, where the last vector +// specifies the element in the output vector. All input vectors must have a max sequence +// value of Nelements. Output is a single array in (vec), with vector lengths of +// Nelements for ease, require that the order of the names match & number of vectors match +Vector **MergeVectorsByIndex (Vector **vec, int *Nvec, Vector **invec, int Ninvec, int Nelements) { + + int i, j; + + // on first call, allocate a new vector, excluding the index + int newArray = FALSE; + if (vec == NULL) { + ALLOCATE (vec, Vector *, Ninvec - 1); + *Nvec = Ninvec - 1; + newArray = TRUE; + } + + myAssert (*Nvec == Ninvec - 1, "programming error (1)"); + + // find the index vector + int idx = Ninvec - 1; + myAssert (!strcmp(invec[idx]->name, "index"), "failed to find index vector"); + + for (i = 0; i < Ninvec - 1; i++) { + // on first call, create the output vector + if (newArray) { + vec[i] = InitVector(); + strcpy (vec[i]->name, invec[i]->name); + ResetVector (vec[i], invec[i]->type, Nelements); + for (j = 0; j < Nelements; j++) { + if (vec[i][0].type == OPIHI_FLT) { + vec[i][0].elements.Flt[j] = NAN; + } else { + vec[i][0].elements.Int[j] = 0; // or NAN_INT? + } + } + } + + myAssert (!strcmp(vec[i]->name, invec[i]->name), "programming error (2)"); + myAssert (vec[i]->type == invec[i]->type, "programming error (2)"); + + // copy vector elements from input to output, matching location + if (vec[i]->type == OPIHI_FLT) { + for (j = 0; j < invec[i]->Nelements; j++) { + int seq = invec[idx]->elements.Int[j]; + vec[i]->elements.Flt[seq] = invec[i]->elements.Flt[j]; + } + } else { + for (j = 0; j < invec[i]->Nelements; j++) { + int seq = invec[idx]->elements.Int[j]; + vec[i]->elements.Int[seq] = invec[i]->elements.Int[j]; + } + } + } + return vec; +} +