"""
The ska_oso_scripting.api.functional.devicecontrol module contains code that controls
SKA Tango devices, translating from the Tango-free 'science domain' objects to
the Tango-required control system domain.
"""
import logging
from pathlib import Path
import tango
from ska_oso_pdm import SBDefinition
from ska_tmc_cdm.messages.central_node.assign_resources import AssignResourcesRequest
from ska_tmc_cdm.messages.central_node.release_resources import ReleaseResourcesRequest
from ska_tmc_cdm.messages.subarray_node.configure import ConfigureRequest
from ska_oso_scripting.api.functional import execution_block
from ska_oso_scripting.api.functional.devicecontrol import (
resource_control,
subarray_control,
subarray_recovery,
telescope_control,
)
from ska_oso_scripting.engineering.scripting_workarounds import TEMP_WORKAROUNDS
from ska_oso_scripting.types import SubarrayID
LOGGER = logging.getLogger(__name__)
__all__ = [
"abort",
"assign_resources_from_cdm",
"assign_resources_from_file",
"assign_resources_from_json",
"configure_from_cdm",
"configure_subarray_quality_monitor",
"configure_from_file",
"configure_from_json",
"end",
"init_qa_wait",
"obsreset",
"release_all_resources",
"release_resources",
"restart",
"scan",
"telescope_off",
"telescope_on",
]
[docs]
def init_qa_wait(initial_state: bool) -> None:
"""
Initialise ADR-111 QA readiness waiting state and subscribe persistent
handlers for operator enable/disable messages.
:param initial_state: initial enabled/disabled state for QA waiting
"""
subarray_control.init_qa_wait(initial_state)
[docs]
@execution_block.capture_request_response
def assign_resources_from_file(
subarray_id: SubarrayID,
request_path: str,
with_processing: bool = True,
timeout: float = None,
):
"""
Allocate resources to a sub-array using a JSON file.
:param subarray_id: the ID of the sub-array to control
:param request_path: JSON file path
:param with_processing: False if JSON should be passed through
to CentralNode directly without any validation or processing
:param timeout: custom timeout provided while execution of assign resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
resource_control.assign_resources(
subarray_id, Path(request_path), with_processing, timeout=timeout
)
[docs]
@execution_block.capture_request_response
def assign_resources_from_cdm(
subarray_id: SubarrayID,
request: AssignResourcesRequest,
timeout: float = None,
):
"""
Allocate resources to a sub-array using a CDM object.
:param subarray_id: the id of the sub-array to allocate the resources
:param request: the CDM AssignResourcesRequest object
:param timeout: custom timeout provided while execution of assign resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
resource_control.assign_resources(subarray_id, request, timeout=timeout)
[docs]
@execution_block.capture_request_response
def assign_resources_from_json(
subarray_id: SubarrayID,
request: str,
with_processing: bool = True,
timeout: float = None,
):
"""
Allocate resources to a sub-array using a JSON string representation of
AssignResources request. Validity of given JSON string will be checked
by default unless with_processing is explicitly set to False. In this case
the string will be passed directly to central node without any processing.
This function will block until the sub-array with the given ID has been
allocated the resources and has transitioned to obsState IDLE.
:param subarray_id: the id of the sub-array to allocate the resources
:param request: the JSON string to AssignResources with
:param with_processing: False if JSON should be passed through
to CentralNode directly without any validation or processing
:param timeout: custom timeout provided while execution of assign resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
resource_control.assign_resources(
subarray_id, request, with_processing, timeout=timeout
)
[docs]
@execution_block.capture_request_response
def release_all_resources(subarray_id: SubarrayID, timeout: float = None):
"""
Release resources on the sub-array with the given ID. If release_all is set to False,
ReleaseResourcesRequest CDM object specified in request parameter will be sent to
the sub-array. If release_all is set to True, the request will not be used.
This function blocks until resources have been released on the given sub-array
and its obsState has transitioned to EMPTY.
:param subarray_id: the sub-array to control
:param timeout: custom timeout provided while execution of release resorce command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
resource_control.release_resources(subarray_id, release_all=True, timeout=timeout)
[docs]
@execution_block.capture_request_response
def release_resources(
subarray_id: SubarrayID,
request: ReleaseResourcesRequest,
timeout: float = None,
):
"""
Release resources on the sub-array with the given ID. If release_all is set to False,
ReleaseResourcesRequest CDM object specified in request parameter will be sent to
the sub-array. If release_all is set to True, the request will not be used.
This function blocks until resources have been released on the given sub-array
and its obsState has transitioned to EMPTY.
:param subarray_id: the sub-array to control
:param request: CDM ReleaseResourcesRequest object specifying resources to release.
Only required if release_all is False
:param timeout: custom timeout provided while execution of release resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
resource_control.release_resources(
subarray_id, release_all=False, request=request, timeout=timeout
)
[docs]
@execution_block.capture_request_response
def scan(subarray_id: SubarrayID, timeout: float = None):
"""
Execute a scan.
:param subarray_id: the sub-array to control
:return: the response from sending the command to configure sub-array
:param timeout: custom timeout provided while execution of scan command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
subarray_control.scan(subarray_id, timeout=timeout)
[docs]
@execution_block.capture_request_response
def end(subarray_id: SubarrayID):
"""
Send the 'end' command to the SubArrayNode, marking the end of the
current observation.
:param subarray_id: the subarray to command
:return:
"""
subarray_control.end(subarray_id)
[docs]
@execution_block.capture_request_response
def telescope_on():
"""
Start up the telescope.
This command powers up Dishes that are currently in standby.
:return:
"""
telescope_control.telescope_on()
[docs]
@execution_block.capture_request_response
def telescope_off(final_state: tango.DevState = tango.DevState.STANDBY):
"""
Instruct telescope devices to switch to STANDBY mode.
If expect_standby is True (default) then this function will wait for the
telescope to reach STANDBY state. Otherwise, this function will wait for
the OFF state to be reached.
Background:
At the time of writing (PI15), CentralNode has new behaviour where it
reports the aggregated state of the systems it controls. Hence. it is not
yet clear which state CentralNode will reach when commanded to turn the
telescope off, as this state could change dependning on what subsystems are
connected to TMC. For instance, if CSP reports STANDBY, CentralNode will
report STANDBY. If CSP was not present, CentralNode might report OFF.
:param final_state: Final Tango state to expect (default=STANDBY)
:return:
"""
telescope_control.telescope_off(final_state)
[docs]
@execution_block.capture_request_response
def abort(subarray_id: SubarrayID):
"""
Send the 'abort' command to the SubArrayNode, halt the subarray
activity.
:param subarray_id: the subarray to command
:return:
"""
subarray_recovery.abort(subarray_id)
[docs]
@execution_block.capture_request_response
def obsreset(subarray_id: SubarrayID):
"""
Send the 'ObsReset' command to the SubArrayNode, which resets
the SubArrayNode state to IDLE.
:param subarray_id: the subarray to command
:return:
"""
subarray_recovery.obsreset(subarray_id)
[docs]
@execution_block.capture_request_response
def restart(subarray_id: SubarrayID):
"""
Send the 'restart' command to the SubArrayNode which sets
the SubArrayNode from ABORTED or FAULT state to EMPTY.
:param subarray_id: the subarray to command
:return:
"""
subarray_recovery.restart(subarray_id)