Index: trunk/Ohana/src/dvopsps/Makefile =================================================================== --- trunk/Ohana/src/dvopsps/Makefile (revision 35098) +++ trunk/Ohana/src/dvopsps/Makefile (revision 35098) @@ -0,0 +1,52 @@ +default: dvopsps dvopsps_client +help: +@echo "make options: dvopsps default help install (dvopsps)" + +include ../../Makefile.System +HOME = $(ROOT)/src/dvopsps +BIN = $(HOME)/bin +LIB = $(HOME)/lib +SRC = $(HOME)/src +MAN = $(HOME)/doc +INC = $(HOME)/include +include ../../Makefile.Common + +# programs may add their own internal requirements here +FULL_CFLAGS = $(BASE_CFLAGS) +FULL_CPPFLAGS = $(BASE_CPPFLAGS) +FULL_LDFLAGS = -ldvo -lFITS -lohana $(BASE_LDFLAGS) + +dvopsps: $(BIN)/dvopsps.$(ARCH) +dvopsps_client: $(BIN)/dvopsps_client.$(ARCH) + +install: $(DESTBIN)/dvopsps $(DESTBIN)/dvopsps_client + +DVOPSPS = \ +$(SRC)/dvopsps.$(ARCH).o \ +$(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 \ +$(SRC)/insert_objects_dvopsps_catalog.$(ARCH).o \ +$(SRC)/insert_detections_dvopsps.$(ARCH).o \ +$(SRC)/insert_detections_dvopsps_catalog.$(ARCH).o + +# $(SRC)/insert_detections_dvopsps_catalog.$(ARCH).o + +$(DVOPSPS): $(INC)/dvopsps.h +$(BIN)/dvopsps.$(ARCH): $(DVOPSPS) + +DVOPSPS_CLIENT = \ +$(SRC)/dvopsps_client.$(ARCH).o \ +$(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 \ +$(SRC)/insert_objects_dvopsps_catalog.$(ARCH).o \ +$(SRC)/insert_detections_dvopsps.$(ARCH).o \ +$(SRC)/insert_detections_dvopsps_catalog.$(ARCH).o + +$(DVOPSPS_CLIENT): $(INC)/dvopsps.h +$(BIN)/dvopsps_client.$(ARCH): $(DVOPSPS_CLIENT) Index: trunk/Ohana/src/dvopsps/include/dvopsps.h =================================================================== --- trunk/Ohana/src/dvopsps/include/dvopsps.h (revision 35098) +++ trunk/Ohana/src/dvopsps/include/dvopsps.h (revision 35098) @@ -0,0 +1,97 @@ +# include +# include +# include +# include "mysql.h" + +# define DVO_MAX_PATH 1024 +# define MAX_BUFFER 0x080000 + +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; +} Detections; + +/* global variables set in parameter file */ +char ImageCat[DVO_MAX_PATH]; + +char *CATDIR; +int HOST_ID; +char *HOSTDIR; +int VERBOSE; + +int PARALLEL; +int PARALLEL_MANUAL; +int PARALLEL_SERIAL; + +char *SINGLE_CPT; + +char *RESULT_FILE; + +char *DATABASE_HOST; +char *DATABASE_USER; +char *DATABASE_PASS; +char *DATABASE_NAME; + +SkyRegion UserPatch; + +/***** prototypes ****/ +int main PROTO((int argc, char **argv)); + +void GetConfig PROTO((char *config, char *field, char *format, int N, void *ptr)); +void ConfigInit PROTO((int *argc, char **argv)); + +void usage_dvopsps PROTO(()); +void initialize_dvopsps PROTO((int argc, char **argv)); +int args_dvopsps PROTO((int argc, char **argv)); + +void usage_dvopsps_client PROTO(()); +void initialize_dvopsps_client PROTO((int argc, char **argv)); +int args_dvopsps_client PROTO((int argc, char **argv)); + +MYSQL *mysql_dvopsps_connect PROTO((MYSQL *mysqlBase)); + +int insert_detections_dvopsps PROTO(()); +int insert_detections_dvopsps_parallel PROTO((SkyTable *sky)); +int insert_detections_dvopsps_catalog PROTO((Catalog *catalog, MYSQL *mysql)); + +int insert_detections_mysql_commit PROTO((IOBuffer *buffer, MYSQL *mysql)); +int insert_detections_mysql_value PROTO((IOBuffer *buffer, Average *average, Measure *measure)); +int insert_detections_mysql_init PROTO((IOBuffer *buffer)); + +int insert_detections_mysql_array PROTO((MYSQL *mysql, 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((SkyList *sky)); +int insert_objects_dvopsps_catalog PROTO((Catalog *catalog, char *basename, MYSQL *mysql)); + +int insert_objects_mysql_create_tables PROTO((char *basename, MYSQL *mysql)); + +int insert_objects_mysql_commit PROTO((IOBuffer *ave_buffer, IOBuffer *sec_buffer, MYSQL *mysql)); +int insert_objects_mysql_value PROTO((IOBuffer *ave_buffer, IOBuffer *sec_buffer, Average *average, SecFilt *secfilt, int Nsecfilt)); +int insert_objects_mysql_init PROTO((IOBuffer *ave_buffer, IOBuffer *sec_buffer, char *basename)); + +int insert_skytable PROTO(()); + +int insert_skytable_mysql_commit PROTO((IOBuffer *buffer, MYSQL *mysql)); +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: trunk/Ohana/src/dvopsps/src/DetectionOps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/DetectionOps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/DetectionOps.c (revision 35098) @@ -0,0 +1,240 @@ +# 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) \ + TYPE *OUT = gfits_get_bintable_column_data (&theader, &ftable, NAME, type, &Nrow, &Ncol); \ + myAssert (!strcmp(type, #TYPE), "wrong column type"); + +Detections *DetectionsLoad(char *filename, int *Ndetections) { + + int i, Ncol; + off_t Nrow; + char type[16]; + Header header; + Header theader; + Matrix matrix; + FTable ftable; + + header.buffer = NULL; + matrix.buffer = NULL; + ftable.buffer = NULL; + theader.buffer = NULL; + Detections *detections = NULL; + + FILE *f = fopen (filename, "r"); + if (!f) { + fprintf (stderr, "ERROR: cannot open image subset file %s\n", filename); + return NULL; + } + + /* load in PHU segment (ignore) */ + if (!gfits_fread_header (f, &header)) { + if (VERBOSE) fprintf (stderr, "can't read image subset header\n"); + goto escape; + } + if (!gfits_fread_matrix (f, &matrix, &header)) { + if (VERBOSE) fprintf (stderr, "can't read image subset matrix\n"); + goto escape; + } + + ftable.header = &theader; + + // load data for this header + if (!gfits_load_header (f, &theader)) goto escape; + + // read the fits table bytes + if (!gfits_fread_ftable_data (f, &ftable, FALSE)) goto escape; + + // need to create and assign to flat-field correction + GET_COLUMN(imageID , "imageID", int); + GET_COLUMN(ippDetectID , "ippDetectID", int); + GET_COLUMN(detectID , "detectID", int64_t); + GET_COLUMN(ippObjID , "ippObjID", int64_t); + GET_COLUMN(objID , "objID", int64_t); + GET_COLUMN(flags , "flags", int); + GET_COLUMN(zp , "zp", float); + GET_COLUMN(zpErr , "zpErr", float); + GET_COLUMN(airMass , "airMass", float); + GET_COLUMN(expTime , "expTime", float); + GET_COLUMN(ra , "ra", double); // XXX signed vs unsigned? + GET_COLUMN(dec , "dec", double); + GET_COLUMN(raErr , "raErr", float); + GET_COLUMN(decErr , "decErr", float); + gfits_free_header (&theader); + gfits_free_table (&ftable); + + ALLOCATE (detections, Detections, Nrow); + for (i = 0; i < Nrow; i++) { + detections[i].imageID = imageID[i]; + detections[i].ippDetectID = ippDetectID[i]; + detections[i].detectID = detectID[i]; + detections[i].ippObjID = ippObjID[i]; + detections[i].objID = objID[i]; + detections[i].flags = flags[i]; + detections[i].zp = zp[i]; + detections[i].zpErr = zpErr[i]; + detections[i].airMass = airMass[i]; + detections[i].expTime = expTime[i]; + detections[i].ra = ra[i]; + detections[i].dec = dec[i]; + detections[i].raErr = raErr[i]; + detections[i].decErr = decErr[i]; + } + fprintf (stderr, "loaded data for %lld detections\n", (long long) Nrow); + + 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_free_header (&header); + gfits_free_matrix (&matrix); + fclose (f); + + *Ndetections = Nrow; + return detections; + + escape: + gfits_free_header (&header); + gfits_free_matrix (&matrix); + gfits_free_header (&theader); + gfits_free_table (&ftable); + if (detections) free (detections); + + fclose (f); + return NULL; +} + +// we are passed a Detections structure, write it to a FITS table (3 ext) +int DetectionsSave(char *filename, Detections *detections, int Ndetections) { + + int i; + Header header; + Header theader; + Matrix matrix; + FTable ftable; + + gfits_init_header (&header); + header.extend = TRUE; + gfits_create_header (&header); + gfits_create_matrix (&header, &matrix); + + FILE *f = fopen (filename, "w"); + if (!f) { + fprintf (stderr, "ERROR: cannot open image subset file for output %s\n", filename); + return FALSE; + } + + gfits_fwrite_header (f, &header); + gfits_fwrite_matrix (f, &matrix); + gfits_free_header (&header); + 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, 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, 0); + gfits_define_bintable_column (&theader, "K", "ippObjID", NULL, NULL, 1.0, 0); + gfits_define_bintable_column (&theader, "K", "objID", NULL, NULL, 1.0, 0); + // 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; +} + Index: trunk/Ohana/src/dvopsps/src/dvopsps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/dvopsps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/dvopsps.c (revision 35098) @@ -0,0 +1,32 @@ +# include "dvopsps.h" + +int main (int argc, char **argv) { + + int status; + + /* get configuration info, args, lockfile */ + initialize_dvopsps (argc, argv); + + status = -1; + if (!strcasecmp (argv[1], "detections")) { + status = insert_detections_dvopsps (); + } + if (!strcasecmp (argv[1], "objects")) { + status = insert_objects_dvopsps (); + } + if (!strcasecmp (argv[1], "skytable")) { + status = insert_skytable (); + } + + if (status == -1) { + fprintf (stderr, "invalid mode, should be : detections, skytable\n"); + exit (2); + } + + if (!status) exit (1); + exit (0); +} + + +/* dvopsps : extract the ids and other needed fields from dvo and insert in the mysql db + */ Index: trunk/Ohana/src/dvopsps/src/dvopsps_client.c =================================================================== --- trunk/Ohana/src/dvopsps/src/dvopsps_client.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/dvopsps_client.c (revision 35098) @@ -0,0 +1,31 @@ +# include "dvopsps.h" + +// dvopsps_client is run on a remote host and is responsible for updating the catalogs +// owned by that host. + +// dvopsps_client -hostname (hostname) -catdir (catdir) -hostdir (hostdir) -images (images) +// load the SkyTable and HostTable from (catdir) [contains the host responsibilities] +// loop over tables from SkyTable with my host ID +// the calling program tells the client which host they are (or host ID?); we do not actually have to run this on the real host + +int main (int argc, char **argv) { + + int status; + + // get configuration info, args, lockfile (set CATDIR, HOST_ID, HOSTDIR, IMAGES) + initialize_dvopsps_client (argc, argv); + + if (!strcasecmp (argv[1], "detections")) { + status = insert_detections_dvopsps (); + } + if (!strcasecmp (argv[1], "objects")) { + status = insert_objects_dvopsps (); + } + if (status == -1) { + fprintf (stderr, "invalid mode, should be : detections, skytable\n"); + exit (2); + } + + if (!status) exit (1); + exit (0); +} Index: trunk/Ohana/src/dvopsps/src/initialize_dvopsps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/initialize_dvopsps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/initialize_dvopsps.c (revision 35098) @@ -0,0 +1,330 @@ +# include "dvopsps.h" + +void usage_dvopsps () { + fprintf (stderr, "USAGE: dvopsps (mode) [dbinfo] (mode) [options]\n"); + fprintf (stderr, " mysql database info is supplied with these:\n"); + fprintf (stderr, " -dbhost : database host machine\n"); + fprintf (stderr, " -dbuser : database username\n"); + fprintf (stderr, " -dbpass : database password\n"); + fprintf (stderr, " -dbname : database name\n\n"); + fprintf (stderr, " options:\n"); + fprintf (stderr, " -v : verbose mode\n"); + fprintf (stderr, " -region Rmin Rmax Dmin Dmax\n"); + fprintf (stderr, " -parallel : run in parallel mode\n"); + fprintf (stderr, " -h : this help list\n"); + fprintf (stderr, " -help : this help list\n"); + fprintf (stderr, " --h : this help list\n"); + fprintf (stderr, " --help : this help list\n"); + fprintf (stderr, " Note that the dvo db can be specified by -D CATDIR (directory)\n"); + exit (2); +} + +void initialize_dvopsps (int argc, char **argv) { + + int N; + struct stat statbuffer; + char CatdirPhotcodeFile[256]; + + if ((N = get_argument (argc, argv, "-h"))) usage_dvopsps(); + if ((N = get_argument (argc, argv, "-help"))) usage_dvopsps(); + if ((N = get_argument (argc, argv, "--h"))) usage_dvopsps(); + if ((N = get_argument (argc, argv, "--help"))) usage_dvopsps(); + + /* are these set correctly? */ + ConfigInit (&argc, argv); + args_dvopsps (argc, argv); + + if (stat (CATDIR, &statbuffer)) { + fprintf (stderr, "error accessing dvo database directory '%s'\n", CATDIR); + exit (1); + } + + // load the photcode table + sprintf (CatdirPhotcodeFile, "%s/Photcodes.dat", CATDIR); + if (!LoadPhotcodes (CatdirPhotcodeFile, NULL, FALSE)) { + fprintf (stderr, "error loading photcode table %s\n", CatdirPhotcodeFile); + exit (1); + } +} + +int args_dvopsps (int argc, char **argv) { + + int N; + + if ((N = get_argument (argc, argv, "-dbhost"))) { + remove_argument (N, &argc, argv); + DATABASE_HOST = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps(); + + if ((N = get_argument (argc, argv, "-dbuser"))) { + remove_argument (N, &argc, argv); + DATABASE_USER = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps(); + + if ((N = get_argument (argc, argv, "-dbpass"))) { + remove_argument (N, &argc, argv); + DATABASE_PASS = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps(); + + if ((N = get_argument (argc, argv, "-dbname"))) { + remove_argument (N, &argc, argv); + DATABASE_NAME = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps(); + + RESULT_FILE = NULL; // only used by dvopsps_client + SINGLE_CPT = NULL; + if ((N = get_argument (argc, argv, "-cpt"))) { + remove_argument (N, &argc, argv); + SINGLE_CPT = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } + + /* specify portion of the sky : allow default of all sky? */ + UserPatch.Rmin = 0; + UserPatch.Rmax = 360; + UserPatch.Dmin = -90; + UserPatch.Dmax = +90; + if ((N = get_argument (argc, argv, "-region"))) { + remove_argument (N, &argc, argv); + UserPatch.Rmin = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Rmax = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Dmin = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Dmax = atof (argv[N]); + remove_argument (N, &argc, argv); + } + + VERBOSE = FALSE; + if ((N = get_argument (argc, argv, "-v"))) { + VERBOSE = TRUE; + remove_argument (N, &argc, argv); + } + + HOST_ID = 0; + PARALLEL = FALSE; + if ((N = get_argument (argc, argv, "-parallel"))) { + PARALLEL = TRUE; + remove_argument (N, &argc, argv); + } + // this is a test mode : rather than launching the remote jobs and waiting for completion, + // dvopsps will simply list the remote command and wait for the user to signal completion + PARALLEL_MANUAL = FALSE; + if ((N = get_argument (argc, argv, "-parallel-manual"))) { + PARALLEL = TRUE; // -parallel-manual implies -parallel + PARALLEL_MANUAL = TRUE; + remove_argument (N, &argc, argv); + } + // this is a test mode : rather than launching the dvopsps_client jobs remotely, they are + // run in serial via 'system' + PARALLEL_SERIAL = FALSE; + if ((N = get_argument (argc, argv, "-parallel-serial"))) { + if (PARALLEL_MANUAL) { + fprintf (stderr, "ERROR: cannot mix -parallel-manual and -parallel-serial\n"); + exit (1); + } + PARALLEL = TRUE; // -parallel-serial implies -parallel + PARALLEL_SERIAL = TRUE; + remove_argument (N, &argc, argv); + } + + if (argc != 2) usage_dvopsps(); + + return (TRUE); +} + +void usage_dvopsps_client () { + fprintf (stderr, "USAGE: dvopsps_client (mode) [dbinfo] [hostinfo] [options]\n"); + fprintf (stderr, " dvo host info:\n"); + fprintf (stderr, " -hostID (ID) : dvo host ID\n"); + fprintf (stderr, " -hostdir (dir) : local dvo catdir\n"); + fprintf (stderr, " -catdir (dir) : master dvo catdir\n\n"); + fprintf (stderr, " mysql database info is supplied with these:\n"); + fprintf (stderr, " -dbhost : database host machine\n"); + fprintf (stderr, " -dbuser : database username\n"); + fprintf (stderr, " -dbpass : database password\n"); + fprintf (stderr, " -dbname : database name\n\n"); + fprintf (stderr, " options:\n"); + fprintf (stderr, " -v : verbose mode\n"); + fprintf (stderr, " -region Rmin Rmax Dmin Dmax\n"); + fprintf (stderr, " -cpt filename\n"); + fprintf (stderr, " -parallel : run in parallel mode\n"); + fprintf (stderr, " -h : this help list\n"); + fprintf (stderr, " -help : this help list\n"); + fprintf (stderr, " --h : this help list\n"); + fprintf (stderr, " --help : this help list\n"); + fprintf (stderr, " Note that the dvo db can be specified by -D CATDIR (directory)\n"); + exit (2); +} + +void initialize_dvopsps_client (int argc, char **argv) { + + int N; + struct stat statbuffer; + char CatdirPhotcodeFile[256]; + + if ((N = get_argument (argc, argv, "-h"))) usage_dvopsps_client(); + if ((N = get_argument (argc, argv, "-help"))) usage_dvopsps_client(); + if ((N = get_argument (argc, argv, "--h"))) usage_dvopsps_client(); + if ((N = get_argument (argc, argv, "--help"))) usage_dvopsps_client(); + + SetZeroPoint (25.0); // XXX is this needed? + args_dvopsps_client (argc, argv); + + if (stat (CATDIR, &statbuffer)) { + fprintf (stderr, "error accessing dvo database directory '%s'\n", CATDIR); + exit (1); + } + + // load the photcode table : XXX needed? + sprintf (CatdirPhotcodeFile, "%s/Photcodes.dat", CATDIR); + if (!LoadPhotcodes (CatdirPhotcodeFile, NULL, FALSE)) { + fprintf (stderr, "error loading photcode table %s\n", CatdirPhotcodeFile); + exit (1); + } +} + +int args_dvopsps_client (int argc, char **argv) { + + int N; + + // by definition, the client is not parallel + PARALLEL = FALSE; + PARALLEL_MANUAL = FALSE; + PARALLEL_SERIAL = FALSE; + + if ((N = get_argument (argc, argv, "-dbhost"))) { + remove_argument (N, &argc, argv); + DATABASE_HOST = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps_client(); + + if ((N = get_argument (argc, argv, "-dbuser"))) { + remove_argument (N, &argc, argv); + DATABASE_USER = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps_client(); + + if ((N = get_argument (argc, argv, "-dbpass"))) { + remove_argument (N, &argc, argv); + DATABASE_PASS = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } else usage_dvopsps_client(); + + if ((N = get_argument (argc, argv, "-dbname"))) { + remove_argument (N, &argc, argv); + DATABASE_NAME = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } 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"))) { + remove_argument (N, &argc, argv); + SINGLE_CPT = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } + + /* specify portion of the sky : allow default of all sky? */ + UserPatch.Rmin = 0; + UserPatch.Rmax = 360; + UserPatch.Dmin = -90; + UserPatch.Dmax = +90; + if ((N = get_argument (argc, argv, "-region"))) { + remove_argument (N, &argc, argv); + UserPatch.Rmin = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Rmax = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Dmin = atof (argv[N]); + remove_argument (N, &argc, argv); + UserPatch.Dmax = atof (argv[N]); + remove_argument (N, &argc, argv); + } + + VERBOSE = FALSE; + if ((N = get_argument (argc, argv, "-v"))) { + VERBOSE = TRUE; + remove_argument (N, &argc, argv); + } + + HOST_ID = 0; + if ((N = get_argument (argc, argv, "-hostID"))) { + remove_argument (N, &argc, argv); + HOST_ID = atoi (argv[N]); + remove_argument (N, &argc, argv); + } + if (!HOST_ID) usage_dvopsps_client(); + + HOSTDIR = NULL; + if ((N = get_argument (argc, argv, "-hostdir"))) { + remove_argument (N, &argc, argv); + HOSTDIR = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } + if (!HOSTDIR) usage_dvopsps_client(); + + if ((N = get_argument (argc, argv, "-catdir"))) { + remove_argument (N, &argc, argv); + CATDIR = strcreate (argv[N]); + remove_argument (N, &argc, argv); + } + if (!CATDIR) usage_dvopsps_client(); + + if (argc != 2) usage_dvopsps_client(); + return (TRUE); +} + +void ConfigInit (int *argc, char **argv) { + + double ZERO_POINT; + char *config, *file; + + /*** load configuration info ***/ + file = SelectConfigFile (argc, argv, "ptolemy"); + config = LoadConfigFile (file); + if (config == (char *) NULL) { + fprintf (stderr, "ERROR: can't find configuration file %s\n", file); + if (file != (char *) NULL) free (file); + exit (0); + } + if (VERBOSE) fprintf (stderr, "loaded config file: %s\n", file); + + // force CATDIR to be absolute (so parallel mode will work) + char *tmpcatdir = NULL; + ALLOCATE (tmpcatdir, char, DVO_MAX_PATH); + GetConfig (config, "CATDIR", "%s", 0, tmpcatdir); + CATDIR = abspath (tmpcatdir, DVO_MAX_PATH); + free (tmpcatdir); + + sprintf (ImageCat, "%s/Images.dat", CATDIR); + + ScanConfig (config, "ZERO_PT", "%lf", 0, &ZERO_POINT); + SetZeroPoint (ZERO_POINT); + + free (config); + free (file); +} + +void GetConfig (char *config, char *field, char *format, int N, void *ptr) { + + char *status; + + status = ScanConfig (config, field, format, N, ptr); + if (status == NULL) { + fprintf (stderr, "error in config, cannot find %s\n", field); + exit (1); + } + return; +} Index: trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps.c (revision 35098) @@ -0,0 +1,251 @@ +# 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 +int insert_detections_dvopsps () { + + SkyTable *sky = NULL; + SkyList *skylist = NULL; + Catalog catalog; + off_t i; + + // load the current sky table (layout of all SkyRegions) + sky = SkyTableLoadOptimal (CATDIR, NULL, NULL, FALSE, -1, VERBOSE); + SkyTableSetFilenames (sky, CATDIR, "cpt"); + + if (PARALLEL && !HOST_ID) { + int status = insert_detections_dvopsps_parallel (sky); + return status; + } + + // determine the populated SkyRegions overlapping the requested area (default depth) + if (SINGLE_CPT) { + skylist = SkyRegionByCPT (sky, SINGLE_CPT); + } else { + skylist = SkyListByPatch (sky, -1, &UserPatch); + } + myAssert (skylist, "ooops!"); + +# if (USE_MYSQL) + // NOTE: mysql connection happens here since each dvopsps_client makes its own connection + MYSQL mysqlBase; + MYSQL *mysqlReal = NULL; + if (!HOST_ID || !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 + for (i = 0; i < skylist[0].Nregions; i++) { + + // does this host ID match the desired location for the table? + if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; + + char localFilename[1024]; + snprintf (localFilename, 1024, "%s/%s.cpt", HOSTDIR, skylist->regions[i]->name); + + // set up the basic catalog info + catalog.filename = HOST_ID ? localFilename : skylist[0].filename[i]; + catalog.catformat = DVO_FORMAT_UNDEF; // not needed since we skip empty catalogs + catalog.catmode = DVO_MODE_UNDEF; // not needed since we skip empty catalogs + catalog.catflags = LOAD_AVES | LOAD_MEAS | LOAD_SECF; + catalog.Nsecfilt = GetPhotcodeNsecfilt (); + + if (!dvo_catalog_open (&catalog, skylist[0].regions[i], VERBOSE, "w")) { + fprintf (stderr, "ERROR: failure reading catalog %s\n", catalog.filename); + exit (1); + } + if (!catalog.Naves_disk) { + if (VERBOSE) fprintf (stderr, "no data in %s, skipping\n", catalog.filename); + dvo_catalog_unlock (&catalog); + dvo_catalog_free (&catalog); + continue; + } + + // NOTE: this is where the real action happens + if (SAVE_REMOTE && HOST_ID) { + append_detections_dvopsps_catalog (&catalog); + } else { + insert_detections_dvopsps_catalog (&catalog, mysqlReal); + } + + // NOTE : unlike setastrom or relphot, this program is read-only wrt dvo + dvo_catalog_unlock (&catalog); + dvo_catalog_free (&catalog); + } + + if (SAVE_REMOTE && HOST_ID) { + save_detections_dvopsps (); + } + + return (TRUE); +} + +int insert_detections_dvopsps_parallel (SkyTable *sky) { + + // launch the dvopsps_client jobs to the parallel hosts + + // load the list of hosts + HostTable *table = HostTableLoad (CATDIR, sky->hosts); + + int i; + for (i = 0; i < table->Nhosts; i++) { + + // ensure that the paths are absolute path names + char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH); + free (table->hosts[i].pathname); + 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); + + char tmpline[1024]; + snprintf (tmpline, 1024, "%s -dbhost %s", command, DATABASE_HOST); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbuser %s", command, DATABASE_USER); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbpass %s", command, DATABASE_PASS); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbname %s", command, DATABASE_NAME); 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; + + if (PARALLEL_SERIAL) { + int status = system (command); + if (status) { + fprintf (stderr, "ERROR running dvopsps_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) { + 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 + } + } + + // wait for the remote jobs to be completed + if (PARALLEL_MANUAL) { + fprintf (stderr, "run the dvopsps_client commands above. when these are done, hit return\n"); + getchar(); + } + if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) { + int status = HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE); + if (!status) return FALSE; + } + + if (SAVE_REMOTE) { + int Ndetections = 0; + Detections *detections = NULL; + + // 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); + } + + 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; + + insert_detections_mysql_array (mysql, detections, Ndetections); + + free (detections); + } + } + + return (TRUE); +} + +int insert_detections_mysql_array (MYSQL *mysql, Detections *detections, int Ndetections) { + + int i; + IOBuffer buffer; + buffer.Nalloc = 0; + + struct timeval start, stop; + + if (Ndetections == 0) return TRUE; + + 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) { + 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: trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c =================================================================== --- trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c (revision 35098) @@ -0,0 +1,208 @@ +# include "dvopsps.h" +# define DEBUG 0 + +# define MARKTIME(MSG,...) { \ + float dtime; \ + gettimeofday (&stop, (void *) NULL); \ + 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->dD / 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) { + + off_t i, j; + int missingID = 0; + struct timeval start, stop; + + IOBuffer buffer; + buffer.Nalloc = 0; + + Average *average = catalog->average; + Measure *measure = catalog->measure; + + off_t found = 0; + + gettimeofday (&start, (void *) NULL); + insert_detections_mysql_init (&buffer); + int Ninsert = 0; + + // NOTE for testing, just do a few objects + if (!mysql) { + catalog[0].Naverage = 5; + } + for (i = 0; i < catalog[0].Naverage; i++) { + + if (average[i].extID == 0) { + missingID ++; + } + + off_t m = average[i].measureOffset; + for (j = 0; j < average[i].Nmeasure; j++) { + + // XXX check return status + insert_detections_mysql_value (&buffer, &average[i], &measure[m+j]); + + if (buffer.Nbuffer > MAX_BUFFER) { + insert_detections_mysql_commit (&buffer, mysql); + if (DEBUG) fprintf (stderr, "inserted %d rows\n", Ninsert); + Ninsert = 0; + if (0) { + FlushIOBuffer (&buffer); + } else { + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + } + insert_detections_mysql_init (&buffer); + } + + Ninsert ++; + found ++; + } + } + insert_detections_mysql_commit (&buffer, mysql); + if (VERBOSE) fprintf (stderr, "inserted %d rows\n", Ninsert); + FreeIOBuffer (&buffer); + + MARKTIME("-- inserted "OFF_T_FMT" rows in %f sec, skipped %d without IDs\n", found, dtime, missingID); + return (TRUE); +} + +int insert_detections_mysql_init (IOBuffer *buffer) { + + if (buffer->Nalloc == 0) { + InitIOBuffer (buffer, 1024); + } + + PrintIOBuffer (buffer, "INSERT INTO dvoDetectionFull (imageID, ippDetectID, detectID, ippObjID, objID, flags, zp, zpErr, airMass, expTime, ra, dec_, raErr, decErr) VALUES \n"); + + return TRUE; +} + +int insert_detections_mysql_value (IOBuffer *buffer, Average *average, Measure *measure) { + + // XXX I am changing the def of ippObjID if I use the code below + PhotCode *code = GetPhotcodebyCode(measure->photcode); + PrintIOBuffer (buffer, " (%d, %u, %lu, %lu, %lu, %u, %f, %f, %f, %f, %lf, %lf, %f, %f),\n", + 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 + ); + return TRUE; +} + +int insert_detections_mysql_commit (IOBuffer *buffer, MYSQL *mysql) { + + MYSQL_RES *result; + + // check that the last two chars are ,\n and replace with ;\n + if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) { + buffer->buffer[buffer->Nbuffer-2] = ';'; + } else { + fprintf (stderr, "invalid sql?\n"); + return FALSE; + } + + // XXX check return status + if (mysql) { + if (DEBUG) fprintf (stderr, "%s\n", buffer->buffer); + int status = mysql_query(mysql, buffer->buffer); + if (status) { + fprintf (stderr, "error with insert:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", buffer->Nbuffer); + } + result = mysql_store_result (mysql); + mysql_free_result (result); + } else { + fprintf (stderr, "%s\n", buffer->buffer); + } + + return TRUE; +} Index: trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps.c (revision 35098) @@ -0,0 +1,175 @@ +# include "dvopsps.h" +# define USE_MYSQL 1 + +// determine the relevant catalogs, launch parallel clients if desired +int insert_objects_dvopsps () { + + SkyTable *sky = NULL; + SkyList *skylist = NULL; + Catalog catalog; + off_t i; + + // load the current sky table (layout of all SkyRegions) + sky = SkyTableLoadOptimal (CATDIR, NULL, NULL, FALSE, -1, VERBOSE); + SkyTableSetFilenames (sky, CATDIR, "cpt"); + + // determine the populated SkyRegions overlapping the requested area (default depth) + if (SINGLE_CPT) { + skylist = SkyRegionByCPT (sky, SINGLE_CPT); + } else { + skylist = SkyListByPatch (sky, -1, &UserPatch); + } + myAssert (skylist, "ooops!"); + + // pass the skylist to the parallel version, and only launch on relevant hosts + if (PARALLEL && !HOST_ID) { + int status = insert_objects_dvopsps_parallel (skylist); + return status; + } + +# if (USE_MYSQL) + // 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); + } +# else + MYSQL *mysqlReal = NULL; +# endif + + // select measurements for each populated catalog + for (i = 0; i < skylist[0].Nregions; i++) { + + // does this host ID match the desired location for the table? + if (!HostTableTestHost(skylist[0].regions[i], HOST_ID)) continue; + + char localFilename[1024]; + snprintf (localFilename, 1024, "%s/%s.cpt", HOSTDIR, skylist->regions[i]->name); + + // set up the basic catalog info + catalog.filename = HOST_ID ? localFilename : skylist[0].filename[i]; + catalog.catformat = DVO_FORMAT_UNDEF; // not needed since we skip empty catalogs + catalog.catmode = DVO_MODE_UNDEF; // not needed since we skip empty catalogs + catalog.catflags = LOAD_AVES | LOAD_SECF; + catalog.Nsecfilt = GetPhotcodeNsecfilt (); + + if (!dvo_catalog_open (&catalog, skylist[0].regions[i], VERBOSE, "w")) { + fprintf (stderr, "ERROR: failure reading catalog %s\n", catalog.filename); + exit (1); + } + if (!catalog.Naves_disk) { + if (VERBOSE) fprintf (stderr, "no data in %s, skipping\n", catalog.filename); + dvo_catalog_unlock (&catalog); + dvo_catalog_free (&catalog); + continue; + } + + // NOTE: this is where the real action happens + insert_objects_dvopsps_catalog (&catalog, skylist->regions[i]->name, mysqlReal); + + // NOTE : unlike setastrom or relphot, this program is read-only wrt dvo + dvo_catalog_unlock (&catalog); + dvo_catalog_free (&catalog); + } + + 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); + } + + return (TRUE); +} + +# define DEBUG 1 + +int insert_objects_dvopsps_parallel (SkyList *sky) { + + // launch the dvopsps_client jobs to the parallel hosts + + // load the list of hosts + HostTable *table = HostTableLoad (CATDIR, sky->hosts); + + // if sky->Nregions < 0.5*Nhosts, check if the host used before launching... + + int i, j; + for (i = 0; i < table->Nhosts; i++) { + + if (sky->Nregions < table->Nhosts) { + // do any of the regions want this host? + int wantThisHost = FALSE; + for (j = 0; j < sky->Nregions; j++) { + if (HostTableTestHost (sky->regions[j], table->hosts[i].hostID)) { + wantThisHost = TRUE; + break; + } + } + if (!wantThisHost) { + // fprintf (stderr, "skip host %s\n", table->hosts[i].hostname); + continue; + } + } + + // ensure that the paths are absolute path names + char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH); + free (table->hosts[i].pathname); + table->hosts[i].pathname = tmppath; + + char command[1024]; + snprintf (command, 1024, "dvopsps_client objects -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); + + char tmpline[1024]; + snprintf (tmpline, 1024, "%s -dbhost %s", command, DATABASE_HOST); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbuser %s", command, DATABASE_USER); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbpass %s", command, DATABASE_PASS); strcpy (command, tmpline); + snprintf (tmpline, 1024, "%s -dbname %s", command, DATABASE_NAME); strcpy (command, tmpline); + + if (VERBOSE) { snprintf (tmpline, 1024, "%s -v", command); strcpy (command, tmpline); } + if (SINGLE_CPT) { snprintf (tmpline, 1024, "%s -cpt %s", command, SINGLE_CPT); strcpy (command, tmpline); } + + fprintf (stderr, "command: %s\n", command); + + if (PARALLEL_MANUAL) continue; + + if (PARALLEL_SERIAL) { + int status = system (command); + if (status) { + fprintf (stderr, "ERROR running dvopsps_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 + } + } + + // wait for the remote jobs to be completed + if (PARALLEL_MANUAL) { + fprintf (stderr, "run the dvopsps_client commands above. when these are done, hit return\n"); + getchar(); + } + if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) { + int status = HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE); + if (!status) return FALSE; + } + + return (TRUE); +} Index: trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps_catalog.c =================================================================== --- trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps_catalog.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/insert_objects_dvopsps_catalog.c (revision 35098) @@ -0,0 +1,366 @@ +# include "dvopsps.h" +# define DEBUG 0 + +# define MARKTIME(MSG,...) { \ + float dtime; \ + gettimeofday (&stop, (void *) NULL); \ + dtime = DTIME (stop, start); \ + fprintf (stderr, MSG, __VA_ARGS__); } + +int insert_objects_dvopsps_catalog (Catalog *catalog, char *basename, MYSQL *mysql) { + + off_t i; + int missingID = 0; + struct timeval start, stop; + + IOBuffer ave_buffer, sec_buffer; + ave_buffer.Nalloc = 0; + sec_buffer.Nalloc = 0; + + Average *average = catalog->average; + SecFilt *secfilt = catalog->secfilt; + int Nsecfilt = catalog->Nsecfilt; + + off_t found = 0; + + char *cleanname = strcreate (basename); + for (i = 0; i < strlen(cleanname); i++) { + if (cleanname[i] == '.') cleanname[i] = '_'; + if (cleanname[i] == '/') cleanname[i] = '_'; + } + insert_objects_mysql_create_tables (cleanname, mysql); + + gettimeofday (&start, (void *) NULL); + insert_objects_mysql_init (&ave_buffer, &sec_buffer, cleanname); + int Ninsert = 0; + + // NOTE for testing, just do a few objects + if (!mysql) { + catalog[0].Naverage = 5; + } + for (i = 0; i < catalog[0].Naverage; i++) { + + if (average[i].extID == 0) { + missingID ++; + } + + // XXX check return status + insert_objects_mysql_value (&ave_buffer, &sec_buffer, &average[i], &secfilt[i*Nsecfilt], Nsecfilt); + + // Average is a bigger table, but Nsecfilt*secfilt might be bigger, so check both + if ((ave_buffer.Nbuffer > MAX_BUFFER) || (sec_buffer.Nbuffer > MAX_BUFFER)) { + insert_objects_mysql_commit (&ave_buffer, &sec_buffer, mysql); + if (DEBUG) fprintf (stderr, "inserted %d rows\n", Ninsert); + Ninsert = 0; + if (0) { + FlushIOBuffer (&ave_buffer); + FlushIOBuffer (&sec_buffer); + } else { + ave_buffer.Nbuffer = 0; + bzero (ave_buffer.buffer, ave_buffer.Nalloc); + sec_buffer.Nbuffer = 0; + bzero (sec_buffer.buffer, sec_buffer.Nalloc); + } + insert_objects_mysql_init (&ave_buffer, &sec_buffer, cleanname); + } + Ninsert ++; + found ++; + } + insert_objects_mysql_commit (&ave_buffer, &sec_buffer, mysql); + if (VERBOSE) fprintf (stderr, "inserted "OFF_T_FMT" average, "OFF_T_FMT" secfilt\n", catalog[0].Naverage, Nsecfilt*catalog[0].Naverage); + FreeIOBuffer (&ave_buffer); + FreeIOBuffer (&sec_buffer); + free (cleanname); + + MARKTIME("-- inserted "OFF_T_FMT" objects in %f sec, skipped %d without IDs\n", found, dtime, missingID); + return (TRUE); +} + +int insert_objects_mysql_create_tables (char *basename, MYSQL *mysql) { + + int status; + IOBuffer buffer; + InitIOBuffer (&buffer, 1024); + // MYSQL_RES *result; + + PrintIOBuffer (&buffer, "DROP TABLE %s_cpt\n", basename); + status = mysql_query(mysql, buffer.buffer); + if (status) { + fprintf (stderr, "failed to drop table:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + } + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + + PrintIOBuffer (&buffer, "CREATE TABLE %s_cpt (" + "RA FLOAT, " + "DEC_ FLOAT, " + "RA_ERR REAL, " + "DEC_ERR REAL, " + "U_RA REAL, " + "U_DEC REAL, " + "V_RA_ERR REAL, " + "V_DEC_ERR REAL, " + "PAR REAL, " + "PAR_ERR REAL, " + "CHISQ_POS REAL, " + "CHISQ_PM REAL, " + "CHISQ_PAP REAL, " + "MEAN_EPOCH INT, " + "TIME_RANGE INT, " + "PSF_QF REAL, " + "PSF_QF_PERF REAL, " + "STARGAL_SEP REAL, " + "NUMBER_POS SMALLINT, " + "NMEASURE SMALLINT, " + "NMISSING SMALLINT, " + "NEXTEND SMALLINT, " + "OFF_MEASURE INT, " + "OFF_MISSING INT, " + "OFF_EXTEND INT, " + "FLAGS INT, " + "PHOTFLAGS_U INT, " + "PHOTFLAGS_L INT, " + "OBJ_ID INT, " + "CAT_ID INT, " + "EXT_ID BIGINT" + ")\n", basename); + + if (DEBUG) fprintf (stderr, "%s\n", buffer.buffer); + status = mysql_query(mysql, buffer.buffer); + if (status) { + fprintf (stderr, "failed to create table:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", buffer.Nbuffer); + } + // result = mysql_store_result (mysql); + // mysql_free_result (result); + + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + + PrintIOBuffer (&buffer, "DROP TABLE %s_cps\n", basename); + status = mysql_query(mysql, buffer.buffer); + if (status) { + fprintf (stderr, "failed to drop table:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + } + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + + PrintIOBuffer (&buffer, "CREATE TABLE %s_cps (" + "MAG REAL, " + "MAG_AP REAL, " + "MAG_KRON REAL, " + "MAG_KRON_ERR REAL, " + "MAG_ERR REAL, " + "MAG_CHI REAL, " + "FLUX_PSF REAL, " + "FLUX_PSF_ERR REAL, " + "FLUX_KRON REAL, " + "FLUX_KRON_ERR REAL, " + "FLAGS INT, " + "NCODE SMALLINT, " + "NUSED SMALLINT, " + "MAG_20 SMALLINT, " + "MAG_80 SMALLINT, " + "UBERCAL_DIST SMALLINT, " + "MAG_STDEV SMALLINT, " + "STACK_DETECT_ID BIGINT" + ")\n", basename); + + if (DEBUG) fprintf (stderr, "%s\n", buffer.buffer); + status = mysql_query(mysql, buffer.buffer); + if (status) { + fprintf (stderr, "failed to create table:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", buffer.Nbuffer); + } + // result = mysql_store_result (mysql); + // mysql_free_result (result); + + FreeIOBuffer (&buffer); + return TRUE; +} + +int insert_objects_mysql_init (IOBuffer *ave_buffer, IOBuffer *sec_buffer, char *basename) { + + if (ave_buffer->Nalloc == 0) { + InitIOBuffer (ave_buffer, 1024); + } + + PrintIOBuffer (ave_buffer, "INSERT INTO %s_cpt (" + "RA, " + "DEC_, " + "RA_ERR, " + "DEC_ERR, " + "U_RA, " + "U_DEC, " + "V_RA_ERR, " + "V_DEC_ERR, " + "PAR, " + "PAR_ERR, " + "CHISQ_POS, " + "CHISQ_PM, " + "CHISQ_PAP, " + "MEAN_EPOCH, " + "TIME_RANGE, " + "PSF_QF, " + "PSF_QF_PERF, " + "STARGAL_SEP, " + "NUMBER_POS, " + "NMEASURE, " + "NMISSING, " + "NEXTEND, " + "OFF_MEASURE, " + "OFF_MISSING, " + "OFF_EXTEND, " + "FLAGS, " + "PHOTFLAGS_U, " + "PHOTFLAGS_L, " + "OBJ_ID, " + "CAT_ID, " + "EXT_ID" + ") VALUES \n", basename); + + if (sec_buffer->Nalloc == 0) { + InitIOBuffer (sec_buffer, 1024); + } + + PrintIOBuffer (sec_buffer, "INSERT INTO %s_cps (" + "MAG, " + "MAG_AP, " + "MAG_KRON, " + "MAG_KRON_ERR, " + "MAG_ERR, " + "MAG_CHI, " + "FLUX_PSF, " + "FLUX_PSF_ERR, " + "FLUX_KRON, " + "FLUX_KRON_ERR, " + "FLAGS, " + "NCODE, " + "NUSED, " + "MAG_20, " + "MAG_80, " + "UBERCAL_DIST, " + "MAG_STDEV, " + "STACK_DETECT_ID " + ") VALUES \n", basename); + return TRUE; +} + +int insert_objects_mysql_value (IOBuffer *ave_buffer, IOBuffer *sec_buffer, Average *average, SecFilt *secfilt, int Nsecfilt) { + + int i; + +// XXX this bit could/should be autocoded... + PrintIOBuffer (ave_buffer, " ("); + if (isnan(average->R)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%lf, ", average->R); + if (isnan(average->D)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%lf, ", average->D); + if (isnan(average->dR)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->dR); + if (isnan(average->dD)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->dD); + if (isnan(average->uR)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->uR); + if (isnan(average->uD)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->uD); + if (isnan(average->duR)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->duR); + if (isnan(average->duD)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->duD); + if (isnan(average->P)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->P); + if (isnan(average->dP)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->dP); + if (isnan(average->ChiSqAve)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->ChiSqAve); + if (isnan(average->ChiSqPM)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->ChiSqPM); + if (isnan(average->ChiSqPar)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->ChiSqPar); + PrintIOBuffer (ave_buffer, "%d, ", average->Tmean); + PrintIOBuffer (ave_buffer, "%d, ", average->Trange); + if (isnan(average->psfQF)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->psfQF); + if (isnan(average->psfQFperf)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->psfQFperf); + if (isnan(average->stargal)) PrintIOBuffer (ave_buffer, "NULL, "); else PrintIOBuffer (ave_buffer, "%f, ", average->stargal); + PrintIOBuffer (ave_buffer, "%hu, ", average->Npos); + PrintIOBuffer (ave_buffer, "%hu, ", average->Nmeasure); + PrintIOBuffer (ave_buffer, "%hu, ", average->Nmissing); + PrintIOBuffer (ave_buffer, "%hu, ", average->Nextend); + PrintIOBuffer (ave_buffer, "%d, ", average->measureOffset); + PrintIOBuffer (ave_buffer, "%d, ", average->missingOffset); + PrintIOBuffer (ave_buffer, "%d, ", average->extendOffset); + PrintIOBuffer (ave_buffer, "%u, ", average->flags); + PrintIOBuffer (ave_buffer, "%u, ", average->photFlagsUpper); + PrintIOBuffer (ave_buffer, "%u, ", average->photFlagsLower); + PrintIOBuffer (ave_buffer, "%u, ", average->objID); + PrintIOBuffer (ave_buffer, "%u, ", average->catID); + PrintIOBuffer (ave_buffer, "%lu", average->extID); + PrintIOBuffer (ave_buffer, "),\n"); + +// XXX this bit could/should be autocoded... + for (i = 0; i < Nsecfilt; i++) { + PrintIOBuffer (sec_buffer, " ("); + if (isnan(secfilt->M)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->M); + if (isnan(secfilt->Map)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->Map); + if (isnan(secfilt->Mkron)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->Mkron); + if (isnan(secfilt->dMkron)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->dMkron); + if (isnan(secfilt->dM)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->dM); + if (isnan(secfilt->Xm)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->Xm); + if (isnan(secfilt->FluxPSF)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->FluxPSF); + if (isnan(secfilt->dFluxPSF)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->dFluxPSF); + if (isnan(secfilt->FluxKron)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->FluxKron); + if (isnan(secfilt->dFluxKron)) PrintIOBuffer (sec_buffer, "NULL, "); else PrintIOBuffer (sec_buffer, "%f, ", secfilt->dFluxKron); + PrintIOBuffer (sec_buffer, "%u, ", secfilt->flags); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->Ncode); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->Nused); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->M_20); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->M_80); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->ubercalDist); + PrintIOBuffer (sec_buffer, "%hd, ", secfilt->Mstdev); + PrintIOBuffer (sec_buffer, "%lu ", secfilt->stackDetectID); + PrintIOBuffer (sec_buffer, "),\n"); + secfilt ++; + } + return TRUE; +} + +int insert_objects_mysql_commit (IOBuffer *ave_buffer, IOBuffer *sec_buffer, MYSQL *mysql) { + + MYSQL_RES *result; + + // check that the last two chars are ,\n and replace with ;\n + if (!strcmp(&ave_buffer->buffer[ave_buffer->Nbuffer-2], ",\n")) { + ave_buffer->buffer[ave_buffer->Nbuffer-2] = ';'; + } else { + fprintf (stderr, "invalid sql?\n"); + return FALSE; + } + + // check that the last two chars are ,\n and replace with ;\n + if (!strcmp(&sec_buffer->buffer[sec_buffer->Nbuffer-2], ",\n")) { + sec_buffer->buffer[sec_buffer->Nbuffer-2] = ';'; + } else { + fprintf (stderr, "invalid sql?\n"); + return FALSE; + } + + // XXX check return status + if (mysql) { + int status; + if (DEBUG) fprintf (stderr, "%s\n", ave_buffer->buffer); + status = mysql_query(mysql, ave_buffer->buffer); + if (status) { + fprintf (stderr, "error with insert:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", ave_buffer->Nbuffer); + } + result = mysql_store_result (mysql); + if (result) mysql_free_result (result); + + if (DEBUG) fprintf (stderr, "%s\n", sec_buffer->buffer); + status = mysql_query(mysql, sec_buffer->buffer); + if (status) { + fprintf (stderr, "error with insert:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", sec_buffer->Nbuffer); + } + result = mysql_store_result (mysql); + if (result) mysql_free_result (result); + } else { + fprintf (stderr, "%s\n", sec_buffer->buffer); + } + + return TRUE; +} Index: trunk/Ohana/src/dvopsps/src/insert_skytable.c =================================================================== --- trunk/Ohana/src/dvopsps/src/insert_skytable.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/insert_skytable.c (revision 35098) @@ -0,0 +1,118 @@ +# include "dvopsps.h" +# define DEBUG 1 +# define USE_MYSQL 1 + +# define MARKTIME(MSG,...) { \ + float dtime; \ + gettimeofday (&stop, (void *) NULL); \ + dtime = DTIME (stop, start); \ + fprintf (stderr, MSG, __VA_ARGS__); } + +// determine the relevant catalogs, launch parallel clients if desired +int insert_skytable () { + + off_t i; + struct timeval start, stop; + SkyTable *sky = NULL; + + IOBuffer buffer; + buffer.Nalloc = 0; + + // load the current sky table (layout of all SkyRegions) + sky = SkyTableLoadOptimal (CATDIR, NULL, NULL, FALSE, -1, VERBOSE); + +# if (USE_MYSQL) + // 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); + } +# else + MYSQL *mysqlReal = NULL; +# endif + + gettimeofday (&start, (void *) NULL); + insert_skytable_mysql_init (&buffer); + int Ninsert = 0; + + for (i = 0; i < sky->Nregions; i++) { + insert_skytable_mysql_value (&buffer, &sky->regions[i]); + + // if (buffer.Nbuffer > 1024) { + if (buffer.Nbuffer > MAX_BUFFER) { + insert_skytable_mysql_commit (&buffer, mysqlReal); + if (DEBUG) fprintf (stderr, "inserted %d rows\n", Ninsert); + Ninsert = 0; + if (0) { + FlushIOBuffer (&buffer); + } else { + buffer.Nbuffer = 0; + bzero (buffer.buffer, buffer.Nalloc); + } + insert_skytable_mysql_init (&buffer); + } + Ninsert ++; + } + + insert_skytable_mysql_commit (&buffer, mysqlReal); + FreeIOBuffer (&buffer); + + MARKTIME("-- inserted %d rows in %f sec\n", Ninsert, dtime); + return (TRUE); +} + +int insert_skytable_mysql_init (IOBuffer *buffer) { + + if (buffer->Nalloc == 0) { + InitIOBuffer (buffer, 1024); + } + + PrintIOBuffer (buffer, "INSERT INTO dvoSkyTable (R_MIN, R_MAX, D_MIN, D_MAX, INDEX_, NAME) VALUES \n"); + + return TRUE; +} + +int insert_skytable_mysql_value (IOBuffer *buffer, SkyRegion *region) { + + PrintIOBuffer (buffer, " (%f, %f, %f, %f, %d, '%s'),\n", + region->Rmin, // Rmin + region->Rmax, // Rmax + region->Dmin, // Dmin + region->Dmax, // Dmax + region->index, // index + region->name // name + ); + return TRUE; +} + +int insert_skytable_mysql_commit (IOBuffer *buffer, MYSQL *mysql) { + + MYSQL_RES *result; + + // check that the last two chars are ,\n and replace with ;\n + if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) { + buffer->buffer[buffer->Nbuffer-2] = ';'; + } else { + fprintf (stderr, "invalid sql?\n"); + return FALSE; + } + + // XXX check return status + if (mysql) { + // if (DEBUG) fprintf (stderr, "%s\n", buffer->buffer); + int status = mysql_query(mysql, buffer->buffer); + if (status) { + fprintf (stderr, "error with insert:\n"); + fprintf (stderr, "%s\n", mysql_error(mysql)); + fprintf (stderr, "Nbuffer: %d\n", buffer->Nbuffer); + } + result = mysql_store_result (mysql); + mysql_free_result (result); + } else { + fprintf (stderr, "%s\n", buffer->buffer); + } + + return TRUE; +} Index: trunk/Ohana/src/dvopsps/src/mysql_dvopsps.c =================================================================== --- trunk/Ohana/src/dvopsps/src/mysql_dvopsps.c (revision 35098) +++ trunk/Ohana/src/dvopsps/src/mysql_dvopsps.c (revision 35098) @@ -0,0 +1,80 @@ +# include "dvopsps.h" + +int dump_result (MYSQL *connection) { + + MYSQL_RES *result = mysql_store_result (connection); + if (!result) return FALSE; + int Nrows = mysql_num_rows(result); + int Ncols = mysql_num_fields(result); + + int i, j; + for (j = 0; j < Nrows; j++) { + MYSQL_ROW row = mysql_fetch_row(result); + for (i = 0; i < Ncols; i++) { + fprintf (stderr, "%s ", row[i]); + } + fprintf (stderr, "\n"); + } + mysql_free_result (result); + return (TRUE); +} + +// DATABASE_* are supplied on the command line and are global variables in dvopsps.h +MYSQL *mysql_dvopsps_connect (MYSQL *mysqlBase) { + + char query[256]; + + mysql_init (mysqlBase); + MYSQL *connection = mysql_real_connect (mysqlBase, DATABASE_HOST, DATABASE_USER, DATABASE_PASS, DATABASE_NAME, 0, 0, 0); + + if (connection == NULL) { + fprintf (stderr, "failed to connect to database\n"); + fprintf (stderr, "%s\n", mysql_error (mysqlBase)); + return NULL; + } + + sprintf (query, "set @@interactive_timeout = 30000;"); + if (mysql_query (connection, query)) { + fprintf (stderr, "failed to set interactive timout\n"); + fprintf (stderr, "%s\n", mysql_error (connection)); + return NULL; + } + dump_result (connection); + + sprintf (query, "set @@wait_timeout = 30000;"); + if (mysql_query (connection, query)) { + fprintf (stderr, "failed to set wait timout\n"); + fprintf (stderr, "%s\n", mysql_error (connection)); + return NULL; + } + dump_result (connection); + + // sprintf (query, "set max_allowed_packet = %d;", 2*MAX_BUFFER); + // if (mysql_query (connection, query)) { + // fprintf (stderr, "failed to set max_allowed_packet\n"); + // fprintf (stderr, "%s\n", mysql_error (connection)); + // return NULL; + // } + // dump_result (connection); + + sprintf (query, "show variables like 'max_allowed_packet';"); + if (mysql_query (connection, query)) { + fprintf (stderr, "failed to set max_allowed_packet\n"); + fprintf (stderr, "%s\n", mysql_error (connection)); + return NULL; + } + dump_result (connection); + + if (0) { + sprintf (query, "set autocommit=0;"); + if (mysql_query (connection, query)) { + fprintf (stderr, "failed to turn off autocommit\n"); + fprintf (stderr, "%s\n", mysql_error (connection)); + return NULL; + } + dump_result (connection); + } + + return connection; +} + Index: trunk/Ohana/src/dvopsps/test/dbadmin.sh =================================================================== --- trunk/Ohana/src/dvopsps/test/dbadmin.sh (revision 35098) +++ trunk/Ohana/src/dvopsps/test/dbadmin.sh (revision 35098) @@ -0,0 +1,97 @@ +#!/bin/csh -f + +## this script creates a database for testing dvopsps + +if ($#argv < 3) goto usage; + +if ("$1" == "user") then + if ($#argv != 4) goto usage; + set dbhost = $2 + set dbuser = $3 + set dbpass = $4 + + mysql -h $dbhost -u root -p < /dev/null + USE $dbname; + describe dvoDetection; +EOF + + if ($status) then + echo "database does not contain dvoDetections, not deleting" + exit 1; + endif + + echo "database is valid, deleting" + mysql -h $dbhost -u $dbuser $dbpass <