# -*- 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 providing the API to be communicate with the RECV process.
The :py:class:`PstReceiveProcessApiSimulator` is used in testing or
simulation mode, while the :py:class:`PstReceiveProcessApiGrpc` is used
to connect to a remote application that exposes a gRPC API.
"""
from __future__ import annotations
import copy
from ska_pst.grpc.lmc.ska_pst_lmc_pb2 import (
BeamConfiguration,
MonitorData,
ReceiveBeamConfiguration,
ReceiveMonitorData,
ReceiveScanConfiguration,
ReceiveSubbandResources,
ScanConfiguration,
)
from ska_pst.lmc.component import MonitorDataCallback
from ska_pst.lmc.component.grpc_lmc_client import protobuf_to_ascii_header
from ska_pst.lmc.receive.receive_model import ReceiveData
from ska_pst.lmc.receive.receive_util import generate_recv_configure_scan_request
__all__ = [
"PstReceiveGrpcApiStrategy",
]
[docs]class PstReceiveGrpcApiStrategy:
"""Implementation of the GrpcApiStrategy for the RECV subcomponent."""
[docs] def get_beam_configuration_msg(
self: PstReceiveGrpcApiStrategy, *, configuration: dict
) -> BeamConfiguration:
"""
Get the gRPC BeamConfiguration Protobuf message for RECV.
:param configuration: the PST beam configuration
:type configuration: dict
:return: the gRPC BeamConfiguration Protobuf message for RECV.
:rtype: BeamConfiguration
"""
configuration = copy.deepcopy(configuration)
del configuration["subband"]["start_centre_freq_mhz"]
subband_resources = ReceiveSubbandResources(**configuration["subband"])
return BeamConfiguration(
receive=ReceiveBeamConfiguration(subband_resources=subband_resources, **configuration["common"])
)
[docs] def get_scan_configuration_msg(
self: PstReceiveGrpcApiStrategy, *, configuration: dict
) -> ScanConfiguration:
"""
Get the gRPC ScanConfiguration Protobuf message for RECV.
:param configuration: the PST scan configuration
:type configuration: dict
:return: the gRPC ScanConfiguration Protobuf message for RECV.
:rtype: ScanConfiguration
"""
return ScanConfiguration(
receive=ReceiveScanConfiguration(**generate_recv_configure_scan_request(**configuration))
)
[docs] def handle_monitor_response(
self: PstReceiveGrpcApiStrategy, *, data: MonitorData, callback: MonitorDataCallback
) -> None:
"""
Handle the gRPC monitoring data response.
:param data: the gRPC/Protobuf monitoring data message from server
:type data: MonitorData
:param callback: the callback used to update the LMC subcomponent model
:type callback: MonitorDataCallback
"""
receive_monitor_data: ReceiveMonitorData = data.receive
ascii_header = protobuf_to_ascii_header(receive_monitor_data.header)
callback(subband_id=1, subband_data=ascii_header.get_dataclass_obj(ReceiveData))