Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/Makefile =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/Makefile (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/Makefile (revision 35069) @@ -26,4 +26,5 @@ $(SRC)/initialize_dvopsps.$(ARCH).o \ $(SRC)/mysql_dvopsps.$(ARCH).o \ +$(SRC)/DetectionOps.$(ARCH).o \ $(SRC)/insert_skytable.$(ARCH).o \ $(SRC)/insert_objects_dvopsps.$(ARCH).o \ @@ -41,4 +42,5 @@ $(SRC)/initialize_dvopsps.$(ARCH).o \ $(SRC)/mysql_dvopsps.$(ARCH).o \ +$(SRC)/DetectionOps.$(ARCH).o \ $(SRC)/insert_skytable.$(ARCH).o \ $(SRC)/insert_objects_dvopsps.$(ARCH).o \ Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/include/dvopsps.h =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/include/dvopsps.h (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/include/dvopsps.h (revision 35069) @@ -8,18 +8,18 @@ typedef struct { - int imageID; - unsigned int ippDetectID; - uint64_t detectID; - uint64_t ippObjID; - uint64_t objID; - unsigned int flags; - float zp; - float zpErr; - float airMass; - float expTime; - double ra; - double dec_; - float raErr; - float decErr; + int imageID; + unsigned int ippDetectID; + uint64_t detectID; + uint64_t ippObjID; + uint64_t objID; + unsigned int flags; + float zp; + float zpErr; + float airMass; + float expTime; + double ra; + double dec; + float raErr; + float decErr; } Detections; @@ -37,4 +37,6 @@ char *SINGLE_CPT; + +char *RESULT_FILE; char *DATABASE_HOST; @@ -69,4 +71,11 @@ int insert_detections_mysql_init PROTO((IOBuffer *buffer)); +int insert_detections_mysql_array PROTO((Detections *detections, int Ndetections)); +int insert_detections_mysql_detvalue PROTO((IOBuffer *buffer, Detections *detection)); + +int init_detections PROTO(()); +int append_detections_dvopsps_catalog PROTO((Catalog *catalog)); +int save_detections_dvopsps PROTO(()); + int insert_objects_dvopsps PROTO(()); int insert_objects_dvopsps_parallel PROTO((SkyTable *sky)); @@ -84,2 +93,5 @@ int insert_skytable_mysql_value PROTO((IOBuffer *buffer, SkyRegion *region)); int insert_skytable_mysql_init PROTO((IOBuffer *buffer)); + +Detections *DetectionsLoad PROTO((char *filename, int *Ndetections)); +int DetectionsSave PROTO((char *filename, Detections *detections, int Ndetections)); Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/DetectionOps.c =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/DetectionOps.c (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/DetectionOps.c (revision 35069) @@ -1,3 +1,6 @@ # include "dvopsps.h" +# define BZERO_INT16 1.0*0x8000 +# define BZERO_INT32 1.0*0x80000000 +# define BZERO_INT64 1.0*0x8000000000000000 # define GET_COLUMN(OUT,NAME,TYPE) \ @@ -5,5 +8,5 @@ myAssert (!strcmp(type, #TYPE), "wrong column type"); -Detections *DetectionsLoad(char *filename) { +Detections *DetectionsLoad(char *filename, int *Ndetections) { int i, Ncol; @@ -101,7 +104,8 @@ fclose (f); + *Ndetections = Nrow; return detections; -escape: + escape: gfits_free_header (&header); gfits_free_matrix (&matrix); @@ -115,5 +119,5 @@ // we are passed a Detections structure, write it to a FITS table (3 ext) -int DetectionsSave(char *filename, Detections *detections) { +int DetectionsSave(char *filename, Detections *detections, int Ndetections) { int i; @@ -139,330 +143,95 @@ gfits_free_matrix (&matrix); - gfits_create_table_header (&theader, "BINTABLE", "DETECTIONS"); - - // XXX need to get the bzero values right - gfits_define_bintable_column (&theader, "J", "imageID", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "J", "ippDetectID", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "K", "detectID", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "K", "ippObjID", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "K", "objID", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "J", "flags", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "zp", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "zpErr", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "airMass", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "expTime", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "D", "ra", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "D", "dec", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "raErr", NULL, NULL, 1.0, 0.0); - gfits_define_bintable_column (&theader, "E", "decErr", NULL, NULL, 1.0, 0.0); - - // generate the output array that carries the data - gfits_create_table (&theader, &ftable); - - // create intermediate storage arrays - float *dR ; ALLOCATE (dR , float, catalog->Nmeasure); - float *dD ; ALLOCATE (dD , float, catalog->Nmeasure); - float *M ; ALLOCATE (M , float, catalog->Nmeasure); - float *Mcal ; ALLOCATE (Mcal , float, catalog->Nmeasure); - float *dM ; ALLOCATE (dM , float, catalog->Nmeasure); - float *airmass ; ALLOCATE (airmass , float, catalog->Nmeasure); - float *Xccd ; ALLOCATE (Xccd , float, catalog->Nmeasure); - float *Yccd ; ALLOCATE (Yccd , float, catalog->Nmeasure); - float *dt ; ALLOCATE (dt , float, catalog->Nmeasure); - int *t ; ALLOCATE (t , int , catalog->Nmeasure); - int *averef ; ALLOCATE (averef , int , catalog->Nmeasure); - int *imageID ; ALLOCATE (imageID , int , catalog->Nmeasure); - int *dbFlags ; ALLOCATE (dbFlags , int , catalog->Nmeasure); - int *photFlags ; ALLOCATE (photFlags, int , catalog->Nmeasure); - int *catID ; ALLOCATE (catID , int , catalog->Nmeasure); - short *photcode ; ALLOCATE (photcode , short, catalog->Nmeasure); - - // assign the storage arrays - MeasureTiny *measure = catalog->measure; - for (i = 0; i < catalog->Nmeasure; i++) { - imageID[i] = detections[i].imageID ; - ippDetectID[i] = detections[i].ippDetectID ; - detectID[i] = detections[i].detectID ; - ippObjID[i] = detections[i].ippObjID ; - objID[i] = detections[i].objID ; - flags[i] = detections[i].flags ; - zp[i] = detections[i].zp ; - zpErr[i] = detections[i].zpErr ; - airMass[i] = detections[i].airMass ; - expTime[i] = detections[i].expTime ; - ra[i] = detections[i].ra ; - dec[i] = detections[i].dec ; - raErr[i] = detections[i].raErr ; - decErr[i] = detections[i].decErr ; - } - - // add the columns to the output array - gfits_set_bintable_column (&theader, &ftable, "RA_OFF", dR, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "DEC_OFF", dD, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "MAG_SYS", M, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "MAG_CAL", Mcal, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "MAG_ERR", dM, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "AIRMASS", airmass, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "X_CCD", Xccd, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "Y_CCD", Yccd, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "EXPTIME", dt, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "TIME", t, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "AVE_REF", averef, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "IMAGE_ID", imageID, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "DB_FLAGS", dbFlags, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "PHOT_FLAGS", photFlags, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "CAT_ID", catID, catalog->Nmeasure); - gfits_set_bintable_column (&theader, &ftable, "PHOTCODE", photcode, catalog->Nmeasure); - - free (dR ); - free (dD ); - free (M ); - free (Mcal ); - free (dM ); - free (airmass ); - free (Xccd ); - free (Yccd ); - free (dt ); - free (t ); - free (averef ); - free (imageID ); - free (dbFlags ); - free (photFlags ); - free (catID ); - free (photcode); - - gfits_fwrite_Theader (f, &theader); - gfits_fwrite_table (f, &ftable); - gfits_free_header (&theader); - gfits_free_table (&ftable); + gfits_create_table_header (&theader, "BINTABLE", "DETECTIONS"); + + // XXX need to get the bzero values right + gfits_define_bintable_column (&theader, "J", "imageID", NULL, NULL, 1.0, BZERO_INT32); + gfits_define_bintable_column (&theader, "J", "ippDetectID", NULL, NULL, 1.0, BZERO_INT32); + gfits_define_bintable_column (&theader, "K", "detectID", NULL, NULL, 1.0, BZERO_INT64); + gfits_define_bintable_column (&theader, "K", "ippObjID", NULL, NULL, 1.0, BZERO_INT64); + gfits_define_bintable_column (&theader, "K", "objID", NULL, NULL, 1.0, BZERO_INT64); + gfits_define_bintable_column (&theader, "J", "flags", NULL, NULL, 1.0, BZERO_INT32); + gfits_define_bintable_column (&theader, "E", "zp", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "E", "zpErr", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "E", "airMass", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "E", "expTime", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "D", "ra", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "D", "dec", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "E", "raErr", NULL, NULL, 1.0, 0.0); + gfits_define_bintable_column (&theader, "E", "decErr", NULL, NULL, 1.0, 0.0); + + // generate the output array that carries the data + gfits_create_table (&theader, &ftable); + + // create intermediate storage arrays + int32_t *imageID ; ALLOCATE (imageID , int32_t, Ndetections); + uint32_t *ippDetectID ; ALLOCATE (ippDetectID , uint32_t, Ndetections); + uint64_t *detectID ; ALLOCATE (detectID , uint64_t, Ndetections); + uint64_t *ippObjID ; ALLOCATE (ippObjID , uint64_t, Ndetections); + uint64_t *objID ; ALLOCATE (objID , uint64_t, Ndetections); + uint32_t *flags ; ALLOCATE (flags , uint32_t, Ndetections); + float *zp ; ALLOCATE (zp , float, Ndetections); + float *zpErr ; ALLOCATE (zpErr , float, Ndetections); + float *airMass ; ALLOCATE (airMass , float, Ndetections); + float *expTime ; ALLOCATE (expTime , float, Ndetections); + double *ra ; ALLOCATE (ra , double, Ndetections); + double *dec ; ALLOCATE (dec , double, Ndetections); + float *raErr ; ALLOCATE (raErr , float, Ndetections); + float *decErr ; ALLOCATE (decErr , float, Ndetections); + + // assign the storage arrays + for (i = 0; i < Ndetections; i++) { + imageID[i] = detections[i].imageID ; + ippDetectID[i] = detections[i].ippDetectID ; + detectID[i] = detections[i].detectID ; + ippObjID[i] = detections[i].ippObjID ; + objID[i] = detections[i].objID ; + flags[i] = detections[i].flags ; + zp[i] = detections[i].zp ; + zpErr[i] = detections[i].zpErr ; + airMass[i] = detections[i].airMass ; + expTime[i] = detections[i].expTime ; + ra[i] = detections[i].ra ; + dec[i] = detections[i].dec ; + raErr[i] = detections[i].raErr ; + decErr[i] = detections[i].decErr ; + } + + // add the columns to the output array + gfits_set_bintable_column (&theader, &ftable, "imageID", imageID , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "ippDetectID", ippDetectID , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "detectID", detectID , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "ippObjID", ippObjID , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "objID", objID , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "flags", flags , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "zp", zp , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "zpErr", zpErr , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "airMass", airMass , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "expTime", expTime , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "ra", ra , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "dec", dec , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "raErr", raErr , Ndetections); + gfits_set_bintable_column (&theader, &ftable, "decErr", decErr , Ndetections); + + free (imageID ); + free (ippDetectID ); + free (detectID ); + free (ippObjID ); + free (objID ); + free (flags ); + free (zp ); + free (zpErr ); + free (airMass ); + free (expTime ); + free (ra ); + free (dec ); + free (raErr ); + free (decErr ); + + gfits_fwrite_Theader (f, &theader); + gfits_fwrite_table (f, &ftable); + gfits_free_header (&theader); + gfits_free_table (&ftable); return TRUE; } -// merge a list of catalogs into a single Detections array set -Detections *DetectionsMerge (Catalog *catalog, int Ncatalog) { - - off_t i, j, k; - - Detections *bcatalog = NULL; - ALLOCATE (bcatalog, Detections, 1); - - int Nmeas = 0; - int Naves = 0; - for (i = 0; i < Ncatalog; i++) { - Nmeas += catalog[i].Nmeasure; - Naves += catalog[i].Naverage; - } - - // XXX this prevents different catalogs from having different Nsecfilt values - int Nsecfilt = GetPhotcodeNsecfilt(); - - ALLOCATE (bcatalog[0].measure, MeasureTiny, Nmeas); - ALLOCATE (bcatalog[0].average, AverageTiny, Naves); - ALLOCATE (bcatalog[0].secfilt, SecFilt, Naves*Nsecfilt); - - int Nm = 0; - int Na = 0; - for (i = 0; i < Ncatalog; i++) { - if (!catalog[i].Naverage) continue; - for (j = 0; j < catalog[i].Naverage; j++) { - bcatalog[0].average[Na] = catalog[i].averageT[j]; - for (k = 0; k < Nsecfilt; k++) { - bcatalog[0].secfilt[Nsecfilt*Na + k] = catalog[i].secfilt[Nsecfilt*j + k]; - } - Na++; - assert (Na <= Naves); - } - for (j = 0; j < catalog[i].Nmeasure; j++) { - bcatalog[0].measure[Nm] = catalog[i].measureT[j]; - Nm++; - assert (Nm <= Nmeas); - } - } - bcatalog->Naverage = Na; - bcatalog->Nmeasure = Nm; - return bcatalog; -} - -// distribute a bright catalog across separate catalogs -CatalogSplitter *DetectionsSplitInit (int Nsecfilt) { - - int i; - - CatalogSplitter *catalogs = NULL; - - ALLOCATE (catalogs, CatalogSplitter, 1); - - // as we see new Detectionss, we will update maxID and extend this array - catalogs->maxID = 0; - ALLOCATE (catalogs->index, int, catalogs->maxID + 1); - for (i = 0; i <= catalogs->maxID; i++) catalogs->index[i] = -1; - - catalogs->Nsecfilt = Nsecfilt; - - catalogs->Ncatalog = 0; - catalogs->NCATALOG = 16; - catalogs->catalog = NULL; - ALLOCATE (catalogs->catalog, Catalog, catalogs->NCATALOG); - - ALLOCATE (catalogs->catIDs, int, catalogs->NCATALOG); - ALLOCATE (catalogs->NAVERAGE, off_t, catalogs->NCATALOG); - ALLOCATE (catalogs->NMEASURE, off_t, catalogs->NCATALOG); - - for (i = 0; i < catalogs->NCATALOG; i++) { - dvo_catalog_init (&catalogs->catalog[i], TRUE); - catalogs->catIDs[i] = 0; - catalogs->NAVERAGE[i] = 256; - catalogs->NMEASURE[i] = 256; - catalogs->catalog[i].Naverage = 0; - catalogs->catalog[i].Nmeasure = 0; - ALLOCATE (catalogs->catalog[i].averageT, AverageTiny, catalogs->NAVERAGE[i]); - ALLOCATE (catalogs->catalog[i].measureT, MeasureTiny, catalogs->NMEASURE[i]); - ALLOCATE (catalogs->catalog[i].secfilt, SecFilt, catalogs->NAVERAGE[i]*Nsecfilt); - } - return catalogs; -} - -// distribute a bright catalog across separate catalogs -int DetectionsSplitFree (CatalogSplitter *catalogs) { - - // XXX don't free the catalogs : - // for (i = 0; i < NCATALOG; i++) { - // free (catalogs->catalog[i].averageT); - // free (catalogs->catalog[i].measureT); - // free (catalogs->catalog[i].secfilt); - // } - // free (catalogs->catalog); - - free (catalogs->catIDs); - free (catalogs->NAVERAGE); - free (catalogs->NMEASURE); - free (catalogs->index); - free (catalogs); - return TRUE; -} - -// distribute a bright catalog across separate catalogs -int DetectionsSplit (CatalogSplitter *catalogs, Detections *bcatalog) { - - int i; - - int D_NCATALOG = 16; - - int Nsecfilt = catalogs->Nsecfilt; - - // find the max value of catID in this Detections - int catIDmax = 0; - for (i = 0; i < bcatalog->Naverage; i++) { - catIDmax = MAX(catIDmax, bcatalog->average[i].catID); - } - // XXX validate the measure ID range here - - int maxIDold = catalogs->maxID; - catalogs->maxID = MAX (maxIDold, catIDmax); - - // extend the index array and init - REALLOCATE (catalogs->index, int, catalogs->maxID + 1); - for (i = maxIDold + 1; i <= catalogs->maxID; i++) catalogs->index[i] = -1; - - // identify the new catID values - for (i = 0; i < bcatalog->Naverage; i++) { - int catID = bcatalog->average[i].catID; - assert (catID > 0); - assert (catID < catalogs->maxID + 1); - int idx = catalogs->index[catID]; - if (idx != -1) continue; // already have seen this one - - // a new catID value - int Ncat = catalogs->Ncatalog; // the next available slot - catalogs->catIDs[Ncat] = catID; - assert (Ncat >= 0); - assert (Ncat < catalogs->NCATALOG); - - catalogs->catalog[Ncat].Nsecfilt = Nsecfilt; - catalogs->catalog[Ncat].catID = catID; - - catalogs->index[catID] = Ncat; - assert (catID > 0); - assert (catID < catalogs->maxID + 1); - - catalogs->Ncatalog ++; - - if (catalogs->Ncatalog >= catalogs->NCATALOG) { - catalogs->NCATALOG += D_NCATALOG; - - // fprintf (stderr, "realloc catalogs->catalog: old: %llx ", (long long) catalogs->catalog); - REALLOCATE (catalogs->catalog, Catalog, catalogs->NCATALOG); - // fprintf (stderr, "new: %llx - %llx\n", (long long) catalogs->catalog, (long long) (catalogs->catalog + sizeof(Catalog)*catalogs->NCATALOG)); - - // fprintf (stderr, "realloc catalogs->NAVERAGE: old: %llx ", (long long) catalogs->NAVERAGE); - REALLOCATE (catalogs->NAVERAGE, off_t, catalogs->NCATALOG); - // fprintf (stderr, "new: %llx - %llx\n", (long long) catalogs->NAVERAGE, (long long) (catalogs->NAVERAGE + sizeof(off_t)*catalogs->NCATALOG)); - - // fprintf (stderr, "realloc catalogs->NMEASURE: old: %llx ", (long long) catalogs->NMEASURE); - REALLOCATE (catalogs->NMEASURE, off_t, catalogs->NCATALOG); - // fprintf (stderr, "new: %llx - %llx\n", (long long) catalogs->NMEASURE, (long long) (catalogs->NMEASURE + sizeof(off_t)*catalogs->NCATALOG)); - - REALLOCATE (catalogs->catIDs, int, catalogs->NCATALOG); - - int j; - for (j = catalogs->NCATALOG - D_NCATALOG; j < catalogs->NCATALOG; j++) { - dvo_catalog_init (&catalogs->catalog[j], TRUE); - catalogs->catIDs[j] = 0; - catalogs->NAVERAGE[j] = 256; - catalogs->NMEASURE[j] = 256; - catalogs->catalog[j].Naverage = 0; - catalogs->catalog[j].Nmeasure = 0; - ALLOCATE (catalogs->catalog[j].averageT, AverageTiny, catalogs->NAVERAGE[j]); - ALLOCATE (catalogs->catalog[j].measureT, MeasureTiny, catalogs->NMEASURE[j]); - ALLOCATE (catalogs->catalog[j].secfilt, SecFilt, catalogs->NAVERAGE[j]*Nsecfilt); - } - D_NCATALOG = MAX(2000, 2*D_NCATALOG); - } - } - - // each bcatalog (from each host) has a different set of catID ranges - // they also (probably) have contiguous ranges. But, it is a bit tricky to be sure I - // can use that info, so perhaps ignore it - - // assign the averages to the corresponding catalog - for (i = 0; i < bcatalog->Naverage; i++) { - int ID = bcatalog->average[i].catID; - int Nc = catalogs->index[ID]; - assert (Nc > -1); - assert (Nc < catalogs->NCATALOG); - - int Na = catalogs->catalog[Nc].Naverage; - catalogs->catalog[Nc].averageT[Na] = bcatalog->average[i]; - - // secfilt entries are grouped in blocks for each average - int k; - for (k = 0; k < Nsecfilt; k++) { - catalogs->catalog[Nc].secfilt[Na*Nsecfilt + k] = bcatalog->secfilt[i*Nsecfilt + k]; - } - - catalogs->catalog[Nc].Naverage ++; - if (catalogs->catalog[Nc].Naverage >= catalogs->NAVERAGE[Nc]) { - catalogs->NAVERAGE[Nc] += MAX(catalogs->NAVERAGE[Nc], 1024); - REALLOCATE (catalogs->catalog[Nc].averageT, AverageTiny, catalogs->NAVERAGE[Nc]); - REALLOCATE (catalogs->catalog[Nc].secfilt, SecFilt, catalogs->NAVERAGE[Nc]*Nsecfilt); - } - } - - // assign the measures to the corresponding catalog - // XXX what about averef and related links? Do I need them? - for (i = 0; i < bcatalog->Nmeasure; i++) { - int ID = bcatalog->measure[i].catID; - int Nc = catalogs->index[ID]; - assert (Nc > -1); - assert (Nc < catalogs->NCATALOG); - int Na = catalogs->catalog[Nc].Nmeasure; - catalogs->catalog[Nc].measureT[Na] = bcatalog->measure[i]; - catalogs->catalog[Nc].Nmeasure ++; - if (catalogs->catalog[Nc].Nmeasure >= catalogs->NMEASURE[Nc]) { - catalogs->NMEASURE[Nc] += MAX(catalogs->NMEASURE[Nc], 4096); - REALLOCATE (catalogs->catalog[Nc].measureT, MeasureTiny, catalogs->NMEASURE[Nc]); - } - } - - return TRUE; -} Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/initialize_dvopsps.c =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/initialize_dvopsps.c (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/initialize_dvopsps.c (revision 35069) @@ -76,4 +76,5 @@ } else usage_dvopsps(); + RESULT_FILE = NULL; // only used by dvopsps_client SINGLE_CPT = NULL; if ((N = get_argument (argc, argv, "-cpt"))) { @@ -216,4 +217,11 @@ } else usage_dvopsps_client(); + RESULT_FILE = NULL; // only used by dvopsps_client + if ((N = get_argument (argc, argv, "-save"))) { + remove_argument (N, &argc, argv); + RESULT_FILE = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } + SINGLE_CPT = NULL; if ((N = get_argument (argc, argv, "-cpt"))) { Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps.c =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps.c (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps.c (revision 35069) @@ -1,4 +1,15 @@ # include "dvopsps.h" # define USE_MYSQL 1 +# define SAVE_REMOTE 1 + +# define MARKTIME(MSG,...) { \ + float dtime; \ + gettimeofday (&stop, (void *) NULL); \ + dtime = DTIME (stop, start); \ + fprintf (stderr, MSG, __VA_ARGS__); } + +// we have two ways of writing to the mysql database: +// SAVE_REMOTE = 0 : each client sends the detections directly to the mysql server +// SAVE_REMOTE = 1 : each client saves the detections to disk and these are loaded by the main program and sent to the mysql server // determine the relevant catalogs, launch parallel clients if desired @@ -30,12 +41,19 @@ // NOTE: mysql connection happens here since each dvopsps_client makes its own connection MYSQL mysqlBase; - MYSQL *mysqlReal = mysql_dvopsps_connect (&mysqlBase); - if (!mysqlReal) { - fprintf (stderr, "failed to connect to mysql\n"); - exit (1); + MYSQL *mysqlReal = NULL; + if (!SAVE_REMOTE) { + mysqlReal = mysql_dvopsps_connect (&mysqlBase); + if (!mysqlReal) { + fprintf (stderr, "failed to connect to mysql\n"); + exit (1); + } } # else MYSQL *mysqlReal = NULL; # endif + + if (SAVE_REMOTE && HOST_ID) { + init_detections (); + } // select measurements for each populated catalog @@ -67,9 +85,8 @@ // NOTE: this is where the real action happens - if (1) { - insert_detections_dvopsps_catalog (&catalog, mysqlReal); + if (SAVE_REMOTE && HOST_ID) { + append_detections_dvopsps_catalog (&catalog); } else { - // NOTE: this is where the real action happens - save_detections_dvopsps_catalog (&catalog, "???"); + insert_detections_dvopsps_catalog (&catalog, mysqlReal); } @@ -79,18 +96,8 @@ } - if (0) { - char query[256]; - MYSQL_RES *result; - - sprintf (query, "commit;"); - if (mysql_query (mysqlReal, query)) { - fprintf (stderr, "failed to turn off autocommit\n"); - fprintf (stderr, "%s\n", mysql_error (mysqlReal)); - return FALSE; - } - result = mysql_store_result (mysqlReal); - mysql_free_result (result); - } - + if (SAVE_REMOTE && HOST_ID) { + save_detections_dvopsps (); + } + return (TRUE); } @@ -113,10 +120,14 @@ table->hosts[i].pathname = tmppath; + if (SAVE_REMOTE) { + // XXX need to uniquify the result file + ALLOCATE (table->hosts[i].results, char, 1024); + snprintf (table->hosts[i].results, 1024, "%s/dvopsps_detections.dat", table->hosts[i].pathname); + } + char command[1024]; snprintf (command, 1024, "dvopsps_client detections -hostID %d -catdir %s -hostdir %s -region %f %f %f %f", table->hosts[i].hostID, CATDIR, table->hosts[i].pathname, UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax); - - fprintf (stderr, "command: %s\n", command); char tmpline[1024]; @@ -126,5 +137,8 @@ snprintf (tmpline, 1024, "%s -dbname %s", command, DATABASE_NAME); strcpy (command, tmpline); - if (VERBOSE) { snprintf (tmpline, 1024, "%s -v", command); strcpy (command, tmpline); } + if (VERBOSE) { snprintf (tmpline, 1024, "%s -v", command); strcpy (command, tmpline); } + if (SAVE_REMOTE) { snprintf (tmpline, 1024, "%s -save %s", command, table->hosts[i].results); strcpy (command, tmpline); } + + fprintf (stderr, "command: %s\n", command); if (PARALLEL_MANUAL) continue; @@ -158,20 +172,81 @@ } -# if (0) - for (i = 0; i < table->Nhosts; i++) { - while ((detections = RemoteDetectionsLoad (table->hosts[i].results)) == NULL) { - // failed to get the data from this host. This can happen for various reasons. Give the user a chance to try again... - fprintf (stderr, "failed to read data from %s\n", table->hosts[i].hostname); - fprintf (stderr, "you may run the command manually\n"); - } - free (table->hosts[i].results); - table->hosts[i].results = NULL; + if (SAVE_REMOTE) { + int Ndetections = 0; + Detections *detections = NULL; + + for (i = 0; i < table->Nhosts; i++) { + while ((detections = DetectionsLoad (table->hosts[i].results, &Ndetections)) == NULL) { + // failed to get the data from this host. This can happen for various reasons. Give the user a chance to try again... + fprintf (stderr, "failed to read data from %s\n", table->hosts[i].hostname); + fprintf (stderr, "you may run the command manually\n"); + } + free (table->hosts[i].results); + table->hosts[i].results = NULL; - RemoveDetectionsInsert (detections); - - free (detections); - } -# endif + insert_detections_mysql_array (detections, Ndetections); + + free (detections); + } + } return (TRUE); } + +int insert_detections_mysql_array (Detections *detections, int Ndetections) { + + int i; + IOBuffer buffer; + buffer.Nalloc = 0; + + struct timeval start, stop; + + // NOTE: mysql connection happens here since each dvopsps_client makes its own connection + MYSQL mysqlBase; + MYSQL *mysql = NULL; + mysql = mysql_dvopsps_connect (&mysqlBase); + if (!mysql) { + fprintf (stderr, "failed to connect to mysql\n"); + exit (1); + } + + insert_detections_mysql_init (&buffer); + + gettimeofday (&start, (void *) NULL); + for (i = 0; i < Ndetections; i++) { + + // XXX check return status + insert_detections_mysql_detvalue (&buffer, &detections[i]); + // if (buffer.Nbuffer > MAX_BUFFER) { + if (buffer.Nbuffer > 2048) { + insert_detections_mysql_commit (&buffer, mysql); + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + insert_detections_mysql_init (&buffer); + } + } + insert_detections_mysql_commit (&buffer, mysql); + FreeIOBuffer (&buffer); + + MARKTIME("-- inserted %d rows in %f sec\n", Ndetections, dtime); + return (TRUE); +} + +int insert_detections_mysql_detvalue (IOBuffer *buffer, Detections *detection) { + + PrintIOBuffer (buffer, "(%d, ", detection->imageID); // imageID + PrintIOBuffer (buffer, "%u, ", detection->ippDetectID); // ippDetectID + PrintIOBuffer (buffer, "%lu, ", detection->detectID); // detectID + PrintIOBuffer (buffer, "%lu, ", detection->ippObjID); // ippObjID + PrintIOBuffer (buffer, "%lu, ", detection->objID); // objID + PrintIOBuffer (buffer, "%u, ", detection->flags); // flags + if (isnan(detection->zp)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ", detection->zp); // zp + if (isnan(detection->zpErr)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ", detection->zpErr); // zpErr + if (isnan(detection->airMass)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ", detection->airMass); // + if (isnan(detection->expTime)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ", detection->expTime); // expTime + if (isnan(detection->ra)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%lf, ", detection->ra); // ra + if (isnan(detection->dec)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%lf, ", detection->dec); // dec + if (isnan(detection->raErr)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ", detection->raErr); // estimate of raErr + if (isnan(detection->decErr)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f),\n", detection->decErr); // estimate of decErr + return TRUE; +} Index: /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c =================================================================== --- /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c (revision 35068) +++ /branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c (revision 35069) @@ -1,7 +1,4 @@ # include "dvopsps.h" -# define DEBUG 0 -# define SQL_INSERT 1 -// 1 = write data via sql insert values = (), -// 0 = write data via load from file +# define DEBUG 1 # define MARKTIME(MSG,...) { \ @@ -10,4 +7,79 @@ dtime = DTIME (stop, start); \ fprintf (stderr, MSG, __VA_ARGS__); } + +int Ndetections = 0; +int NDETECTIONS = 0; +Detections *detections = NULL; + +int init_detections () { + NDETECTIONS = 1000; + ALLOCATE (detections, Detections, NDETECTIONS); + return TRUE; +} + +int append_detections_dvopsps_catalog (Catalog *catalog) { + + off_t i, j; + + int missingID = 0; + + if (Ndetections + catalog[0].Nmeasure >= NDETECTIONS) { + NDETECTIONS = Ndetections + catalog[0].Nmeasure + 1000; + REALLOCATE (detections, Detections, NDETECTIONS); + } + + for (i = 0; i < catalog[0].Naverage; i++) { + + if (catalog[0].average[i].extID == 0) { + missingID ++; + } + + off_t m = catalog[0].average[i].measureOffset; + for (j = 0; j < catalog[0].average[i].Nmeasure; j++) { + + Average *average = &catalog->average[i]; + Measure *measure = &catalog->measure[m + j]; + PhotCode *code = GetPhotcodebyCode(measure->photcode); + + detections[Ndetections].imageID = measure->imageID; // imageID + detections[Ndetections].ippDetectID = measure->detID; // ippDetectID + detections[Ndetections].detectID = measure->extID; // detectID + detections[Ndetections].ippObjID = ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID; // ippObjID + // NOTE: this is better, but the above is the current ippToPsps value + // detections[Ndetections]. ((uint64_t)average->catID << 32) + (uint64_t)average->objID; // ippObjID + detections[Ndetections].objID = average->extID; // objID + detections[Ndetections].flags = measure->dbFlags; // flags + detections[Ndetections].zp = code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal; // zp + detections[Ndetections].zpErr = measure->dMcal; // zpErr + detections[Ndetections].airMass = measure->airmass; + detections[Ndetections].expTime = pow(10.0, 0.4 * measure->dt); // expTime + detections[Ndetections].ra = average->R - measure->dR / 3600.; // ra + detections[Ndetections].dec = average->D - measure->dR / 3600.; // dec + detections[Ndetections].raErr = measure->dXccd * 0.01 * fabs(measure->pltscale); // estimate of raErr + detections[Ndetections].decErr = measure->dYccd * 0.01 * fabs(measure->pltscale); // estimate of decErr + + Ndetections ++; + myAssert (Ndetections <= NDETECTIONS, "programming error"); + } + } + + fprintf (stderr, "write %d detections, %d missing extID\n", Ndetections, missingID); + return TRUE; +} + +// XXX write a single table for all detections or a series of tables? +// if I write a single table, then I have to store all data in memory +// if I write multiple tables, then I need to (a) iterate over the tables and (b) +// incur seek hits on the file +int save_detections_dvopsps () { + + if (!DetectionsSave (RESULT_FILE, detections, Ndetections)) { + fprintf (stderr, "failed to save detection file %s\n", RESULT_FILE); + } + + free (detections); + Ndetections = 0; + return (TRUE); +} int insert_detections_dvopsps_catalog (Catalog *catalog, MYSQL *mysql) { @@ -76,9 +148,5 @@ } -// XXX I need to add the fileID (catID) here - -# if (SQL_INSERT) PrintIOBuffer (buffer, "INSERT INTO dvoDetectionFull (imageID, ippDetectID, detectID, ippObjID, objID, flags, zp, zpErr, airMass, expTime, ra, dec_, raErr, decErr) VALUES \n"); -# endif return TRUE; @@ -89,31 +157,22 @@ // XXX I am changing the def of ippObjID if I use the code below PhotCode *code = GetPhotcodebyCode(measure->photcode); -# if (SQL_INSERT) PrintIOBuffer (buffer, " (%d, %u, %lu, %lu, %lu, %u, %f, %f, %f, %f, %lf, %lf, %f, %f),\n", -# else - PrintIOBuffer (buffer, " %d %u %lu %lu %lu %u %f %f %f %f %lf %lf %f %f\n", -# endif - measure->imageID, // imageID - measure->detID, // ippDetectID - measure->extID, // detectID - ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID, // ippObjID + measure->imageID, // imageID + measure->detID, // ippDetectID + measure->extID, // detectID + ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID, // ippObjID // NOTE: this is better, but the above is the current ippToPsps value // ((uint64_t)average->catID << 32) + (uint64_t)average->objID, // ippObjID - average->extID, // objID - measure->dbFlags, // flags - code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal, // zp - measure->dMcal, // zpErr - measure->airmass, - pow(10.0, 0.4 * measure->dt), // expTime - average->R - measure->dR / 3600., // ra - average->D - measure->dR / 3600., // dec - measure->dXccd * 0.01 * fabs(measure->pltscale), // estimate of raErr - measure->dYccd * 0.01 * fabs(measure->pltscale) // estimate of decErr -// XXX fileID?? -# if (SQL_INSERT) - ); -# else + average->extID, // objID + measure->dbFlags, // flags + code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal, // zp + measure->dMcal, // zpErr + measure->airmass, + pow(10.0, 0.4 * measure->dt), // expTime + average->R - measure->dR / 3600., // ra + average->D - measure->dR / 3600., // dec + measure->dXccd * 0.01 * fabs(measure->pltscale), // estimate of raErr + measure->dYccd * 0.01 * fabs(measure->pltscale) // estimate of decErr ); -# endif return TRUE; } @@ -123,5 +182,4 @@ MYSQL_RES *result; -# if (SQL_INSERT) // check that the last two chars are ,\n and replace with ;\n if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) { @@ -131,26 +189,9 @@ return FALSE; } -# endif // XXX check return status if (mysql) { -# if (SQL_INSERT) if (DEBUG) fprintf (stderr, "%s\n", buffer->buffer); int status = mysql_query(mysql, buffer->buffer); -# else - FILE *f = fopen ("/tmp/dvopsps.dat", "w"); - if (!f) { - fprintf (stderr, "failed to write buffer file\n"); - return FALSE; - } - int fd = fileno (f); - int Nbytes = write (fd, buffer->buffer, buffer->Nbuffer); - if (Nbytes != buffer->Nbuffer) { - fprintf (stderr, "error writing data to buffer file\n"); - } - char sql[256]; - snprintf (sql, 256, "load data infile '/tmp/dvopsps.dat' into table dvoDetectionFull;"); - int status = mysql_query(mysql, sql); -# endif if (status) { fprintf (stderr, "error with insert:\n");