Writing scripts for the OET

The Observation Execution Tool (OET) can run observing scripts in a headless non-interactive manner. For efficiency, OET script execution is split into two phases: an initialisation phase and an execution phase. Scripts that are expected to be run by the OET should be structured to have two entry points corresponding to these two phases, as the template below:

Observing script template
 1def init(subarray: int, *args, **kwargs):
 2    # Called by the OET when the script is loaded and initialised by someone
 3    # calling 'oet prepare'. Add your script initialisation code here. Note that
 4    # the target subarray is supplied to this function as the first argument.
 5    pass
 6
 7def main(*args, **kwargs):
 8    # Called by the OET when the prepared script is told to run by someone
 9    # calling 'oet start'. Add the main body of your script to this function.
10    pass

The initialisation phase occurs when the script is loaded and the script’s init function is called (if defined) to perform any preparation and/or initialisation. Expensive and slow operations that can be performed ahead of the main body of script execution can be run in the initialisation phase. Typical actions performed in init are I/O intensive operations, e.g., cloning a git repository, creating multiple Tango device proxies, subscribing to Tango events, etc. When run by the Observation Execution Tool (OET), the init function is passed an integer subarray ID declaring which subarray the control script is intended to control.

Subsequently, at some point a user may call oet start, requesting that the initialised script begin the main body of its execution. When this occurs, the OET calls the script’s main function, which should performs the main function of the script. For an observing script, this would involve the configuration and control of a subarray.

below is the real example script in the scripts folder of this project.

SKA : Allocate Resources and Perform Observation

Allocating resources and performing scans requires communication with TMC CentralNode and TMC SubarrayNode, and targets a specific subarray. This script’s init function pre-applies the subarray ID argument to the main function. Note that this script does not perform any Tango calls directly, but uses ska_oso_scripting.functions.devicecontrol functions to perform all the required Tango interactions (command invocation; event subscriptions; event monitoring).

Resource allocation and perform observation script for an SKA MID/LOW subarray
  1"""
  2Example script for running an SB-driven observation. Last updated
  307/08/24. ConfigureRequest is created from the SB in its entirety.
  45PointScan observations are supported.
  5"""
  6import functools
  7import logging
  8import os
  9
 10from ska_oso_pdm import SBDefinition
 11
 12from ska_oso_scripting import oda_helper
 13from ska_oso_scripting.functions import (
 14    devicecontrol,
 15    messages,
 16    pdm_transforms,
 17    sb,
 18)
 19from ska_oso_scripting.event import user_topics
 20from ska_oso_scripting.functions.devicecontrol import release_all_resources
 21from ska_oso_scripting.functions.devicecontrol.common import ValueTransitionError
 22from ska_oso_scripting.objects import SubArray
 23
 24LOG = logging.getLogger(__name__)
 25FORMAT = "%(asctime)-15s %(message)s"
 26
 27logging.basicConfig(level=logging.INFO, format=FORMAT)
 28
 29
 30def init(subarray_id: int):
 31    """
 32    Initialise the script, binding the sub-array ID to the script.
 33    """
 34    LOG.debug(f"Initializing script {__name__} with subarray_id={subarray_id}")
 35    global main
 36    main = functools.partial(_main, subarray_id)
 37    LOG.info(f"Script bound to sub-array {subarray_id}")
 38
 39
 40def assign_resources(subarray: SubArray, sbi: SBDefinition):
 41    """
 42    assign resources to a target sub-array using a Scheduling Block (SB).
 43    :param subarray: subarray ID
 44    :param sbi: ska_oso_pdm.SBDefinition
 45    :return:
 46    """
 47    LOG.info(
 48        f"Running assign_resources(subarray={subarray.id} sbi.sbd_id={sbi.sbd_id})"
 49    )
 50
 51    cdm_allocation = pdm_transforms.create_cdm_assign_resources_request_from_scheduling_block(
 52            subarray.id, sbi
 53        )
 54
 55    response = devicecontrol.assign_resources_from_cdm(subarray.id, cdm_allocation)
 56    LOG.info(f"Resources Allocated: {response}")
 57
 58    LOG.info("Allocation complete")
 59
 60
 61def observe(subarray: SubArray, sbi: SBDefinition):
 62    """
 63    Observe using a Scheduling Block (SB) and template CDM file.
 64
 65    :param subarray:  SubArray instance containing subarray ID
 66    :param sbi: Instance of a SBDefinition
 67    :return:
 68    """
 69
 70    LOG.info(
 71        f"Starting observing for Scheduling Block: {sbi.sbd_id}, subarray_id={subarray.id})"
 72    )
 73
 74    if not sbi.scan_sequence:
 75        LOG.info(f"No scans defined in Scheduling Block {sbi.sbd_id}. No observation performed.")
 76        return
 77    cdm_configure_requests = (
 78        pdm_transforms.create_cdm_configure_request_from_scheduling_block(sbi)
 79    )
 80
 81    for scan_definition_id in sbi.scan_sequence:
 82        cdm_configs = cdm_configure_requests[scan_definition_id]
 83        for index, cdm_config in enumerate(cdm_configs):
 84            scan_id_string = (f"{scan_definition_id} "
 85                              f"({str(index + 1)}/{str(len(cdm_configs))})" if len(cdm_configs) > 1 else '')
 86            try:
 87                # With the CDM modified, we can now issue the Configure instruction...
 88                LOG.info(f"Configuring subarray {subarray.id} for scan: {scan_id_string}")
 89                messages.send_message(
 90                    user_topics.script.announce,
 91                    msg=f"Configuring subarray {subarray.id} for scan: {scan_id_string}"
 92                )
 93                devicecontrol.configure_from_cdm(subarray.id, cdm_config)
 94            except ValueTransitionError as err:
 95                LOG.error(f"Error configuring subarray: {err}")
 96                messages.send_message(
 97                    user_topics.script.announce,
 98                    msg=f"Error configuring subarray for scan {scan_id_string}"
 99                )
