# -*- 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.vcc_unit.vcc_unit_component_manager import (
VccUnitComponentManager,
)
[docs]
class VccUnit(CbfDevice):
"""
Vcc Unit TANGO device class
"""
# ----------------------------------------------------------------------- #
# Attributes #
# ----------------------------------------------------------------------- #
# --- Device Properties --- #
ControllerFQDN = device_property(
doc="Fully Qualified Domain Name (FQDN) for the MCS CbfController device.",
dtype=str,
mandatory=True,
)
VccAllBandsFQDNs = device_property(
doc="Fully Qualified Domain Names (FQDNs) for the FHS VCCAllBandsController devices for this VCC Unit.",
dtype=[str],
mandatory=True,
)
FhsHostControllerFQDN = device_property(
doc="Fully Qualified Domain Name (FQDN) for the FHSHostController devices for this VCC 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.
"""
health_states_vcc_signal: AttrSignal[str] = AttrSignal(
stored=True, initial_value=""
)
"""
Signal for the healthStatesVcc 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="VCC-UNIT Hardware Health State",
)
"""healthStateHardware device attribute"""
healthStatesVcc: attribute_from_signal = attribute_from_signal(
health_states_vcc_signal,
access=AttrWriteType.READ,
dtype=str,
description="VCC-UNIT VCC Resource Health States in JSON format",
)
"""healthStatesVcc device attribute"""
# ----------------------------------------------------------------------- #
# Methods #
# ----------------------------------------------------------------------- #
# --- Device Attributes --- #
@attribute(
dtype=str,
label="ResourceStatus",
doc=" JSON string that contains ResourceStatus for all VCCs available to the this unit."
" The JSON string will be formatted according to the ResourceStatus"
" schema found in SKA Telmodel.",
)
def resourceStatus(self: VccUnit) -> 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: VCC Unit's ResourceStatus as a JSON string
:rtype: str
"""
return self.component_manager.resource_status
# --- Device Initialization --- #
[docs]
def create_component_manager(self: VccUnit) -> VccUnitComponentManager:
self.logger.debug(
"Entering vcc_unit_device.create_component_manager()"
)
return VccUnitComponentManager(
device=self,
device_id=self.DeviceID,
mid_cbf_controller_fqdn=self.ControllerFQDN,
fhs_host_controller_fqdn=self.FhsHostControllerFQDN,
vcc_all_bands_fqdns=self.VccAllBandsFQDNs,
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, VccUnit.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()