##################### generated by xml-casa (v2) from hifa_timegaincal.xml ##########
##################### 87c20af92490128b22f8350b74593e2f ##############################
from __future__ import absolute_import
from casashell.private.stack_manip import find_local as __sf__
from casashell.private.stack_manip import find_frame as _find_frame
from casatools.typecheck import validator as _pc
from casatools.coercetype import coerce as _coerce
from pipeline.hifa.cli import hifa_timegaincal as _hifa_timegaincal_t
from collections import OrderedDict
import numpy
import sys
import os
import shutil
[docs]def static_var(varname, value):
def decorate(func):
setattr(func, varname, value)
return func
return decorate
class _hifa_timegaincal:
"""
hifa_timegaincal ---- Determine temporal gains from calibrator observations
The time-dependent complex gains for each antenna/spwid are determined
from the raw data (DATA column) divided by the model (MODEL column), for the
specified fields. The gains are computed independently for each specified
spectral window. One gain solution is computed for the calibrator source
targets and one for the science targets.
Previous calibrations are applied on the fly.
The complex gains are derived from the data column (raw data) divided by the
model column (usually set with hif_setjy). The gains are obtained for the
specified solution intervals, spw combination and field combination. One gain
solution is computed for the science targets and one for the calibrator
targets.
Good candidate reference antennas can be determined using the hif_refant
task.
Previous calibrations that have been stored in the pipeline context are
applied on the fly. Users can interact with these calibrations via the
hif_export_calstate and hif_import_calstate tasks.
Output
results -- If pipeline mode is 'getinputs' then None is returned. Otherwise
the results object for the pipeline task is returned.
--------- parameter descriptions ---------------------------------------------
vis The list of input MeasurementSets. Defaults to the list of
MeasurementSets specified in the pipeline context.
example: vis=['M82A.ms', 'M82B.ms']
calamptable The list of output diagnostic calibration amplitude tables for
the calibration targets. Defaults to the standard pipeline naming
convention.
example: calamptable=['M82.gacal', 'M82B.gacal']
offsetstable The list of output diagnostic phase offset tables for the
calibration targets. Defaults to the standard pipeline naming convention.
example: offsetstable=['M82.offsets.gacal', 'M82B.offsets.gacal']
calphasetable The list of output calibration phase tables for the
calibration targets. Defaults to the standard pipeline naming convention.
example: calphasetable=['M82.gcal', 'M82B.gcal']
targetphasetable The list of output phase calibration tables for the science
targets. Defaults to the standard pipeline naming convention.
example: targetphasetable=['M82.gcal', 'M82B.gcal']
amptable The list of output calibration amplitude tables for the
calibration and science targets.
Defaults to the standard pipeline naming convention.
example: amptable=['M82.gcal', 'M82B.gcal']
field The list of field names or field ids for which gain solutions are to
be computed. Defaults to all fields with the standard intent.
example: field='3C279', field='3C279, M82'
intent A string containing a comma delimited list of intents against which
the selected fields are matched. Defaults to the equivalent of
'AMPLITUDE,PHASE,BANDPASS'.
example: intent='', intent='PHASE'
spw The list of spectral windows and channels for which gain solutions are
computed. Defaults to all science spectral windows.
example: spw='3C279', spw='3C279, M82'
antenna
calsolint Time solution interval in CASA syntax for calibrator source
solutions.
example: calsolint='inf', calsolint='int', calsolint='100sec'
targetsolint Time solution interval in CASA syntax for target source
solutions.
example: targetsolint='inf', targetsolint='int', targetsolint='100sec'
combine Data axes to combine for solving. Options are '', 'scan', 'spw',
'field' or any comma-separated combination.
default: ''
example: combine=''
refant Reference antenna name(s) in priority order. Defaults to most recent
values set in the pipeline context. If no reference antenna is defined in
the pipeline context use the CASA defaults.
example: refant='DV01', refant='DV05,DV07'
refantmode Controls how the refant is applied. Currently available
choices are 'flex', 'strict', and the default value of ''.
Setting to '' allows the pipeline to select the appropriate
mode based on the state of the reference antenna list.
Examples: refantmode='strict', refantmode=''
solnorm Normalise the gain solutions.
minblperant Minimum number of baselines required per antenna for each solve.
Antennas with fewer baselines are excluded from solutions.
example: minblperant=2
calminsnr Solutions below this SNR are rejected for calibrator solutions.
targetminsnr Solutions below this SNR are rejected for science target
solutions.
smodel Point source Stokes parameters for source model (experimental)
Defaults to using standard MODEL_DATA column data.
example: smodel=[1,0,0,0] - (I=1, unpolarized)
pipelinemode The pipeline operating mode. In 'automatic' mode the pipeline
determines the values of all context defined pipeline inputs automatically.
In interactive mode the user can set the pipeline context defined
parameters manually. In 'getinputs' mode the user can check the settings of
all pipeline parameters without running the task.
dryrun Run the commands (True) or generate the commands to be run but do not
execute (False).
acceptresults Add the results of the task to the pipeline context (True) or
reject them (False).
--------- examples -----------------------------------------------------------
1. Compute standard per scan gain solutions that will be used to calibrate
the target:
hifa_timegaincal()
"""
_info_group_ = """pipeline"""
_info_desc_ = """Determine temporal gains from calibrator observations"""
__schema = {'vis': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'calamptable': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'offsetstable': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'calphasetable': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'targetphasetable': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'amptable': {'type': 'cStrVec', 'coerce': [_coerce.to_list,_coerce.to_strvec]}, 'field': {'type': 'cStr', 'coerce': _coerce.to_str}, 'intent': {'type': 'cStr', 'coerce': _coerce.to_str}, 'spw': {'type': 'cStr', 'coerce': _coerce.to_str}, 'antenna': {'type': 'cStr', 'coerce': _coerce.to_str}, 'calsolint': {'type': 'cVariant', 'coerce': [_coerce.to_variant]}, 'targetsolint': {'type': 'cVariant', 'coerce': [_coerce.to_variant]}, 'combine': {'type': 'cStr', 'coerce': _coerce.to_str}, 'refant': {'type': 'cStr', 'coerce': _coerce.to_str}, 'refantmode': {'type': 'cStr', 'coerce': _coerce.to_str, 'allowed': [ '', 'flex', 'strict' ]}, 'solnorm': {'type': 'cBool'}, 'minblperant': {'type': 'cInt'}, 'calminsnr': {'type': 'cFloat', 'coerce': _coerce.to_float}, 'targetminsnr': {'type': 'cFloat', 'coerce': _coerce.to_float}, 'smodel': {'type': 'cFloatVec', 'coerce': [_coerce.to_list,_coerce.to_floatvec]}, 'pipelinemode': {'type': 'cStr', 'coerce': _coerce.to_str, 'allowed': [ 'automatic', 'interactive', 'getinputs' ]}, 'dryrun': {'type': 'cBool'}, 'acceptresults': {'type': 'cBool'}}
def __init__(self):
self.__stdout = None
self.__stderr = None
self.__root_frame_ = None
def __globals_(self):
if self.__root_frame_ is None:
self.__root_frame_ = _find_frame( )
assert self.__root_frame_ is not None, "could not find CASAshell global frame"
return self.__root_frame_
def __to_string_(self,value):
if type(value) is str:
return "'%s'" % value
else:
return str(value)
def __validate_(self,doc,schema):
return _pc.validate(doc,schema)
def __do_inp_output(self,param_prefix,description_str,formatting_chars):
out = self.__stdout or sys.stdout
description = description_str.split( )
prefix_width = 23 + 19 + 4
output = [ ]
addon = ''
first_addon = True
while len(description) > 0:
## starting a new line.....................................................................
if len(output) == 0:
## for first line add parameter information............................................
if len(param_prefix)-formatting_chars > prefix_width - 1:
output.append(param_prefix)
continue
addon = param_prefix + ' #'
first_addon = True
addon_formatting = formatting_chars
else:
## for subsequent lines space over prefix width........................................
addon = (' ' * prefix_width) + '#'
first_addon = False
addon_formatting = 0
## if first word of description puts us over the screen width, bail........................
if len(addon + description[0]) - addon_formatting + 1 > self.term_width:
## if we're doing the first line make sure it's output.................................
if first_addon: output.append(addon)
break
while len(description) > 0:
## if the next description word puts us over break for the next line...................
if len(addon + description[0]) - addon_formatting + 1 > self.term_width: break
addon = addon + ' ' + description[0]
description.pop(0)
output.append(addon)
out.write('\n'.join(output) + '\n')
#--------- return nonsubparam values ----------------------------------------------
def __combine_dflt( self, glb ):
return ''
def __combine( self, glb ):
if 'combine' in glb: return glb['combine']
return ''
def __calsolint_dflt( self, glb ):
return 'int'
def __calsolint( self, glb ):
if 'calsolint' in glb: return glb['calsolint']
return 'int'
def __calminsnr_dflt( self, glb ):
return float(2.0)
def __calminsnr( self, glb ):
if 'calminsnr' in glb: return glb['calminsnr']
return float(2.0)
def __minblperant_dflt( self, glb ):
return int(4)
def __minblperant( self, glb ):
if 'minblperant' in glb: return glb['minblperant']
return int(4)
def __targetsolint_dflt( self, glb ):
return 'inf'
def __targetsolint( self, glb ):
if 'targetsolint' in glb: return glb['targetsolint']
return 'inf'
def __pipelinemode_dflt( self, glb ):
return 'automatic'
def __pipelinemode( self, glb ):
if 'pipelinemode' in glb: return glb['pipelinemode']
return 'automatic'
def __targetminsnr_dflt( self, glb ):
return float(3.0)
def __targetminsnr( self, glb ):
if 'targetminsnr' in glb: return glb['targetminsnr']
return float(3.0)
#--------- return inp/go default --------------------------------------------------
def __antenna_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __dryrun_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return bool(False)
return None
def __field_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __intent_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __targetphasetable_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __offsetstable_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __calamptable_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __refant_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __calphasetable_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __vis_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __acceptresults_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return bool(True)
return None
def __solnorm_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return bool(False)
if self.__pipelinemode( glb ) == "getinputs": return bool(False)
return None
def __refantmode_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __smodel_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
def __spw_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return ""
if self.__pipelinemode( glb ) == "getinputs": return ""
return None
def __amptable_dflt( self, glb ):
if self.__pipelinemode( glb ) == "interactive": return []
if self.__pipelinemode( glb ) == "getinputs": return []
return None
#--------- return subparam values -------------------------------------------------
def __vis( self, glb ):
if 'vis' in glb: return glb['vis']
dflt = self.__vis_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __calamptable( self, glb ):
if 'calamptable' in glb: return glb['calamptable']
dflt = self.__calamptable_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __offsetstable( self, glb ):
if 'offsetstable' in glb: return glb['offsetstable']
dflt = self.__offsetstable_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __calphasetable( self, glb ):
if 'calphasetable' in glb: return glb['calphasetable']
dflt = self.__calphasetable_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __targetphasetable( self, glb ):
if 'targetphasetable' in glb: return glb['targetphasetable']
dflt = self.__targetphasetable_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __amptable( self, glb ):
if 'amptable' in glb: return glb['amptable']
dflt = self.__amptable_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __field( self, glb ):
if 'field' in glb: return glb['field']
dflt = self.__field_dflt( glb )
if dflt is not None: return dflt
return ''
def __intent( self, glb ):
if 'intent' in glb: return glb['intent']
dflt = self.__intent_dflt( glb )
if dflt is not None: return dflt
return ''
def __spw( self, glb ):
if 'spw' in glb: return glb['spw']
dflt = self.__spw_dflt( glb )
if dflt is not None: return dflt
return ''
def __antenna( self, glb ):
if 'antenna' in glb: return glb['antenna']
dflt = self.__antenna_dflt( glb )
if dflt is not None: return dflt
return ''
def __refant( self, glb ):
if 'refant' in glb: return glb['refant']
dflt = self.__refant_dflt( glb )
if dflt is not None: return dflt
return ''
def __refantmode( self, glb ):
if 'refantmode' in glb: return glb['refantmode']
dflt = self.__refantmode_dflt( glb )
if dflt is not None: return dflt
return ''
def __solnorm( self, glb ):
if 'solnorm' in glb: return glb['solnorm']
dflt = self.__solnorm_dflt( glb )
if dflt is not None: return dflt
return False
def __smodel( self, glb ):
if 'smodel' in glb: return glb['smodel']
dflt = self.__smodel_dflt( glb )
if dflt is not None: return dflt
return [ ]
def __dryrun( self, glb ):
if 'dryrun' in glb: return glb['dryrun']
dflt = self.__dryrun_dflt( glb )
if dflt is not None: return dflt
return False
def __acceptresults( self, glb ):
if 'acceptresults' in glb: return glb['acceptresults']
dflt = self.__acceptresults_dflt( glb )
if dflt is not None: return dflt
return True
#--------- subparam inp output ----------------------------------------------------
def __vis_inp(self):
if self.__vis_dflt( self.__globals_( ) ) is not None:
description = 'List of input MeasurementSets'
value = self.__vis( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'vis': value},{'vis': self.__schema['vis']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('vis',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __calamptable_inp(self):
if self.__calamptable_dflt( self.__globals_( ) ) is not None:
description = 'List of diagnostic output amplitude caltables for calibrator targets'
value = self.__calamptable( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'calamptable': value},{'calamptable': self.__schema['calamptable']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('calamptable',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __offsetstable_inp(self):
if self.__offsetstable_dflt( self.__globals_( ) ) is not None:
description = 'List of diagnostic output phase offset caltables for calibrator targets'
value = self.__offsetstable( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'offsetstable': value},{'offsetstable': self.__schema['offsetstable']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('offsetstable',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __calphasetable_inp(self):
if self.__calphasetable_dflt( self.__globals_( ) ) is not None:
description = 'List of output phase caltables for calibrator targets'
value = self.__calphasetable( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'calphasetable': value},{'calphasetable': self.__schema['calphasetable']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('calphasetable',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __targetphasetable_inp(self):
if self.__targetphasetable_dflt( self.__globals_( ) ) is not None:
description = 'List of output phase caltables for science targets'
value = self.__targetphasetable( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'targetphasetable': value},{'targetphasetable': self.__schema['targetphasetable']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('targetphasetable',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __amptable_inp(self):
if self.__amptable_dflt( self.__globals_( ) ) is not None:
description = 'List of output amp caltables for science targets'
value = self.__amptable( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'amptable': value},{'amptable': self.__schema['amptable']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('amptable',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __field_inp(self):
if self.__field_dflt( self.__globals_( ) ) is not None:
description = 'Set of data selection field names or ids'
value = self.__field( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'field': value},{'field': self.__schema['field']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('field',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __intent_inp(self):
if self.__intent_dflt( self.__globals_( ) ) is not None:
description = 'Set of data selection observing intents'
value = self.__intent( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'intent': value},{'intent': self.__schema['intent']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('intent',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __spw_inp(self):
if self.__spw_dflt( self.__globals_( ) ) is not None:
description = 'Set of data selection spectral window/channels'
value = self.__spw( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'spw': value},{'spw': self.__schema['spw']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('spw',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __antenna_inp(self):
if self.__antenna_dflt( self.__globals_( ) ) is not None:
description = 'Set of data selection antenna ids'
value = self.__antenna( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'antenna': value},{'antenna': self.__schema['antenna']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('antenna',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __calsolint_inp(self):
description = 'Phase solution interval for calibrator sources'
value = self.__calsolint( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'calsolint': value},{'calsolint': self.__schema['calsolint']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('calsolint',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __targetsolint_inp(self):
description = 'Phase solution interval for science target sources'
value = self.__targetsolint( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'targetsolint': value},{'targetsolint': self.__schema['targetsolint']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('targetsolint',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __combine_inp(self):
description = 'Data axes which to combine for solve (scan, spw, and/or field)'
value = self.__combine( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'combine': value},{'combine': self.__schema['combine']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('combine',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __refant_inp(self):
if self.__refant_dflt( self.__globals_( ) ) is not None:
description = 'Reference antenna names'
value = self.__refant( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'refant': value},{'refant': self.__schema['refant']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('refant',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __refantmode_inp(self):
if self.__refantmode_dflt( self.__globals_( ) ) is not None:
description = 'Reference antenna selection mode'
value = self.__refantmode( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'refantmode': value},{'refantmode': self.__schema['refantmode']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('refantmode',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __solnorm_inp(self):
if self.__solnorm_dflt( self.__globals_( ) ) is not None:
description = 'Normalize average solution amplitudes to 1.0'
value = self.__solnorm( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'solnorm': value},{'solnorm': self.__schema['solnorm']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('solnorm',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __minblperant_inp(self):
description = 'Minimum baselines per antenna required for solve'
value = self.__minblperant( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'minblperant': value},{'minblperant': self.__schema['minblperant']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('minblperant',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __calminsnr_inp(self):
description = 'Reject solutions below this SNR for calibrator solutions'
value = self.__calminsnr( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'calminsnr': value},{'calminsnr': self.__schema['calminsnr']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('calminsnr',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __targetminsnr_inp(self):
description = 'Reject solutions below this SNR for science solutions'
value = self.__targetminsnr( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'targetminsnr': value},{'targetminsnr': self.__schema['targetminsnr']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('%-19.19s = %s%-23s%s' % ('targetminsnr',pre,self.__to_string_(value),post),description,0+len(pre)+len(post))
def __smodel_inp(self):
if self.__smodel_dflt( self.__globals_( ) ) is not None:
description = 'Point source Stokes parameters for source model'
value = self.__smodel( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'smodel': value},{'smodel': self.__schema['smodel']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('smodel',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __pipelinemode_inp(self):
description = 'The pipeline operating mode'
value = self.__pipelinemode( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'pipelinemode': value},{'pipelinemode': self.__schema['pipelinemode']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output('\x1B[1m\x1B[47m%-19.19s =\x1B[0m %s%-23s%s' % ('pipelinemode',pre,self.__to_string_(value),post),description,13+len(pre)+len(post))
def __dryrun_inp(self):
if self.__dryrun_dflt( self.__globals_( ) ) is not None:
description = 'Run task (False) or display the command(True)'
value = self.__dryrun( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'dryrun': value},{'dryrun': self.__schema['dryrun']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('dryrun',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
def __acceptresults_inp(self):
if self.__acceptresults_dflt( self.__globals_( ) ) is not None:
description = 'Automatically accept results into the context'
value = self.__acceptresults( self.__globals_( ) )
(pre,post) = ('','') if self.__validate_({'acceptresults': value},{'acceptresults': self.__schema['acceptresults']}) else ('\x1B[91m','\x1B[0m')
self.__do_inp_output(' \x1B[92m%-16.16s =\x1B[0m %s%-23s%s' % ('acceptresults',pre,self.__to_string_(value),post),description,9+len(pre)+len(post))
#--------- global default implementation-------------------------------------------
@static_var('state', __sf__('casa_inp_go_state'))
def set_global_defaults(self):
self.set_global_defaults.state['last'] = self
glb = self.__globals_( )
if 'antenna' in glb: del glb['antenna']
if 'dryrun' in glb: del glb['dryrun']
if 'targetminsnr' in glb: del glb['targetminsnr']
if 'field' in glb: del glb['field']
if 'pipelinemode' in glb: del glb['pipelinemode']
if 'intent' in glb: del glb['intent']
if 'calminsnr' in glb: del glb['calminsnr']
if 'targetphasetable' in glb: del glb['targetphasetable']
if 'calsolint' in glb: del glb['calsolint']
if 'offsetstable' in glb: del glb['offsetstable']
if 'calamptable' in glb: del glb['calamptable']
if 'refant' in glb: del glb['refant']
if 'calphasetable' in glb: del glb['calphasetable']
if 'vis' in glb: del glb['vis']
if 'acceptresults' in glb: del glb['acceptresults']
if 'minblperant' in glb: del glb['minblperant']
if 'solnorm' in glb: del glb['solnorm']
if 'combine' in glb: del glb['combine']
if 'refantmode' in glb: del glb['refantmode']
if 'targetsolint' in glb: del glb['targetsolint']
if 'smodel' in glb: del glb['smodel']
if 'spw' in glb: del glb['spw']
if 'amptable' in glb: del glb['amptable']
#--------- inp function -----------------------------------------------------------
def inp(self):
print("# hifa_timegaincal -- %s" % self._info_desc_)
self.term_width, self.term_height = shutil.get_terminal_size(fallback=(80, 24))
self.__vis_inp( )
self.__calamptable_inp( )
self.__offsetstable_inp( )
self.__calphasetable_inp( )
self.__targetphasetable_inp( )
self.__amptable_inp( )
self.__field_inp( )
self.__intent_inp( )
self.__spw_inp( )
self.__antenna_inp( )
self.__calsolint_inp( )
self.__targetsolint_inp( )
self.__combine_inp( )
self.__refant_inp( )
self.__refantmode_inp( )
self.__solnorm_inp( )
self.__minblperant_inp( )
self.__calminsnr_inp( )
self.__targetminsnr_inp( )
self.__smodel_inp( )
self.__pipelinemode_inp( )
self.__dryrun_inp( )
self.__acceptresults_inp( )
#--------- tget function ----------------------------------------------------------
@static_var('state', __sf__('casa_inp_go_state'))
def tget(self,file=None):
from casashell.private.stack_manip import find_frame
from runpy import run_path
filename = None
if file is None:
if os.path.isfile("hifa_timegaincal.last"):
filename = "hifa_timegaincal.last"
elif isinstance(file, str):
if os.path.isfile(file):
filename = file
if filename is not None:
glob = find_frame( )
newglob = run_path( filename, init_globals={ } )
for i in newglob:
glob[i] = newglob[i]
self.tget.state['last'] = self
else:
print("could not find last file, setting defaults instead...")
self.set_global_defaults( )
def __call__( self, vis=None, calamptable=None, offsetstable=None, calphasetable=None, targetphasetable=None, amptable=None, field=None, intent=None, spw=None, antenna=None, calsolint=None, targetsolint=None, combine=None, refant=None, refantmode=None, solnorm=None, minblperant=None, calminsnr=None, targetminsnr=None, smodel=None, pipelinemode=None, dryrun=None, acceptresults=None ):
def noobj(s):
if s.startswith('<') and s.endswith('>'):
return "None"
else:
return s
_prefile = os.path.realpath('hifa_timegaincal.pre')
_postfile = os.path.realpath('hifa_timegaincal.last')
_return_result_ = None
_arguments = [vis,calamptable,offsetstable,calphasetable,targetphasetable,amptable,field,intent,spw,antenna,calsolint,targetsolint,combine,refant,refantmode,solnorm,minblperant,calminsnr,targetminsnr,smodel,pipelinemode,dryrun,acceptresults]
_invocation_parameters = OrderedDict( )
if any(map(lambda x: x is not None,_arguments)):
# invoke python style
# set the non sub-parameters that are not None
local_global = { }
if calsolint is not None: local_global['calsolint'] = calsolint
if targetsolint is not None: local_global['targetsolint'] = targetsolint
if combine is not None: local_global['combine'] = combine
if minblperant is not None: local_global['minblperant'] = minblperant
if calminsnr is not None: local_global['calminsnr'] = calminsnr
if targetminsnr is not None: local_global['targetminsnr'] = targetminsnr
if pipelinemode is not None: local_global['pipelinemode'] = pipelinemode
# the invocation parameters for the non-subparameters can now be set - this picks up those defaults
_invocation_parameters['calsolint'] = self.__calsolint( local_global )
_invocation_parameters['targetsolint'] = self.__targetsolint( local_global )
_invocation_parameters['combine'] = self.__combine( local_global )
_invocation_parameters['minblperant'] = self.__minblperant( local_global )
_invocation_parameters['calminsnr'] = self.__calminsnr( local_global )
_invocation_parameters['targetminsnr'] = self.__targetminsnr( local_global )
_invocation_parameters['pipelinemode'] = self.__pipelinemode( local_global )
# the sub-parameters can then be set. Use the supplied value if not None, else the function, which gets the appropriate default
_invocation_parameters['vis'] = self.__vis( _invocation_parameters ) if vis is None else vis
_invocation_parameters['calamptable'] = self.__calamptable( _invocation_parameters ) if calamptable is None else calamptable
_invocation_parameters['offsetstable'] = self.__offsetstable( _invocation_parameters ) if offsetstable is None else offsetstable
_invocation_parameters['calphasetable'] = self.__calphasetable( _invocation_parameters ) if calphasetable is None else calphasetable
_invocation_parameters['targetphasetable'] = self.__targetphasetable( _invocation_parameters ) if targetphasetable is None else targetphasetable
_invocation_parameters['amptable'] = self.__amptable( _invocation_parameters ) if amptable is None else amptable
_invocation_parameters['field'] = self.__field( _invocation_parameters ) if field is None else field
_invocation_parameters['intent'] = self.__intent( _invocation_parameters ) if intent is None else intent
_invocation_parameters['spw'] = self.__spw( _invocation_parameters ) if spw is None else spw
_invocation_parameters['antenna'] = self.__antenna( _invocation_parameters ) if antenna is None else antenna
_invocation_parameters['refant'] = self.__refant( _invocation_parameters ) if refant is None else refant
_invocation_parameters['refantmode'] = self.__refantmode( _invocation_parameters ) if refantmode is None else refantmode
_invocation_parameters['solnorm'] = self.__solnorm( _invocation_parameters ) if solnorm is None else solnorm
_invocation_parameters['smodel'] = self.__smodel( _invocation_parameters ) if smodel is None else smodel
_invocation_parameters['dryrun'] = self.__dryrun( _invocation_parameters ) if dryrun is None else dryrun
_invocation_parameters['acceptresults'] = self.__acceptresults( _invocation_parameters ) if acceptresults is None else acceptresults
else:
# invoke with inp/go semantics
_invocation_parameters['vis'] = self.__vis( self.__globals_( ) )
_invocation_parameters['calamptable'] = self.__calamptable( self.__globals_( ) )
_invocation_parameters['offsetstable'] = self.__offsetstable( self.__globals_( ) )
_invocation_parameters['calphasetable'] = self.__calphasetable( self.__globals_( ) )
_invocation_parameters['targetphasetable'] = self.__targetphasetable( self.__globals_( ) )
_invocation_parameters['amptable'] = self.__amptable( self.__globals_( ) )
_invocation_parameters['field'] = self.__field( self.__globals_( ) )
_invocation_parameters['intent'] = self.__intent( self.__globals_( ) )
_invocation_parameters['spw'] = self.__spw( self.__globals_( ) )
_invocation_parameters['antenna'] = self.__antenna( self.__globals_( ) )
_invocation_parameters['calsolint'] = self.__calsolint( self.__globals_( ) )
_invocation_parameters['targetsolint'] = self.__targetsolint( self.__globals_( ) )
_invocation_parameters['combine'] = self.__combine( self.__globals_( ) )
_invocation_parameters['refant'] = self.__refant( self.__globals_( ) )
_invocation_parameters['refantmode'] = self.__refantmode( self.__globals_( ) )
_invocation_parameters['solnorm'] = self.__solnorm( self.__globals_( ) )
_invocation_parameters['minblperant'] = self.__minblperant( self.__globals_( ) )
_invocation_parameters['calminsnr'] = self.__calminsnr( self.__globals_( ) )
_invocation_parameters['targetminsnr'] = self.__targetminsnr( self.__globals_( ) )
_invocation_parameters['smodel'] = self.__smodel( self.__globals_( ) )
_invocation_parameters['pipelinemode'] = self.__pipelinemode( self.__globals_( ) )
_invocation_parameters['dryrun'] = self.__dryrun( self.__globals_( ) )
_invocation_parameters['acceptresults'] = self.__acceptresults( self.__globals_( ) )
try:
with open(_prefile,'w') as _f:
for _i in _invocation_parameters:
_f.write("%-16s = %s\n" % (_i,noobj(repr(_invocation_parameters[_i]))))
_f.write("#hifa_timegaincal( ")
count = 0
for _i in _invocation_parameters:
_f.write("%s=%s" % (_i,noobj(repr(_invocation_parameters[_i]))))
count += 1
if count < len(_invocation_parameters): _f.write(",")
_f.write(" )\n")
except: pass
try:
_return_result_ = _hifa_timegaincal_t( _invocation_parameters['vis'],_invocation_parameters['calamptable'],_invocation_parameters['offsetstable'],_invocation_parameters['calphasetable'],_invocation_parameters['targetphasetable'],_invocation_parameters['amptable'],_invocation_parameters['field'],_invocation_parameters['intent'],_invocation_parameters['spw'],_invocation_parameters['antenna'],_invocation_parameters['calsolint'],_invocation_parameters['targetsolint'],_invocation_parameters['combine'],_invocation_parameters['refant'],_invocation_parameters['refantmode'],_invocation_parameters['solnorm'],_invocation_parameters['minblperant'],_invocation_parameters['calminsnr'],_invocation_parameters['targetminsnr'],_invocation_parameters['smodel'],_invocation_parameters['pipelinemode'],_invocation_parameters['dryrun'],_invocation_parameters['acceptresults'] )
except Exception as e:
from traceback import format_exc
from casatasks import casalog
casalog.origin('hifa_timegaincal')
casalog.post("Exception Reported: Error in hifa_timegaincal: %s" % str(e),'SEVERE')
casalog.post(format_exc( ))
_return_result_ = False
try:
os.rename(_prefile,_postfile)
except: pass
return _return_result_
hifa_timegaincal = _hifa_timegaincal( )