# -*- coding: utf-8 -*-
#
# This file is part of the ska-mid-cbf-mcs project
#
# Distributed under the terms of the BSD 3-Clause license.
# See LICENSE for more info.
from __future__ import annotations
from typing import cast
from ska_control_model import HealthState
from ska_tango_base.software_bus import AttrSignal, attribute_from_signal
from tango import AttrWriteType
from tango.server import attribute, device_property
from ska_mid_cbf_mcs.base.base_device import CbfDevice
from ska_mid_cbf_mcs.fsp_unit.fsp_unit_component_manager import (
FspUnitComponentManager,
)
[docs]
class FspUnit(CbfDevice):
"""
Fsp Unit TANGO device class
"""
# ----------------------------------------------------------------------- #
# Attributes #
# ----------------------------------------------------------------------- #
# --- Device Properties --- #
FspFQDN = device_property(
doc="Fully Qualified Domain Name (FQDN) of the MCS Fsp device corresponding to this FSP Unit.",
dtype=str,
)
FhsHostControllerFQDNs = device_property(
doc="Fully Qualified Domain Names (FQDNs) for all of the FHSHostController devices for this FSP Unit",
dtype=[str],
)
# --- Device Attributes & Signals --- #
health_state_hw_signal: AttrSignal[HealthState] = AttrSignal(
stored=True, initial_value=HealthState.UNKNOWN
)
"""
Signal for the healthStateHardware attribute.
Values are emitted for this signal whenever a client changes the attribute.
"""
healthStateHardware: attribute_from_signal = attribute_from_signal(
health_state_hw_signal,
access=AttrWriteType.READ,
dtype="DevEnum",
enum_labels=[state.name for state in HealthState],
description="FSP-UNIT Hardware Health State",
)
"""healthStateHardware device attribute"""
# ----------------------------------------------------------------------- #
# Methods #
# ----------------------------------------------------------------------- #
# --- Device Attributes --- #
@attribute(
dtype=str,
label="ResourceStatus",
doc=" JSON string that contains ResourceStatus for all FSPs available to the this unit."
" The JSON string will be formatted according to the ResourceStatus"
" schema found in SKA Telmodel.",
)
def resourceStatus(self: FspUnit) -> str:
"""
Reads and return the ResourceStatus JSON string value.
The JSON string will be formatted according to the ResourceStatus
schema found in SKA Telmodel.
:return: FSP Unit's ResourceStatus as a JSON string
:rtype: str
"""
return self.component_manager.resource_status
# --- Device Initialization --- #
[docs]
def create_component_manager(self: FspUnit) -> FspUnitComponentManager:
self.logger.debug(
"Entering fsp_unit_device.create_component_manager()"
)
return FspUnitComponentManager(
device=self,
device_id=self.DeviceID,
fsp_fqdn=self.FspFQDN,
fhs_host_controller_fqdns=self.FhsHostControllerFQDNs,
logger=self.logger,
health_state_callback=self._update_health_state,
admin_mode_callback=self._admin_mode_perform_action,
communication_state_callback=self._communication_state_changed,
component_state_callback=self._component_state_changed,
)
# --- Run Device Server --- #
def main(*args: str, **kwargs: str) -> int:
"""
Entry point for module.
:param args: positional arguments
:param kwargs: keyword arguments
:return: exit code
"""
return cast(int, FspUnit.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()