Source code for ska_sdp_wflow_pointing_offset.export_data

"""
Functions for exporting pointing offset data to text file
"""

import logging
import os
from typing import Sequence

import numpy
import ska_sdp_datamodels.calibration
from ska_sdp_datamodels.calibration import PointingTable
from ska_sdp_dataproduct_metadata import MetaData, ObsCore

from ska_sdp_wflow_pointing_offset.utils import (
    antenna_and_offsets_reorder,
    convert_dict_to_numpy_array,
)

LOG = logging.getLogger("ska-sdp-pointing-offset")

POINTING_TABLE_HDF5_DESCRIPTION = {
    "time": "The middle timestamp in MJD of the central scan of "
    "the pointing observation. This corresponds to the time at "
    "which the commanded_pointing is calculated. If the central "
    "scan is not found then commanded_pointing cannot be "
    "calculated and the median of the middle timestamps from "
    "all scans is used. In the two-dish mode scenario, this "
    "timestamp for the two sets of observations are stored",
    "frequency": "The central frequency in Hz if fitting to gains "
    "and frequency at the higher end of the band if fitting to "
    "visibilities",
    "weight": "The inverse square of the standard error in the "
    "fitted pointing values in radians",
    "pointing": "The pointing offsets in cross-elevation and "
    " elevation in radians for all antennas in units of radians",
    "expected_width": "The theoretical voltage beam sizes for all "
    "antennas in radians in the horizontal and "
    "vertical co-polarisations",
    "fitted_width": "The fitted voltage beam sizes for all antennas "
    "in radians in the horizontal and vertical co-polarisations",
    "fitted_width_std": "The standard error on the fitted_width in radians",
    "fitted_height": "The fitted Gaussian height for all antennas "
    "in arbitrary units",
    "fitted_height_std": "The standard error on the "
    "fitted_height in arbitrary units",
    "band_type": "Observing band",
    "scan_mode": "Pointing observation mode",
    "track_duration": "How long each scan position was tracked for in seconds",
    "discrete_offset": "Discrete offset of each scan from the central scan, "
    "in degrees (xel-el)",
    "commanded_pointing": "The commanded pointings at the middle "
    "timestamp of the central scan of the pointing observation in radians. "
    "In the two-dish mode scenario, these commanded pointings for the two "
    "sets of observations are stored",
}




def export_pointing_offset_data(filename, offsets, visibility):
    """
    Writes the results of the pointing offset calibration
    to a text file.

    :param filename: Output filename
    :param offsets: A dictionary of the weighted-average pointing
        offsets output by :func:`array_data_func.weighted_average`
    :param visibility: Visibility object

    :return: True-Success, False-Failed
    """
    header = (
        "AntennaName,"
        "CrossElevationOffset,CrossElevationOffsetStd,"
        "ElevationOffset,ElevationOffsetStd,"
        "ExpectedWidthH,ExpectedWidthV,"
        "FittedWidthH,FittedWidthHStd,"
        "FittedWidthV,FittedWidthVStd,"
        "FittedHeight,FittedHeightStd"
    )
    if (
        offsets["antenna_names"]
        != visibility.attrs["configuration"].names.data
    ).all():
        LOG.info(
            "Antenna names and antenna names in Configuration "
            "do not match. Fixing it and their corresponding "
            "fitted parameters before writing to CSV file"
        )
        offsets = antenna_and_offsets_reorder(offsets)

    numpy.savetxt(
        filename,
        convert_dict_to_numpy_array(offsets),
        fmt="%s",
        header=header,
        delimiter=",",
    )





def create_metadata(metadata_file_path):
    """
    Create SDP dataproduct metadata file based on the
    ska-sdp-dataproduct-metadata library

    :param metadata_file_path: file (full path) to metadata file
    """
    if os.path.exists(metadata_file_path):
        metadata = MetaData(path=metadata_file_path)
        metadata.output_path = metadata_file_path

    else:
        metadata = MetaData()
        metadata.output_path = metadata_file_path
        # load basic data from processing block
        metadata.load_processing_block()

        # add obscore attributes
        data = metadata.get_data()
        data.obscore.dataproduct_type = ObsCore.DataProductType.POINTING
        data.obscore.calib_level = ObsCore.CalibrationLevel.LEVEL_0
        data.obscore.obs_collection = (
            f"{ObsCore.SKA}/"
            f"{ObsCore.SKA_MID}/"
            f"{ObsCore.DataProductType.POINTING}"
        )
        data.obscore.access_format = ObsCore.AccessFormat.HDF5
        data.obscore.facility_name = ObsCore.SKA
        data.obscore.instrument_name = ObsCore.SKA_MID

        try:
            metadata.write()
        except MetaData.ValidationError as err:
            LOG.error("Validation failed with error(s): %s", err.errors)
            raise err

    return metadata





def make_results_dir_name(
    results_dir_base: str, scan_ids: Sequence[int | str]
) -> str:
    """Generate the name of the results directory for an SDP Pipeline run.

    :param results_dir_base: base results directory
    :param scan_ids: sequence of scan IDs

    :return: Generated directory name
    """
    first_scan_id = scan_ids[0]
    last_scan_id = scan_ids[-1]
    directory_name = f"scan{first_scan_id}-{last_scan_id}"
    return os.path.join(results_dir_base, directory_name)





def export_pointingtable_to_hdf5(
    pointing_table: PointingTable, results_file_hdf: str | os.PathLike
):
    """
    Export a PointingTable to an HDF5 file.

    Parameters
    ----------
    pointing_table : PointingTable
        The pointing table to be written to disk.
    results_file_hdf : str or os.PathLike
        Path to the output HDF5 file.

    Notes
    -----
    This function is a thin wrapper around
    `ska_sdp_datamodels.calibration.export_pointingtable_to_hdf5` that adds a
    description of the data fields as HDF5 attributes.
    """
    return ska_sdp_datamodels.calibration.export_pointingtable_to_hdf5(
        pointing_table, results_file_hdf, **POINTING_TABLE_HDF5_DESCRIPTION
    )