Source code for ska_pst.send.scan_factory

# -*- coding: utf-8 -*-
#
# This file is part of the SKA PST project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.

"""Module class for managing scans of recorded by PST."""

from __future__ import annotations

import json
import logging
import pathlib
from typing import Any, Type

from ska_control_model import PstProcessingMode
from ska_pst.common.constants import SCAN_CONFIG_FILE_NAME
from ska_pst.common.pst_configuration import convert_csp_config_to_pst_config

from .detected_filterbank_scan import DetectedFilterbankScan
from .flow_through_scan import FlowThroughScan
from .processing_context import ProcessingContext
from .scan import Scan
from .voltage_recorder_scan import VoltageRecorderScan

__all__ = [
    "Scan",
]

_SCAN_TYPE_CLASSES: dict[PstProcessingMode, Type[Scan]] = {
    PstProcessingMode.VOLTAGE_RECORDER: VoltageRecorderScan,
    PstProcessingMode.FLOW_THROUGH: FlowThroughScan,
    PstProcessingMode.DETECTED_FILTERBANK: DetectedFilterbankScan,
}


def create_scan(
    *,
    ctx: ProcessingContext,
    relative_scan_path: pathlib.Path,
    logger: logging.Logger | None = None,
    **kwargs: Any,
) -> Scan | None:
    """
    Construct a Scan sub-class from the processing/observing mode scan_configuration.json file.

    :param ctx: the processing context for the DLM transfer
    :type ctx: ProcessingContext
    :param relative_scan_path: the path of the scan, relative to the ``ctx.local_path``.
    :type relative_scan_path: pathlib.Path
    :return: Subclass of Scan that matches the PST processing mode
    :rtype: Scan | None
    """
    logger = logger or logging.getLogger(__name__)
    scan_config_file = ctx.local_path / relative_scan_path / SCAN_CONFIG_FILE_NAME
    if not scan_config_file.exists():
        return None

    with open(scan_config_file) as json_file:
        csp_config = json.load(json_file)

    pst_scan_config = convert_csp_config_to_pst_config(ctx.telescope_config, csp_config)
    pst_processing_mode: PstProcessingMode = pst_scan_config["pst_processing_mode"]

    cls: Type[Scan]

    try:
        cls = _SCAN_TYPE_CLASSES[pst_processing_mode]
    except KeyError:
        logger.error(f"PST Processing mode {pst_processing_mode.name} type not yet supported")
        return None

    return cls(
        ctx=ctx,
        relative_scan_path=relative_scan_path,
        logger=logger,
        pst_processing_mode=pst_processing_mode,
        pst_scan_config=pst_scan_config,
        **kwargs,
    )