Source
iterations_tuple = reduce( lambda acc, v: (acc[0]+v, acc[1] + [acc[0]+v+curr_iters_sum]), new_subsm_rec['iterations'], (0,[]) )
########################################################################3
# _gclean.py
#
# Copyright (C) 2021,2022
# Associated Universities, Inc. Washington DC, USA.
#
# This script is free software; you can redistribute it and/or modify it
# under the terms of the GNU Library General Public License as published by
# the Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public
# License for more details.
#
# You should have received a copy of the GNU Library General Public License
# along with this library; if not, write to the Free Software Foundation,
# Inc., 675 Massachusetts Ave, Cambridge, MA 02139, USA.
#
# Correspondence concerning AIPS++ should be adressed as follows:
# Internet email: aips2-request@nrao.edu.
# Postal address: AIPS++ Project Office
# National Radio Astronomy Observatory
# 520 Edgemont Road
# Charlottesville, VA 22903-2475 USA
#
import os
import asyncio
from functools import reduce
import copy
# from casatasks.private.imagerhelpers._gclean import gclean
class gclean:
'''gclean(...) creates a stream of convergence records which indicate
the convergence quaility of the tclean process. The initial record
describes the initial dirty image.
It is designed for use with the interactive clean GUI, but it could
be used independently. It can be used as a regular generator:
for rec in gclean( vis='refim_point_withline.ms', imagename='test', imsize=512, cell='12.0arcsec',
specmode='cube', interpolation='nearest', nchan=5, start='1.0GHz', width='0.2GHz',
pblimit=-1e-05, deconvolver='hogbom', niter=500, cyclefactor=3, scales=[0, 3, 10] ):
# use rec to decide when to stop, for example to check stopcode or peak residual:
# if (rec[0] > 1) or (min(rec[1][0][0]['peakRes']) < 0.001):
# break
print(rec)
or as an async generator:
async for rec in gclean( vis='refim_point_withline.ms', imagename='test', imsize=512, cell='12.0arcsec',
specmode='cube', interpolation='nearest', nchan=5, start='1.0GHz', width='0.2GHz',
pblimit=-1e-05, deconvolver='hogbom', niter=500, cyclefactor=3, scales=[0, 3, 10] ):
# use rec to decide when to stop
print(rec)
See also: __next__(...) for a description of the returned rec
TODO: do we need to preserve any hidden state between tclean calls for the iterbotsink and/or synthesisimager tools?
'''
def _tclean( self, *args, **kwargs ):
from casatasks import tclean
arg_s = ', '.join( map( lambda a: self._history_filter(len(self._exe_cmds), None, repr(a)), args ) )
kw_s = ', '.join( map( lambda kv: self._history_filter(len(self._exe_cmds), kv[0], "%s=%s" % (kv[0],repr(kv[1]))), kwargs.items()) )
if len(arg_s) > 0 and len(ks_s) > 0:
parameters = arg_s + ", " + kw_s
else:
parameters = arg_s + kw_s
self._exe_cmds.append( "tclean( %s )" % parameters )
return tclean( *args, **kwargs )
def cmds( self ):
return self._exe_cmds
def update( self, msg ):
""" Interactive clean parameters update.
msg: dict with possible keys 'niter', 'cycleniter', 'threshold', 'cyclefactor' and 'mask'
"""
if 'niter' in msg:
try:
self._niter = int(msg['niter'])
except ValueError:
pass
if 'cycleniter' in msg:
try:
self._cycleniter = int(msg['cycleniter'])
except ValueError:
pass
if 'nmajor' in msg:
try:
self._nmajor = int(msg['nmajor'])