Configuring the Script

This section describes the configuration of the pointing-offset script, covering the options available in the JSON configuration string supplied to AssignResources, the required environment variables, the helm chart arguments, and setting receive addresses.

Configuration parameters

The pointing-offset script can be configured using the parameters described at Processing block parameters, which can be added to the AssignResources configuration string (see example_configuration_string.json). Below, we describe some of the important points to note about these parameters.

The pointing-offset helm chart requires command arguments (args) and environment variables (env). The command arguments define the options passed to the pointing pipeline.

The basic set of mandatory command arguments used by the script consists of:

eb_id is the execution block ID passed to the script by the Processing Block

num_scans is the expected number of pointing scans with MS data to be processed, defined by the num_scans configuration parameter (with default value of 5)

Additional command line arguments can be provided by setting the additional_args parameter in the Processing Block configuration parameters. This must be a list of strings, e.g.: ["--num_chunks", "1", "--thresh_width", "1.25"]. See the Pointing offset CLI module for a full list of options. The additional arguments add to (or override) the basic mandatory set.

Configuring path to Measurement Sets

The calibration pipeline loads data from the directory defined by the vis-receive data product flow pointing to the written Measurement Set data. The pointing-offset script points the pipeline to this data flow by retrieving the processing block ID of vis-receive.

If there is a vis-receive script running, then the receive_pb_id needs to match the vis-receive script PB, since the MS data will be written into that script’s associated directory. If no vis-receive script is running, then the pointing-offset script will configure the pointing pipeline to load MS data from its own processing block directory. The PB ID of the vis-receive script is passed to the pointing-offset script via the dependencies argument of the Processing Block.

Telescope model data and supplying an RFI mask

You may specify the data you want the pointing pipeline to use from the Telescope Model Data Repository by specifying the sources and file path needed. Currently only loading static RFI masks is supported, which can be set up in the configuration string as the following example shows:

"parameters": {
    "telescope_model": {
        "telmodel_sources": ["gitlab://gitlab.com/ska-telescope/ska-telmodel-data?yan-1618-add-rfi-model#tmdata"],
        "static_rfi_key": "instrument/ska1_mid/static-rfi/rfi_mask.h5"
    }
}

In addition, set the "--apply_mask" additional argument to actually use the mask:

"additional_args": ["--apply_mask", "1"]

Minimum allowed pointing pipeline version

When developing with different versions of the pointing offset pipeline, version needs to be at least the minimum allowable version, which can be found in the pointing_offset.py script as MIN_PIPELINE_VERSION. If the version is incompatible, it is reset to a default version provided by the DEFAULT_PIPELINE_VERSION variable.

Environment variables

The following default set of environment variables are added to the pointing-offset helm chart env parameter by the processing script:

Default set of environment variables used by the pointing-offset script.
Name	Description	Default
`SDP_PB_ID`	Processing block ID (get from SDP)	`None`
`SDP_PROCESSING_SCRIPT`	Tells the pointing pipeline that it is run as an SDP script	`True`
`SDP_CONFIG_HOST`	Host address of the Configuration DB	`127.0.0.1`
`SDP_CONFIG_PORT`	Port of the Configuration DB	`2379`
`SDP_KAFKA_TOPIC`	Kafka topic to send pointing offsets to	`pointing_offset`
`SDP_KAFKA_SERVER`	Kafka server (host)	`localhost:9092`
`SKA_TELMODEL_SOURCES`	A list of telescope model sources as provided by the `telmodel_sources` configuration parameter	`[]`

SDP_PROCESSING_SCRIPT always needs to be “True”, otherwise the pointing pipeline will not check the Configuration DB for scan information, instead it will run in a stand-alone mode, assuming the user provides input data manually.

SDP_CONFIG_HOST and SDP_CONFIG_PORT are needed for establishing a connection to the Configuration DB and are specified directly by SDP when the script is run inside SDP.

SDP_KAFKA_TOPIC is updated using the kafka_topic Processing Block configuration parameter if it exists, or defaults to the values in the table. SDP_KAFKA_SERVER is based on the KAFKA_HOST environment variable passed to the script by the SDP Processing Controller at runtime.

