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,

        field=field,

        spw=spw,

        array=array,

        scan=scan,

        intent=intent,

        observation=observation,

    )

​

    colnames = _get_col_names(vis)

    # Check if CORRECTED column exists, when requested

    datacolumn = datacolumn.upper()

    if datacolumn == "CORRECTED":

        if "CORRECTED_DATA" not in colnames:

            casalog.post(

                "Input data column CORRECTED_DATA does not exist. Will use DATA", "WARN"

            )

            datacolumn = "DATA"

​

    casalog.post(f"Will use datacolumn = {datacolumn}", "DEBUG")

    config["datacolumn"] = datacolumn

​

    # Call MSTransform framework with tviphaseshift=True

    config["tviphaseshift"] = True

    config["reindex"] = False

    tviphaseshift_config = {"phasecenter": phasecenter}

    config["tviphaseshiftlib"] = tviphaseshift_config

​

    # Configure the tool