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

1import os 

2import time 

3import copy 

4 

5from casatools import agentflagger 

6from casatasks import casalog 

7 

8def flagmanager( 

9 vis=None, 

10 mode=None, 

11 versionname=None, 

12 oldname=None, 

13 comment=None, 

14 merge=None, 

15 ): 

16 

17 casalog.origin('flagmanager') 

18 aflocal = agentflagger() 

19 

20 try: 

21 if type(vis) == str and os.path.exists(vis): 

22 if mode != 'rename': 

23 aflocal.open(vis) 

24 else: 

25 raise ValueError('Visibility data set not found - please verify the name') 

26 if mode == 'list': 

27 flist = [] 

28 flist = aflocal.getflagversionlist() 

29 

30 # Get the name of the MS and properly add it to the dictionary 

31 if "\nMS : " in flist[0]: 

32 MS = flist.pop(0) 

33 MS = MS.strip("\nMS : ") 

34 

35 flist.remove('main : working copy in main table') 

36 fdict = dict(enumerate(flist)) 

37 fversionsdict = copy.deepcopy(fdict) 

38 for k in fdict: 

39 singleversion = {} 

40 # split each flagversion into versionname and comment 

41 # The below partitioning is a big fragile. If the string contains 

42 # other entries of the character ':', the spliting will fail 

43 (versionname, middle, comment) = fdict[k].partition(':') 

44 singleversion['name'] = versionname.rstrip() 

45 singleversion['comment'] = comment.lstrip() 

46 fversionsdict[k] = singleversion 

47 

48 fversionsdict['MS'] = MS 

49 return fversionsdict 

50 

51 elif mode == 'save': 

52 if versionname == '': 

53 raise ValueError("Illegal empty versionname: ''") 

54 

55 newdir = vis+'.flagversions/flags.'+versionname 

56 if os.path.exists(newdir): 

57 tt = int(time.time()) 

58 tmpname = versionname+'.old.'+str(tt) 

59 casalog.post("Version name \'%s\' already exist. Will rename it to %s"%(versionname,tmpname), 'WARN') 

60 

61 tmpdir = newdir+'.old.'+str(tt) 

62 # Rename existing versionname to old name 

63 os.rename(newdir, tmpdir) 

64 

65 # Edit entry in .flagversions/FLAG_VERSION_LIST 

66 # For realistic usecases, this file is short enough to keep in memory 

67 file = vis + '.flagversions/FLAG_VERSION_LIST' 

68 fd = open(file) 

69 lines = fd.readlines() 

70 fd.close() 

71 

72 for i in range(len(lines)): 

73 if (lines[i])[:len(versionname) + 3] == versionname + ' : ': 

74 lines[i] = tmpname + ' : ' + comment + '\n' 

75 break 

76 

77 fd = open(file, 'w') 

78 fd.writelines(lines) 

79 fd.close() 

80 

81 casalog.post('Save current flagversions to ' + versionname) 

82 aflocal.saveflagversion(versionname=versionname, 

83 comment=comment, merge=merge) 

84 elif mode == 'restore': 

85 if versionname == '': 

86 raise ValueError("Illegal versionname: ''") 

87 

88 casalog.post('Restore flagversions ' + versionname) 

89 aflocal.restoreflagversion(versionname=versionname, 

90 merge=merge) 

91 elif mode == 'delete': 

92 if versionname == '': 

93 raise ValueError("Illegal versionname: ''") 

94 

95 aflocal.deleteflagversion(versionname=versionname) 

96 elif mode == 'rename': 

97 if versionname == '': 

98 raise ValueError("Illegal versionname: ''") 

99 

100 if oldname == '': 

101 raise ValueError("Illegal oldname: ''") 

102 

103 # The directory structure is unlikely to change 

104 olddir = vis + '.flagversions/flags.' + oldname 

105 newdir = vis + '.flagversions/flags.' + versionname 

106 if not os.path.isdir(olddir): 

107 raise ValueError('No such flagversions: ' + str(oldname)) 

108 

109 if os.path.exists(newdir): 

110 raise ValueError('Flagversions ' + str(versionname) + ' already exists!') 

111 

112 casalog.post('Rename flagversions "%s" to "%s"' % (oldname,versionname)) 

113 

114 os.rename(olddir, newdir) 

115 

116 # Edit entry in .flagversions/FLAG_VERSION_LIST 

117 # For realistic usecases, this file is short enough to keep in memory 

118 file = vis + '.flagversions/FLAG_VERSION_LIST' 

119 fd = open(file) 

120 lines = fd.readlines() 

121 fd.close() 

122 

123 for i in range(len(lines)): 

124 if (lines[i])[:len(oldname) + 3] == oldname + ' : ': 

125 lines[i] = versionname + ' : ' + comment + '\n' 

126 break 

127 

128 fd = open(file, 'w') 

129 fd.writelines(lines) 

130 fd.close() 

131 

132 else: 

133 raise ValueError('Unknown mode' + str(mode)) 

134 

135 finally: 

136 aflocal.done()