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:
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).
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_oet.event import topics
11from ska_oso_oet.tango import SCAN_ID_GENERATOR
12from ska_oso_pdm import SBDefinition
13from ska_oso_pdm._shared import TelescopeType
14
15from ska_oso_scripting import WORKAROUNDS, oda_helper
16from ska_oso_scripting.functions import (
17 devicecontrol,
18 environment,
19 messages,
20 pdm_transforms,
21 sb,
22)
23from ska_oso_scripting.functions.devicecontrol import release_all_resources
24from ska_oso_scripting.functions.devicecontrol.common import ValueTransitionError
25from ska_oso_scripting.objects import SubArray
26
27LOG = logging.getLogger(__name__)
28FORMAT = "%(asctime)-15s %(message)s"
29
30logging.basicConfig(level=logging.INFO, format=FORMAT)
31
32
33def init(subarray_id: int):
34 """
35 Initialise the script, binding the sub-array ID to the script.
36 """
37 LOG.debug(f"Initializing script {__name__} with subarray_id={subarray_id}")
38 global main
39 main = functools.partial(_main, subarray_id)
40 LOG.info(f"Script bound to sub-array {subarray_id}")
41
42
43def assign_resources(subarray: SubArray, sbi: SBDefinition):
44 """
45 assign resources to a target sub-array using a Scheduling Block (SB).
46 :param subarray: subarray ID
47 :param sbi: ska_oso_pdm.SBDefinition
48 :return:
49 """
50 LOG.info(
51 f"Running assign_resources(subarray={subarray.id} sbi.sbd_id={sbi.sbd_id})"
52 )
53
54 cdm_allocation = pdm_transforms.create_cdm_assign_resources_request_from_scheduling_block(
55 subarray.id, sbi
56 )
57
58 response = devicecontrol.assign_resources_from_cdm(subarray.id, cdm_allocation)
59 LOG.info(f"Resources Allocated: {response}")
60
61 messages.send_message(topics.sb.lifecycle.allocated, sb_id=sbi.sbd_id)
62 LOG.info("Allocation complete")
63
64
65def observe(subarray: SubArray, sbi: SBDefinition):
66 """
67 Observe using a Scheduling Block (SB) and template CDM file.
68
69 :param subarray: SubArray instance containing subarray ID
70 :param sbi: Instance of a SBDefinition
71 :return:
72 """
73
74 LOG.info(
75 f"Starting observing for Scheduling Block: {sbi.sbd_id}, subarray_id={subarray.id})"
76 )
77
78 if not sbi.scan_sequence:
79 LOG.info(f"No scans defined in Scheduling Block {sbi.sbd_id}. No observation performed.")
80 return
81
82 cdm_configure_requests = (
83 pdm_transforms.create_cdm_configure_request_from_scheduling_block(sbi)
84 )
85
86 for scan_definition_id in sbi.scan_sequence:
87 LOG.info(f"Configuring for scan definition: {scan_definition_id}")
88 messages.send_message(topics.sb.lifecycle.observation.started, sb_id=sbi.sbd_id)
89 cdm_configs = cdm_configure_requests[scan_definition_id]
90 for cdm_config in cdm_configs:
91 # Get the scan ID. This is only used for logging, not for any
92 # business logic.
93 scan_id = SCAN_ID_GENERATOR.next()
94 try:
95 # With the CDM modified, we can now issue the Configure instruction...
96 LOG.info(f"Configuring subarray {subarray.id} for scan {scan_id}")
97 messages.send_message(
98 topics.scan.lifecycle.configure.started,
99 sb_id=sbi.sbd_id,
100 scan_id=scan_id,
101 )
102 devicecontrol.configure_from_cdm(subarray.id, cdm_config)
103 except ValueTransitionError as err:
104 LOG.error(f"Error configuring subarray: {err}")
105 messages.send_message(
106 topics.scan.lifecycle.configure.failed,
107 sb_id=sbi.sbd_id,
108 scan_id=scan_id,
109 )
110 messages.send_message(
111 topics.sb.lifecycle.observation.finished.failed, sb_id=sbi.sbd_id
112 )
113 raise err
114 else:
115 messages.send_message(
116 topics.scan.lifecycle.configure.complete,
117 sb_id=sbi.sbd_id,
118 scan_id=scan_id,
119 )
120
121 try:
122 # with configuration complete, we can begin the scan.
123 LOG.info(f"Starting scan {scan_id}")
124 messages.send_message(
125 topics.scan.lifecycle.start, sb_id=sbi.sbd_id, scan_id=scan_id
126 )
127 devicecontrol.scan(subarray.id)
128 except ValueTransitionError as err:
129 LOG.error(f"Error when executing scan {scan_id}: {err}")
130 messages.send_message(
131 topics.scan.lifecycle.end.failed, sb_id=sbi.sbd_id, scan_id=scan_id
132 )
133 messages.send_message(
134 topics.sb.lifecycle.observation.finished.failed, sb_id=sbi.sbd_id
135 )
136 raise err
137 else:
138 messages.send_message(
139 topics.scan.lifecycle.end.succeeded,
140 sb_id=sbi.sbd_id,
141 scan_id=scan_id,
142 )
143
144 # All scans are complete. Observations are concluded with an 'end'
145 # command.
146 messages.send_message(
147 topics.sb.lifecycle.observation.finished.succeeded, sb_id=sbi.sbd_id
148 )
149 LOG.info(f"End scheduling block: {sbi.sbd_id}")
150 devicecontrol.end(subarray.id)
151
152 LOG.info("Observation script complete")
153
154
155def _main(subarray_id: int, sb_json: str, sbi_id: str):
156 LOG.info(f"Running OS process {os.getpid()}")
157 LOG.info(f"Called with main(subarray_id={subarray_id}, sbi_id={sbi_id})")
158 LOG.debug(f"main() sb_json={sb_json}")
159 sbd: SBDefinition = sb.load_sbd(sb_json)
160 eb_id = oda_helper.create_eb(sbd.telescope, sbi_ref=sbi_id)
161 LOG.info(f"Created Execution Block {eb_id}")
162 subarray = SubArray(subarray_id)
163 assign_resources(subarray, sbd)
164 observe(subarray, sbd)
165 release_all_resources(subarray_id)