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

1# msuvbin task 

2# Copyright (C) 2022 

3# Associated Universities, Inc. Washington DC, USA. 

4# 

5# This library is free software; you can redistribute it and/or modify it 

6# under the terms of the GNU General Public License as published by 

7# the Free Software Foundation; either version 3 of the License, or (at your 

8# option) any later version. 

9# 

10# This library is distributed in the hope that it will be useful, but WITHOUT 

11# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

12# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public 

13# License for more details. 

14# 

15# https://www.gnu.org/licenses/ 

16# 

17# Queries concerning CASA should be submitted at 

18# https://help.nrao.edu 

19# 

20# Postal address: CASA Project Manager 

21# National Radio Astronomy Observatory 

22# 520 Edgemont Road 

23# Charlottesville, VA 22903-2475 USA 

24# 

25# $Id$ 

26# * Created on: Mar 07, 2022 

27# * Author: kgolap 

28# * 

29 

30import os 

31import shutil 

32import typing 

33from typing import Tuple, List, Union, Optional 

34 

35from . import flaghelper as fh 

36 

37from casatasks import casalog 

38from casatools import msuvbinner as msbin 

39from casatools import msmetadata 

40from casatools import ms 

41 

42ms = ms() 

43msmd = msmetadata() 

44from casatasks.private.imagerhelpers.input_parameters import ( 

45 saveparams2last, 

46 determineFreqRange, 

47) 

48 

49 

50@saveparams2last(multibackup=True) 

51def msuvbin( 

52 vis: Optional[str] = None, 

53 field: Optional[str] = None, 

54 spw: Optional[str] = None, 

55 taql: Optional[str] = None, 

56 outputvis: Optional[str] = None, 

57 phasecenter: Optional[str] = None, 

58 imsize: Optional[Union[List[int], List[float], int, float]] = None, 

59 cell: Optional[str] = None, 

60 ncorr: Optional[int] = None, 

61 nchan: Optional[int] = None, 

62 start: Optional[str] = None, 

63 width: Optional[str] = None, 

64 wproject: Optional[bool] = None, 

65 memfrac: Optional[float] = None, 

66 mode: Optional[str] = None, 

67 flagbackup: Optional[bool] = None, 

68) -> None: 

69 fstart = start 

70 fstep = width 

71 casalog.origin("msuvbin ") 

72 if wproject: 

73 casalog.post( 

74 "The wprojection option is extremely slow; you may consider running without it", 

75 "WARN", 

76 "task_msuvbin" 

77 ) 

78 if field == "": 

79 field = "*" 

80 fieldid = 0 

81 fieldid = ms.msseltoindex(vis=vis, field=field)["field"][0] 

82 if isinstance(imsize, (int, float)): 

83 nx = imsize 

84 ny = imsize 

85 else: 

86 nx = imsize[0] 

87 ny = imsize[0] if (len(imsize)==1) else imsize[1] 

88 if phasecenter == "": 

89 msmd.open(vis) 

90 phcen = msmd.phasecenter(fieldid) 

91 msmd.done() 

92 phasecenter = ( 

93 phcen["refer"] 

94 + " " 

95 + str(phcen["m0"]["value"]) 

96 + str(phcen["m0"]["unit"]) 

97 + " " 

98 + str(phcen["m1"]["value"]) 

99 + str(phcen["m1"]["unit"]) 

100 ) 

101 if spw == "": 

102 spw = "*" 

103 if(nchan < 1): 

104 casalog.post( 

105 "nchan has to be larger than 0", "ERROR", 

106 "task_msuvbin" 

107 ) 

108 

109 if (not start) or (start == ""): 

110 (fbeg, fwidth) = determineFreqRange(vis=vis, fieldid=fieldid, spw=spw) 

111 fstart = f"{fbeg}Hz" 

112 if (not width) or (width == ""): 

113 fstep = f"{fwidth/nchan}Hz" 

114 #print(f"fstart={fstart}, fstep={fstep}") 

115 doflag= "write_flags" in mode 

116 if(doflag and flagbackup): 

117 fh.backupFlags(aflocal=None, msfile=vis, prename='msuvbin') 

118 msbinner = msbin.msuvbinner( 

119 phasecenter=phasecenter, 

120 nx=nx, 

121 ny=ny, 

122 ncorr=ncorr, 

123 nchan=nchan, 

124 cellx=cell, 

125 celly=cell, 

126 fstart=fstart, 

127 fstep=fstep, 

128 memfrac=memfrac, 

129 wproject=wproject, 

130 doflag=doflag, 

131 ) 

132 msbinner.selectdata(msname=vis, spw=spw, field=field, taql=taql) 

133 msbinner.setoutputms(outputvis) 

134 msbinner.filloutputms() 

135 del msbinner