casa6

from typing import Dict, Optional, Tuple, Union

​

import numpy as np

from .mstools import write_history

from casatools import table, ms, mstransformer

from casatools import measures as me

from casatasks import casalog

from .parallel.parallel_data_helper import ParallelDataHelper

​

​

def phaseshift(

    vis: str,

    outputvis: str,

    keepmms: bool,

    field: Optional[str],

    spw: Optional[str],

    scan: Optional[str],

    intent: Optional[str],

    array: Optional[str],

    observation: Optional[str],

    datacolumn: Optional[str],

    phasecenter: Union[str, dict],

):

    """

    Changes the phase center for either short or large

    offsets/angles w.r.t. the original

    """

    casalog.origin("phaseshift")

​

    if len(phasecenter) == 0:

        raise ValueError("phasecenter parameter must be specified")

    # Initiate the helper class

    pdh = ParallelDataHelper("phaseshift", locals())

​

    # Validate input and output parameters

    try:

        pdh.setupIO()

    except Exception as instance:

        casalog.post(str(instance), "ERROR")

        raise RuntimeError(str(instance))

​

    # Input vis is an MMS

    if pdh.isMMSAndNotServer(vis) and keepmms:

        if not pdh.validateInputParams():

            raise RuntimeError("Unable to continue with MMS processing")

​

        pdh.setupCluster("phaseshift")

​

        # Execute the jobs

        try:

            pdh.go()

        except Exception as instance:

            casalog.post(str(instance), "ERROR")

            raise RuntimeError(str(instance))

        return

​

    # Actual task code starts here

    # Gather all the parameters in a dictionary.

    config = {}

​

    config = pdh.setupParameters(

        inputms=vis,

        outputms=outputvis,