You may specify additional environment variables in the configuration string env parameter. These user defined environment variables are included with the default set and passed to the helm chart. The optional environment variables are:

LOG_LEVEL: logging level, set to INFO by default
WATCHER_TIMEOUT: timeout in seconds for the pipeline when waiting for pointing scans, set to 60 by default
POINTING_METADATA_FILE: name of the pointing metadata yaml file, set to ska-data-product.yaml by default

For example, to lower the logging level, set the following Processing Block configuration parameter:

"env": [
    {
        "name": "LOG_LEVEL",
        "value": "WARN"
    }
]

Script Helm chart

The execution engine for the pointing offset calibration pipeline is the pointing-offset chart. For more customizable chart variables, follow the link to the documentation.

Configuring receive addresses

The pointing-offset script is expected to provide the pointing offset Fully Qualified Domain Name (FQDN) in the receiveAddresses tango attribute of the SDP subarray. The script itself writes the information to its processing block state, and the subarray device takes this information and combines it with the host and port values generated by the vis-receive script, which is then stored in the tango attribute.

The pointing FQDN is stored in the pointing_cal key defined by schema 0.5 (and above) of receive addresses. It is available for every scan_type but only for “visibility”-type beams.

Configuring data flows

ADR-81 describes how data flows within the SDP system. Each flow is configured by specifying its sources and sinks, i.e. where the raw data comes from, what needs to process it and where the processed data goes to at different stages.

The pointing-offset script configures three different data flow objects. Two of these are used by the pointing offset calibration pipeline, one is used by the QueueConnector device.

Flows configuring the pointing calibration pipeline:

data-product: describes where the pointing calibration results stored in HDF5 format need to be saved on storage
data-queue: describes what Kafka queue the output offsets need to be sent to

Flow configuring the QueueConnector device:

tango-flow: describes how to distribute incoming pointing calibration results on dish-specific tango attributes of the QueueConnector device.

Each flow points to another flow object as its source. E.g. the “data” flows use the MeasurementSet(s) written by vis-receive and the source of these is loaded from the data-product flow of vis-receive. The source of the “tango-flow” is the data sent to the Kafka queue, i.e. the “data-queue” flow.

When the script finishes running, the data flow objects (including their states) are removed from the Configuration Database.

Processing block parameters

pydantic settings PointingOffsetParams

Pointing offset script parameters

Show JSON schema

