Changeset 33286

Timestamp:

Feb 16, 2012, 2:42:17 PM (14 years ago)

Author:

bills

Message:

build response file when all of the jobs for a request have finished

File:

• trunk/pstamp/scripts/dquery_finish.pl (modified) (6 diffs)

trunk/pstamp/scripts/dquery_finish.pl

@@ -13 +13 @@
 use File::Temp qw( tempfile );
 use File::Copy;
+use Astro::FITS::CFITSIO qw( :constants );
+Astro::FITS::CFITSIO::PerlyUnpacking(1);
 
 use PS::IPP::Metadata::Config;
@@ -69 +71 @@
     exit 0;
 }
-if (0) {
-my $prod_dir = "$outputDataStoreRoot/$product";
-if (! -e $prod_dir ) {
-    # something must have gone wrong at the parse stage
-    update_request($req_id, $PS_EXIT_SYS_ERROR, $verbose);
-    die "product directory does not exist $prod_dir";
-}
-my $outdir = "$prod_dir/$req_name";
-}
+
 if (! -e $outdir ) {
     # something must have gone wrong at the parse stage
@@ -97 +91 @@
 {
     my $command = "$pstamptool -listjob -req_id $req_id";
+    $command .= " -jobType detect_query";   # temporary
     $command .= " -dbname $dbname" if $dbname;
     $command .= " -dbserver $dbserver" if $dbserver;
@@ -117 +112 @@
 }
 
-# XXX: have the jobs produce the reglist as with postage stamp requests
-my ($REGLIST, $reg_list) = tempfile("$outdir/reqlist.XXXX", UNLINK => !$save_temps);
-
+my ($REGLIST, $reg_list) = tempfile("$outdir/reglist.XXXX", UNLINK => !$save_temps);
+my $response_file = "response.$req_id.fits";
+my $response_path = "$outdir/$response_file";
+
+print $REGLIST "$response_file|||table|\n";
+
+my @columns = qw(ROWNUM PROC_ERROR NPIX QFACTOR FLUX FLUX_SIG FPA_ID STAGE STAGE_ID COMPONENT CMFFILE);
+my %colData;
+foreach my $col (@columns) {
+    $colData{$col} = [];
+}
+    
+my $i = 0;
 foreach my $job (@jobs) {
-    next if $job->{parent_id};
     my $job_id = $job->{job_id};
-    my $response_file = "response.${job_id}.fits";
-    my $response_path = "$outdir/$response_file";
-
-    if (-e $response_path) {
-        # the job generated a response file put it into the Data Store
-        print $REGLIST "$response_file|||table|\n";
-
-        # do the same if we have an error file.  Should the parse data be uploaded as well?
-        my $err_file = "parse_error.txt";
-        if (-e "$outdir/$err_file") {
-            print $REGLIST "$err_file|||text|\n";
-        }
-    } else  {
-        print STDERR "detect_query response file for job $job_id not found\n" if $verbose;
-        $request_fault = $PS_EXIT_UNKNOWN_ERROR;
-    }
+
+    my $outputBase = $job->{outputBase};
+    my $results = "${outputBase}results.txt";
+
+    read_results($results, \%colData);
+
+    my $cmf = $colData{CMFFILE}->[$i];
+    print $REGLIST "$cmf|||table|\n";
+
+    $i++;
 }
 close $REGLIST;
 
+write_response_file($response_path, $req_name, \%colData);
+
 if (-s $reg_list) {
-    my $command = "$dsreg --add $req_name --product $product --list $reg_list";
+    my $command = "echo Skipping $dsreg --add $req_name --product $product --list $reg_list";
     $command .= " --copy --datapath $outdir";
     $command .= " --type MOPS_DETECTABILITY_RESPONSE";
@@ -162 +162 @@
 exit 0;
 
+sub read_results {
+    my $results_file = shift;
+    my $results = shift;
+
+    open IN, "<$results_file" or my_die("failed to open $results_file", $PS_EXIT_UNKNOWN_ERROR);
+
+    foreach my $line (<IN>) {
+        # skip header
+        next if $line =~ /ROWNUM/;
+        chomp $line;
+        my @words = split " ", $line;
+        my ($rownum, $proc_error, $npix, $qfactor, $flux, $flux_sig, $fpa_id, $stage, $stage_id, $component, $cmf) = @words;
+        push @{$results->{ROWNUM}}, $rownum;
+        push @{$results->{PROC_ERROR}}, $proc_error;
+        push @{$results->{NPIX}}, $npix;
+        push @{$results->{QFACTOR}}, $qfactor;
+        push @{$results->{FLUX}}, $flux;
+        push @{$results->{FLUX_SIG}}, $flux_sig;
+        push @{$results->{FPA_ID}}, $fpa_id;
+        push @{$results->{STAGE}}, $stage;
+        push @{$results->{STAGE_ID}}, $stage_id;
+        push @{$results->{COMPONENT}}, $component;
+        push @{$results->{CMFFILE}}, $cmf;
+    }
+    # XXX: make sure that the size of each of these arrays is the same
+}
+
+sub write_response_file {
+    my $output = shift;
+    my $query_id = shift;
+    my $results = shift;
+
+    my $columns;
+    my $headers;
+
+    my $EXTVER_IS_1 = 0 ; # (scalar(keys(%query)) == 1);
+#    print "EXTVER: $EXTVER_IS_1\n";
+    my ($FPA_ID,$MJD_OBS,$filter,$obscode,$status);
+    if ($EXTVER_IS_1 == 1) {
+        # Specification of columns to write
+        $columns = [
+            # matching rownum from detectability original request
+            { name => 'ROWNUM',   type => '20A', writetype => TSTRING }, 
+            # any errors that occurred during processing
+            { name => 'ERROR_CODE',   type => 'V', writetype => TULONG }, 
+            # number of pixels used in hypothetical PSF for the query detection
+            { name => 'DETECT_N', type => 'V',   writetype => TULONG },
+            # detectibility, indicating the fraction of PSF pixels detetable by IPP
+            { name => 'DETECT_F', type => 'D',   writetype => TDOUBLE },
+            # flux of the target source
+            { name => 'TARGET_FLUX', type => 'D', writetype => TDOUBLE },
+            # error in the flux of the target source
+            { name => 'TARGET_FLUX_SIG', type => 'D', writetype => TDOUBLE },
+            ];
+        
+        # Header translation table
+        $headers = {
+            'QUERY_ID' => { name => 'QUERY_ID', type => TSTRING, 
+                            comment => 'MOPS Query ID for this batch query' },
+            'FPA_ID' => { name => 'FPA_ID',   type => TSTRING, 
+                          comment => 'original FPA_ID used at ingest' },
+#           'MJD-OBS' => { name => 'FPA_ID',   type => TSTRING, 
+#                         comment => 'original FPA_ID used at ingest' },
+#           'FILTER' => { name => 'FPA_ID',   type => TSTRING, 
+#                         comment => 'original FPA_ID used at ingest' },
+#           'OBSCODE' => { name => 'FPA_ID',   type => TSTRING, 
+#                         comment => 'original FPA_ID used at ingest' },
+        };
+    }
+    else {
+        # Specification of columns to write
+        $columns = [
+            # matching rownum from detectability original request
+            { name => 'ROWNUM',   type => '20A', writetype => TSTRING}, 
+            # any errors that occurred during processing
+            { name => 'ERROR_CODE',   type => 'V', writetype => TULONG }, 
+            # number of pixels used in hypothetical PSF for the query detection
+            { name => 'DETECT_N', type => 'V',   writetype => TULONG },
+            # detectibility, indicating the fraction of PSF pixels detetable by IPP
+            { name => 'DETECT_F', type => 'D',   writetype => TDOUBLE },
+            # flux of the target source
+            { name => 'TARGET_FLUX', type => 'D', writetype => TDOUBLE },
+            # error in the flux of the target source
+            { name => 'TARGET_FLUX_SIG', type => 'D', writetype => TDOUBLE },
+            # The FPA That would be in the header if it were to be there.
+            { name => 'FPA_ID',   type => '20A', writetype => TSTRING },
+            ];
+        
+        # Header translation table
+        $headers = {
+            'QUERY_ID' => { name => 'QUERY_ID', type => TSTRING, 
+                            comment => 'MOPS Query ID for this batch query' },
+        };
+    }   
+
+    # Parse the list of columns
+    my @colNames;                   # Names of columns
+    my @colTypes;                   # Types of columns
+    my %colData;                    # Data for each column
+    my @colWriteType;                 # type to use to write
+    foreach my $colSpec ( @$columns) {
+        push @colNames, $colSpec->{name};
+        push @colTypes, $colSpec->{type};
+        push @colWriteType, $colSpec->{writetype};
+        $colData{$colSpec->{name}} = [];
+    }
+
+    my $inHeader = { };
+
+    # Hack to force the data to match the detect_response_create formats
+
+    $inHeader->{QUERY_ID}->{value} = $query_id;
+    if ($EXTVER_IS_1 == 1) {
+        my $fpa_id = $results->{FPA_ID}->[0];
+        $inHeader->{FPA_ID}->{value} = $fpa_id;
+    }
+    
+    # Fill the table columns with the data, making sure the flux is defined
+
+    @{$colData{'ROWNUM'}} = @{ $results->{ROWNUM} };
+    @{$colData{'ERROR_CODE'}} = @{ $results->{PROC_ERROR} };
+    @{$colData{'DETECT_N'}} = @{ $results->{NPIX} };
+    @{$colData{'DETECT_F'}} = @{ $results->{QFACTOR} };
+    @{$colData{'TARGET_FLUX'}} = @{ $results->{FLUX} };
+    @{$colData{'TARGET_FLUX_SIG'}} = @{ $results->{FLUX_SIG} };
+    @{$colData{'FPA_ID'}} = @{ $results->{FPA_ID} };
+
+    my $numRows = 0;
+    # Back to detect_response_create:
+    $status = 0;
+#    print "$output\n";
+    if (-e $output) {
+        unlink $output or die "failed to unlink existing response file $output\n";
+    }
+    my $outFits = Astro::FITS::CFITSIO::create_file( $output, $status );
+    check_fitsio( $status );
+    $outFits->create_img( 16, 0, undef, $status );
+    check_fitsio( $status );
+    my $EXTNAME = 'MOPS_DETECTABILITY_RESPONSE';
+    
+    $outFits->create_tbl( BINARY_TBL(), $numRows, scalar @colNames, \@colNames, \@colTypes, undef, $EXTNAME, $status);
+    check_fitsio( $status );
+
+    # TODO: get the extension version number from somewhere common
+    $outFits->write_key( TINT, 'EXTVER', 2, 'Version of this Extension', $status );
+    check_fitsio( $status );
+
+    # Write the Extension keywords
+    foreach my $keyword ( keys %$headers ) {
+        my $value = $inHeader->{$keyword}->{value}; # Header keyword value
+        unless (defined $value) {
+            print "Can't find header keyword $keyword\n";
+            next;
+        }
+        $value =~ s/\'//g;
+        my $name    = $headers->{$keyword}->{name}; # New name
+        my $type    = $headers->{$keyword}->{type}; # Type
+        my $comment = $headers->{$keyword}->{comment}; # Comment
+        $outFits->write_key( $type, $name, $value, $comment, $status );
+        check_fitsio( $status );
+    }
+
+    for (my $i = 0; $i < scalar @colNames; $i++) {
+        my $colName = $colNames[$i];# Column name
+        my $writeType = $colWriteType[$i];
+        my $numRows = scalar(@{$colData{$colName}});
+        unless(defined($writeType)) {
+            print "write type undefined for $colName\n";
+        }
+        unless(defined($numRows)) {
+            print "num Rows undefined for $colName\n";
+        }
+        unless(defined($status)) {
+            print "status undefined for $colName\n";
+        }
+        unless(defined($colData{$colName})) {
+            print "col data undefined for $colName\n";
+        }
+        unless(defined($colName)) {
+            print "column name undefined for $i\n";
+        }
+        print STDERR "$writeType $i $numRows $colName $status @{ $colData{$colName} }\n";
+        $outFits->write_col( $writeType, $i + 1, 1, 1, $numRows, $colData{$colName}, $status );
+        print STDERR "$writeType $i $numRows $colName $status\n";
+        check_fitsio( $status );
+    }
+    $outFits->close_file( $status );
+    return($status);
+}
+
+# From Astro::FITS::CFITSIO demo
+sub check_fitsio
+{
+    my $status = shift;         # Status of FITSIO calls
+
+    if ($status != 0) {
+        my $msg;                # Message to output
+        Astro::FITS::CFITSIO::fits_get_errstatus( $status , $msg );
+        carp("CFITSIO error: $status => $msg");
+        die "CFITSIO error: $msg\n";
+    }
+}
 sub update_request {
     my $req_id = shift;
@@ -181 +383 @@
     }
 }
+
+sub my_die {
+    my $msg = shift;
+    my $fault = shift;
+
+    print STDERR $msg;
+
+    update_request($req_id, $fault);
+
+    exit $fault;
+}