Coverage for /wheeldirectory/casa-6.7.0-12-py3.10.el8/lib/py/lib/python3.10/site-packages/casatasks/private/vishead_util.py: 15%

1#""" 

2#Helper functions for the vishead task that might also be useful outside it, 

3#when working with measurement sets as tables. 

4#""" 

5 

6import os 

7 

8from casatools import table, quanta 

9_tb = table( ) 

10_qa = quanta( ) 

11 

12# basestring is use in the CASA5 code, now use str and hide it a bit from any other basestring 

13# this is really a python3 difference 

14_basestring = str 

15 

16def getput_keyw(mode, vis, key, hdindex, hdvalue='', hdref=None): 

17 table = vis + '/' + key[0] 

18 

19 col = key[1] 

20 

21 _tb.open(table, nomodify = (mode == 'get')) 

22 colinfo = _tb.getcolkeywords(col) 

23 

24 if mode == 'get': 

25 try: 

26 i = int(hdindex) 

27 if i < 0: 

28 # allowed by python, but... 

29 raise Exception("Illegal index " + str(i)) 

30 

31 value = _tb.getcell(col, i) # throws exception if index too large 

32 except (ValueError, TypeError): # This is almost certainly from 

33 if(_tb.isvarcol(col)): # int('') or int(None). Default 

34 value = _tb.getvarcol(col) # to returning the full column. 

35 else: 

36 value = _tb.getcol(col) 

37 

38 elif mode == 'put': 

39 if(_tb.isvarcol(col)): 

40 _tb.close() 

41 raise Exception("vishead does not yet read/write variably sized columns") 

42 else: 

43 #TODO: Apply colinfo and hdref. 

44 

45 i = None 

46 try: 

47 i = int(hdindex) 

48 except (ValueError, TypeError): 

49 i = None # hdindex is not convertable to an int. 

50 

51 if isinstance(i, int): 

52 # Get full column, change one element, write it back. Not 

53 # efficient but columns used by this task are short 

54 

55 c = _tb.getcol(col) 

56 

57 # Must be careful here: 

58 if isinstance(c[0], _basestring): 

59 # The new hdvalue may be longer than 

60 # the current string. 

61 # numpy arrays do *not* expand flexibly, 

62 # therefore convert to a python list 

63 c = list(c) 

64 # else: For numerical values, 

65 # the size of the array needs to be unchanged, 

66 # otherwise the following _tb.putcol() will fail, 

67 # therefore let c still be a numpy array 

68 

69 c[i] = hdvalue 

70 

71 _tb.putcol(col, c) 

72 else: 

73 _tb.putcol(col, hdvalue) # hdvalue expected to be an array 

74 

75 value = None 

76 else: 

77 _tb.close() 

78 raise Exception("Assertion error") 

79 

80 # casalog.post("Will return: " + value) 

81 

82 _tb.close() 

83 return value, colinfo 

84 

85 

86def keyword_exists(vis, key): 

87 table = vis + '/' + key[0] 

88 col = key[1] 

89 

90 if not os.path.exists(table): 

91 return False 

92 

93 try: 

94 # Throws StandardError if subtable 

95 # does not exist 

96 _tb.open(table) 

97 except: 

98 return False 

99 

100 

101 return (col in _tb.colnames()) 

102 

103def dict2direction_strs(raddict, csys='J2000', units=('rad', 'rad')): 

104 """ 

105 Returns a list containing the values of raddict, sorted by the keys, and 

106 converted to directions if possible. 

107 """ 

108 retlist = [] 

109 rkeys = list(raddict.keys()) 

110 rkeys.sort() 

111 for rk in rkeys: 

112 val = raddict[rk] 

113 if hasattr(val, 'flatten'): # So we don't have to do val[0][0][0] 

114 val = val.flatten() # and val[1][0][0] for arrays. 

115 lon = _qa.formxxx(_qa.toangle('%f%s' % (val[0], units[0])), format='hms') 

116 lat = _qa.formxxx(_qa.toangle('%f%s' % (val[1], units[1])), format='dms') 

117 retlist.append("%s %s %s" % (csys, lon, lat)) 

118 return retlist 

119 

120def getrefunits(d, defunits=None): 

121 """ 

122 Given a dictionary d, this tries to extract a reference system and units 

123 from it. Returns some combination of ('UNKNOWN', defunits) on failure. 

124 """ 

125 rsys = 'UNKNOWN' 

126 try: 

127 if 'MEASINFO' in d: 

128 rsys = d['MEASINFO'].get('Ref', 'UNKNOWN') 

129 except: 

130 casalog.post("d =" + d) 

131 return rsys, d.get('QuantumUnits', defunits) 

132 

133def valref2direction_strs(valreftuple): 

134 """ 

135 Splits a (values, ref_desc) pair and passes it on to dict2direction_strs(). 

136 """ 

137 coordsys, angunits = getrefunits(valreftuple[1], ('rad', 'rad')) 

138 return dict2direction_strs(valreftuple[0], csys=coordsys, units=angunits) 

139 

140def secArray2localDate(secArray, timesys='UTC', timeunit='s'): 

141 """ 

142 Given an array containing a float assumed to be timesys timeunits, returns a 

143 string of it as a local date. 

144 """ 

145 return _qa.time( {'unit': timeunit, 'value': secArray[0]}, 

146 form=['ymd', 'local'] )[0] 

147 

148def valref2localDate(valreftuple): 

149 """ 

150 Splits a (values, ref_desc) pair and passes it on to secArray2localDate(). 

151 """ 

152 timeref, tunits = getrefunits(valreftuple[1], ['s']) 

153 return secArray2localDate(valreftuple[0], timesys=timeref, timeunit=tunits[0]) 

154 

155def strip_r1(scheddict): 

156 """ 

157 Given a dictionary with an 'r1' key, remove the r1 layer. 

158 """ 

159 return scheddict.get('r1', scheddict) 

160 

161def digest(tup): 

162 """ 

163 Given a (val, dict) tuple, returns a string with the boring stuff removed. 

164 """ 

165 t0 = tup[0] 

166 if hasattr(t0, 'shape') and len(t0.shape) < 2: 

167 t0 = list(t0.flatten()) 

168 elif hasattr(t0, 'get'): 

169 t0 = strip_r1(t0) 

170 retval = str(t0) 

171 if len(tup[1].keys()) > 0: 

172 retval += " " + str(tup[1]) 

173 return retval