{
   "title": "pointing-offset",
   "description": "Pointing offset script parameters",
   "type": "object",
   "properties": {
      "version": {
         "default": "1.0.1",
         "description": "Version of the pointing offset calibration pipeline used by the helm chart for deployment",
         "pattern": "^[a-zA-Z0-9_\\.-]+$",
         "title": "Pointing offset calibration pipeline Version",
         "type": "string"
      },
      "image": {
         "default": "artefact.skao.int/ska-sdp-wflow-pointing-offset",
         "description": "The OCI image used by the helm chart to launch the pointing offset calibration pipeline",
         "pattern": "^[a-zA-Z0-9_:\\./-]+$",
         "title": "OCI image of the pointing offset calibration pipeline",
         "type": "string"
      },
      "imagePullPolicy": {
         "default": "IfNotPresent",
         "description": "Kubernetes image pull policy used by the helm chart (Always, IfNotPresent, Never)",
         "pattern": "^(Always|IfNotPresent|Never)$",
         "title": "Pull policy of the image",
         "type": "string"
      },
      "command": {
         "const": "pointing-offset",
         "default": "pointing-offset",
         "description": "The command the helm chart deploys the pointing offset pipeline with",
         "title": "Run command for the pointing offset calibration pipeline",
         "type": "string"
      },
      "encoding": {
         "const": "msgpack_numpy",
         "default": "msgpack_numpy",
         "description": "Encoding used by the QueueConnector to decode pointing data from Kafka",
         "title": "Pointing data encoding",
         "type": "string"
      },
      "kube_namespaces": {
         "default": null,
         "description": "Kubernetes namespace where the pointing offset calibration pipeline is deployed. The default is given by the SDP_HELM_NAMESPACE environment variable.",
         "title": "Kubernetes namespace",
         "type": "string"
      },
      "dataProductStorage": {
         "$ref": "#/$defs/DataProductStorage",
         "title": "Data Product storage PVC"
      },
      "kafka_topic": {
         "default": "pointing_offset",
         "description": "Kafka topic that the QueueConnector will use to load pointing offset data from",
         "title": "Kafka topic",
         "type": "string"
      },
      "telescope_model": {
         "$ref": "#/$defs/TelescopeModel",
         "description": "Dictionary of path and file values to specify static data for the pipeline coming from Telescope Model data",
         "title": "Telescope Model reference"
      },
      "env": {
         "description": "User-defined environment variables",
         "items": {
            "$ref": "#/$defs/Env"
         },
         "title": "Environment variables",
         "type": "array"
      },
      "num_scans": {
         "default": 5,
         "description": "Expected number of scans in a pointing observation",
         "title": "Number of scans",
         "type": "integer"
      },
      "additional_args": {
         "default": [
            "--use_source_offset_column"
         ],
         "description": "Additional arguments to be passed to the pointing offset calibration pipeline",
         "items": {
            "type": "string"
         },
         "title": "Additional arguments",
         "type": "array"
      },
      "tango_attribute": {
         "default": "pointing_offset",
         "description": "Prefix to the tango attribute name which will hold the pointing offsets on the QueueConnector device. (i.e. '<tango_attribute>_{dish_id}')",
         "title": "Tango attribute",
         "type": "string"
      },
      "args": {
         "description": "CLI arguments for the pointing offset calibration pipeline.",
         "items": {
            "type": "string"
         },
         "title": "Pointing offset calibration pipeline arguments",
         "type": "array"
      }
   },
   "$defs": {
      "DataProductStorage": {
         "description": "PVC name and path to data product storage",
         "properties": {
            "name": {
               "default": null,
               "description": "PVC name pointing to the data product storage.The default value comes from the SDP_DATA_PVC_NAME environment variable.",
               "title": "PVC name",
               "type": "string"
            },
            "mountPath": {
               "default": "/mnt/data",
               "description": "Path where the data product storage PVC is mounted in the pointing pipeline container",
               "title": "Path where the PVC is mounted in the pointing pipeline container",
               "type": "string"
            }
         },
         "title": "DataProductStorage",
         "type": "object"
      },
      "Env": {
         "description": "User-defined Environment variables",
         "properties": {
            "name": {
               "description": "User-defined environment variable name",
               "title": "Environment variable name",
               "type": "string"
            },
            "value": {
               "description": "User-defined environment variable value",
               "title": "Environment variable value",
               "type": "string"
            }
         },
         "required": [
            "name",
            "value"
         ],
         "title": "Env",
         "type": "object"
      },
      "TelescopeModel": {
         "description": "Telescope Model Information",
         "properties": {
            "telmodel_sources": {
               "description": "Telescope model sources",
               "items": {
                  "type": "string"
               },
               "title": "Telescope Model sources",
               "type": "array"
            },
            "static_rfi_key": {
               "default": "",
               "description": "Filename/key to static RFI mask in Telescope Model",
               "title": "Static RFI key",
               "type": "string"
            }
         },
         "title": "TelescopeModel",
         "type": "object"
      }
   },
   "additionalProperties": false
}

Config:

strict: bool = True
extra: str = forbid
arbitrary_types_allowed: bool = False
validate_assignment: bool = True
title: str = pointing-offset

Fields:

additional_args (list[str])
args (list[str])
command (Literal['pointing-offset'])
dataProductStorage (pointing_offset_params.DataProductStorage)
encoding (Literal['msgpack_numpy'])
env (list[pointing_offset_params.Env])
image (str)
imagePullPolicy (str)
kafka_topic (str)
kube_namespaces (str)
num_scans (int)
tango_attribute (str)
telescope_model (pointing_offset_params.TelescopeModel)
version (str)

field additional_args: list[str] = ['--use_source_offset_column']: Additional arguments to be passed to the pointing offset calibration pipeline

