"""
Used for checking CSP configuration strings for conformance
"""
from inspect import cleandoc
from schema import And, Optional, Or, Regex, Schema
from .._common import TMSchema, mk_if, split_interface_version
from ..pst.schema import _get_pst_config_schema
from . import validators as validators
from . import version as csp_version
from .common_schema import (
_get_common_config_schema_without_band,
add_common_frequency_band,
use_camel_case,
)
from .pss_schema import get_pss_config_schema
[docs]def get_vlbi_config_schema(version: str, strict: bool):
"""VLBI specific items
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON schema for the MID.CBF VLBI configuration.
"""
return TMSchema.new(
"VLBI config",
version,
strict,
schema={Optional("dummy_param"): str},
description=cleandoc(
"""
Very Long Baseline Interferometry specific parameters. To
be borrowed from IICD This section contains the parameters
relevant only for VLBI. This section is forwarded only to
CSP subelement.
"""
),
as_reference=True,
)
[docs]def get_fsp_config_schema(version: str, strict: bool):
"""Frequency slice processor configuration schema
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON schema for the MID.CBF FSP configuration.
"""
if_strict = mk_if(strict)
camel_case = use_camel_case(version)
elems = TMSchema.new("FSP config", version, strict, as_reference=True)
elems.add_field(
("fsp_id" if camel_case else "fspID"),
int,
check_strict=lambda n: n >= 1 and n <= 27,
)
elems.add_field(
("function_mode" if camel_case else "functionMode"),
str,
check_strict=Or("CORR", "PSS-BF", "PST-BF", "VLBI"),
)
elems.add_opt_field(
"receptors",
[
Regex(
r"""^(SKA(00[1-9]|0[1-9][0-9]|1[0-2][0-9]|13[0-3]))|
(MKT(0[0-5][0-9]|06[0-3]))$"""
)
if strict
else str
],
description=cleandoc(
"""
Optionally a subset of receptors to be correlated can be
specified. If not specified, all receptors that belong to
the subarray are cross-correlated (i.e. visibilities for
all the baselines in the subarray are generated and
transmitted to SDP).
Valid receptor IDs include:
SKA dishes: "SKAnnn", where nnn is a zero padded integer in the
range of 001 to 133.
MeerKAT dishes: "MKTnnn", where nnn is a zero padded integer in the
range of 000 to 063.
"""
),
)
elems.add_field(
("frequency_slice_id" if camel_case else "frequencySliceID"),
int,
check_strict=lambda n: n >= 1 and n <= 26,
description=cleandoc(
"""
Frequency Slice to be processed on this FSP (valid range
depends on the Frequency Band).
"""
),
)
elems.add_field(
("zoom_factor" if camel_case else "corrBandwidth"),
int,
check_strict=lambda n: n >= 0 and n <= 6,
description=cleandoc(
"""
Bandwidth to be correlated calculated as FSBW/2n, where n is in
range [0..6].
When n=0 the full Frequency Slice bandwidth is correlated.
BW > 0 implies ‘Zoom Window’ configuration; the spectral
Zoom Window tuning must be specified.
"""
),
)
elems.add_opt_field(
("zoom_window_tuning" if camel_case else "zoomWindowTuning"),
int,
description=cleandoc(
"""
The Zoom Window tuning provided in absolute terms as RF
center frequency. Based on that, CSP_Mid calculates
tuning within the data stream received from the receptor.
Must be selected so that the entire Zoom Window is within
the Frequency Slice. If partially out of the FS a warning
is generated. If completely outside of the FS an
exception is generated.
Step size <= 0.01MHz.
The Frequency Band Offset can be used to shift the entire
observed band in order to accommodate a Zoom Window that
spans across a Frequency Slice boundary.
"""
),
)
elems.add_field(
("integration_factor" if camel_case else "integrationTime"),
(
And(int, if_strict(validators.validate_integration_factor))
if camel_case
else 1400
),
description="Integration time for the "
"correlation products, defines multiple of 140 milliseconds.",
)
elems.add_opt_field(
("channel_averaging_map" if camel_case else "channelAveragingMap"),
[And([int], if_strict(lambda lst: len(lst) == 2))],
description=cleandoc(
"""
Table of up to 20 x 2 integers. Each of entries contains:
* Start channel ID, and
* averaging factor.
Explanation: Each FSP produces 14880 (TBC) fine channels
across the correlated bandwidth (Frequency Slice or Zoom
Window). Channels are evenly spaced in frequency.
TM shall provide the table that for each FSP and each
group of 744 channels (there are 20 groups per FSP)
indicates the channel averaging factor. More precisely,
for each group the TMC provided table specifies:
* the channel ID (integer) of the first channel, and
* the averaging factor, as follows:
* 0 means do not send channels to SDP,
* 1 means no averaging,
* 2 means average two adjacent channels,
* 3 means average three adjacent channels,
and so on.
If no entry is present for an FSP, the averaging settings
of the previous FSP are still applicable.
"""
),
)
elems.add_opt_field(
("channel_offset" if camel_case else "fspChannelOffset"),
int,
description=cleandoc(
"""
Channel ID to use for visibilities of the first channel produced
by this FSP. For example, if the channel offset is 5000 the first
channel group would span IDs 5000-5743.
Note that this offset does not apply to channel maps in
this structure (such as `channelAveragingMap` or
`outputHost`).
"""
),
)
elems.add_opt_field(
("output_link_map" if camel_case else "outputLinkMap"),
[And([int, str], if_strict(lambda lst: len(lst) == 2))],
description=cleandoc(
"""
Output links to emit visibilities on for every channel, given as a
list of start channel ID to link ID. Where no value is
given for concrete channel, the previous value should be
used.
"""
),
)
elems.add_opt_field(
("output_host" if camel_case else "outputHost"),
[And([int, str], if_strict(lambda lst: len(lst) == 2))],
description=cleandoc(
"""
Output host to send visibilities to for every channel, given as a
list of start channel ID to host IP addresses in
dot-decimal notation. Where no value is given for a concrete
channel, the previous value should be used.
"""
),
)
elems.add_opt_field(
("output_port" if camel_case else "outputPort"),
[And([int], if_strict(lambda lst: len(lst) >= 2 and len(lst) <= 3))],
description=cleandoc(
"""
Output port to send visibilities to for every channel, given as a
list of start channel ID to port number. Where no value is
given for a concrete channel, the previous value should be
used.
"""
),
)
elems.add_opt_field(
("output_mac" if camel_case else "outputMac"),
[And([int, str], if_strict(lambda lst: len(lst) == 2))],
description=cleandoc(
"""
Output MAC address to send visibilities to for every channel,
given as a list of start channel ID to IEEE 802 MAC
addresses. Where no value is given for a concrete channel,
the previous value should be used.
"""
),
)
return elems
[docs]def get_cbf_config_schema(version: str, strict: bool) -> Schema:
"""Correlator and Beamformer configuration schema
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the MID.CBF configuration.
"""
# Sub-schemas
searchWindow = get_search_window_config_schema(version, strict)
# cbf specific items
camel_case = use_camel_case(version)
elems = TMSchema.new(
"CBF config",
version,
strict,
description=cleandoc(
"""
Correlator and Beamformer specific parameters. This section
contains the parameters relevant only for CBF
sub-element. This section is forwarded only to CBF
subelement. Most of it to be borrowed from IICD
"""
),
as_reference=True,
)
elems.add_opt_field(
(
"frequency_band_offset_stream1"
if camel_case
else "frequencyBandOffsetStream1"
),
int,
description=cleandoc(
"""
Optionally, an offset can be specified so that the entire observed
band is shifted (to accommodate a Zoom Window that crosses
a ‘natural’ Frequency Slice boundary). If specified,
applies for all the receptors in the sub-array.
Bands 1, 2, 3 and 4: input from the receptor consists of a
single data stream; the Frequency Band Offset (FBO) should
be specified for Stream 1 only.
Bands 5a and 5b: input from the receptor consists of two
data streams; the FBO can be specified for each stream
independently. Note: For Band 5a and 5b the frequency
shift is performed by the receptor (DISH).
Note: This is optional and does not need to be implemented
in PI3, but would be great for demo; if Team Buttons is
looking for opportunities to showcase interesting GUIs,
Zoom Windows are perfect opportunity (would require TMC
and CSP to support these two parameters, corrBandwidth
values > 0 and zoom window tuning.)
"""
),
)
elems.add_opt_field(
(
"frequency_band_offset_stream2"
if camel_case
else "frequencyBandOffsetStream2"
),
int,
description="See `frequencyBandOffsetStream1`",
)
elems.add_opt_field(
(
"delay_model_subscription_point"
if camel_case
else "delayModelSubscriptionPoint"
),
str,
description=cleandoc(
"""
FQDN of TMC.DelayModel TANGO attribute which exposes
delay values for all the dishes assigned to a Subarray
in JSON format. Delay values are updated every 10 seconds.
"""
),
)
elems.add_opt_field(
(
"doppler_phase_corr_subscription_point"
if camel_case
else "dopplerPhaseCorrSubscriptionPoint"
),
str,
description=cleandoc(
"""
The same model applies for all receptors that belong to
the subarray. Delivered by TMC using publish-subscribe
mechanism (see ICD Section 3.8.8.5.3). The Doppler phase
correction, by default, applies only to the CSP_Mid
Processing Mode Correlation; optionally may apply to other
Processing Modes as well.
"""
),
)
elems.add_opt_field(
("rfi_flagging_mask" if camel_case else "rfiFlaggingMask"),
{},
description=cleandoc(
"""
Specified as needed in advance of the scan start and/or
during the scan. Delivered using publish-subscribe
mechanism (see ICD Section 3.8.8.5.7).
"""
),
)
elems.add_field("fsp", [get_fsp_config_schema(version, strict)])
elems.add_opt_field("vlbi", get_vlbi_config_schema(version, strict))
elems.add_opt_field("search_window", [searchWindow])
return elems
[docs]def get_search_window_config_schema(version: str, strict: bool) -> Schema:
"""SearchWindow configuration schema
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the MID.CBF SearchWindow configuration.
"""
# Search window schema
camel_case = use_camel_case(version)
elems = TMSchema.new(
"Search window config",
version,
strict,
description=cleandoc(
"""
Up to two 300 MHz Search Windows can be optionally configured and
used as input for Transient Data Capture and/or Pulsar Search
beam-forming.
"""
),
as_reference=True,
)
elems.add_field(
("search_window_id" if camel_case else "searchWindowID"),
int,
description="Identifier of the 300MHz Search Window. "
"Unique within a sub-array.",
)
elems.add_field(
("search_window_tuning" if camel_case else "searchWindowTuning"),
int,
description=cleandoc(
"""
The Search Window tuning is provided in absolute terms as RF
center frequency. The Search Window must be placed
within the observed band. If partially out of the
observed Band a warning is generated. If completely
outside of the observed Band an exception is
generated.
"""
),
)
elems.add_field(
("tdc_enable" if camel_case else "tdcEnable"),
bool,
description="Enable / disable Transient Data Capture"
"for the Search Window.",
)
elems.add_opt_field(
("tdc_num_bits" if camel_case else "tdcNumBits"),
int,
description=cleandoc(
"""
Number of bits per sample (for the Transient Data Capture).
Required if TDC is enabled, otherwise not specified.
"""
),
)
elems.add_opt_field(
("tdc_period_before_epoch" if camel_case else "tdcPeriodBeforeEpoch"),
int,
description=cleandoc(
"""
Users can trade the period of time for which data are saved and
transmitted for the sample bit-width and/or the number of
Search Windows. The exact information regarding the
memory capacity per receptor and supported range will be
provided in construction.
The epoch is specified in the command that triggers TDC
off-loading (transmission of data).
"""
),
)
elems.add_opt_field(
("tdc_period_after_epoch" if camel_case else "tdcPeriodAfterEpoch"),
int,
description="see `tdcPeriodBeforeEpoch`",
)
elems.add_opt_field(
("tdc_destination_address" if camel_case else "tdcDestinationAddress"),
[int, str, str, str],
description=cleandoc(
"""
Destination addresses (MAC, IP, port) for off-loading of the
content of the Transient Data Capture Buffer, specified
per receptor. The destination addresses for the content of
the Transient Data Capture can be provided either as a
part of the scan configuration or by the command that
triggers transmission of the captured data. The latter, if
provided, overrides previously set addresses.
Required if TDC is enabled, otherwise not specified.
"""
),
)
return elems
[docs]def get_subarray_config_schema(version: str, strict: bool) -> Schema:
"""CSP Subarray configuration schema
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the CSP subarray specific configuraiton.
"""
# # subarray specific items
camel_case = use_camel_case(version)
elems = TMSchema.new(
"subarray",
version,
strict,
description=cleandoc(
"""
subarray section, containing the parameters relevant only for
the current sub-array device. This section is not forwarded
to any subelement.
"""
),
)
elems.add_field(
("subarray_name" if camel_case else "subarrayName"),
str,
description=cleandoc(
"""
Name and scope of current subarray
the sub-array.
"""
),
)
return elems
[docs]def get_common_config_schema(version: str, strict: bool) -> Schema:
"""CSP Subarray common configuration schema.
This section is valid for Mid CSP because it includes
some parameters that are Mid CBF specific.
The set of parameters that are common to Mid and Low
CSP, are retrieved by the _get_common_config_schema
method defined in the common_schema python module.
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the CSP subarray common
configuration (ADR-18).
"""
camel_case = use_camel_case(version)
(major, minor) = split_interface_version(version)
elems = _get_common_config_schema_without_band(version, strict)
elems.add_opt_field(
("band_5_tuning" if camel_case else "band5Tuning"),
[float],
description=cleandoc(
"""
Center frequency for the Band-of-Interest. Required if Band is 5a
or 5b; not specified for other Bands (not configurable for
Band 1, 2, 3 and 4).
Input for Band 5a and 5b consists of two 2.5 GHz streams;
the center frequency can be independently tuned for each
stream.
The following nomenclature is used to refer to Band 5a and
5b streams: 5a1, 5a2, 5b1, 5b2.
"""
),
)
if (major, minor) <= (2, 3):
elems.add_field(
("frequency_band" if camel_case else "frequencyBand"),
(Regex(r"^(1|2|3|4|5(a|b))$") if strict else str),
description=cleandoc(
"""
Frequency band applies for all the receptors (VCCs)
that belong to the sub-array.
"""
),
)
else:
add_common_frequency_band(version, strict, elems)
# In versions before 1.0, FSPs were part of the common schema
if version.startswith(csp_version.CSP_CONFIG_VER0):
elems.add_field("fsp", [get_fsp_config_schema(version, strict)])
# In version 1.0, subarrayID was added
elif version.startswith(csp_version.CSP_CONFIG_VER1):
elems.add_field("subarrayID", int, description="Subarray number")
# And(int, if_strict(lambda n: n >= 1 and n <= 17)),
# all subelements common schema
return elems
[docs]def get_csp_config_schema(version: str, strict: bool) -> Schema:
"""
Returns a schema to verify a CSP configuration
:param version: Interface version
:param strict: Strict mode - refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the CSP configuration.
:raise: `ValueError` exception on mismatch major version or invalid JSON
Schema URI
"""
# Convert version to standard format
version = csp_version.normalize_csp_config_version(version)
# Valid?
csp_version.check_csp_interface_version(
version, csp_version.CSP_CONFIG_PREFIX
)
# Pre-ADR-18 version with everything in the top level?
elems = TMSchema.new("CSP config", version, strict)
if version.startswith(csp_version.CSP_CONFIG_VER0):
# Overall configuration schema
elems.update(get_common_config_schema(version, strict))
return elems
elif version.startswith(csp_version.CSP_CONFIG_VER1):
# Overall configuration schema
elems.add_opt_field("interface", str)
elems.add_opt_field(
"subarray", get_subarray_config_schema(version, strict)
)
elems.add_field("common", get_common_config_schema(version, strict))
elems.add_field("cbf", get_cbf_config_schema(version, strict))
elems.add_opt_field("pss", get_pss_config_schema(version, strict))
elems.add_opt_field(
"pst",
_get_pst_config_schema(version, strict, beamid_required=False),
)
return elems
elif version.startswith(csp_version.CSP_CONFIG_VER2):
# Overall configuration schema
elems.add_field("interface", str)
elems.add_opt_field(
"subarray", get_subarray_config_schema(version, strict)
)
elems.add_field("common", get_common_config_schema(version, strict))
elems.add_field("cbf", get_cbf_config_schema(version, strict))
elems.add_opt_field("pss", get_pss_config_schema(version, strict))
elems.add_opt_field(
"pst",
_get_pst_config_schema(version, strict, beamid_required=False),
)
return elems
else:
major_version, _ = split_interface_version(version)
raise ValueError(f"Unknown major schema version: {major_version}")
[docs]def get_csp_scan_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP scan command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
scan_schema = TMSchema.new("CSP scan", version, strict)
scan_schema.add_field(
"interface",
str,
description="URI of JSON schema applicable to this JSON payload.",
)
scan_schema.add_field(
"scan_id",
int,
description="Scan ID to associate with the data.",
)
return scan_schema
[docs]def get_csp_assignresources_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP assignresources command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
assignresources_schema = TMSchema.new(
"CSP assignresources", version, strict
)
assignresources_schema.add_field(
"interface",
str,
description="URI of JSON schema applicable to this JSON payload.",
)
assignresources_schema.add_field(
"subarray_id",
int,
description=cleandoc(
"""
The Subarray ID that the list of receptors will be assigned to.
For Mid, there are a maximum of 16 subarrays.
"""
),
)
dish_elements = TMSchema.new("dish assignresources", version, strict)
dish_elements.add_field(
"receptor_ids",
[
Regex(
r"""^(SKA(00[1-9]|0[1-9][0-9]|1[0-2][0-9]|13[0-3]))|
(MKT(0[0-5][0-9]|06[0-3]))$"""
)
if strict
else str
],
description=cleandoc(
"""
The list of receptors that will be assigned to the Subarray ID.
Receptor IDs can be any string, not necessarily numbers.
Valid receptor IDs include:
SKA dishes: "SKAnnn", where nnn is a zero padded integer in the
range of 001 to 133.
MeerKAT dishes: "MKTnnn", where nnn is a zero padded integer in the
range of 000 to 063.
"""
),
)
assignresources_schema.add_field(
"dish",
dish_elements,
)
return assignresources_schema
[docs]def get_csp_endscan_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP endscan command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
endscan_schema = TMSchema.new("CSP endscan", version, strict)
endscan_schema.add_field(
"interface",
str,
description="URI of JSON schema applicable to this JSON payload.",
)
endscan_schema.add_field(
"scan_id",
int,
description="Scan ID to end.",
)
return endscan_schema
[docs]def get_csp_releaseresources_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP releaseresources command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
releaseresources_schema = TMSchema.new(
"CSP releaseresources", version, strict
)
releaseresources_schema.add_field(
"interface",
str,
description="URI of JSON schema applicable to this JSON payload.",
)
releaseresources_schema.add_field(
"subarray_id",
int,
description="Subarray ID which will have its resource(s) released.",
)
releaseresources_schema.add_opt_field(
"release_all",
bool,
description=cleandoc(
"""
Set to true if you wish to release all resources assigned to the
Subarray.
"""
),
)
releaseresources_schema.add_opt_field(
"receptor_ids",
[
Regex(
r"""^(SKA(00[1-9]|0[1-9][0-9]|1[0-2][0-9]|13[0-3]))|
(MKT(0[0-5][0-9]|06[0-3]))$"""
)
if strict
else str
],
description=cleandoc(
"""
The list of receptors that will be released from the Subarray ID.
Receptor IDs can be any string, not necessarily numbers.
Valid receptor IDs include:
SKA dishes: "SKAnnn", where nnn is a zero padded integer in the
range of 001 to 133.
MeerKAT dishes: "MKTnnn", where nnn is a zero padded integer in the
range of 000 to 063.
"""
),
)
return releaseresources_schema
[docs]def get_csp_delaymodel_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP delaymodel command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
(major, minor) = split_interface_version(version)
delaymodel_schema = TMSchema.new(
"CSP delaymodel",
version,
strict,
)
delaymodel_schema.add_field(
"interface",
str,
description="URI of JSON schema applicable to this JSON payload.",
)
if (major, minor) >= (3, 0):
delaymodel_schema.add_field(
"start_validity_sec",
float,
check_strict=lambda n: n > 0.0,
description=cleandoc(
"""
Time when delay model becomes valid (when Mid.CBF shall apply
the new model), specified as the number of seconds since the
1999-12-31T23:59:28Z UTC (SKA epoch).
Range: Non-zero positive number
"""
),
)
delaymodel_schema.add_field(
"cadence_sec",
float,
check_strict=lambda n: n > 0.0,
description=cleandoc(
"""
The time in SI seconds of the planned validity
period of the delay model, measured from start_validity_sec.
Also indicates that the next delay model should be issued no
more than cadence_sec later than the current delay model that
was issued. This is a configurable field and may change during
operations, but the expected value for Mid.CBF is 10 seconds.
Mid.CBF will expect the next delay model it receives to have a
start_validity_sec <= (current start_validity_sec +
cadence_sec). If such a delay model does not arrive, Mid.CBF
will continue to use the current delay model, up to the
maximum acceptable validity period, which is
validity_period_sec. At that point, if a new delay model still
hasn't arrived, Mid.CBF will stop processing (including
outputting products) and will issue an error message.
Range: Non-zero positive number
"""
),
)
delaymodel_schema.add_field(
"validity_period_sec",
float,
check_strict=lambda n: n > 0.0,
description=cleandoc(
"""
The maximum acceptable delay model validity period in SI
seconds, starting at start_validity_sec. This is a
configurable field and may change during operations, but
the expected value for Mid.CBF is 30 seconds.
If Mid.CBF has not received, as expected, a new delay model
with a new start_validity_sec <= (start_validity_sec +
cadence_sec), it will continue to use the current delay model
for up to validity_period_sec seconds. At that point, if
a new delay model still hasn't arrived, Mid.CBF will stop
processing (including outputting products) and will issue an
error message.
Range: Non-zero positive number
"""
),
)
delaymodel_schema.add_field(
"config_id",
str,
description=cleandoc(
"""
The configuration ID of the scan that this delay model applies
to. Corresponds to "config_id" provided in the scan
configuration. This field is used to ensure that the CBF does
not use delay models from a previous observation at the start
of a new observation.
"""
),
)
delaymodel_schema.add_field(
"subarray",
int,
check_strict=lambda n: n >= 1 and n <= 16,
description=cleandoc(
"""
The subarray to which the delay models apply.
Range: Integer from 1-16 inclusive
"""
),
)
delaymodel_schema.add_field(
"receptor_delays",
[get_csp_delay_details_schema(version, strict)],
)
elif (major, minor) == (2, 2):
delaymodel_schema.add_field(
"epoch",
float,
description=cleandoc(
"""
Time when delay model becomes valid
(when Mid.CBF shall apply the new model) specified as an
offset in seconds, expressed as a float number, from
1999-12-31T23:59:28Z UTC (which is called the 'SKA epoch').
Range: 64-bit number
"""
),
)
delaymodel_schema.add_field(
"validity_period",
float,
description=cleandoc(
"""
validity period of the delay model (starting at epoch) [s]
Range: positive number
"""
),
)
delaymodel_schema.add_field(
"delay_details",
[get_csp_delay_details_schema(version, strict)],
)
return delaymodel_schema
[docs]def get_csp_delay_details_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema with the CSP delay details
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
(major, minor) = split_interface_version(version)
delaydetails_schema = TMSchema.new(
"delay details",
version,
strict,
as_reference=True,
)
if (major, minor) >= (3, 0):
delaydetails_schema.add_field(
"receptor",
(
Regex(
r"""^(SKA(00[1-9]|0[1-9][0-9]|1[0-2][0-9]|13[0-3]))|
(MKT(0[0-5][0-9]|06[0-3]))$"""
)
if strict
else str
),
description=cleandoc(
"""
The Receptor (Dish) ID to which the xypol_coeffs_ns and
ypol_offset_ns apply.
Valid receptor IDs include:
SKA dishes: "SKAnnn", where nnn is a zero padded integer in
the range of 001 to 133.
MeerKAT dishes: "MKTnnn", where nnn is a zero padded integer
in the range of 000 to 063.
"""
),
)
delaydetails_schema.add_field(
"xypol_coeffs_ns",
[float],
description=cleandoc(
"""
The delay model for a receptor is specified as a set of
coefficients for a 5th order polynomial. Coefficients of the
polynomial are specified as an array. The Coefficients apply
to both X and Y polarizations.
The delay at time t, where t is measured with respect the
beginning of the validity interval, is calculated as:
d(t) = c0 + c1*t + c2*t^2 + c3*t^3 + c4*t^4 + c5*t^5
Units for coefficients c0,c1,...,c5: ns/s^k where:
k=0,1,...,5
ns=nanoseconds
s=seconds
Type: 64 bit floating point number
"""
),
)
delaydetails_schema.add_field(
"ypol_offset_ns",
float,
description=cleandoc(
"""
Constant delay offset of polarization Y with respect to
polarization X, in nanoseconds.
Type: 64 bit floating point number
"""
),
)
elif (major, minor) == (2, 2):
delaydetails_schema.add_field(
"receptor",
(
Regex(
r"""^(SKA(00[1-9]|0[1-9][0-9]|1[0-2][0-9]|13[0-3]))|
(MKT(0[0-5][0-9]|06[0-3]))$"""
)
if strict
else str
),
description=cleandoc(
"""
The Receptor (Dish) ID to which the poly_info coeffs apply.
Valid receptor IDs include:
SKA dishes: "SKAnnn", where nnn is a zero padded integer in
the range of 001 to 133.
MeerKAT dishes: "MKTnnn", where nnn is a zero padded integer
in the range of 000 to 063.
Range: any string
"""
),
)
delaydetails_schema.add_field(
"poly_info",
[get_csp_poly_info_schema(version, strict)],
)
return delaydetails_schema
[docs]def get_csp_poly_info_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema with the CSP delay details
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
polyinfo_schema = TMSchema.new(
"poly info",
version,
strict,
as_reference=True,
)
polyinfo_schema.add_field(
"polarization",
str,
description=cleandoc(
"""
Polarization of the delay model entry
Range: X or Y
"""
),
)
polyinfo_schema.add_field(
"coeffs",
[float],
description=cleandoc(
"""
Delay Model is specified as coefficients
for a 5th order polynomial.
Coefficients of the polynomial are specified as an array.
The delay at time t, where t is measured with respect the
beginning of the validity interval is calculated as:
d(t) = c0 + c1*t + c2*t^2 + c3*t^3 + c4*t^4 + c5*t^5
Units for coefficients c0,c1,..c5:
ns/s^k where k=0,1,..5
Range for coefficients: 64 bit number
"""
),
)
return polyinfo_schema
[docs]def get_csp_low_delaymodel_schema(version: str, strict: bool) -> Schema:
"""
Returns the schema to verify the CSP low delaymodel command.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
delaymodel_schema = TMSchema.new(
"CSP low delaymodel",
version,
strict,
)
delaymodel_schema.add_field(
"interface",
str,
description=cleandoc(
"""
URI of JSON schema applicable to this JSON payload.
Type: str
"""
),
)
delaymodel_schema.add_field(
"start_validity_sec",
float,
description=cleandoc(
"""
Time when delay model becomes valid
Start point of polynomial validity
no sensible default. It indicates an epoch, which could be anytime.
Type: float
"""
),
)
delaymodel_schema.add_field(
"cadence_sec",
float,
check_strict=lambda n: n > 0.0,
description=cleandoc(
"""
The time in seconds between updates/publications of the
delay polynomials.
Type: float
Range: Non-zero positive number
"""
),
)
delaymodel_schema.add_field(
"validity_period_sec",
float,
check_strict=lambda n: n > 0.0,
description=cleandoc(
"""
Validity period of the delay model (starting at epoch) [s]
Type: float
Range: Non-zero positive number
"""
),
)
delaymodel_schema.add_field(
"config_id",
str,
description=cleandoc(
"""
A string, should be the same as the equivalent value in the
last "configure" JSON. If not it indicates that these are
not yet valid polys for the current configuration.
Type: str
"""
),
)
delaymodel_schema.add_field(
"station_beam",
int,
check_strict=lambda n: n >= 1 and n <= 48, # range
description=cleandoc(
"""
The station beams for which the delay
polynomials apply to.
Type: int
Range: Integer from 1-48 inclusive
"""
),
)
delaymodel_schema.add_field(
"subarray",
int,
check_strict=lambda n: n >= 1 and n <= 16,
description=cleandoc(
"""
The subarray for which the delay polynomials
apply to.
Type: int
Range: Integer from 1-16 inclusive
"""
),
)
delaymodel_schema.add_field(
"station_beam_delays",
[get_low_csp_station_beam_details_schema(version, strict)],
)
return delaymodel_schema
[docs]def get_low_csp_station_beam_details_schema(
version: str, strict: bool
) -> Schema:
"""
Returns the schema with the Low CSP delay details
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
stationbeamdetail_schema = TMSchema.new(
"station beam delays",
version,
strict,
as_reference=True,
)
stationbeamdetail_schema.add_field(
"station_id",
int,
check_strict=lambda n: n >= 1 and n <= 512,
description=cleandoc(
"""
The station ids for which the delay polynomials
apply to.
Type: int
Range: Integer from 1-512 inclusive
"""
),
)
stationbeamdetail_schema.add_field(
"substation_id",
int,
description=cleandoc(
"""
The substation ids for which the delay polynomials
apply to.
Type: int
"""
),
)
stationbeamdetail_schema.add_field(
"xypol_coeffs_ns",
[float],
description=cleandoc(
"""
X coefficient set
Delay Model is specified as coefficients
for a 5th order polynomial.
Coefficients of the polynomial are specified as an array.
The delay at time t, where t is measured with respect the
beginning of the validity interval is calculated as:
d(t) = c0 + c1*t + c2*t^2 + c3*t^3 + c4*t^4 + c5*t^5
Units for coefficients c0,c1,..c5:
ns/s^k where k=0,1,..5
Type: float
Range for coefficients: 64 bit number
"""
),
)
stationbeamdetail_schema.add_field(
"ypol_offset_ns",
float,
description=cleandoc(
"""
Offset for the Y polarisation
Type: float
"""
),
)
return stationbeamdetail_schema