"""
This module contains code related to running the OSO simulators in a test Tango
context.
"""
import logging
from ska_control_model import ObsState
from ska_ser_logging import configure_logging
from ska_tango_testing.harness import TangoTestHarness
from ska_tango_transducer.transducer import Transducer
from ska_oso_tmcsim import (
CbfSimulator,
CentralNode,
SubArrayNode,
get_cbf_trl,
get_centralnode_trl,
get_subarraynode_trl,
)
from ska_oso_tmcsim.cspleafnode import CSPLeafNode, get_csp_leaf_node_trl
from ska_oso_tmcsim.sdpcontrollernode import SDPControllerNode, get_sdp_controller_trl
from ska_oso_tmcsim.subarrayleafnodemccs import (
SubarrayLeafNodeMccs,
get_subarray_leaf_node_mccs_trl,
)
configure_logging(logging.DEBUG)
LOGGER = logging.getLogger(__name__)
[docs]
class TMCSimTestHarness:
"""
TMCSimTestHarness is an integration test harness for OSO's TMC Simulator.
TMCSimTestHarness can populate a Tango test context with simulations of
- TMC CentralNode
- TMC SubArrayNode
- TMC Subarray Quality Monitor
- TMC MCCS subarray leaf node
- CSP Leaf node
- CSP CBF, for the Quality Monitor to react to
- SDP controller
"""
def __init__(self, domain: str):
"""
Create a new TMCSimTestHarness.
:param domain: Tango domain
"""
self._tango_test_harness = TangoTestHarness()
self._domain = domain
[docs]
def add_central_node(self):
"""
Make CentralNode available within the test context.
"""
self._tango_test_harness.add_device(
device_name=get_centralnode_trl(self._domain),
device_class=CentralNode,
domain=self._domain,
)
[docs]
def add_csp_leaf_node(self, subarray_id: int = 1):
"""
Make CSP leaf node available within the test context.
:param subarray_id: The subarray ID for which the leaf node should be added. Defaults to 1.
"""
self._tango_test_harness.add_device(
device_name=get_csp_leaf_node_trl(self._domain, subarray_id),
device_class=CSPLeafNode,
domain=self._domain,
)
[docs]
def add_sdp_controller_node(self, sdp_version=None):
"""
Make an SDP controller node available within the test context.
"""
if sdp_version is None:
sdp_version = '{"version": "1.0.0"}'
device_props = dict(sdp_version=sdp_version)
self._tango_test_harness.add_device(
device_name=get_sdp_controller_trl(self._domain),
device_class=SDPControllerNode,
domain=self._domain,
**device_props,
)
[docs]
def add_subarray(
self, subarray_id: int, initial_obsstate: ObsState = ObsState.EMPTY
):
"""
Add a subarray to the test harness.
The subarray obsState will be set to the default obsState of EMPTY
unless overridden.
:param subarray_id: The subarray ID for which the subarray should be added.
:param initial_obsstate: The initial obsState of the subarray. Defaults to ObsState.EMPTY.
"""
subarray_trl = get_subarraynode_trl(self._domain, subarray_id)
device_props = dict(initial_obsstate=initial_obsstate.value)
self._tango_test_harness.add_device(
device_name=subarray_trl, device_class=SubArrayNode, **device_props
)
[docs]
def add_cbf_device(self, subarray_id: int, ramp_duration=None):
"""
Make a CBF simulator device available within the test context.
:param subarray_id: The subarray ID for the CBF simulator.
:param ramp_duration: Optional ramp duration in seconds.
"""
cbf_trl = get_cbf_trl(self._domain, subarray_id)
subarray_trl = get_subarraynode_trl(self._domain, subarray_id)
device_props = dict(SubarrayNodeTRL=subarray_trl)
if ramp_duration is not None:
device_props["RampDuration"] = ramp_duration
self._tango_test_harness.add_device(
device_name=cbf_trl,
device_class=CbfSimulator,
**device_props,
)
[docs]
def add_subarray_quality_monitor(self, subarray_id: int):
"""
Make a SubarrayQualityMonitor (Transducer) available within the test context.
The device is added with no properties. At runtime, observing scripts
configure it to subscribe to CBF ``processorsReadyPercent`` and derive
``readyToScan`` (``processorsReadyPercent == 100``).
:param subarray_id: The subarray ID for the quality monitor.
"""
trl = f"{self._domain}/subarray-quality-monitor/{subarray_id:02}"
self._tango_test_harness.add_device(
device_name=trl,
device_class=Transducer,
)
[docs]
def add_mccs_subarray_leaf_node(self, subarray_id: int):
"""
Make a MCCS Subarray Leaf Node available within the test context.
:param subarray_id: The subarray ID for the leaf node.
"""
self._tango_test_harness.add_device(
device_name=get_subarray_leaf_node_mccs_trl(subarray_id),
device_class=SubarrayLeafNodeMccs,
)
def __enter__(self):
return self._tango_test_harness.__enter__()
def __exit__(self, exc_type, exception, trace):
return self._tango_test_harness.__exit__(exc_type, exception, trace)