field args: list[str] [Optional]: CLI arguments for the pointing offset calibration pipeline.

field command: Literal['pointing-offset'] = 'pointing-offset': The command the helm chart deploys the pointing offset pipeline with

field dataProductStorage: DataProductStorage [Optional]: PVC name and path to data product storage

field encoding: Literal['msgpack_numpy'] = 'msgpack_numpy': Encoding used by the QueueConnector to decode pointing data from Kafka

field env: list[Env] [Optional]: User-defined environment variables

field image: str = 'artefact.skao.int/ska-sdp-wflow-pointing-offset'

The OCI image used by the helm chart to launch the pointing offset calibration pipeline

Constraints:

pattern = ^[a-zA-Z0-9_:./-]+$

field imagePullPolicy: str = 'IfNotPresent'

Kubernetes image pull policy used by the helm chart (Always, IfNotPresent, Never)

Constraints:

pattern = ^(Always|IfNotPresent|Never)$

field kafka_topic: str = 'pointing_offset': Kafka topic that the QueueConnector will use to load pointing offset data from

field kube_namespaces: str = None: Kubernetes namespace where the pointing offset calibration pipeline is deployed. The default is given by the SDP_HELM_NAMESPACE environment variable.

field num_scans: int = 5: Expected number of scans in a pointing observation

field tango_attribute: str = 'pointing_offset': Prefix to the tango attribute name which will hold the pointing offsets on the QueueConnector device. (i.e. ‘<tango_attribute>_{dish_id}’)

field telescope_model: TelescopeModel [Optional]: Dictionary of path and file values to specify static data for the pipeline coming from Telescope Model data

field version: str = '1.0.1'

Version of the pointing offset calibration pipeline used by the helm chart for deployment

Constraints:

pattern = ^[a-zA-Z0-9_.-]+$

pydantic settings DataProductStorage

PVC name and path to data product storage

Show JSON schema

{
   "title": "DataProductStorage",
   "description": "PVC name and path to data product storage",
   "type": "object",
   "properties": {
      "name": {
         "default": null,
         "description": "PVC name pointing to the data product storage.The default value comes from the SDP_DATA_PVC_NAME environment variable.",
         "title": "PVC name",
         "type": "string"
      },
      "mountPath": {
         "default": "/mnt/data",
         "description": "Path where the data product storage PVC is mounted in the pointing pipeline container",
         "title": "Path where the PVC is mounted in the pointing pipeline container",
         "type": "string"
      }
   }
}

Fields:

mountPath (str)
name (str)

field mountPath: str = '/mnt/data': Path where the data product storage PVC is mounted in the pointing pipeline container

field name: str = None: PVC name pointing to the data product storage.The default value comes from the SDP_DATA_PVC_NAME environment variable.

pydantic settings TelescopeModel

Telescope Model Information

Show JSON schema

{
   "title": "TelescopeModel",
   "description": "Telescope Model Information",
   "type": "object",
   "properties": {
      "telmodel_sources": {
         "description": "Telescope model sources",
         "items": {
            "type": "string"
         },
         "title": "Telescope Model sources",
         "type": "array"
      },
      "static_rfi_key": {
         "default": "",
         "description": "Filename/key to static RFI mask in Telescope Model",
         "title": "Static RFI key",
         "type": "string"
      }
   }
}

Fields:

static_rfi_key (str)
telmodel_sources (list[str])

field static_rfi_key: str = '': Filename/key to static RFI mask in Telescope Model

field telmodel_sources: list[str] [Optional]: Telescope model sources

pydantic settings Env

User-defined Environment variables

Show JSON schema

{
   "title": "Env",
   "description": "User-defined Environment variables",
   "type": "object",
   "properties": {
      "name": {
         "description": "User-defined environment variable name",
         "title": "Environment variable name",
         "type": "string"
      },
      "value": {
         "description": "User-defined environment variable value",
         "title": "Environment variable value",
         "type": "string"
      }
   },
   "required": [
      "name",
      "value"
   ]
}

Fields:

name (str)
value (str)

field name: str [Required]: User-defined environment variable name

field value: str [Required]: User-defined environment variable value