100                raise err
101            else:
102                LOG.info(f"Configuration for scan {scan_id_string} complete")
103                messages.send_message(
104                    user_topics.script.announce,
105                    msg=f"Configuration for scan {scan_id_string} complete"
106                )
107            try:
108                # with configuration complete, we can begin the scan.
109                LOG.info(f"Starting scan: {scan_id_string}")
110                messages.send_message(
111                    user_topics.script.announce,
112                    msg=f"Starting scan: {scan_id_string}"
113                )
114                devicecontrol.scan(subarray.id)
115            except ValueTransitionError as err:
116                LOG.error(f"Error when executing scan: {scan_id_string}: {err}")
117                messages.send_message(
118                    user_topics.script.announce,
119                    msg=f"Error when executing scan: {scan_id_string}"
120                )
121                raise err
122            else:
123                LOG.info(f"Scan {scan_id_string} complete")
124                messages.send_message(
125                    user_topics.script.announce,
126                    msg=f"Scan {scan_id_string} complete"
127                )
128
129    # All scans are complete. Observations are concluded with an 'end'
130    # command.
131    LOG.info(f"End scheduling block: {sbi.sbd_id}")
132    devicecontrol.end(subarray.id)
133
134    LOG.info("Observation script complete")
135
136
137def _main(subarray_id: int, sb_json: str, sbi_id: str):
138    LOG.info(f"Running OS process {os.getpid()}")
139    LOG.info(f"Called with main(subarray_id={subarray_id}, sbi_id={sbi_id})")
140    LOG.debug(f"main() sb_json={sb_json}")
141    sbd: SBDefinition = sb.load_sbd(sb_json)
142    eb_id = oda_helper.create_eb(sbd.telescope, sbi_ref=sbi_id)
143    LOG.info(f"Created Execution Block {eb_id}")
144    subarray = SubArray(subarray_id)
145    assign_resources(subarray, sbd)
146    observe(subarray, sbd)
147    release_all_resources(subarray_id)