Changeset 33299 for trunk/pstamp
- Timestamp:
- Feb 16, 2012, 5:59:46 PM (14 years ago)
- File:
-
- 1 edited
-
trunk/pstamp/scripts/dqueryparse.pl (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
trunk/pstamp/scripts/dqueryparse.pl
r33142 r33299 1 #! /usr/bin/env perl1 #! /usr/bin/env perl 2 2 # 3 # parse a MOPS_DETCTABILITY_QUERY table and create a results file 3 # 4 # Create a response to a MOPS_DETECTABILITY_QUERY 5 # 4 6 # 5 7 6 8 use strict; 7 9 use warnings; 10 11 use Sys::Hostname; 8 12 use Carp; 13 use File::Basename; 14 use File::Copy; 15 use IPC::Cmd 0.36 qw( can_run run ); 9 16 use Getopt::Long qw( GetOptions ); 10 17 use Pod::Usage qw( pod2usage ); 11 use IPC::Cmd 0.36 qw( can_run run);18 use File::Temp qw( tempfile tempdir); 12 19 13 20 use PS::IPP::PStamp::RequestFile qw( :standard ); 14 21 use PS::IPP::PStamp::Job qw( :standard ); 15 use PS::IPP::Config qw($PS_EXIT_SUCCESS 16 $PS_EXIT_UNKNOWN_ERROR 17 $PS_EXIT_SYS_ERROR 18 $PS_EXIT_CONFIG_ERROR 19 $PS_EXIT_PROG_ERROR 20 $PS_EXIT_DATA_ERROR 21 $PS_EXIT_TIMEOUT_ERROR 22 metadataLookupStr 23 metadataLookupBool 24 caturi 25 ); 26 27 my ($no_update, $imagedb, $label); 28 my ($req_file, $req_id, $outdir, $product, $mode, $dbname, $dbserver, $verbose, $save_temps); 29 my ($job_id,$rownum); # stuff from the post-update world 30 # 31 # parse args 32 # 33 22 use PS::IPP::Config qw( :standard ); 23 use PS::IPP::Metadata::List qw( parse_md_list ); 24 25 # use Astro::FITS::CFITSIO qw( :constants ); 26 # Astro::FITS::CFITSIO::PerlyUnpacking(1); 27 28 my $host = hostname(); 29 30 my $mode = 'queue_job'; 31 32 my ($req_id, $product, $need_magic, $missing_tools, $project, $label); 33 my ($request_file, $outdir, $dbname, $dbserver, $verbose, $save_temps, $no_update); 34 34 GetOptions( 35 'file=s' => \$req_file, 36 'req_id=s' => \$req_id, 37 'job_id=s' => \$job_id, 38 'rownum=s' => \$rownum, 39 'outdir=s' => \$outdir, 40 'label=s' => \$label, 41 'product=s' => \$product, 42 'mode=s' => \$mode, 43 'dbname=s' => \$dbname, 44 'dbserver=s' => \$dbserver, 45 'verbose' => \$verbose, 46 'save-temps' => \$save_temps, 47 ) or pod2usage(2); 48 49 die "invalid mode '$mode'" unless ($mode eq "list_uri") or ($mode eq "queue_job"); 50 die "--file or --job_id is required" if !$req_file; 51 52 if ($mode ne "list_uri") { 53 die "req_id is required" if !$req_id; 54 die "outdir is required" if !$outdir; 55 die "product is required" if !$product; 56 } 57 my $missing_tools; 58 my $pstamptool = can_run('pstamptool') or (warn "Can't find pstamptool" and $missing_tools =1); 59 my $detectresponse = can_run('detectability_respond.pl') or 60 (warn "Can't find detectability_respond.pl" and $missing_tools = 1); 61 my $fields = can_run('fields') or (warn "Can't find fields" and $missing_tools =1); 62 35 'file=s' => \$request_file, 36 'req_id=s' => \$req_id, 37 'outdir=s' => \$outdir, 38 'label=s' => \$label, 39 'product=s' => \$product, 40 'mode=s' => \$mode, 41 'dbname=s' => \$dbname, 42 'dbserver=s' => \$dbserver, 43 'verbose' => \$verbose, 44 'save-temps' => \$save_temps, 45 'no-update' => \$no_update, 46 ) or pod2usage(2); 47 48 pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2 ) if @ARGV; 49 pod2usage( -msg => "Required options: --file --req_id --outdir --dbname", 50 -exitval => 3, 51 ) unless 52 defined $request_file and defined $outdir and defined $dbname and defined $req_id; 53 54 my $detect_query_read = can_run('detect_query_read') or (warn "Can't find detect_query_read" and $missing_tools = 1); 55 my $pstamptool = can_run('pstamptool') or (warn "Can't find pstamptool" and $missing_tools = 1); 56 my $fields = can_run('fields') or (warn "Can't find fields" and $missing_tools = 1); 63 57 if ($missing_tools) { 64 58 warn("Can't find required tools."); 65 exit ($PS_EXIT_CONFIG_ERROR); 66 } 59 exit($PS_EXIT_CONFIG_ERROR); 60 } 61 62 unless (defined $verbose) { 63 $verbose = 0; 64 } 65 66 my $ipprc = PS::IPP::Config->new(); 67 my $mdcParser = PS::IPP::Metadata::Config->new; # Parser for metadata config files 68 67 69 68 70 if (!$dbserver) { 69 my $ipprc = PS::IPP::Config->new(); 70 $dbserver = metadataLookupStr($ipprc->{_siteConfig}, 'PS_DBSERVER'); 71 } 72 73 # just deal with these arguments once and for all 74 $pstamptool .= " -dbname $dbname" if $dbname; 75 $pstamptool .= " -dbserver $dbserver" if $dbserver; 76 $detectresponse .= " --dbname $dbname" if $dbname; 77 78 $no_update = 1 if $mode eq "list_job"; 79 80 # Unless we're running as a job, write the parse arguments in case we need to rerun this parsing. 81 if (!$job_id) { 82 my $argslist = "$outdir/parse.args"; 83 open ARGSLIST, ">$argslist" or my_die("failed to open argslist file $argslist", $job_id, $PS_EXIT_UNKNOWN_ERROR); 84 print ARGSLIST "--label $label --mode $mode --req_id $req_id --product $product --outdir $outdir --file $req_file\n"; 85 close ARGSLIST; 71 $dbserver = metadataLookupStr($ipprc->{_siteConfig}, 'PS_DBSERVER'); 72 } 73 74 $pstamptool .= " -dbname $dbname -dbserver $dbserver"; 75 76 # Project name is hardcoded to gpc1 for the moment. 77 $project = resolve_project($ipprc,"gpc1",$dbname,$dbserver); 78 my $imagedb = $project->{dbname}; 79 if (!$imagedb) { 80 my_die("failed to find imagedb for project: $project", $PS_EXIT_CONFIG_ERROR); 86 81 } 87 82 … … 89 84 my $fields_output; 90 85 { 91 my $command = "echo $req _file | $fields -x 0 EXTNAME EXTVER QUERY_ID";86 my $command = "echo $request_file | $fields -x 0 EXTNAME EXTVER QUERY_ID"; 92 87 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 93 88 run(command => $command, verbose => $verbose); … … 97 92 my (undef, $extname, $extver, $req_name) = split " ", $fields_output; 98 93 99 my_die("$req _file is missing one of EXTNAME EXTVER or QUERY_ID", $PS_EXIT_PROG_ERROR)94 my_die("$request_file is missing one of EXTNAME EXTVER or QUERY_ID", $PS_EXIT_PROG_ERROR) 100 95 if !(defined($extname) and defined($extver) and defined($req_name)); 101 my_die("$req _file has EXTNAME $extname not MOPS_DETECTABILITY_QUERY table", $PS_EXIT_PROG_ERROR)96 my_die("$request_file has EXTNAME $extname not MOPS_DETECTABILITY_QUERY table", $PS_EXIT_PROG_ERROR) 102 97 if $extname ne "MOPS_DETECTABILITY_QUERY"; 103 my_die("$req _file is version $extver expecting 1", $PS_EXIT_PROG_ERROR)98 my_die("$request_file is version $extver expecting 1 or 2", $PS_EXIT_PROG_ERROR) 104 99 if ($extver ne 1) and ($extver ne 2); 105 100 106 # Set up the workdir for this query. 107 if (! -e $outdir ) { 108 mkdir $outdir or my_die("cannot create output directory $outdir", $PS_EXIT_PROG_ERROR); 109 } elsif (! -d $outdir ) { 110 my_die ("output fileset directory $outdir exists but is not a directory", $PS_EXIT_PROG_ERROR); 111 } 112 113 114 # Pass along the request file to the response generator. 115 my $response_file = "$outdir/${req_name}.dresponse.${req_id}.fits"; 116 my $fault; 117 # my $data_to_update = ''; 101 102 my %query = (); 103 my %image_list_hash = (); 104 118 105 { 119 my $command = "$detectresponse --input $req_file --output $response_file --workdir $outdir"; 120 if ($job_id) { 121 $command .= " --job_id $job_id"; 122 } else { 123 $command .= " --ignore-wisdom"; 124 } 125 $command .= " --save-temps" if $save_temps; 126 $command .= " --verbose" if $verbose; 127 106 # 107 # Parse input request file using detect_query_read 108 # 109 my $dqr_command = "$detect_query_read --dbname $imagedb --input $request_file"; 128 110 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 129 run(command => $command, verbose => $verbose);111 run(command => $dqr_command, verbose => $verbose); 130 112 unless ($success) { 131 warn("Warn! Unable to perform $command error code: $error_code"); 132 } 133 # Use the fault code to see if we can regenerate the missing data. 134 $fault = $error_code >> 8; 135 if ($fault == $PSTAMP_NOT_AVAILABLE) { 136 unless (-e "$outdir/update_request.dat") { 137 my_die ("Update request indicated, but unable to find actual request!", $PS_EXIT_PROG_ERROR); 138 } 139 warn("Some inputs are not available and must be updated."); 140 } 141 142 } 143 144 my $result; 145 unless ($job_id) { 146 # We are running as a parse job 147 # If we returned correctly with a valid response file, get a job ID 148 # for the completed work, and move the response to a standardized name. 149 if ($fault == 0) { 150 my $command = "$pstamptool -addjob -req_id $req_id -outputBase $outdir"; 151 $command .= " -job_type detect_query -state stop -fault 0"; 152 $command .= " -rownum 1"; 153 154 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 155 run(command => $command, verbose => $verbose); 156 if ($success) { 157 $job_id = join "", @$stdout_buf; 158 chomp $job_id; 159 if ($job_id && -e $response_file) { 160 rename $response_file, "$outdir/response.${job_id}.fits"; 113 # This is a problem, because I'm not sure how we handle a failure to read something. 114 # We need to return a $PSTAMP_INVALID_REQUEST, I think, but if we can't read it, 115 # we can't send that response back. 116 die("Unable to perform $dqr_command error code: $error_code"); 117 } 118 my $Nrows = 0; 119 { 120 my @column_names = (); 121 # split output into lines skip until the line which lists the column names is found 122 # Parse subsequent lines 123 foreach my $entry (split /\n/, (join "", @$stdout_buf)) { 124 if ($entry =~ /^#/) { 125 @column_names = split /\s+/, $entry; 126 shift(@column_names); # Dump the hash sign. 127 } 128 elsif (scalar @column_names) { 129 # ROWNUM RA1_DEG DEC1_DEG RA2_DEG DEC2_DEG MAG QUERY_ID FPA_ID MJD-OBS FILTER OBSCODE STAGE 130 my %row_data; 131 @row_data{@column_names} = (split /\s+/, $entry); 132 for (my $i = 0; $i <= $#column_names; $i++) { 133 push @{ $query{$row_data{"FPA_ID"}}{$column_names[$i]} }, $row_data{$column_names[$i]}; 134 $Nrows = scalar(keys(%query)); 135 # print "$row_data{'FPA_ID'} $Nrows $i $column_names[$i] $row_data{$column_names[$i]}\n"; 136 } 137 } 138 } 139 } 140 # 141 # Identify target images. This should properly collate targets on a single imfile. 142 # 143 my $query_id; 144 foreach my $fpa_id (keys %query) { 145 my %temp_hash; 146 my $query_style = 'byexp'; 147 my $stage; 148 my $filter; 149 my $mjd; 150 # Confirm that we only have one stage/filter/mjd 151 if ($fpa_id ne 'Not_Set') { # We only need to check things that aren't the known odd case. 152 for (my $i = 0; $i <= $#{ $query{$fpa_id}{STAGE} }; $i++) { 153 $temp_hash{STAGE}{$query{$fpa_id}{STAGE}[$i]} = 1; 154 $temp_hash{FILTER}{$query{$fpa_id}{FILTER}[$i]} = 1; 155 $temp_hash{'MJD-OBS'}{$query{$fpa_id}{'MJD-OBS'}[$i]} = 1; 156 } 157 if (scalar(keys(%{ $temp_hash{STAGE} })) == 1) { 158 $stage = (keys(%{ $temp_hash{STAGE} }))[0]; 159 } 160 else { 161 exit_with_failure(21,"Too many STAGEs specified"); 162 } 163 if (scalar(keys(%{ $temp_hash{FILTER} })) == 1) { 164 $filter = (keys(%{ $temp_hash{FILTER} }))[0]; 165 } 166 else { 167 exit_with_failure(21,"Too many FILTERs specified"); 168 } 169 if (scalar(keys(%{ $temp_hash{'MJD-OBS'} })) == 1) { 170 $mjd = (keys(%{ $temp_hash{'MJD-OBS'} }))[0]; 171 } 172 else { 173 exit_with_failure(21,"Too many MJD-OBS specified"); 174 } 175 } 176 177 # Set up a rowList with default values 178 my @rowList; 179 for (my $i = 0; $i <= $#{ $query{$fpa_id}{STAGE} }; $i++) { 180 $rowList[$i]->{CENTER_X} = $query{$fpa_id}{RA1_DEG}[$i]; 181 $rowList[$i]->{CENTER_Y} = $query{$fpa_id}{DEC1_DEG}[$i]; 182 $rowList[$i]->{ID} = $query{$fpa_id}{ROWNUM}[$i]; 183 $rowList[$i]->{COORD_MASK} = 0; 184 # Set default values 185 $query{$fpa_id}{BAD_COMPONENT}[$i] = 1; 186 $query{$fpa_id}{IMAGE}[$i] = 'no_image'; 187 $query{$fpa_id}{MASK}[$i] = 'no_mask'; 188 $query{$fpa_id}{WEIGHT}[$i] = 'no_weight'; 189 $query{$fpa_id}{PSF}[$i] = 'no_psf'; 190 191 $query{$fpa_id}{STAGE_ID}[$i] = 'no_id'; 192 $query{$fpa_id}{IMAGE_DB}[$i] = 'no_imdb'; 193 $query{$fpa_id}{NEED_MAGIC}[$i] = 'no_magic'; 194 $query{$fpa_id}{MAGICKED}[$i] = 'no_magic'; 195 $query{$fpa_id}{CATALOG}[$i] = 'no_catalog'; 196 $query{$fpa_id}{COMPONENT_ID}[$i] = 'no_component'; 197 $query{$fpa_id}{CLASS_ID}[$i] = 'no_class'; 198 199 $query{$fpa_id}{STATE}[$i] = 'no_state'; 200 $query{$fpa_id}{DATA_STATE}[$i] = 'no_dstate'; 201 $query{$fpa_id}{FAULT}[$i] = 'no_fault'; 202 $query{$fpa_id}{BURNTOOL_STATE}[$i] = 'no_btstate'; 203 } 204 205 # Determine the query style for this fpa_id 206 if ($fpa_id =~ /o.*g.*o/) { 207 $query_style = 'byexp'; 208 } 209 elsif ($fpa_id =~ /\d+/) { 210 $query_style = 'byid'; 211 } 212 elsif ($fpa_id eq 'Not_Set') { 213 # no matching file was found skip 214 next; 215 } 216 else { 217 exit_with_failure(21,"Parse error in request file"); 218 } 219 # Set common request components 220 my $option_mask |= 1; 221 $option_mask |= $PSTAMP_SELECT_IMAGE; 222 $option_mask |= $PSTAMP_SELECT_MASK; 223 $option_mask |= $PSTAMP_SELECT_VARIANCE; 224 $option_mask |= $PSTAMP_SELECT_PSF; 225 my $need_magic = 1; 226 if ($stage eq 'stack') { 227 $need_magic = 0; 228 } 229 my $mjd_min = int $mjd; 230 my $mjd_max = $mjd + 1; 231 232 my $req_filter; 233 if ($filter ne 'Not_Set') { 234 $req_filter = $filter . '%'; 235 } 236 237 # Call the PStamp code to find the images that contain the target on the given MJD in the specified filter. 238 my $pstamp_images_ref = locate_images($ipprc,$imagedb, 239 \@rowList, 240 $query_style,$stage, 241 $fpa_id,undef,undef, 242 $option_mask,$need_magic, 243 $mjd_min,$mjd_max,$req_filter,undef,$verbose); 244 245 foreach my $this_image_ref (@{ $pstamp_images_ref }) { 246 247 if (0) { 248 foreach my $key (sort (keys %{ $this_image_ref } )) { 249 my $value = ${ $this_image_ref }{$key}; 250 if ($key eq 'row_index') { 251 $value = join ' ', @{ $this_image_ref->{$key} }; 252 } 253 print "$this_image_ref $key $value\n"; 254 } 161 255 } 162 $result = 0; 256 257 258 foreach my $valid_index (@{ $this_image_ref->{row_index} }) { 259 $query{$fpa_id}{IMAGE}[$valid_index] = $this_image_ref->{image}; 260 $query{$fpa_id}{MASK}[$valid_index] = $this_image_ref->{mask}; 261 $query{$fpa_id}{WEIGHT}[$valid_index] = $this_image_ref->{weight}; 262 $query{$fpa_id}{PSF}[$valid_index] = $this_image_ref->{psf}; 263 $query{$fpa_id}{STAGE_ID}[$valid_index] = $this_image_ref->{stage_id}; 264 $query{$fpa_id}{IMAGE_DB}[$valid_index] = $this_image_ref->{imagedb}; 265 $query{$fpa_id}{NEED_MAGIC}[$valid_index] = $need_magic; 266 $query{$fpa_id}{BAD_COMPONENT}[$valid_index] = 0; 267 268 if (exists($this_image_ref->{astrom})) { 269 $query{$fpa_id}{CATALOG}[$valid_index] = $this_image_ref->{astrom}; 270 } 271 else { 272 $query{$fpa_id}{CATALOG}[$valid_index] = $this_image_ref->{cmf}; 273 } 274 if (exists($this_image_ref->{class_id})) { 275 $query{$fpa_id}{COMPONENT_ID}[$valid_index] = $this_image_ref->{class_id}; 276 $query{$fpa_id}{CLASS_ID}[$valid_index] = $this_image_ref->{class_id}; 277 278 } 279 else { 280 $query{$fpa_id}{COMPONENT_ID}[$valid_index] = $this_image_ref->{skycell_id}; 281 $query{$fpa_id}{CLASS_ID}[$valid_index] = 'fpa'; 282 } 283 284 $query{$fpa_id}{STATE}[$valid_index] = $this_image_ref->{state}; 285 if (exists($this_image_ref->{data_state})) { 286 $query{$fpa_id}{DATA_STATE}[$valid_index] = $this_image_ref->{data_state}; 287 } 288 else { 289 $query{$fpa_id}{DATA_STATE}[$valid_index] = $this_image_ref->{state}; 290 } 291 $query{$fpa_id}{FAULT}[$valid_index] = 0; 292 $query{$fpa_id}{MAGICKED}[$valid_index] = $this_image_ref->{magicked}; 293 if ($stage eq 'chip') { 294 $query{$fpa_id}{BURNTOOL_STATE}[$valid_index] = $this_image_ref->{burntool_state}; 295 } 296 297 # Determine if the data exists. 298 if (($query{$fpa_id}{STATE}[$valid_index] eq 'goto_purged') or 299 ($query{$fpa_id}{DATA_STATE}[$valid_index] eq 'purged') or 300 ($query{$fpa_id}{STATE}[$valid_index] eq 'drop') or 301 ($query{$fpa_id}{STATE}[$valid_index] eq 'error_cleaned') or 302 ($query{$fpa_id}{STATE}[$valid_index] eq 'goto_scrubbed') or 303 ($query{$fpa_id}{DATA_STATE}[$valid_index] eq 'scrubbed')) { 304 305 # image is gone and it's not coming back 306 $query{$fpa_id}{FAULT}[$valid_index] = $PSTAMP_GONE; 307 } 308 elsif ($need_magic and ($query{$fpa_id}{MAGICKED}[$valid_index] eq 0)) { 309 $query{$fpa_id}{FAULT}[$valid_index] = $PSTAMP_NOT_DESTREAKED; 310 } 311 elsif (($query{$fpa_id}{DATA_STATE}[$valid_index] ne 'full')) { 312 $query{$fpa_id}{FAULT}[$valid_index] = $PSTAMP_NOT_AVAILABLE; 313 314 # updating stacks isn't implemented 315 if (($stage eq 'stack')) { 316 $query{$fpa_id}{FAULT}[$valid_index] = $PSTAMP_NOT_IMPLEMENTED; 317 } 318 # updating old burntool data isn't implemented 319 elsif ($stage eq 'chip') { 320 if ($query{$fpa_id}{BURNTOOL_STATE}[$valid_index] and 321 (abs($query{$fpa_id}{BURNTOOL_STATE}[$valid_index]) < 14)) { 322 $query{$fpa_id}{FAULT}[$valid_index] = $PSTAMP_NOT_IMPLEMENTED; 323 } 324 } 325 } # End determining error faults. 326 } 327 } 328 } 329 } 330 331 my %update_request; 332 my %rows_for_component; 333 my %faulted_rows; 334 335 # now build a hash using image name as key which contains the list of rows for that image 336 foreach my $fpa_id (keys %query) { 337 for (my $i = 0; $i <= $#{ $query{$fpa_id}{ROWNUM} }; $i++) { 338 # print WISDOM "$fpa_id\t"; 339 # foreach my $key (keys %{ $query{$fpa_id} }) { 340 # print WISDOM "$key $query{$fpa_id}{$key}[$i]\t"; 341 # } 342 # print WISDOM "\n"; 343 # my $data_state = $query{$fpa_id}{DATA_STATE}[$i]; 344 # if ($query{$fpa_id}{BAD_COMPONENT}[$i] == 0) { 345 # if ($data_state ne 'full') { 346 347 # @{ $update_request{$query{$fpa_id}{IMAGE}[$i]}{$query{$fpa_id}{FAULT}[$i]} } = 348 # ($query{$fpa_id}{STATE}[$i],$query{$fpa_id}{STAGE}[$i],$query{$fpa_id}{STAGE_ID}[$i], 349 # $query{$fpa_id}{COMPONENT_ID}[$i],$query{$fpa_id}{NEED_MAGIC}[$i],$query{$fpa_id}{IMAGE_DB}[$i]); 350 # } 351 # } 352 if ($query{$fpa_id}{BAD_COMPONENT}[$i] == 0) { 353 push @{ $rows_for_component{$fpa_id}{$query{$fpa_id}{IMAGE}[$i]} }, $i; 163 354 } else { 164 warn("Unable to perform $command error code: $error_code"); 165 $result = $error_code >> 8; 166 } 167 } 168 elsif ($fault == $PSTAMP_NOT_AVAILABLE) { 169 # Failed to run correctly, which means that we need to queue a job and flag data for updating. 170 # first create a parent job for the actual detectabilty query 171 my $command = "$pstamptool -addjob -req_id $req_id -outputBase $outdir/1_"; 172 $command .= " -job_type detect_query -state run -fault 0 -is_parent"; 173 $command .= " -rownum 1"; 174 175 my $parent_job_id; 176 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 177 run(command => $command, verbose => $verbose); 178 if ($success) { 179 # pstamptool returns the job_id of the new job 180 $parent_job_id = join "", @$stdout_buf; 181 chomp $parent_job_id; 182 if ($parent_job_id) { 183 $result = 0; 355 $faulted_rows{$query{$fpa_id}{ROWNUM}[$i]} = $query{$fpa_id}{FAULT}[$i]; 356 } 357 } 358 } 359 360 my $job_num = 0; 361 foreach my $fpa_id (keys %rows_for_component) { 362 foreach my $component (keys % {$rows_for_component{$fpa_id} } ) { 363 my $i = $rows_for_component{$fpa_id}{$component}[0]; 364 my $dep_id = 0; 365 my $data_state = $query{$fpa_id}{DATA_STATE}[$i]; 366 if ($data_state ne 'full') { 367 # Queue this data to be updated 368 my $stage = $query{$fpa_id}{STAGE}[$i]; 369 my $stage_id = $query{$fpa_id}{STAGE_ID}[$i]; 370 my $component = $query{$fpa_id}{STAGE_ID}[$i]; 371 print "Need to UPDATE $stage $stage_id $component from $data_state\n"; 372 $dep_id = queue_update_run($req_id, $outdir, $label, $data_state, $stage, $stage_id, $component, 0, $imagedb); 373 } 374 375 my $outputBase = "$outdir/$job_num" . "_"; 376 my $params_file = $outputBase . "params.mdc"; 377 378 open PARAMS, ">$params_file" or my_die("failed to create $params_file", $PS_EXIT_UNKNOWN_ERROR); 379 print PARAMS "dqueryJobParams METADATA\n"; 380 foreach my $key (keys %{ $query{$fpa_id} } ) { 381 my $value = $query{$fpa_id}{$key}[0]; 382 print PARAMS "\t$key\t\tSTR\t$value\n"; 383 } 384 print PARAMS "END\n\n"; 385 print PARAMS "dqueryCoord MULTI\n"; 386 387 foreach my $row ( @{ $rows_for_component{$fpa_id}{$component} }) { 388 my @coordkeys = qw(ROWNUM RA1_DEG DEC1_DEG RA2_DEG DEC2_DEG MAG); 389 print PARAMS "\ndqueryCoord METADATA\n"; 390 foreach my $key (@coordkeys) { 391 my $value = $query{$fpa_id}{$key}[$row]; 392 print PARAMS "\t$key\t\tSTR\t$value\n"; 393 } 394 print PARAMS "END\n"; 395 } 396 close(PARAMS); 397 398 # use first rownum for this component as the rownum for the job 399 my $rownum = $query{$fpa_id}{ROWNUM}[$rows_for_component{$fpa_id}{$component}[0]]; 400 401 my $command = "$pstamptool -addjob -req_id $req_id -outputBase $outputBase -job_type detect_query -state run -rownum $rownum"; 402 $command .= " -dep_id $dep_id" if $dep_id; 403 unless ($no_update) { 404 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 405 run(command => $command, verbose => $verbose); 406 if ($success) { 407 my $job_id = join "", @$stdout_buf; 408 print " Queued job: $job_id\n"; 184 409 } else { 185 print STDERR "pstamptool did not return a valid job_id for parent job\n"; 186 $result = $PS_EXIT_UNKNOWN_ERROR; 410 my $exit_status = $error_code >> 8; 411 $exit_status = $PS_EXIT_UNKNOWN_ERROR unless $exit_status; 412 my_die("Unable to perform $command error code: $error_code", $exit_status); 187 413 } 188 414 } else { 189 warn("Unable to perform $command error code: $error_code"); 190 $result = $error_code >> 8; 191 } 192 # now create child jobs for each dependent. All these jobs do is finish once the dependency 193 # is satisfied 194 if (!$result) { 195 # Get the dependency id for the data we need to have updated. 196 my $job_num = 2; 197 open(UPDATE_REQUEST,"$outdir/update_request.dat") || my_die ("Update request indicated, but unable to find actual request!", $PS_EXIT_PROG_ERROR); 198 while (<UPDATE_REQUEST>) { 199 my $data_to_update = $_; 200 chomp($data_to_update); 201 my $dep_id = queue_update_run($req_id,$job_id,$outdir,$label,$data_to_update); 202 203 # Link this request to a job and link that job to the dependent 204 my $command = "$pstamptool -addjob -req_id $req_id -outputBase $outdir/${job_num}_"; 205 $command .= " -job_type child -state run -fault 0 -parent_id $parent_job_id"; 206 $command .= " -rownum 1"; # XXX: we should choose a correct rownum 207 $command .= " -dep_id $dep_id" if $dep_id; 208 209 $job_num++; 210 211 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 212 run(command => $command, verbose => $verbose); 213 214 if ($success) { 215 $job_id = join "", @$stdout_buf; 216 chomp $job_id; 217 $result = 0; 218 } else { 219 warn("Unable to perform $command error code: $error_code"); 220 $result = $error_code >> 8; 221 last; 222 } 223 } 224 close(UPDATE_REQUEST); 225 } 226 } 227 else { 228 my_die ("Parse fault!!", $fault); 229 } 230 # This does not set the request state to stop. That will happen with the request_finish.pl script, 231 # which will notice that we've inserted the stopped job and decide we're finished. Easy enough. 232 { 233 my $command = "$pstamptool -updatereq -req_id $req_id -set_name $req_name -set_outProduct $product"; 234 $command .= " -set_fault $result" if $result; 235 415 print STDERR "skipping $command\n"; 416 } 417 $job_num++; 418 } 419 } 420 421 # XXX: flesh this out once we have a request without duplicate ROWNUMS 422 foreach my $rownum (keys %faulted_rows ) { 423 my $fault = $faulted_rows{$rownum}; 424 $fault = $PSTAMP_NO_IMAGE_MATCH if ($fault eq 'no_fault'); 425 print STDERR "insert faulted job for $rownum: $fault\n"; 426 } 427 428 { 429 my $command = "$pstamptool -updatereq -req_id $req_id -set_name $req_name -set_outProduct $product"; 430 # $command .= " -set_fault $result" if $result; 431 432 unless ($no_update) { 236 433 my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 237 434 run(command => $command, verbose => $verbose); 238 435 unless ($success) { 239 my_die("$command failed",$PS_EXIT_UNKNOWN_ERROR); 240 } 241 } 242 } else { 243 # We are running as a job presumably because an input needed 244 # to be regenerated 245 if ($fault) { 246 # in some cases we will need to finish off the request 247 my_die ("Run fault!!", $fault); 248 } 249 if (-e $response_file) { 250 rename $response_file, "$outdir/response.${job_id}.fits"; 251 } 252 } 253 254 436 my_die("$command failed",$PS_EXIT_UNKNOWN_ERROR); 437 } 438 } else { 439 print STDERR "Skipping $command\n"; 440 } 441 } 442 255 443 exit 0; 256 444 257 258 # If we have to queue an update run, do so and create a new dependent259 445 sub queue_update_run { 260 my ($req_id, $ job_id, $outdir, $label, $data_to_update) = @_;261 262 my ($state, $stage, $stage_id, $component, $need_magic, $imagedb) = split /\s+/, $data_to_update;446 my ($req_id, $outdir, $label, $state, $stage, $stage_id, $component, $need_magic, $imagedb) = @_; 447 448 # my ($state, $stage, $stage_id, $component, $need_magic, $imagedb) = split /\s+/, $data_to_update; 263 449 264 450 if (($state ne 'cleaned') and ($state ne 'update') and ($state ne 'goto_cleaned')) { … … 293 479 return($dep_id); 294 480 } 295 296 481 297 482 298 483 sub my_die { 299 my $msg = shift; 300 my $fault = shift; 301 302 carp $msg; 303 304 # we don't fault the request here pstamp_parser_run.pl handles that if necessary 305 exit $fault; 306 } 484 my $message = shift; 485 my $exit_code = shift; 486 487 $exit_code = $PS_EXIT_PROG_ERROR unless defined $exit_code; 488 489 carp("$message"); 490 exit($exit_code); 491 } 492 493 sub exit_with_failure { 494 my $status = shift; 495 my $message = shift; 496 my_die($message, $status); 497 } 498
Note:
See TracChangeset
for help on using the changeset viewer.
