Source code for ska_pst.lmc.component.pst_simulator

# -*- 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 for base PST Simulator."""

from __future__ import annotations

import logging
from random import randint
from typing import Any, Dict, Generic, TypeVar, cast

from overrides import EnforceOverrides, final

from .monitor_data_handler import MonitorDataStore

DataStore = TypeVar("DataStore", bound=MonitorDataStore)
SubbandMonitorDataType = TypeVar("SubbandMonitorDataType")


class PstSimulator(EnforceOverrides, Generic[SubbandMonitorDataType, DataStore]):
    """
    A base class for simulators used when the BEAM is in simulation mode.

    A simulator class is needed for each subcomponent. This class
    is generic over the subband monitoring data type and a monitoring
    data store.

    This is an abstract class and sub-classes of this are expected to
    implement the :py:meth:`_update` method.
    """

    def __init__(
        self: PstSimulator,
        *,
        data_store: DataStore,
        num_subbands: int | None = None,
        logger: logging.Logger | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialise the base PST simulator.

        :param data_store: the data store to use within the base simulator
        :type data_store: DataStore
        :param num_subbands: number of subbands, if None a random number is used.
        :type num_subbands: int
        :param logger: the logger to use within the simulator, defaults to None
        :type logger: logging.Logger | None, optional
        """
        self.num_subbands = num_subbands if num_subbands else randint(1, 4)
        self.logger = logger or logging.getLogger(__name__)
        self._scan = False
        self._data_store = data_store

    def configure_scan(self: PstSimulator, configuration: dict) -> None:
        """
        Simulate configuring a scan.

        :param configuration: the configuration to be configured
        :type configuration: dict
        """
        raise NotImplementedError("PstSimulator is abstract")

    @final
    def deconfigure_scan(self: PstSimulator) -> None:
        """Simulate deconfiguring of a scan."""
        self._scan = False

    @final
    def start_scan(self: PstSimulator, **kwargs: Any) -> None:
        """
        Simulate start scanning.

        :param: the scan arguments.
        """
        self._scan = True

    @final
    def stop_scan(self: PstSimulator) -> None:
        """Simulate stop scanning."""
        self._scan = False

    @final
    def abort(self: PstSimulator) -> None:
        """Tell the component to abort whatever it was doing."""
        self._scan = False

    def _update(self: PstSimulator) -> None:
        """Simulate the update of monitoring data."""
        raise NotImplementedError("PstSimulator is abstract")

    @final
    def get_subband_data(self: PstSimulator) -> Dict[int, SubbandMonitorDataType]:
        """Get simulated subband data."""
        if self._scan:
            self._update()

        return {**cast(MonitorDataStore, self._data_store)._subband_data}

    def get_env(self: PstSimulator) -> dict:
        """Get the simulated environment variables."""
        return {}