# -*- 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 SMRB process.
The :py:class:`PstSmrbProcessApiSimulator` is used in testing or
simulation.)
"""
from __future__ import annotations
from ska_pst.grpc.lmc.ska_pst_lmc_pb2 import (
BeamConfiguration,
MonitorData,
ScanConfiguration,
SmrbBeamConfiguration,
SmrbScanConfiguration,
)
from ska_pst.lmc.component import MonitorDataCallback
from ska_pst.lmc.smrb.smrb_model import SmrbSubbandMonitorData
__all__ = [
"PstSmrbGrpcApiStrategy",
]
[docs]class PstSmrbGrpcApiStrategy:
"""Implementation of the GrpcApiStrategy for the SMRB subcomponent."""
[docs] def get_beam_configuration_msg(self: PstSmrbGrpcApiStrategy, *, configuration: dict) -> BeamConfiguration:
"""
Get the gRPC BeamConfiguration Protobuf message for SMRB.
:param configuration: the PST beam configuration
:type configuration: dict
:return: the gRPC BeamConfiguration Protobuf message for SMRB.
:rtype: BeamConfiguration
"""
return BeamConfiguration(smrb=SmrbBeamConfiguration(**configuration))
[docs] def get_scan_configuration_msg(self: PstSmrbGrpcApiStrategy, *, configuration: dict) -> ScanConfiguration:
"""
Get the gRPC ScanConfiguration Protobuf message for SMRB.
:param configuration: the PST scan configuration
:type configuration: dict
:return: the gRPC ScanConfiguration Protobuf message for SMRB.
:rtype: ScanConfiguration
"""
return ScanConfiguration(smrb=SmrbScanConfiguration())
[docs] def handle_monitor_response(
self: PstSmrbGrpcApiStrategy, *, 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
"""
smrb_data_stats = data.smrb.data
smrb_weights_stats = data.smrb.weights
# assume the number of written and read are the same for data and weights
# so total written/read size in bytes is the total buffer size * written/read
buffer_size = smrb_data_stats.bufsz + smrb_weights_stats.bufsz
total_written = buffer_size * smrb_data_stats.written
total_read = buffer_size * smrb_data_stats.read
callback(
subband_id=1,
subband_data=SmrbSubbandMonitorData(
buffer_size=buffer_size,
total_written=total_written,
total_read=total_read,
full=smrb_data_stats.full,
num_of_buffers=smrb_data_stats.nbufs,
),
)