IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 35069


Ignore:
Timestamp:
Jan 28, 2013, 1:50:16 PM (13 years ago)
Author:
eugene
Message:

fix up inserts by local

Location:
branches/eam_branches/ipp-20121219/Ohana/src/dvopsps
Files:
6 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/Makefile

    r35055 r35069  
    2626$(SRC)/initialize_dvopsps.$(ARCH).o \
    2727$(SRC)/mysql_dvopsps.$(ARCH).o \
     28$(SRC)/DetectionOps.$(ARCH).o \
    2829$(SRC)/insert_skytable.$(ARCH).o \
    2930$(SRC)/insert_objects_dvopsps.$(ARCH).o \
    … …  
    4142$(SRC)/initialize_dvopsps.$(ARCH).o \
    4243$(SRC)/mysql_dvopsps.$(ARCH).o \
     44$(SRC)/DetectionOps.$(ARCH).o \
    4345$(SRC)/insert_skytable.$(ARCH).o \
    4446$(SRC)/insert_objects_dvopsps.$(ARCH).o \

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/include/dvopsps.h

    r35064 r35069  
    88
    99typedef struct {
    10     int imageID;
    11     unsigned int ippDetectID;
    12     uint64_t detectID;
    13     uint64_t ippObjID;
    14     uint64_t objID;
    15     unsigned int flags;
    16     float zp;
    17     float zpErr;
    18     float airMass;
    19     float expTime;
    20     double ra;
    21     double dec_;
    22     float raErr;
    23     float decErr;
     10  int imageID;
     11  unsigned int ippDetectID;
     12  uint64_t detectID;
     13  uint64_t ippObjID;
     14  uint64_t objID;
     15  unsigned int flags;
     16  float zp;
     17  float zpErr;
     18  float airMass;
     19  float expTime;
     20  double ra;
     21  double dec;
     22  float raErr;
     23  float decErr;
    2424} Detections;
    2525
    … …  
    3737
    3838char        *SINGLE_CPT;
     39
     40char        *RESULT_FILE;
    3941
    4042char        *DATABASE_HOST;
    … …  
    6971int    insert_detections_mysql_init       PROTO((IOBuffer *buffer));
    7072
     73int    insert_detections_mysql_array      PROTO((Detections *detections, int Ndetections));
     74int    insert_detections_mysql_detvalue   PROTO((IOBuffer *buffer, Detections *detection));
     75
     76int    init_detections                    PROTO(());
     77int    append_detections_dvopsps_catalog  PROTO((Catalog *catalog));
     78int    save_detections_dvopsps            PROTO(());
     79
    7180int    insert_objects_dvopsps             PROTO(());
    7281int    insert_objects_dvopsps_parallel    PROTO((SkyTable *sky));
    … …  
    8493int    insert_skytable_mysql_value        PROTO((IOBuffer *buffer, SkyRegion *region));
    8594int    insert_skytable_mysql_init         PROTO((IOBuffer *buffer));
     95
     96Detections *DetectionsLoad                PROTO((char *filename, int *Ndetections));
     97int         DetectionsSave                PROTO((char *filename, Detections *detections, int Ndetections));

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/DetectionOps.c

    r35064 r35069  
    11# include "dvopsps.h"
     2# define BZERO_INT16 1.0*0x8000
     3# define BZERO_INT32 1.0*0x80000000
     4# define BZERO_INT64 1.0*0x8000000000000000
    25
    36# define GET_COLUMN(OUT,NAME,TYPE)                                      \
    … …  
    58  myAssert (!strcmp(type, #TYPE), "wrong column type");
    69
    7 Detections *DetectionsLoad(char *filename) {
     10Detections *DetectionsLoad(char *filename, int *Ndetections) {
    811
    912  int i, Ncol;
    … …  
    101104  fclose (f);
    102105
     106  *Ndetections = Nrow;
    103107  return detections;
    104108
    105 escape:
     109 escape:
    106110  gfits_free_header (&header);
    107111  gfits_free_matrix (&matrix);
    … …  
    115119
    116120// we are passed a Detections structure, write it to a FITS table (3 ext)
    117 int DetectionsSave(char *filename, Detections *detections) {
     121int DetectionsSave(char *filename, Detections *detections, int Ndetections) {
    118122
    119123  int i;
    … …  
    139143  gfits_free_matrix (&matrix);
    140144
    141     gfits_create_table_header (&theader, "BINTABLE", "DETECTIONS");
    142 
    143     // XXX need to get the bzero values right
    144     gfits_define_bintable_column (&theader, "J", "imageID",     NULL, NULL, 1.0, 0.0);
    145     gfits_define_bintable_column (&theader, "J", "ippDetectID", NULL, NULL, 1.0, 0.0);
    146     gfits_define_bintable_column (&theader, "K", "detectID",    NULL, NULL, 1.0, 0.0);
    147     gfits_define_bintable_column (&theader, "K", "ippObjID",    NULL, NULL, 1.0, 0.0);
    148     gfits_define_bintable_column (&theader, "K", "objID",       NULL, NULL, 1.0, 0.0);
    149     gfits_define_bintable_column (&theader, "J", "flags",       NULL, NULL, 1.0, 0.0);
    150     gfits_define_bintable_column (&theader, "E", "zp",          NULL, NULL, 1.0, 0.0);
    151     gfits_define_bintable_column (&theader, "E", "zpErr",       NULL, NULL, 1.0, 0.0);
    152     gfits_define_bintable_column (&theader, "E", "airMass",     NULL, NULL, 1.0, 0.0);
    153     gfits_define_bintable_column (&theader, "E", "expTime",     NULL, NULL, 1.0, 0.0);
    154     gfits_define_bintable_column (&theader, "D", "ra",          NULL, NULL, 1.0, 0.0);
    155     gfits_define_bintable_column (&theader, "D", "dec",         NULL, NULL, 1.0, 0.0);
    156     gfits_define_bintable_column (&theader, "E", "raErr",       NULL, NULL, 1.0, 0.0);
    157     gfits_define_bintable_column (&theader, "E", "decErr",      NULL, NULL, 1.0, 0.0);
    158 
    159     // generate the output array that carries the data
    160     gfits_create_table (&theader, &ftable);
    161 
    162     // create intermediate storage arrays
    163     float *dR        ; ALLOCATE (dR       ,  float, catalog->Nmeasure);
    164     float *dD        ; ALLOCATE (dD       ,  float, catalog->Nmeasure);
    165     float *M         ; ALLOCATE (M        ,  float, catalog->Nmeasure);
    166     float *Mcal      ; ALLOCATE (Mcal     ,  float, catalog->Nmeasure);
    167     float *dM        ; ALLOCATE (dM       ,  float, catalog->Nmeasure);
    168     float *airmass   ; ALLOCATE (airmass  ,  float, catalog->Nmeasure);
    169     float *Xccd      ; ALLOCATE (Xccd     ,  float, catalog->Nmeasure);
    170     float *Yccd      ; ALLOCATE (Yccd     ,  float, catalog->Nmeasure);
    171     float *dt        ; ALLOCATE (dt       ,  float, catalog->Nmeasure);
    172     int   *t         ; ALLOCATE (t        ,  int  , catalog->Nmeasure);
    173     int   *averef    ; ALLOCATE (averef   ,  int  , catalog->Nmeasure);
    174     int   *imageID   ; ALLOCATE (imageID  ,  int  , catalog->Nmeasure);
    175     int   *dbFlags   ; ALLOCATE (dbFlags  ,  int  , catalog->Nmeasure);
    176     int   *photFlags ; ALLOCATE (photFlags,  int  , catalog->Nmeasure);
    177     int   *catID     ; ALLOCATE (catID    ,  int  , catalog->Nmeasure);
    178     short *photcode  ; ALLOCATE (photcode ,  short, catalog->Nmeasure);
    179 
    180     // assign the storage arrays
    181     MeasureTiny *measure = catalog->measure;
    182     for (i = 0; i < catalog->Nmeasure; i++) {
    183       imageID[i]     = detections[i].imageID     ;
    184       ippDetectID[i] = detections[i].ippDetectID ;
    185       detectID[i]    = detections[i].detectID    ;
    186       ippObjID[i]    = detections[i].ippObjID    ;
    187       objID[i]       = detections[i].objID       ;
    188       flags[i]       = detections[i].flags       ;
    189       zp[i]          = detections[i].zp          ;
    190       zpErr[i]       = detections[i].zpErr       ;
    191       airMass[i]     = detections[i].airMass     ;
    192       expTime[i]     = detections[i].expTime     ;
    193       ra[i]          = detections[i].ra          ;
    194       dec[i]         = detections[i].dec         ;
    195       raErr[i]       = detections[i].raErr       ;
    196       decErr[i]      = detections[i].decErr      ;
    197     }
    198 
    199     // add the columns to the output array
    200     gfits_set_bintable_column (&theader, &ftable, "RA_OFF",     dR,        catalog->Nmeasure);
    201     gfits_set_bintable_column (&theader, &ftable, "DEC_OFF",    dD,        catalog->Nmeasure);
    202     gfits_set_bintable_column (&theader, &ftable, "MAG_SYS",    M,         catalog->Nmeasure);
    203     gfits_set_bintable_column (&theader, &ftable, "MAG_CAL",    Mcal,      catalog->Nmeasure);
    204     gfits_set_bintable_column (&theader, &ftable, "MAG_ERR",    dM,        catalog->Nmeasure);
    205     gfits_set_bintable_column (&theader, &ftable, "AIRMASS",    airmass,   catalog->Nmeasure);
    206     gfits_set_bintable_column (&theader, &ftable, "X_CCD",      Xccd,      catalog->Nmeasure);
    207     gfits_set_bintable_column (&theader, &ftable, "Y_CCD",      Yccd,      catalog->Nmeasure);
    208     gfits_set_bintable_column (&theader, &ftable, "EXPTIME",    dt,        catalog->Nmeasure);
    209     gfits_set_bintable_column (&theader, &ftable, "TIME",       t,         catalog->Nmeasure);
    210     gfits_set_bintable_column (&theader, &ftable, "AVE_REF",    averef,    catalog->Nmeasure);
    211     gfits_set_bintable_column (&theader, &ftable, "IMAGE_ID",   imageID,   catalog->Nmeasure);
    212     gfits_set_bintable_column (&theader, &ftable, "DB_FLAGS",   dbFlags,   catalog->Nmeasure);
    213     gfits_set_bintable_column (&theader, &ftable, "PHOT_FLAGS", photFlags, catalog->Nmeasure);
    214     gfits_set_bintable_column (&theader, &ftable, "CAT_ID",     catID,     catalog->Nmeasure);
    215     gfits_set_bintable_column (&theader, &ftable, "PHOTCODE",   photcode,  catalog->Nmeasure);
    216 
    217     free (dR      );
    218     free (dD      );
    219     free (M       );
    220     free (Mcal    );
    221     free (dM      );
    222     free (airmass );
    223     free (Xccd    );
    224     free (Yccd    );
    225     free (dt      );
    226     free (t       );
    227     free (averef  );
    228     free (imageID );
    229     free (dbFlags );
    230     free (photFlags );
    231     free (catID   );
    232     free (photcode);
    233 
    234     gfits_fwrite_Theader (f, &theader);
    235     gfits_fwrite_table  (f, &ftable);
    236     gfits_free_header (&theader);
    237     gfits_free_table (&ftable);
     145  gfits_create_table_header (&theader, "BINTABLE", "DETECTIONS");
     146
     147  // XXX need to get the bzero values right
     148  gfits_define_bintable_column (&theader, "J", "imageID",     NULL, NULL, 1.0, BZERO_INT32);
     149  gfits_define_bintable_column (&theader, "J", "ippDetectID", NULL, NULL, 1.0, BZERO_INT32);
     150  gfits_define_bintable_column (&theader, "K", "detectID",    NULL, NULL, 1.0, BZERO_INT64);
     151  gfits_define_bintable_column (&theader, "K", "ippObjID",    NULL, NULL, 1.0, BZERO_INT64);
     152  gfits_define_bintable_column (&theader, "K", "objID",       NULL, NULL, 1.0, BZERO_INT64);
     153  gfits_define_bintable_column (&theader, "J", "flags",       NULL, NULL, 1.0, BZERO_INT32);
     154  gfits_define_bintable_column (&theader, "E", "zp",          NULL, NULL, 1.0, 0.0);
     155  gfits_define_bintable_column (&theader, "E", "zpErr",       NULL, NULL, 1.0, 0.0);
     156  gfits_define_bintable_column (&theader, "E", "airMass",     NULL, NULL, 1.0, 0.0);
     157  gfits_define_bintable_column (&theader, "E", "expTime",     NULL, NULL, 1.0, 0.0);
     158  gfits_define_bintable_column (&theader, "D", "ra",          NULL, NULL, 1.0, 0.0);
     159  gfits_define_bintable_column (&theader, "D", "dec",         NULL, NULL, 1.0, 0.0);
     160  gfits_define_bintable_column (&theader, "E", "raErr",       NULL, NULL, 1.0, 0.0);
     161  gfits_define_bintable_column (&theader, "E", "decErr",      NULL, NULL, 1.0, 0.0);
     162
     163  // generate the output array that carries the data
     164  gfits_create_table (&theader, &ftable);
     165
     166  // create intermediate storage arrays
     167   int32_t   *imageID     ; ALLOCATE (imageID     ,   int32_t, Ndetections);
     168  uint32_t   *ippDetectID ; ALLOCATE (ippDetectID ,  uint32_t, Ndetections);
     169  uint64_t   *detectID    ; ALLOCATE (detectID    ,  uint64_t, Ndetections);
     170  uint64_t   *ippObjID    ; ALLOCATE (ippObjID    ,  uint64_t, Ndetections);
     171  uint64_t   *objID       ; ALLOCATE (objID       ,  uint64_t, Ndetections);
     172  uint32_t   *flags       ; ALLOCATE (flags       ,  uint32_t, Ndetections);
     173  float      *zp          ; ALLOCATE (zp          ,  float,    Ndetections);
     174  float      *zpErr       ; ALLOCATE (zpErr       ,  float,    Ndetections);
     175  float      *airMass     ; ALLOCATE (airMass     ,  float,    Ndetections);
     176  float      *expTime     ; ALLOCATE (expTime     ,  float,    Ndetections);
     177  double     *ra          ; ALLOCATE (ra          ,  double,   Ndetections);
     178  double     *dec         ; ALLOCATE (dec         ,  double,   Ndetections);
     179  float      *raErr       ; ALLOCATE (raErr       ,  float,    Ndetections);
     180  float      *decErr      ; ALLOCATE (decErr      ,  float,    Ndetections);
     181
     182  // assign the storage arrays
     183  for (i = 0; i < Ndetections; i++) {
     184    imageID[i]     = detections[i].imageID     ;
     185    ippDetectID[i] = detections[i].ippDetectID ;
     186    detectID[i]    = detections[i].detectID    ;
     187    ippObjID[i]    = detections[i].ippObjID    ;
     188    objID[i]       = detections[i].objID       ;
     189    flags[i]       = detections[i].flags       ;
     190    zp[i]          = detections[i].zp          ;
     191    zpErr[i]       = detections[i].zpErr       ;
     192    airMass[i]     = detections[i].airMass     ;
     193    expTime[i]     = detections[i].expTime     ;
     194    ra[i]          = detections[i].ra          ;
     195    dec[i]         = detections[i].dec         ;
     196    raErr[i]       = detections[i].raErr       ;
     197    decErr[i]      = detections[i].decErr      ;
     198  }
     199
     200  // add the columns to the output array
     201  gfits_set_bintable_column (&theader, &ftable, "imageID",     imageID     , Ndetections);
     202  gfits_set_bintable_column (&theader, &ftable, "ippDetectID", ippDetectID , Ndetections);
     203  gfits_set_bintable_column (&theader, &ftable, "detectID",    detectID    , Ndetections);
     204  gfits_set_bintable_column (&theader, &ftable, "ippObjID",    ippObjID    , Ndetections);
     205  gfits_set_bintable_column (&theader, &ftable, "objID",       objID       , Ndetections);
     206  gfits_set_bintable_column (&theader, &ftable, "flags",       flags       , Ndetections);
     207  gfits_set_bintable_column (&theader, &ftable, "zp",          zp          , Ndetections);
     208  gfits_set_bintable_column (&theader, &ftable, "zpErr",       zpErr       , Ndetections);
     209  gfits_set_bintable_column (&theader, &ftable, "airMass",     airMass     , Ndetections);
     210  gfits_set_bintable_column (&theader, &ftable, "expTime",     expTime     , Ndetections);
     211  gfits_set_bintable_column (&theader, &ftable, "ra",          ra          , Ndetections);
     212  gfits_set_bintable_column (&theader, &ftable, "dec",         dec         , Ndetections);
     213  gfits_set_bintable_column (&theader, &ftable, "raErr",       raErr       , Ndetections);
     214  gfits_set_bintable_column (&theader, &ftable, "decErr",      decErr      , Ndetections);
     215
     216  free (imageID     );
     217  free (ippDetectID );
     218  free (detectID    );
     219  free (ippObjID    );
     220  free (objID       );
     221  free (flags       );
     222  free (zp          );
     223  free (zpErr       );
     224  free (airMass     );
     225  free (expTime     );
     226  free (ra          );
     227  free (dec         );
     228  free (raErr       );
     229  free (decErr      );
     230
     231  gfits_fwrite_Theader (f, &theader);
     232  gfits_fwrite_table  (f, &ftable);
     233  gfits_free_header (&theader);
     234  gfits_free_table (&ftable);
    238235  return TRUE;
    239236}
    240237
    241 // merge a list of catalogs into a single Detections array set
    242 Detections *DetectionsMerge (Catalog *catalog, int Ncatalog) {
    243 
    244   off_t i, j, k;
    245 
    246   Detections *bcatalog = NULL;
    247   ALLOCATE (bcatalog, Detections, 1);
    248  
    249   int Nmeas = 0;
    250   int Naves = 0;
    251   for (i = 0; i < Ncatalog; i++) {
    252     Nmeas += catalog[i].Nmeasure;
    253     Naves += catalog[i].Naverage;
    254   }
    255    
    256   // XXX this prevents different catalogs from having different Nsecfilt values
    257   int Nsecfilt = GetPhotcodeNsecfilt();
    258 
    259   ALLOCATE (bcatalog[0].measure, MeasureTiny, Nmeas);
    260   ALLOCATE (bcatalog[0].average, AverageTiny, Naves);
    261   ALLOCATE (bcatalog[0].secfilt, SecFilt, Naves*Nsecfilt);
    262 
    263   int Nm = 0;
    264   int Na = 0;
    265   for (i = 0; i < Ncatalog; i++) {
    266     if (!catalog[i].Naverage) continue;
    267     for (j = 0; j < catalog[i].Naverage; j++) {
    268       bcatalog[0].average[Na] = catalog[i].averageT[j];
    269       for (k = 0; k < Nsecfilt; k++) {
    270         bcatalog[0].secfilt[Nsecfilt*Na + k] = catalog[i].secfilt[Nsecfilt*j + k];
    271       }
    272       Na++;
    273       assert (Na <= Naves);
    274     }
    275     for (j = 0; j < catalog[i].Nmeasure; j++) {
    276       bcatalog[0].measure[Nm] = catalog[i].measureT[j];
    277       Nm++;
    278       assert (Nm <= Nmeas);
    279     }
    280   }
    281   bcatalog->Naverage = Na;
    282   bcatalog->Nmeasure = Nm;
    283   return bcatalog;
    284 }
    285 
    286 // distribute a bright catalog across separate catalogs
    287 CatalogSplitter *DetectionsSplitInit (int Nsecfilt) {
    288 
    289   int i;
    290 
    291   CatalogSplitter *catalogs = NULL;
    292 
    293   ALLOCATE (catalogs, CatalogSplitter, 1);
    294 
    295   // as we see new Detectionss, we will update maxID and extend this array
    296   catalogs->maxID = 0;
    297   ALLOCATE (catalogs->index, int, catalogs->maxID + 1);
    298   for (i = 0; i <= catalogs->maxID; i++) catalogs->index[i] = -1;
    299 
    300   catalogs->Nsecfilt = Nsecfilt;
    301 
    302   catalogs->Ncatalog =  0;
    303   catalogs->NCATALOG = 16;
    304   catalogs->catalog = NULL;
    305   ALLOCATE (catalogs->catalog, Catalog, catalogs->NCATALOG);
    306 
    307   ALLOCATE (catalogs->catIDs,   int,   catalogs->NCATALOG);
    308   ALLOCATE (catalogs->NAVERAGE, off_t, catalogs->NCATALOG);
    309   ALLOCATE (catalogs->NMEASURE, off_t, catalogs->NCATALOG);
    310 
    311   for (i = 0; i < catalogs->NCATALOG; i++) {
    312     dvo_catalog_init (&catalogs->catalog[i], TRUE);
    313     catalogs->catIDs[i] = 0;
    314     catalogs->NAVERAGE[i] = 256;
    315     catalogs->NMEASURE[i] = 256;
    316     catalogs->catalog[i].Naverage = 0;
    317     catalogs->catalog[i].Nmeasure = 0;
    318     ALLOCATE (catalogs->catalog[i].averageT, AverageTiny, catalogs->NAVERAGE[i]);
    319     ALLOCATE (catalogs->catalog[i].measureT, MeasureTiny, catalogs->NMEASURE[i]);
    320     ALLOCATE (catalogs->catalog[i].secfilt,  SecFilt,     catalogs->NAVERAGE[i]*Nsecfilt);
    321   }
    322   return catalogs;
    323 }
    324 
    325 // distribute a bright catalog across separate catalogs
    326 int DetectionsSplitFree (CatalogSplitter *catalogs) {
    327 
    328   // XXX don't free the catalogs :
    329   // for (i = 0; i < NCATALOG; i++) {
    330   //   free (catalogs->catalog[i].averageT);
    331   //   free (catalogs->catalog[i].measureT);
    332   //   free (catalogs->catalog[i].secfilt);
    333   // }
    334   // free (catalogs->catalog);
    335 
    336   free (catalogs->catIDs);
    337   free (catalogs->NAVERAGE);
    338   free (catalogs->NMEASURE);
    339   free (catalogs->index);
    340   free (catalogs);
    341   return TRUE;
    342 }
    343 
    344 // distribute a bright catalog across separate catalogs
    345 int DetectionsSplit (CatalogSplitter *catalogs, Detections *bcatalog) {
    346 
    347   int i;
    348 
    349   int D_NCATALOG = 16;
    350 
    351   int Nsecfilt = catalogs->Nsecfilt;
    352 
    353   // find the max value of catID in this Detections
    354   int catIDmax = 0;
    355   for (i = 0; i < bcatalog->Naverage; i++) {
    356     catIDmax = MAX(catIDmax, bcatalog->average[i].catID);
    357   }
    358   // XXX validate the measure ID range here
    359    
    360   int maxIDold = catalogs->maxID;
    361   catalogs->maxID = MAX (maxIDold, catIDmax);
    362    
    363   // extend the index array and init
    364   REALLOCATE (catalogs->index, int, catalogs->maxID + 1);
    365   for (i = maxIDold + 1; i <= catalogs->maxID; i++) catalogs->index[i] = -1;
    366 
    367   // identify the new catID values
    368   for (i = 0; i < bcatalog->Naverage; i++) {
    369     int catID = bcatalog->average[i].catID;
    370     assert (catID > 0);
    371     assert (catID < catalogs->maxID + 1);
    372     int idx = catalogs->index[catID];
    373     if (idx != -1) continue; // already have seen this one
    374 
    375     // a new catID value
    376     int Ncat = catalogs->Ncatalog; // the next available slot
    377     catalogs->catIDs[Ncat] = catID;
    378     assert (Ncat >= 0);
    379     assert (Ncat < catalogs->NCATALOG);
    380 
    381     catalogs->catalog[Ncat].Nsecfilt = Nsecfilt;
    382     catalogs->catalog[Ncat].catID = catID;
    383 
    384     catalogs->index[catID] = Ncat;
    385     assert (catID > 0);
    386     assert (catID < catalogs->maxID + 1);
    387 
    388     catalogs->Ncatalog ++;
    389 
    390     if (catalogs->Ncatalog >= catalogs->NCATALOG) {
    391       catalogs->NCATALOG += D_NCATALOG;
    392 
    393       // fprintf (stderr, "realloc catalogs->catalog: old: %llx  ", (long long) catalogs->catalog);
    394       REALLOCATE (catalogs->catalog, Catalog, catalogs->NCATALOG);
    395       // fprintf (stderr, "new: %llx  -  %llx\n", (long long) catalogs->catalog, (long long) (catalogs->catalog + sizeof(Catalog)*catalogs->NCATALOG));
    396 
    397       // fprintf (stderr, "realloc catalogs->NAVERAGE: old: %llx  ", (long long) catalogs->NAVERAGE);
    398       REALLOCATE (catalogs->NAVERAGE, off_t,  catalogs->NCATALOG);
    399       // fprintf (stderr, "new: %llx  -  %llx\n", (long long) catalogs->NAVERAGE, (long long) (catalogs->NAVERAGE + sizeof(off_t)*catalogs->NCATALOG));
    400 
    401       // fprintf (stderr, "realloc catalogs->NMEASURE: old: %llx  ", (long long) catalogs->NMEASURE);
    402       REALLOCATE (catalogs->NMEASURE, off_t,  catalogs->NCATALOG);
    403       // fprintf (stderr, "new: %llx  -  %llx\n", (long long) catalogs->NMEASURE, (long long) (catalogs->NMEASURE + sizeof(off_t)*catalogs->NCATALOG));
    404 
    405       REALLOCATE (catalogs->catIDs,   int,   catalogs->NCATALOG);
    406 
    407       int j;
    408       for (j = catalogs->NCATALOG - D_NCATALOG; j < catalogs->NCATALOG; j++) {
    409         dvo_catalog_init (&catalogs->catalog[j], TRUE);
    410         catalogs->catIDs[j] = 0;
    411         catalogs->NAVERAGE[j] = 256;
    412         catalogs->NMEASURE[j] = 256;
    413         catalogs->catalog[j].Naverage = 0;
    414         catalogs->catalog[j].Nmeasure = 0;
    415         ALLOCATE (catalogs->catalog[j].averageT, AverageTiny, catalogs->NAVERAGE[j]);
    416         ALLOCATE (catalogs->catalog[j].measureT, MeasureTiny, catalogs->NMEASURE[j]);
    417         ALLOCATE (catalogs->catalog[j].secfilt,  SecFilt,     catalogs->NAVERAGE[j]*Nsecfilt);
    418       }
    419       D_NCATALOG = MAX(2000, 2*D_NCATALOG);
    420     }
    421   }
    422 
    423   // each bcatalog (from each host) has a different set of catID ranges
    424   // they also (probably) have contiguous ranges.  But, it is a bit tricky to be sure I
    425   // can use that info, so perhaps ignore it
    426 
    427   // assign the averages to the corresponding catalog
    428   for (i = 0; i < bcatalog->Naverage; i++) {
    429     int ID = bcatalog->average[i].catID;
    430     int Nc = catalogs->index[ID];
    431     assert (Nc > -1);
    432     assert (Nc < catalogs->NCATALOG);
    433 
    434     int Na = catalogs->catalog[Nc].Naverage;
    435     catalogs->catalog[Nc].averageT[Na] = bcatalog->average[i];
    436 
    437     // secfilt entries are grouped in blocks for each average
    438     int k;
    439     for (k = 0; k < Nsecfilt; k++) {
    440       catalogs->catalog[Nc].secfilt[Na*Nsecfilt + k] = bcatalog->secfilt[i*Nsecfilt + k];
    441     }     
    442 
    443     catalogs->catalog[Nc].Naverage ++;
    444     if (catalogs->catalog[Nc].Naverage >= catalogs->NAVERAGE[Nc]) {
    445       catalogs->NAVERAGE[Nc] += MAX(catalogs->NAVERAGE[Nc], 1024);
    446       REALLOCATE (catalogs->catalog[Nc].averageT, AverageTiny, catalogs->NAVERAGE[Nc]);
    447       REALLOCATE (catalogs->catalog[Nc].secfilt,  SecFilt,     catalogs->NAVERAGE[Nc]*Nsecfilt);
    448     }         
    449   }
    450 
    451   // assign the measures to the corresponding catalog
    452   // XXX what about averef and related links?  Do I need them?
    453   for (i = 0; i < bcatalog->Nmeasure; i++) {
    454     int ID = bcatalog->measure[i].catID;
    455     int Nc = catalogs->index[ID];
    456     assert (Nc > -1);
    457     assert (Nc < catalogs->NCATALOG);
    458     int Na = catalogs->catalog[Nc].Nmeasure;
    459     catalogs->catalog[Nc].measureT[Na] = bcatalog->measure[i];
    460     catalogs->catalog[Nc].Nmeasure ++;
    461     if (catalogs->catalog[Nc].Nmeasure >= catalogs->NMEASURE[Nc]) {
    462       catalogs->NMEASURE[Nc] += MAX(catalogs->NMEASURE[Nc], 4096);
    463       REALLOCATE (catalogs->catalog[Nc].measureT, MeasureTiny, catalogs->NMEASURE[Nc]);
    464     }         
    465   }
    466 
    467   return TRUE;
    468 }

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/initialize_dvopsps.c

    r35055 r35069  
    7676  } else usage_dvopsps();
    7777
     78  RESULT_FILE = NULL; // only used by dvopsps_client
    7879  SINGLE_CPT = NULL;
    7980  if ((N = get_argument (argc, argv, "-cpt"))) {
    … …  
    216217  } else usage_dvopsps_client();
    217218
     219  RESULT_FILE = NULL; // only used by dvopsps_client
     220  if ((N = get_argument (argc, argv, "-save"))) {
     221    remove_argument (N, &argc, argv);
     222    RESULT_FILE = strcreate (argv[N]);
     223    remove_argument (N, &argc, argv);
     224  }   
     225
    218226  SINGLE_CPT = NULL;
    219227  if ((N = get_argument (argc, argv, "-cpt"))) {

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps.c

    r35064 r35069  
    11# include "dvopsps.h"
    22# define USE_MYSQL 1
     3# define SAVE_REMOTE 1
     4
     5# define MARKTIME(MSG,...) {                    \
     6    float dtime;                                \
     7    gettimeofday (&stop, (void *) NULL);        \
     8    dtime = DTIME (stop, start);                \
     9    fprintf (stderr, MSG, __VA_ARGS__); }
     10
     11// we have two ways of writing to the mysql database:
     12// SAVE_REMOTE = 0 : each client sends the detections directly to the mysql server
     13// SAVE_REMOTE = 1 : each client saves the detections to disk and these are loaded by the main program and sent to the mysql server
    314
    415// determine the relevant catalogs, launch parallel clients if desired
    … …  
    3041  // NOTE: mysql connection happens here since each dvopsps_client makes its own connection
    3142  MYSQL  mysqlBase;
    32   MYSQL *mysqlReal = mysql_dvopsps_connect (&mysqlBase);
    33   if (!mysqlReal) {
    34     fprintf (stderr, "failed to connect to mysql\n");
    35     exit (1);
     43  MYSQL *mysqlReal = NULL;
     44  if (!SAVE_REMOTE) {
     45    mysqlReal = mysql_dvopsps_connect (&mysqlBase);
     46    if (!mysqlReal) {
     47      fprintf (stderr, "failed to connect to mysql\n");
     48      exit (1);
     49    }
    3650  }
    3751# else
    3852  MYSQL *mysqlReal = NULL;
    3953# endif
     54
     55  if (SAVE_REMOTE && HOST_ID) {
     56    init_detections ();
     57  }
    4058
    4159  // select measurements for each populated catalog
    … …  
    6785
    6886    // NOTE: this is where the real action happens
    69     if (1) {
    70         insert_detections_dvopsps_catalog (&catalog, mysqlReal);
     87    if (SAVE_REMOTE && HOST_ID) {
     88      append_detections_dvopsps_catalog (&catalog);
    7189    } else {
    72         // NOTE: this is where the real action happens
    73         save_detections_dvopsps_catalog (&catalog, "???");
     90      insert_detections_dvopsps_catalog (&catalog, mysqlReal);
    7491    }
    7592
    … …  
    7996  }
    8097
    81   if (0) {
    82     char query[256];
    83     MYSQL_RES *result;
    84 
    85     sprintf (query, "commit;");
    86     if (mysql_query (mysqlReal, query)) {
    87       fprintf (stderr, "failed to turn off autocommit\n");
    88       fprintf (stderr, "%s\n", mysql_error (mysqlReal));
    89       return FALSE;
    90     }
    91     result = mysql_store_result (mysqlReal);
    92     mysql_free_result (result);
    93   }
    94    
     98  if (SAVE_REMOTE && HOST_ID) {
     99    save_detections_dvopsps ();
     100  }
     101
    95102  return (TRUE);
    96103}     
    … …  
    113120    table->hosts[i].pathname = tmppath;
    114121
     122    if (SAVE_REMOTE) {
     123      // XXX need to uniquify the result file
     124      ALLOCATE (table->hosts[i].results, char, 1024);
     125      snprintf (table->hosts[i].results, 1024, "%s/dvopsps_detections.dat", table->hosts[i].pathname);
     126    }
     127
    115128    char command[1024];
    116129    snprintf (command, 1024, "dvopsps_client detections -hostID %d -catdir %s -hostdir %s -region %f %f %f %f",
    117130              table->hosts[i].hostID, CATDIR, table->hosts[i].pathname,
    118131              UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax);
    119 
    120     fprintf (stderr, "command: %s\n", command);
    121132
    122133    char tmpline[1024];
    … …  
    126137    snprintf (tmpline, 1024, "%s -dbname %s", command, DATABASE_NAME); strcpy (command, tmpline);
    127138
    128     if (VERBOSE) { snprintf (tmpline, 1024, "%s -v", command); strcpy (command, tmpline); }
     139    if (VERBOSE)     { snprintf (tmpline, 1024, "%s -v",       command);                          strcpy (command, tmpline); }
     140    if (SAVE_REMOTE) { snprintf (tmpline, 1024, "%s -save %s", command, table->hosts[i].results); strcpy (command, tmpline); }
     141
     142    fprintf (stderr, "command: %s\n", command);
    129143
    130144    if (PARALLEL_MANUAL) continue;
    … …  
    158172  }
    159173
    160 # if (0)
    161   for (i = 0; i < table->Nhosts; i++) {
    162     while ((detections = RemoteDetectionsLoad (table->hosts[i].results)) == NULL) {
    163       // failed to get the data from this host.  This can happen for various reasons.  Give the user a chance to try again...
    164       fprintf (stderr, "failed to read data from %s\n", table->hosts[i].hostname);
    165       fprintf (stderr, "you may run the command manually\n");
    166     }
    167     free (table->hosts[i].results);
    168     table->hosts[i].results = NULL;
     174  if (SAVE_REMOTE) {
     175    int Ndetections = 0;
     176    Detections *detections = NULL;
     177
     178    for (i = 0; i < table->Nhosts; i++) {
     179      while ((detections = DetectionsLoad (table->hosts[i].results, &Ndetections)) == NULL) {
     180        // failed to get the data from this host.  This can happen for various reasons.  Give the user a chance to try again...
     181        fprintf (stderr, "failed to read data from %s\n", table->hosts[i].hostname);
     182        fprintf (stderr, "you may run the command manually\n");
     183      }
     184      free (table->hosts[i].results);
     185      table->hosts[i].results = NULL;
    169186   
    170     RemoveDetectionsInsert (detections);
    171 
    172     free (detections);
    173   }
    174 # endif
     187      insert_detections_mysql_array (detections, Ndetections);
     188
     189      free (detections);
     190    }
     191  }
    175192
    176193  return (TRUE);
    177194}     
     195
     196int insert_detections_mysql_array (Detections *detections, int Ndetections) {
     197
     198  int i;
     199  IOBuffer buffer;
     200  buffer.Nalloc = 0;
     201
     202  struct timeval start, stop;
     203
     204  // NOTE: mysql connection happens here since each dvopsps_client makes its own connection
     205  MYSQL  mysqlBase;
     206  MYSQL *mysql = NULL;
     207  mysql = mysql_dvopsps_connect (&mysqlBase);
     208  if (!mysql) {
     209    fprintf (stderr, "failed to connect to mysql\n");
     210    exit (1);
     211  }
     212
     213  insert_detections_mysql_init (&buffer);
     214
     215  gettimeofday (&start, (void *) NULL);
     216  for (i = 0; i < Ndetections; i++) {
     217
     218    // XXX check return status
     219    insert_detections_mysql_detvalue (&buffer, &detections[i]);
     220    // if (buffer.Nbuffer > MAX_BUFFER) {
     221    if (buffer.Nbuffer > 2048) {
     222      insert_detections_mysql_commit (&buffer, mysql);
     223      buffer.Nbuffer = 0;
     224      bzero (buffer.buffer, buffer.Nalloc);
     225      insert_detections_mysql_init (&buffer);
     226    }
     227  }
     228  insert_detections_mysql_commit (&buffer, mysql);
     229  FreeIOBuffer (&buffer);
     230
     231  MARKTIME("-- inserted %d rows in %f sec\n", Ndetections, dtime);
     232  return (TRUE);
     233}
     234
     235int insert_detections_mysql_detvalue (IOBuffer *buffer, Detections *detection) {
     236
     237  PrintIOBuffer (buffer, "(%d, ", detection->imageID);    // imageID
     238  PrintIOBuffer (buffer, "%u, ", detection->ippDetectID); // ippDetectID
     239  PrintIOBuffer (buffer, "%lu, ", detection->detectID);   // detectID
     240  PrintIOBuffer (buffer, "%lu, ", detection->ippObjID);   // ippObjID
     241  PrintIOBuffer (buffer, "%lu, ", detection->objID);      // objID
     242  PrintIOBuffer (buffer, "%u, ", detection->flags);       // flags
     243  if (isnan(detection->zp))      PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ",  detection->zp);   // zp
     244  if (isnan(detection->zpErr))   PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ",  detection->zpErr);        // zpErr
     245  if (isnan(detection->airMass)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ",  detection->airMass);      //
     246  if (isnan(detection->expTime)) PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ",  detection->expTime);      // expTime
     247  if (isnan(detection->ra))      PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%lf, ", detection->ra);   // ra
     248  if (isnan(detection->dec))     PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%lf, ", detection->dec);          // dec
     249  if (isnan(detection->raErr))   PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f, ",  detection->raErr);        // estimate of raErr
     250  if (isnan(detection->decErr))  PrintIOBuffer (buffer, "NULL, "); else PrintIOBuffer (buffer, "%f),\n",  detection->decErr);     // estimate of decErr
     251  return TRUE;
     252}

  • branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c

    r35022 r35069  
    11# include "dvopsps.h"
    2 # define DEBUG 0
    3 # define SQL_INSERT 1
    4 // 1 = write data via sql insert values = (),
    5 // 0 = write data via load from file
     2# define DEBUG 1
    63
    74# define MARKTIME(MSG,...) {                    \
    … …  
    107    dtime = DTIME (stop, start);                \
    118    fprintf (stderr, MSG, __VA_ARGS__); }
     9
     10int Ndetections = 0;
     11int NDETECTIONS = 0;
     12Detections *detections = NULL;
     13
     14int init_detections () {
     15  NDETECTIONS = 1000;
     16  ALLOCATE (detections, Detections, NDETECTIONS);
     17  return TRUE;
     18}
     19
     20int append_detections_dvopsps_catalog (Catalog *catalog) {
     21
     22  off_t i, j;
     23
     24  int missingID = 0;
     25
     26  if (Ndetections + catalog[0].Nmeasure >= NDETECTIONS) {
     27    NDETECTIONS = Ndetections + catalog[0].Nmeasure + 1000;
     28    REALLOCATE (detections, Detections, NDETECTIONS);
     29  }
     30
     31  for (i = 0; i < catalog[0].Naverage; i++) {
     32
     33    if (catalog[0].average[i].extID == 0) {
     34      missingID ++;
     35    }
     36
     37    off_t m = catalog[0].average[i].measureOffset;
     38    for (j = 0; j < catalog[0].average[i].Nmeasure; j++) {
     39
     40      Average *average = &catalog->average[i];
     41      Measure *measure = &catalog->measure[m + j];
     42      PhotCode *code = GetPhotcodebyCode(measure->photcode);
     43
     44      detections[Ndetections].imageID     = measure->imageID;    // imageID
     45      detections[Ndetections].ippDetectID = measure->detID;   // ippDetectID
     46      detections[Ndetections].detectID    = measure->extID;   // detectID
     47      detections[Ndetections].ippObjID    = ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID; // ippObjID
     48      // NOTE: this is better, but the above is the current ippToPsps value
     49      // detections[Ndetections]. ((uint64_t)average->catID << 32) + (uint64_t)average->objID; // ippObjID
     50      detections[Ndetections].objID        = average->extID;   // objID
     51      detections[Ndetections].flags        = measure->dbFlags;  // flags
     52      detections[Ndetections].zp           = code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal;   // zp
     53      detections[Ndetections].zpErr        = measure->dMcal;    // zpErr
     54      detections[Ndetections].airMass      = measure->airmass;
     55      detections[Ndetections].expTime      = pow(10.0, 0.4 * measure->dt);    // expTime
     56      detections[Ndetections].ra           = average->R - measure->dR / 3600.;    // ra
     57      detections[Ndetections].dec          = average->D - measure->dR / 3600.;    // dec
     58      detections[Ndetections].raErr        = measure->dXccd * 0.01 * fabs(measure->pltscale); // estimate of raErr
     59      detections[Ndetections].decErr       = measure->dYccd * 0.01 * fabs(measure->pltscale); // estimate of decErr
     60
     61      Ndetections ++;
     62      myAssert (Ndetections <= NDETECTIONS, "programming error");
     63    }
     64  }
     65
     66  fprintf (stderr, "write %d detections, %d missing extID\n", Ndetections, missingID);
     67  return TRUE;
     68}
     69
     70// XXX write a single table for all detections or a series of tables?
     71// if I write a single table, then I have to store all data in memory
     72// if I write multiple tables, then I need to (a) iterate over the tables and (b)
     73// incur seek hits on the file
     74int save_detections_dvopsps () {
     75
     76  if (!DetectionsSave (RESULT_FILE, detections, Ndetections)) {
     77    fprintf (stderr, "failed to save detection file %s\n", RESULT_FILE);
     78  }
     79
     80  free (detections);
     81  Ndetections = 0;
     82  return (TRUE);
     83}
    1284
    1385int insert_detections_dvopsps_catalog (Catalog *catalog, MYSQL *mysql) {
    … …  
    76148  }
    77149
    78 // XXX I need to add the fileID (catID) here
    79 
    80 # if (SQL_INSERT)
    81150  PrintIOBuffer (buffer, "INSERT INTO dvoDetectionFull (imageID, ippDetectID, detectID, ippObjID, objID, flags, zp, zpErr, airMass, expTime, ra, dec_, raErr, decErr) VALUES \n");
    82 # endif
    83151
    84152  return TRUE;
    … …  
    89157  // XXX I am changing the def of ippObjID if I use the code below
    90158  PhotCode *code = GetPhotcodebyCode(measure->photcode);
    91 # if (SQL_INSERT)
    92159  PrintIOBuffer (buffer, "    (%d, %u, %lu, %lu, %lu, %u, %f, %f, %f, %f, %lf, %lf, %f, %f),\n",
    93 # else
    94                  PrintIOBuffer (buffer, "    %d %u %lu %lu %lu %u %f %f %f %f %lf %lf %f %f\n",
    95 # endif
    96                                 measure->imageID,    // imageID
    97                                 measure->detID,   // ippDetectID
    98                                 measure->extID,   // detectID
    99                                 ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID, // ippObjID
     160                 measure->imageID,    // imageID
     161                 measure->detID,   // ippDetectID
     162                 measure->extID,   // detectID
     163                 ((uint64_t)average->catID * 1000000000) + (uint64_t)average->objID, // ippObjID
    100164// NOTE: this is better, but the above is the current ippToPsps value
    101165//                              ((uint64_t)average->catID << 32) + (uint64_t)average->objID, // ippObjID
    102                                 average->extID,   // objID
    103                                 measure->dbFlags,  // flags
    104                                 code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal,   // zp
    105                                 measure->dMcal,    // zpErr
    106                                 measure->airmass,
    107                                 pow(10.0, 0.4 * measure->dt),    // expTime
    108                                 average->R - measure->dR / 3600.,    // ra
    109                                 average->D - measure->dR / 3600.,    // dec
    110                                 measure->dXccd * 0.01 * fabs(measure->pltscale), // estimate of raErr
    111                                 measure->dYccd * 0.01 * fabs(measure->pltscale) // estimate of decErr
    112 // XXX fileID??
    113 # if (SQL_INSERT)
    114                    );
    115 # else
     166                 average->extID,   // objID
     167                 measure->dbFlags,  // flags
     168                 code->C * 0.001 + code->K * (measure->airmass - 1) - measure->Mcal,   // zp
     169                 measure->dMcal,    // zpErr
     170                 measure->airmass,
     171                 pow(10.0, 0.4 * measure->dt),    // expTime
     172                 average->R - measure->dR / 3600.,    // ra
     173                 average->D - measure->dR / 3600.,    // dec
     174                 measure->dXccd * 0.01 * fabs(measure->pltscale), // estimate of raErr
     175                 measure->dYccd * 0.01 * fabs(measure->pltscale) // estimate of decErr
    116176    );
    117 # endif
    118177  return TRUE;
    119178}
    … …  
    123182  MYSQL_RES *result;
    124183
    125 # if (SQL_INSERT) 
    126184  // check that the last two chars are ,\n and replace with ;\n
    127185  if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) {
    … …  
    131189    return FALSE;
    132190  }
    133 # endif
    134191
    135192  // XXX check return status
    136193  if (mysql) {
    137 # if (SQL_INSERT)   
    138194    if (DEBUG) fprintf (stderr, "%s\n", buffer->buffer);
    139195    int status = mysql_query(mysql, buffer->buffer);
    140 # else
    141     FILE *f = fopen ("/tmp/dvopsps.dat", "w");
    142     if (!f) {
    143       fprintf (stderr, "failed to write buffer file\n");
    144       return FALSE;
    145     }
    146     int fd = fileno (f);
    147     int Nbytes = write (fd, buffer->buffer, buffer->Nbuffer);
    148     if (Nbytes != buffer->Nbuffer) {
    149       fprintf (stderr, "error writing data to buffer file\n");
    150     }
    151     char sql[256];
    152     snprintf (sql, 256, "load data infile '/tmp/dvopsps.dat' into table dvoDetectionFull;");
    153     int status = mysql_query(mysql, sql);
    154 # endif
    155196    if (status) {
    156197      fprintf (stderr, "error with insert:\n");
Note: See TracChangeset for help on using the changeset viewer.