Coverage for /wheeldirectory/casa-6.7.0-12-py3.10.el8/lib/py/lib/python3.10/site-packages/casatasks/private/task_msuvbin.py: 28%
50 statements
« prev ^ index » next coverage.py v7.6.4, created at 2024-11-01 07:19 +0000
« prev ^ index » next coverage.py v7.6.4, created at 2024-11-01 07:19 +0000
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# *
30import os
31import shutil
32import typing
33from typing import Tuple, List, Union, Optional
35from . import flaghelper as fh
37from casatasks import casalog
38from casatools import msuvbinner as msbin
39from casatools import msmetadata
40from casatools import ms
42ms = ms()
43msmd = msmetadata()
44from casatasks.private.imagerhelpers.input_parameters import (
45 saveparams2last,
46 determineFreqRange,
47)
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 )
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