"""
The ska_oso_objects module holds the object-oriented scripting interface for SKA
telescope and subarray control.
"""
import logging
from typing import Optional, Union
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.functions import devicecontrol
from ska_oso_scripting.functions.devicecontrol.exception import EventTimeoutError
from ska_oso_scripting.types import SubarrayID
LOGGER = logging.getLogger(__name__)
__all__ = [
"SubArray",
"Telescope",
]
[docs]class Telescope:
"""
Telescope represents an SKA telescope.
Telescope methods will control the SKA telescope appropriate to the environment.
That is, calling Telescope.on() inside an SKA MID environment will control the SKA
MID devices, while calling the method inside an SKA LOW environment will control
SKA LOW devices.
"""
def __repr__(self):
return "<Telescope>"
[docs] @staticmethod
def on() -> None:
"""
Power up all telescope devices.
"""
devicecontrol.telescope_on()
[docs] @staticmethod
def off() -> None:
"""
Power down the telescope, setting all devices to standby.
"""
devicecontrol.telescope_off()
[docs]class SubArray:
"""
SubArray represents an SKA telescope subarray.
SubArrays are identified by a numeric ID, which is an integer between 1 and 16.
This ID, accessible as SubArray.id, corresponds to the TMC SubArrayNode of the
same ID that is controlled by invoking methods on a SubArray instance. SubArray
objects with the same numeric ID are considered equal.
SubArrays are used to:
- allocate resources to a subarray
- configure a subarray
- instruct a subarray to commence scanning
- end an SB
- release subarray resources
"""
def __init__(self, identifier: Union[int, str, SubarrayID]):
"""
Create a new SubArray object.
:param identifier: the numeric subarray ID
"""
# As a user-facing class, handle both strings and ints
try:
identifier = int(identifier)
except (TypeError, ValueError) as error:
raise ValueError(
"SubArray identifier must be a positive integer"
) from error
if identifier < 1:
raise ValueError("SubArray identifier must be a positive integer")
self.id = identifier # pylint: disable=invalid-name
def __repr__(self):
return f"<SubArray({self.id})>"
def __eq__(self, other) -> bool:
if isinstance(other, SubArray) and self.id == other.id:
return True
else:
return False
[docs] def assign_from_file(
self, path: str, with_processing: bool = True, timeout: float = None
) -> None:
"""
Allocate resources to a subarray using a JSON file.
This method reads in a file, verifies that the file contents are a valid CDM
resource allocation string, before using that JSON as the basis of a resource
allocation control sequence. This method returns once the targeted subarray
indicates that resources have been allocated or an error occurs.
Note that some JSON content will be replaced (Execution Block IDs, etc.) as
required by the telescope control model, so what is sent to the telescope
control system may not exactly match the contents of the file. To disable this
JSON processing and send the raw contents of the file to the system, set
with_processing to False. Note that you will be responsible for ensuring
JSON correctness, Execution Block ID uniqueness, etc.
:param path: path to the CDM JSON file
:param with_processing: perform JSON validation and processing (default=True)
:param timeout: custom timeout provided while execution of assign resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
devicecontrol.assign_resources_from_file(
self.id, path, with_processing=with_processing, timeout=timeout
)
[docs] def assign_from_cdm(
self, request: AssignResourcesRequest, timeout: float = None
) -> None:
"""
Allocate resources to this subarray using a CDM AssignResourcesRequest object.
Given a well-formed CDM AssignResourcesRequest object, this method will
perform a resource allocation control sequence, returning once the targeted
subarray indicates that resources have been allocated.
:param request: CDM resource allocation request
:param timeout: custom timeout provided while execution of assign resource command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
devicecontrol.assign_resources_from_cdm(self.id, request, timeout=timeout)
[docs] def release(
self,
request: Optional[ReleaseResourcesRequest] = None,
timeout: float = None,
) -> None:
"""
Release subarray resources.
This method blocks until the subarray has released resources or an error
occurs.
By default, all subarray resources will be released. To release some but not
all of the subarray resources (partial release), give this method a CDM
ReleaseResourcesRequest instance defining which resources should be
deallocated.
:param request: optional CDM request object for partial deallocation
:param timeout: custom timeout provided while execution of release resource command's
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
if request is None:
devicecontrol.release_all_resources(self.id, timeout=timeout)
else:
devicecontrol.release_resources(self.id, request=request, timeout=timeout)
[docs] def scan(self, timeout: float = None) -> None:
"""
Start a scan.
This method returns once the scan is complete or an error occurs.
The subarray should have resources allocated and configured before this
command is called.
:param timeout: custom timeout provided while execution of scan command
if systems do not respond within reasonable timescales then method raised EventTimeoutError.
"""
devicecontrol.scan(self.id, timeout=timeout)
[docs] def end(self) -> None:
"""
End Scheduling Block, marking the end of the active observing sequence.
This method returns once the underlying subarray becomes idle or if an error
occurs.
"""
devicecontrol.end(self.id)
[docs] def abort(self) -> None:
"""
Abort subarray activity.
This method returns once the underlying subarray state indicates that all
activity has stopped or an error occurs.
"""
devicecontrol.abort(self.id)
[docs] def reset(self) -> None:
"""
Reset the subarray.
Reset the SubArray from ABORTED or FAULT state to IDLE.
"""
devicecontrol.obsreset(self.id)
[docs] def restart(self) -> None:
"""
Reset the SubArray from ABORTED or FAULT state to EMPTY.
"""
devicecontrol.restart(self.id)