Source code for pipeline.hifa.cli.gotasks.hifa_lock_refant

##################### generated by xml-casa (v2) from hifa_lock_refant.xml ##########
##################### 49194355e4aceddbdfaebcc969ab0335 ##############################
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_lock_refant as _hifa_lock_refant_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_lock_refant: """ hifa_lock_refant ---- Lock reference antenna list hifa_lock_refant locks the reference antenna list, preventing modification of the refant list by subsequent tasks. After executing hifa_lock_refant, the default gaincal refantmode is set to 'strict'. """ _info_group_ = """pipeline""" _info_desc_ = """Lock reference antenna list""" __schema = { } 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 + 3 + 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 ---------------------------------------------- #--------- return inp/go default -------------------------------------------------- #--------- return subparam values ------------------------------------------------- #--------- subparam inp output ---------------------------------------------------- #--------- 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_( ) #--------- inp function ----------------------------------------------------------- def inp(self): print("# hifa_lock_refant -- %s" % self._info_desc_) self.term_width, self.term_height = shutil.get_terminal_size(fallback=(80, 24)) pass #--------- 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_lock_refant.last"): filename = "hifa_lock_refant.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 ): def noobj(s): if s.startswith('<') and s.endswith('>'): return "None" else: return s _prefile = os.path.realpath('hifa_lock_refant.pre') _postfile = os.path.realpath('hifa_lock_refant.last') _return_result_ = None _arguments = [] _invocation_parameters = OrderedDict( ) try: with open(_prefile,'w') as _f: for _i in _invocation_parameters: _f.write("%-3s = %s\n" % (_i,noobj(repr(_invocation_parameters[_i])))) _f.write("#hifa_lock_refant( ") 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_lock_refant_t( ) except Exception as e: from traceback import format_exc from casatasks import casalog casalog.origin('hifa_lock_refant') casalog.post("Exception Reported: Error in hifa_lock_refant: %s" % str(e),'SEVERE') casalog.post(format_exc( )) _return_result_ = False try: os.rename(_prefile,_postfile) except: pass return _return_result_ hifa_lock_refant = _hifa_lock_refant( )