"""
Used for checking CSP configuration strings for conformance
"""
from inspect import cleandoc
from schema import And, Or
from ska_telmodel.skydirection import get_skydirection
from .._common import TMSchema, mk_if, split_interface_version
[docs]
def get_pst_config_schema(version: str, strict: bool) -> TMSchema:
"""
Return the PST configure schema.
This PST schema could be used to validate the JSON script
received during configuration.This schema includes the
common section with the mandatory fields.
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the PST configuration.
"""
# Need a delay import to avoid circular import
from ska_telmodel.csp.common_schema import (
_get_common_config_schema_without_band,
add_common_frequency_band,
)
major, minor = split_interface_version(version)
main_pst_schema = TMSchema.new(
"PST configuration schema",
version,
strict,
description="Schema to validate the Pulsar Timing configuration.",
)
pst_schema = _get_pst_config_schema(version, strict, beamid_required=True)
main_pst_schema.add_field("pst", pst_schema)
if (major, minor) >= (2, 4):
main_pst_schema.add_field(
"interface",
str,
description=(
"URI of JSON schema for this command's JSON payload.."
),
)
common_schema = _get_common_config_schema_without_band(version, strict)
add_common_frequency_band(version, strict, common_schema)
main_pst_schema.add_field("common", common_schema)
return main_pst_schema
def _get_pst_config_schema(
version: str, strict: bool, beamid_required=True
) -> TMSchema:
"""PST specific items
:param version: Interface Version URI
:param strict: Schema strictness
:param beamid_required: whether the timing_beam_id field is mandatory
(CSP-PST interface) or optional (TMC-CSP interface).
:return: the JSON Schema for the PST configuration.
"""
major, minor = split_interface_version(version)
elems = TMSchema.new(
"PST configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters.
To be borrowed from IICD
"""
),
)
if (major, minor) < (2, 2):
elems.add_opt_field("dummy_param", str)
elif (major, minor) < (3, 0):
elems.add_opt_field(
"scan",
get_pst_scan_config_schema(
version, strict, beamid_required=beamid_required
),
)
elems.add_opt_field(
"beam", get_pst_beam_config_schema(version, strict)
)
else:
elems.add_opt_field(
"scan",
get_pst_scan_config_schema(
version, strict, beamid_required=beamid_required
),
)
return elems
[docs]
def get_pst_beam_config_schema(version: str, strict: bool) -> TMSchema:
"""PST specific beam configuration
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the PST beam configuration.
"""
major, minor = split_interface_version(version)
elems = TMSchema.new(
"PST beam configuration",
version,
strict,
description=cleandoc(
"""
PST specific beam configuration parameters.
**Deprecated:** As of version 2.3 this schema has no elements
and is deprecated, and has been removed in version 3.0 of the
schema.
"""
),
as_reference=True,
)
if (major, minor) < (2, 3):
elems.add_field(
"activation_time",
str,
description=cleandoc(
"""
Date and time when to start the PST reconfiguration in UTC.
**Keyword:** ACTIVATION_TIME
"""
),
)
elems.add_field(
"num_channelization_stages",
int,
check_strict=lambda n: n >= 1 and n <= 2,
description=cleandoc(
"""
The number of stages used to channelize the data: e.g.
* for Low, there are 2 stages: 1 in LFAA and 1 in CBF
* for Mid, there are 2 stages: 1 in FSP and 1 in PST BF.
**Keyword:** NSTAGE
"""
),
)
elems.add_field(
"channelization_stages",
[_get_pst_channelisation_schema(version, strict)],
description="List of configuration for each channelization stage.",
)
return elems
def _get_pst_channelisation_schema(version: str, strict: bool) -> TMSchema:
"""PST specific configuration for a channelisation stage.
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the PST channelisation stage.
"""
elems = TMSchema.new(
"PST channelization stage configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for channelization
stage configuration.
**Deprecated:** This field has been deprecated and has been
removed in version 3.0 of PST schema.
"""
),
as_reference=True,
)
elems.add_field(
"num_filter_taps",
int,
check_strict=lambda n: n >= 1 and n <= 2**32,
description=cleandoc(
"""
Total number of taps in the prototype filter (i.e. over
all arms) used in the stage.
**Deprecated:** This field has been deprecated and has been
removed in version 3.0 of PST schema. This field can also be
inferred from the length of the ``filter_coefficients`` field.
**Keyword:** NSTAP_k
"""
),
)
elems.add_field(
"filter_coefficients",
[float],
description=cleandoc(
"""
An array of filter coefficients that define the (time
domain) response function of the prototype filter used
in the stage.
Length of this is num_filter_taps.
**Deprecated:** This field has been deprecated and has been
removed in version 3.0 of PST schema.
**Keyword:** COEFF_k
"""
),
)
elems.add_field(
"num_frequency_channels",
int,
check_strict=lambda n: n >= 1 and n <= 2**32,
description=cleandoc(
"""
The number of frequency channels output by each polyphase
filter bank (PFB) for this stage.
**Deprecated:** This field has been deprecated and has been
removed in version 3.0 of PST schema. This field can be
determined by which telescope and frequency band.
**Keyword:** NCHAN_PFB_k
"""
),
)
elems.add_field(
"oversampling_ratio",
[int],
check_strict=lambda x: len(x) == 2,
description=cleandoc(
"""
The oversampling ratio expressed as a fraction as an array
of int, with the first value the numerator and the second is
the denominator. (e.g. 8/7 is assigned as [8,7]).
**Deprecated:** This field has been deprecated and has been
removed in version 3.0 of PST schema. This field can be
determined by which telescope and frequency band.
**Keyword:** OVERSAMP_k
"""
),
)
return elems
[docs]
def get_pst_scan_config_schema(
version: str, strict: bool, beamid_required=True
) -> TMSchema:
"""PST specific scan configuration
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the PST scan configuration.
"""
(major, minor) = split_interface_version(version)
if (major, minor) >= (3, 0):
return _get_pst_scan_config_schema(version, strict, beamid_required)
else:
return _get_pst_scan_config_schema_legacy(
version, strict, beamid_required
)
def _get_pst_scan_config_schema_legacy(
version: str, strict: bool, beamid_required: bool
) -> TMSchema:
if_strict = mk_if(strict)
(major, minor) = split_interface_version(version)
elems = TMSchema.new(
"PST scan configuration",
version,
strict,
description="PST specific scan configuration parameters.",
as_reference=True,
)
elems.add_field(
"activation_time",
str,
description=cleandoc(
"""
Date and time when to start the PST reconfiguration.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This field has never been used by PST and
was always ignored.
**Units:** UTC timestamp
**Keyword:** ACTIVATION_TIME
"""
),
)
if (major, minor) < (2, 3):
elems.add_field(
"capability",
str,
description=cleandoc(
"""
Identifier of the capability PST Beam to be used for this
configuration.
**Keyword:** CAPABILITY
"""
),
)
elems.add_field(
"scan_id",
int,
description=cleandoc(
"""
The identifier for the scan to be configured.
This is a 64bits long.
**Keyword:** SCAN_ID
"""
),
)
elems.add_field(
"subarray_id",
str,
description=cleandoc(
"""
The ID for the sub-array.
**Keyword:** SUBARRAY_ID
"""
),
)
# timing_beam_id is mandatory in the CSP-PST interface, while it can be
# optional in the TMC-CSP interface.
pst_beam_id_desc = """
Identifier assigned by LMC/TMC used to identify the beam
configuration.
PST selects which PST server to use for this scan and timing
beam, and provides a mapping from the timing beam identifier
by the TM to PST capability id.
**Keyword:** BEAM
"""
if beamid_required:
elems.add_field(
"timing_beam_id",
str,
description=cleandoc(pst_beam_id_desc),
)
else:
elems.add_opt_field(
"timing_beam_id",
str,
description=cleandoc(pst_beam_id_desc),
)
elems.add_field(
"bits_per_sample",
int,
check_strict=lambda n: n in [16, 24, 32],
description=cleandoc(
"""
The number of bits per complex-values time sample in the CBF
output data.
Valid values are 16, 24, or 32.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Keyword:** NBIT
"""
),
)
elems.add_field(
"num_of_polarizations",
int,
check_strict=lambda n: n in [1, 2],
description=cleandoc(
"""
The number of polarizations in the CBF output data.
Valid values are 1 or 2.
**Note:** all SKA receivers produce two polarisation inputs, and
the correlator produces products for both polarisations. The
value of 1 was added to allow for future capability if the need
arises.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Keyword:** NPOL
"""
),
default=2,
)
elems.add_field(
"udp_nsamp",
int,
check_strict=lambda n: n in [4, 32],
description=cleandoc(
"""
The number of time samples for each single polarization
and the a single frequency in each UDP packet sent by CBF.
Note: this must be an integer multiple of WT_NSAMP
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Range:** 32 (Low), 4 (Mid)
**Keyword:** UDP_NSAMP
"""
),
)
elems.add_field(
"wt_nsamp",
int,
check_strict=lambda n: n in [4, 32],
description=cleandoc(
"""
The number of time samples described by as single relative
weight. There is a unique relative weight for each frequency
channel, and each relative weight describes both polarizations.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Range:** 32 (Low), 4 (Mid)
**Keyword:** WT_NSAMP
"""
),
)
elems.add_field(
"udp_nchan",
int,
check_strict=lambda n: n in [24, 185],
description=cleandoc(
"""
The number of contiguous frequency channels in each UDP packet
sent by CBF.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Range:** 24 (Low), 185 (Mid)
**Keyword:** UDP_NCHAN
"""
),
)
elems.add_field(
"num_frequency_channels",
int,
check_strict=lambda n: n >= 1 and n <= 82944,
description=cleandoc(
"""
The total number of frequency channels into which the total
critical bandwidth has been divided.
This must be an integer multiple of udp_nchan
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band. It is also required to be a multiple of the
udp_chan which is also now deprecated for the same reason.
**Range:** 1 to 82944
**Keyword:** OBSNCHAN
"""
),
)
elems.add_field(
"centre_frequency",
float,
check_strict=lambda x: x >= 50e6 and x <= 12800e6,
description=cleandoc(
"""
Centre frequency of the total (critical) bandwidth spanned by
the frequency channels.
**Units:** Hz
**Range:** 50e6 to 12800e6
**Keyword:** OBSFREQ
"""
),
)
elems.add_field(
"total_bandwidth",
float,
check_strict=lambda x: x >= 3610 and x <= 2.5e9,
description=cleandoc(
"""
Total (critical) bandwidth spanned by the channels of the
observation.
Low: 0.00361 to 300 MHz
Mid: 0.053.76 to 2500 MHz
**Units:** Hz
**Range:** 3610 to 2.5e9
**Keyword:** OBSBW
"""
),
)
if (major, minor) < (2, 3):
elems.add_field(
"observation_mode",
str,
check_strict=Or(
"PULSAR_TIMING", "DYNAMIC_SPECTRUM", "FLOW_THROUGH"
),
description=cleandoc(
"""
The observation mode used for the scan.
**Range:** PULSAR_TIMING, DYNAMIC_SPECTRUM, or FLOW_THROUGH
**Keyword:** OBSMODE
"""
),
)
else:
elems.add_field(
"observation_mode",
str,
check_strict=Or(
"PULSAR_TIMING",
"DYNAMIC_SPECTRUM",
"FLOW_THROUGH",
"VOLTAGE_RECORDER",
),
description=cleandoc(
"""
The PST processing mode used for the scan.
The value VOLTAGE_RECORDER is added for AA0.5, while the other
values will be needed in the future for data processing.
**Deprecated:** this field has been renamed to
``pst_processing_mode`` in version 3.0 of the PST schema.
**Keyword:** OBSMODE
"""
),
)
elems.add_field(
"observer_id",
str,
description=cleandoc(
"""
The observer in charge of the observations.
**Keyword:** OBSERVER
"""
),
)
elems.add_field(
"project_id",
str,
description=cleandoc(
"""
The project that the observations are for.
**Keyword:** PROJID
"""
),
)
elems.add_opt_field(
"pointing_id",
str,
description=cleandoc(
"""
The ID for the sub-array pointing.
**Deprecated:** this field has been removed in version 3.0 of the
PST schema.
**Keyword:** PNT_ID
"""
),
)
elems.add_field(
"source",
str,
description=cleandoc(
"""
The name of the source.
**Keyword:** SRC_NAME
"""
),
)
elems.add_field(
"itrf",
[float],
check_strict=lambda x: len(x) == 3,
description=cleandoc(
"""
The International Terrestrial Reference Frame (ITRF) coordinates
of the telescope delay centre.
**Deprecated:** this field has been renamed to "delay_centre" in
version 3.0 of the PST schema.
**Units:** metres
**Keyword:** ITRF
"""
),
)
elems.add_field(
"receiver_id",
str,
description=cleandoc(
"""
The receiver name or ID.
**Keyword:** FRONTEND
"""
),
)
elems.add_field(
"feed_polarization",
str,
check_strict=Or("LIN", "CIRC"),
description=cleandoc(
"""
The native polarization of feed.
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
**Range:** LIN or CIRC
**Keyword:** FD_POLN
"""
),
default="LIN",
)
elems.add_field(
"feed_handedness",
int,
check_strict=Or(-1, 1),
description=cleandoc(
"""
Code for sense of feed.
For value of +1 for XYZ forming RH set with Z in the
direction of propagation. Looking up into the feed of a
prime-focus receiver or at the sky).
For FD_HAND = +1, the rotation from A (or X) to B (or Y)
is counter clockwise or in the direction of increasing
Feed Angle (FA) or Position Angle (PA).
For circular feeds, FD_HAND = +1 for IEEE LCP on the A (or
X) probe.
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
**Range:** -1 or +1
**Keyword:** FD_HAND
"""
),
)
elems.add_field(
"feed_angle",
float,
check_strict=lambda x: x >= -180 and x <= 180,
description=cleandoc(
"""
Feed angle of the E-vector for an equal in-phase response from the
A(X) and B(Y) probes, measured in the direction of
increasing feed angle or position angle (clockwise when
looking down on a prime focus receiver).
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
**Units:** degrees
**Range:** -180 to 180.
**Keyword:** FD_SANG
"""
),
)
elems.add_field(
"feed_tracking_mode",
str,
check_strict=Or("FA", "CPA", "SPA", "TPA"),
description=cleandoc(
"""
The tracking mode for the feed:
* **FA** - constant feed angle and that the feed stays
fixed with respect to the telescope's reference frame.
* **CPA** - the feed rotates to maintain a constant phase
angle (i.e. it tracks the variation of the parallactic
angle.). When the coordinate mode is GALACTIC, PA is with
respect to Galactic north and similarly for coordinate
mode ECLIPTIC then PA is with respect to ecliptic north.
* **SPA** - the feed angle is held fixed at an angle such
that the requested PA is obtained at the mid-point of
the observation.
* **TPA** - is only relevant for scan observations - the
feed is rotated to maintain a constant angle with
respect to the scan direction.
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
**Range:** FA, CPA, SPA, or TPA
**Keyword:** FD_MODE
"""
),
)
elems.add_field(
"feed_position_angle",
float,
check_strict=lambda x: x >= -180 and x <= 180,
description=cleandoc(
"""
The requested angle of feed reference.
For feed_mode = 'FA' this is respect to the telescope's reference
frame (feed_angle = 0), and for feed_mode = 'CPA' this is with
respect to the celestial north (parallactic angle = 0) or with
respect to the Galactic north for coordinate_mode = 'GALACTIC'.
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
**Range:** -180 to +180.
**Keyword:** FA_REQ
"""
),
)
elems.add_field(
"oversampling_ratio",
[int],
check_strict=lambda x: len(x) == 2,
description=cleandoc(
"""
The oversampling ratio expressed as a fraction as an array
of int, with the first value the numerator and the second is
the denominator. (e.g. 8/7 is assigned as [8,7]).
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This can be inferred from the telescope and
and frequency band.
**Range:** 8/7 or 4/3
**Keyword:** OVERSAMP
"""
),
)
elems.add_field(
"coordinates",
_get_pst_equatorial_coordinates(version, strict),
description=cleandoc(
"""
The tied-array beam's tracking co-ordinates.
As of version 2.2 of the schema this only handles
equatorial tracking which means uses RA/Dec J2000.0 coords
but PST may support different tracking modes and
coordinates the future.
"""
),
)
elems.add_field(
"max_scan_length",
float,
check_strict=lambda n: n >= 30 and n <= 43200,
description=cleandoc(
"""
The maximum length of the observation.
**Units:** seconds
**Range:** 30 - 43200
**Keyword:** SCANLEN_MAX
"""
),
)
elems.add_field(
"subint_duration",
float,
check_strict=lambda n: n >= 1 and n <= 60,
description=cleandoc(
"""
The length of each output sub-integration.
**Units:** seconds
**Range:** 1 - 60
**Keyword:** OUTSUBINT
"""
),
)
elems.add_field(
"receptors",
[str],
check_strict=lambda x: len(x) >= 1 and len(x) <= 512,
description=cleandoc(
"""
An array of receptor IDs for the receptors included in the
sub-array.
**Keyword:** ANTENNA
"""
),
)
elems.add_field(
"receptor_weights",
[float],
description=cleandoc(
"""
Weight for each receptor.
**Range:** 0 - 1.0
**Keyword:** ANT_WEIGHTS
"""
),
)
elems.add_opt_field(
"num_rfi_frequency_masks",
int,
check_strict=lambda n: n >= 0 and n <= 1024,
description=cleandoc(
"""
The number of frequency ranges to be masked.
**Deprecated:** this field has been removed in version 3.0
of the PST schema. This is will be inferred by length of
``rfi_frequency_masks``.
**Range:** 0 - 1024
**Keyword:** NMASK
"""
),
default=0,
)
elems.add_opt_field(
"rfi_frequency_masks",
[And([float], if_strict(lambda x: len(x) == 2))],
description=cleandoc(
"""
A two-dimensional array of length of num_frequency_mask of known
RFI frequency ranges to excise from the data.
The array contains mask pairs of [f_min, f_max] pairs for known
frequency ranges containing RFI not excised by the CBF.
The overall dimension of this array is num_frequency_mask x 2.
**Units:** Hz
**Keyword:** FREQ_MASK
"""
),
)
if (major, minor) < (2, 3):
# calibration is not supported. If needed in future this needs logic
# that if cal_mod is not "OFF" then the other fields are required.
elems.add_opt_field(
"cal_mode",
And(str, Or("OFF", "SYNC", "EXT1", "EXT2")),
description=cleandoc(
"""
Operation mode for the injected calibration:
* OFF: there is no injected calibration.
* SYNC: the calibration is pulsed synchronously with the
folding frequency.
* EXT1/EXT2: the calibration is driven by one of two
possible user defined external signals.
**Range:** [OFF, SYNC, EXT1, EXT2]
**Keyword:** CAL_MODE
"""
),
)
elems.add_opt_field(
"calibration_modulation_frequency",
float,
check_strict=lambda x: x >= 0.001 and x <= 1000.0,
description=cleandoc(
"""
The modulation frequency for the injected calibration signal.
**Range:** 0.001 - 1000
**Units:** Hertz
**Keyword:** CAL_FREQ
"""
),
)
elems.add_opt_field(
"calibration_duty_cycle",
float,
check_strict=lambda x: x >= 0.0 and x <= 1.0,
description=cleandoc(
"""
Duty cycle for the injected calibration signal.
**Range:** 0.0 - 1.0
**Keyword:** CAL_DCYC
"""
),
)
elems.add_opt_field(
"calibration_phase",
float,
check_strict=lambda x: x >= 0.0 and x <= 1.0,
description=cleandoc(
"""
The calibration phase with respect to time.
Phase of the leasing edge of the injected calibration signal
in calibration SYNC mode.
**Range:** 0.0 - 1.0
**Keyword:** CAL_PHS
"""
),
)
elems.add_opt_field(
"calibration_num_phase_states",
float,
description=cleandoc(
"""
The number of pulses in one period of the calibration phase.
**Keyword:** CAL_NPHS
"""
),
)
elems.add_opt_field(
"destination_address",
[
str,
int,
],
description=cleandoc(
"""
The destination address for the PST output data.
Includes IPv4 Address, port number.
**Deprecated:** this field has been removed in version 3.0
of the PST schema.
"""
),
)
elems.add_opt_field(
"test_vector_id",
str,
description=cleandoc(
"""
Identifier for a test vector that will be present in the
tied-array beam data stream beam CBF and PST.
**Deprecated:** This field has never been used and has
been removed in version 3.0 of the PST schema.
**Keyword:** TEST_VECTOR
"""
),
)
elems.add_opt_field("pt", _get_pst_pt_mode_schema(version, strict))
elems.add_opt_field("ds", _get_pst_df_mode_schema(version, strict))
elems.add_opt_field("ft", _get_pst_ft_mode_schema(version, strict))
if (major, minor) >= (2, 3):
elems.add_field(
"num_channelization_stages",
int,
check_strict=lambda n: n >= 1 and n <= 2,
description=cleandoc(
"""
The number of stages used to channelize the data: e.g.
* for Low, there are 2 stages: 1 in LFAA and 1 in CBF
* for Mid, there are 2 stages: 1 in FSP and 1 in PST BF.
**Deprecated:** This field has been removed in version 3.0
of the PST schema. The information for this field and
``channelization_stages`` can be inferred from the telescope
and frequency band.
**Keyword:** NSTAGE
"""
),
)
elems.add_field(
"channelization_stages",
[_get_pst_channelisation_schema(version, strict)],
description=cleandoc(
"""
List of configuration for each channelization stage.
**Deprecated:** This field has been removed in version 3.0
of the PST schema. The information for this field and
``num_channelization_stages`` can be inferred from the
telescope and frequency band.
"""
),
)
return elems
def _get_pst_scan_config_schema(
version: str, strict: bool, beamid_required=True
) -> TMSchema:
"""
Get the latest version of the Pulsar Timing scan configuration schema.
:param version: the version string
:type version: str
:param strict: schema strictness to apply with validation
:type strict: bool
:param beamid_required: whether to include the beam_id field as a required
field in the schema, defaults to True
:type beamid_required: bool, optional
:return: the latest version of the Pulsar Timing scan configuration schema.
:rtype: TMSchema
"""
if_strict = mk_if(strict)
(major, _) = split_interface_version(version)
assert major >= 3, (
"expected call to _get_pst_scan_config_schema valid only for at "
"least version 3.0 of PST scan configuration"
)
elems = TMSchema.new(
"PST scan configuration",
version,
strict,
description="PST specific scan configuration parameters.",
as_reference=True,
)
# timing_beam_id is mandatory in the CSP-PST interface, while it can be
# optional in the TMC-CSP interface.
pst_beam_id_desc = """
Identifier assigned by LMC/TMC used to identify the beam
configuration.
CSP LMC selects which PS`T server to use for this scan and
timing beam, and provides a mapping from the timing beam
identifier by the TM to PST capability id.
**Keyword:** BEAM
"""
if beamid_required:
elems.add_field(
"timing_beam_id",
str,
description=cleandoc(pst_beam_id_desc),
)
else:
elems.add_opt_field(
"timing_beam_id",
str,
description=cleandoc(pst_beam_id_desc),
)
elems.add_field(
"centre_frequency",
float,
check_strict=lambda x: x >= 50e6 and x <= 12800e6,
description=cleandoc(
"""
Centre frequency of the total (critical) bandwidth spanned by
the frequency channels.
**Units:** Hz
**Range:** 50e6 to 12800e6
**Keyword:** OBSFREQ
"""
),
)
elems.add_field(
"total_bandwidth",
float,
check_strict=lambda x: x >= 3610 and x <= 2.5e9,
description=cleandoc(
"""
Total (critical) bandwidth spanned by the channels of the
observation.
Low: 0.00361 to 300 MHz
Mid: 0.053.76 to 2500 MHz
**Units:** Hz
**Range:** 3610 to 2.5e9
**Keyword:** OBSBW
"""
),
)
elems.add_field(
"pst_processing_mode",
str,
check_strict=Or(
"PULSAR_TIMING",
"DETECTED_FILTERBANK",
"FLOW_THROUGH",
"VOLTAGE_RECORDER",
"VLBI",
),
description=cleandoc(
"""
The PST processing mode used for the scan.
The value VOLTAGE_RECORDER is added for AA0.5, while the other
values will be needed in future array assemblies.
The value of DYNAMIC_SPECTRUM has been renamed in favour of
the value DETECTED_FILTERBANK. This change was made before PST
implemented DETECTED_FILTERBANK mode and no clients should have
been using this.
**Note:** in previous versions of the schema this used to be
called ``observation_mode`` but has been renamed to be clear
that this is specific about how PST will process that data
from the correlator beam former (CBF).
**Note:** The values come from the Python enum
`PstProcessingMode`` as defined in the SKA Control Model project.
**Keyword:** OBSMODE
"""
),
)
elems.add_field(
"observer_id",
str,
description=cleandoc(
"""
The observer in charge of the observations.
**Keyword:** OBSERVER
"""
),
)
elems.add_field(
"project_id",
str,
description=cleandoc(
"""
The project that the observations are for.
**Keyword:** PROJID
"""
),
)
elems.add_field("target", get_skydirection(version, strict, True))
elems.add_field(
"delay_centre",
[float],
check_strict=lambda x: len(x) == 3,
description=cleandoc(
"""
The telescope delay centre in International Terrestrial Reference
Frame (ITRF) coordinates.
**Units:** metres
**Keyword:** DELAY_CENTRE
"""
),
)
elems.add_field(
"receiver_id",
str,
description=cleandoc(
"""
The receiver name or ID.
**Keyword:** FRONTEND
"""
),
)
elems.add_field(
"max_scan_length",
float,
check_strict=lambda n: n >= 30 and n <= 43200,
description=cleandoc(
"""
The maximum length of the observation.
**Units:** seconds
**Range:** 30 - 43200
**Keyword:** SCANLEN_MAX
"""
),
)
elems.add_opt_field(
"subint_duration",
float,
check_strict=lambda n: n >= 1 and n <= 60,
description=cleandoc(
"""
The length of each output sub-integration, in seconds.
As of version 3.0 of the schema this field is now optional
and will default to 10.0 seconds.
For VOLTAGE_RECORDER mode the sub-integration length is always
10 seconds but in other processing modes this value could be
changed by the observer depending on their use case.
**Units:** seconds
**Range:** 1 - 60
**Keyword:** OUTSUBINT
"""
),
default=10.0,
)
elems.add_field(
"receptors",
[str],
check_strict=lambda x: len(x) >= 1 and len(x) <= 512,
description=cleandoc(
"""
An array of receptor IDs for the receptors included in the
sub-array.
**Keyword:** ANTENNA
"""
),
)
elems.add_field(
"receptor_weights",
[float],
description=cleandoc(
"""
Weight for each receptor.
**Range:** 0 - 1.0
**Keyword:** ANT_WEIGHTS
"""
),
)
elems.add_opt_field(
"rfi_frequency_masks",
[And([float], if_strict(lambda x: len(x) == 2))],
description=cleandoc(
"""
A two-dimensional array of length of num_frequency_mask of known
RFI frequency ranges to excise from the data.
The array contains mask pairs of [f_min, f_max] pairs for known
frequency ranges containing RFI not excised by the CBF.
The overall dimension of this array is num_frequency_mask x 2.
**Units:** Hz
**Keyword:** FREQ_MASK
"""
),
)
elems.add_opt_field("pt", _get_pst_pt_mode_schema(version, strict))
elems.add_opt_field("df", _get_pst_df_mode_schema(version, strict))
elems.add_opt_field("ft", _get_pst_ft_mode_schema(version, strict))
return elems
def _get_pst_pt_mode_schema(version: str, strict: bool) -> TMSchema:
"""
Returns a schema to verify a PST mode configuration for the
'PULSAR_TIMING' observing mode.
:param version: Interface version
:param strict: Schema strictness
:return: The JSON Schema for the PST 'PULSAR_TIMING' mode configuration.
"""
elems = TMSchema.new(
"PST PULSAR_TIMING mode configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for the 'PULSAR_TIMING' mode
configuration.
"""
),
as_reference=True,
)
elems.add_field(
"dispersion_measure",
float,
check_strict=lambda n: n >= 0 and n <= 100000,
description=cleandoc(
"""
The dispersion measure for coherent/incoherent de-dispersion.
**Units:** pccm^-3
**Range:** 0 - 100000
**Keyword:** DM
"""
),
)
elems.add_opt_field(
"rotation_measure",
float,
check_strict=lambda n: n >= -2e9 and n <= 2e9,
description=cleandoc(
"""
The rotation measure for phase-coherent Faraday rotation
correction.
**Units:** radians per metre squared
**Keyword:** RM
"""
),
)
elems.add_field(
"ephemeris",
str,
description=cleandoc(
"""
The ephemeris of the pulsar being observed.
**Units:** PSRCAT compatible ASCII string
**Keyword:** EPHEMERIS
"""
),
)
elems.add_field(
"pulsar_phase_predictor",
str,
description=cleandoc(
"""
Pulsar phase predictor generated from ephemeris.
**Units:** TEMPO2 compatible ASCII string
**Keyword:** PREDICTOR
"""
),
)
elems.add_field(
"output_frequency_channels",
int,
description=cleandoc(
"""
The number of output frequency channels. This must
be between 1 and the number of observation channels.
**Keyword:** OUTNCHAN
"""
),
)
elems.add_field(
"output_phase_bins",
int,
check_strict=lambda n: n >= 64 and n <= 2048,
description=cleandoc(
"""
The number of output phase bins.
**Range:** 64 - 2048
**Keyword:** OUTNBIN
"""
),
)
elems.add_field(
"num_sk_config",
int,
description=cleandoc(
"""
The number of spectral kurtosis (SK) configurations to apply.
**Keyword:** N_SK
"""
),
)
elems.add_field(
"sk_config",
[_get_pst_sk_config_schema(version, strict)],
description="List of spectral kurtosis configurations.",
)
elems.add_field(
"target_snr",
float,
description=cleandoc(
"""
The signal-to-noise ratio (SNR) of the on-pulse flux for
the scan. May be used to prematurely end a scan when the
integrated SNR reaches the target. A value of 0 indicates
there is no limit.
**Keyword:** TARGET_SNR
"""
),
)
return elems
def _get_pst_sk_config_schema(version: str, strict: bool) -> TMSchema:
"""
Returns a schema to verify a a spectral kurtosis (SK) configuration
used in the 'PULSAR_TIMING' mode.
:param version: Interface version
:param strict: Schema strictness
:return: The JSON Schema for the PST spectral kurtosis configuration.
"""
elems = TMSchema.new(
"PST spectral kurtosis configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for the spectral kurtosis (SK)
for the 'PULSAR_TIMING' mode.
"""
),
as_reference=True,
)
elems.add_field(
"sk_range",
[float],
check_strict=lambda x: len(x) == 2,
description=cleandoc(
"""
Frequency ranges for each spectral kurtosis (SK) configuration.
**Units:** Hz
**Keyword:** SK_RNG
"""
),
)
elems.add_field(
"sk_integration_limit",
int,
check_strict=lambda n: n >= 64 and n <= 1024,
description=cleandoc(
"""
The number of input time samples integrated into each spectral
kurtosis (SK) statistic.
**Range:** 64 - 1024
**Keyword:** SK_INTS
"""
),
)
elems.add_field(
"sk_excision_limit",
float,
check_strict=lambda n: n >= 1.0 and n <= 100.0,
description=cleandoc(
"""
Spectral kurtosis excision limits (RFI threshold) in
units of standard deviations.
**Range:** 1 - 100
**Keyword:** SK_EXIS
"""
),
)
return elems
def _get_pst_df_mode_schema(version: str, strict: bool) -> TMSchema:
"""
Returns a schema to verify a PST 'DYNAMIC_SPECTRUM' mode configuration.
:param version: Interface version
:param strict: Schema strictness
:return: The JSON Schema for the PST 'DYNAMIC_SPECTRUM' mode configuration.
"""
elems = TMSchema.new(
"PST DYNAMIC_SPECTRUM mode configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for the 'DYNAMIC_SPECTRUM' mode
configuration.
"""
),
as_reference=True,
)
elems.add_field(
"dispersion_measure",
float,
check_strict=lambda n: n >= 0 and n <= 100000,
description=cleandoc(
"""
The dispersion measure for coherent/incoherent de-dispersion.
This is only required for pulsar timing and dynamic spectrum modes.
**Range:** [0, 100000]
**Keyword:** DM
"""
),
)
elems.add_opt_field(
"rotation_measure",
float,
check_strict=lambda n: n >= -2e9 and n <= 2e9,
description=cleandoc(
"""
The rotation measure for phase-coherent Faraday rotation
correction.
**Units:** radians per metre squared
**Keyword:** RM
"""
),
)
elems.add_field(
"output_frequency_channels",
int,
description=cleandoc(
"""
The number of output frequency channels. This must
be between 1 and the number of observation channels.
**Keyword:** OUTNCHAN
"""
),
)
elems.add_field(
"stokes_parameters",
str,
description=cleandoc(
"""
The Stokes parameters to output when in Dynamic
spectrum mode.
**Range:** string with a combination of I, Q, U, and V.
**Keyword:** STOKES_FB
"""
),
)
elems.add_field(
"num_bits_out",
int,
check_strict=Or(1, 2, 4, 8, 16, 32),
description=cleandoc(
"""
The number of bits per output sample.
**Range:** 1, 2, 4, 8, 16 or 32
**Keyword:** NBIT_OUT
"""
),
)
elems.add_field(
"time_decimation_factor",
int,
description=cleandoc(
"""
The number of input samples per output time sample
when in Dynamic Spectrum mode.
**Keyword:** TDEC_FB
"""
),
)
elems.add_field(
"frequency_decimation_factor",
int,
description=cleandoc(
"""
The number of input frequency channels incoherently
added to each output frequency channel in Dynamic
Spectrum.
This is required in addition to output_frequency_channels
because some frequency channels may be merged coherently
to increase temporal resolution.
**Keyword:** FDEC_FB
"""
),
)
elems.add_opt_field(
"num_sk_config",
int,
description=cleandoc(
"""
The number of spectral kurtosis (SK) configurations to apply.
**Keyword:** N_SK
"""
),
)
elems.add_opt_field(
"sk_config",
[_get_pst_sk_config_schema(version, strict)],
description="List of spectral kurtosis configurations.",
)
elems.add_field(
"requantisation_scale",
float,
description=cleandoc(
"""
Scale factor to govern the dynamic range for fixed precision
output to be applied during re-quantisation.
**Keyword:** DIGITIZER_SCALE
"""
),
)
elems.add_field(
"requantisation_length",
float,
description=cleandoc(
"""
Length of data to be used when determining the scaling factors
used for fixed precision output during re-quantisation.
**Units:** seconds
**Keyword:** DIGITIZER_LENGTH
"""
),
)
return elems
def _get_pst_ft_mode_schema_legacy(version, strict: bool) -> TMSchema:
if_strict = mk_if(strict)
(major, minor) = split_interface_version(version)
elems = TMSchema.new(
"PST FLOW_THROUGH mode configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for the 'FLOW_THROUGH' mode
configuration.
"""
),
as_reference=True,
)
elems.add_field(
"num_bits_out",
int,
check_strict=Or(1, 2, 4, 8, 16, 32),
description=cleandoc(
"""
The number of bits per output sample.
**Range:** 1, 2, 4, 8, 16 or 32
**Keyword:** NBIT_OUT
"""
),
)
if (major, minor) <= (2, 4):
channels_field_type = [int]
else:
channels_field_type = And([int], if_strict(lambda lst: len(lst) == 2))
elems.add_field(
"channels",
channels_field_type,
description=cleandoc(
"""
The indices of the first and last (inclusive) frequency
channels that define the single contiguous range of
frequency channels to be recorded.
**Keyword:** CHAN_FT
"""
),
)
elems.add_field(
"requantisation_scale",
float,
description=cleandoc(
"""
Scale factor applied during re-quantisation that modifies the
dynamic range of the fixed precision output.
By default, for 2, 4, and 8 bits per sample, data will be scaled
to minimize scattered power by adopting the Optimum Input
Threshold Spacing for a Uniform Digitizer defined in Table 3 of
Jenet & Anderson (1998; PASP 110:1467).
For more than 8 bits per sample, the values from Table 3 of Jenet
& Anderson (1998) are extrapolated to num_bits_out-2, such that 2
bits are retained as additional headroom for RFI.
For all num_bits_out, the standard deviation is that of either the
real or imaginary part of each complex-valued sample.
The default scale factor is computed such that, after
multiplication by this scale factor, the data would satisfy the
conditions described above. This default scale factor is
multiplied by requantisation_scale. Therefore, a
requantisation_scale value greater than 1 increases the value of
the floating point data before it is cast to a fixed precision
value, thereby reducing the overhead available to represent RFI
and increasing the probability of clipping.
**Keyword:** DIGITIZER_SCALE
"""
),
)
if (major, minor) <= (2, 4):
elems.add_field(
"num_channels",
int,
description=cleandoc(
"""
The number of input channels to be recorded.
This value must be less than or equal to the
output_frequency_channels.
**Keyword:** NCHAN_FT
"""
),
)
elems.add_field(
"requantisation_length",
float,
description=cleandoc(
"""
Length of data to be used when determining the scaling factors
used for fixed precision output during re-quantisation.
**Units:** seconds
**Keyword:** DIGITIZER_LENGTH
"""
),
)
if (major, minor) > (2, 4):
elems.add_field(
"polarizations",
str,
check_strict=Or(
"A",
"B",
"Both",
),
description=cleandoc(
"""
The polarizations to be recorded.
**Valid values:** A, B, or Both
**Keyword:** POLN_FT
"""
),
)
elems.add_field(
"requantisation_init_time",
float,
description=cleandoc(
"""
Time interval spanned by data used at the start of a scan to
determine the scale factors applied before re-quantisation.
**Units:** seconds
**Keyword:** DIGITIZER_INIT_TIME
"""
),
)
return elems
def _get_pst_rescale_schema(version: str, strict: bool) -> TMSchema:
elems = TMSchema.new(
"PST Rescale configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters used for rescaling of
data before requantisation. This is used within the
FLOW_THROUGH processing mode.
"""
),
as_reference=True,
)
elems.add_opt_field(
"algorithm",
str,
check_strict=Or(
"MedianMAD",
"MeanStdDev",
),
description=cleandoc(
"""
The algorithm used to determine the scales and offsets when
rescaling complex voltage data in Flow Through mode.
The value MedianMAD is the default algorithm and PST will
determine the scales and offsets based on the calculating the
median and the median absolute deviation (MAD) as the proxies
for the sample mean and standard deviation. These statistics
are considered more robust against outliers and are less
affected by radio frequency interference (RFI) when computing
the values.
The value MeanStdDev will calculate the sample mean and standard
deviation of the input data and use that to rescale the data.
This algorithm is not robust against RFI which will reduce the
dynamic range when performing requantisation of the data.
**Default:** MedianMAD
"""
),
default="MedianMAD",
)
elems.add_opt_field(
"periodic_update",
bool,
description=cleandoc(
"""
An indicator for whether to recalculate the rescale statistics
periodically.
The rescaling of data within the FLOW_THROUGH processing mode can
either be calculated once and applied across all the data during
the scan or can be calculated periodically. If this value is set
to false then the former is performed while the latter is performed
if the field is set to true.
**Default:** false
"""
),
default=False,
)
elems.add_opt_field(
"timescale",
float,
check_strict=lambda x: x >= 0.0,
description=cleandoc(
"""
The timescale needed to calculate rescale stats, in seconds.
This value is how long in time to sample data before calculating
the rescale statistics. If periodic_update is true, then this
is also the period of how often the rescale stats are recalculated.
If this value is set to 0.0, then PST will use the smallest chunk
of data available to it to perform the statistics calculation.
**Default:** 0.0
"""
),
default=0.0,
)
return elems
def _get_pst_chanpol_select_schema(version: str, strict: bool) -> TMSchema:
if_strict = mk_if(strict)
elems = TMSchema.new(
"PST channel and polarisation selection configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters used for performing
channel and polarisation selection of input data.
This is used within the FLOW_THROUGH processing mode.
"""
),
as_reference=True,
)
elems.add_field(
"channels",
And([int], if_strict(lambda lst: len(lst) == 2)),
description=cleandoc(
"""
The indices of the first and last (inclusive) frequency
channels that define the single contiguous range of
frequency channels to be recorded.
**Keyword:** CHAN_FT
"""
),
)
elems.add_field(
"polarisations",
str,
check_strict=Or(
"A",
"B",
"Both",
),
description=cleandoc(
"""
The polarisations to be recorded.
**Valid values:** A, B, or Both
**Keyword:** POLN_FT
"""
),
)
return elems
def _get_pst_requantisation_schema(version: str, strict: bool) -> TMSchema:
elems = TMSchema.new(
"PST requantisation configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters used for requantisation/
digitisation of data after other steps within the processing
pipeline have been performed.
The PST processing pipeline works on floating point numbers
but the output data are digitised and this step can reduce
the output data size by using small number of bits to
represent the output value.
This is used within the FLOW_THROUGH processing mode.
"""
),
as_reference=True,
)
elems.add_field(
"num_bits_out",
int,
check_strict=Or(1, 2, 4, 8, 16, 32),
description=cleandoc(
"""
The number of bits per output sample.
**Range:** 1, 2, 4, 8, 16 or 32
**Keyword:** NBIT_OUT
"""
),
)
elems.add_field(
"scale",
float,
description=cleandoc(
"""
Scale factor applied during re-quantisation that modifies the
dynamic range of the fixed precision output.
By default, for 2, 4, and 8 bits per sample, data will be scaled
to minimize scattered power by adopting the Optimum Input
Threshold Spacing for a Uniform Digitizer defined in Table 3 of
Jenet & Anderson (1998; PASP 110:1467).
For more than 8 bits per sample, the values from Table 3 of Jenet
& Anderson (1998) are extrapolated to num_bits_out-2, such that 2
bits are retained as additional headroom for RFI.
For all num_bits_out, the standard deviation is that of either the
real or imaginary part of each complex-valued sample.
The default scale factor is computed such that, after
multiplication by this scale factor, the data would satisfy the
conditions described above. This default scale factor is
multiplied by scale. Therefore, a scale value greater than 1
increases the value of the floating point data before it is cast
to a fixed precision value, thereby reducing the overhead available
to represent RFI and increasing the probability of clipping.
**Keyword:** DIGITIZER_SCALE
**Default:** 1.0
"""
),
default=1.0,
)
def _get_pst_ft_mode_schema(version: str, strict: bool) -> TMSchema:
"""
Returns a schema to verify a PST 'FLOW_THROUGH' mode configuration.
:param version: Interface version
:param strict: Schema strictness
:return: The JSON Schema for the PST 'FLOW_THROUGH' mode configuration.
"""
(major, minor) = split_interface_version(version)
if (major, minor) < (3, 0):
return _get_pst_ft_mode_schema_legacy(version, strict)
elems = TMSchema.new(
"PST FLOW_THROUGH mode configuration",
version,
strict,
description=cleandoc(
"""
PST specific parameters for the 'FLOW_THROUGH' mode
configuration.
"""
),
as_reference=True,
)
elems.add_field(
"channel_polarisation_selection",
_get_pst_chanpol_select_schema(version, strict),
)
elems.add_field("rescale", _get_pst_rescale_schema(version, strict))
elems.add_field(
"requantisation", _get_pst_requantisation_schema(version, strict)
)
return elems
def _get_pst_equatorial_coordinates(version: str, strict: bool) -> TMSchema:
"""
Returns a schema to verify RA/Dec coordinates used within PST.
As of version 2.2 of the schema this only handles RA/Dec J2000 coords
but PST may support different tracking modes and coordinates the
future.
:param version: Interface version
:param strict: Schema strictness
:return: The JSON Schema
"""
elems = TMSchema.new(
"PST RA/Dec coordinates",
version,
strict,
description="""
PST specific parameters for RA/Dec tracking coordinates.
""",
as_reference=True,
)
elems.add_opt_field(
"equinox",
float,
check_strict=lambda x: x >= 2000,
description=cleandoc(
"""
The coordinate epoch.
This can be in Julian date or Modified Julian Date.
**Units:** years
**Range:** >= 2000
**Keyword:** EQUINOX
"""
),
default=2000.0,
)
elems.add_field(
"ra",
str,
description=cleandoc(
"""
The Right Accession (RA) of the coordinates used for tracking.
Valid formats is 'hh:mm:ss.sss' or 'ddd.ddd'
**Keyword:** STT_CTD1
"""
),
)
elems.add_field(
"dec",
str,
description=cleandoc(
"""
The declination (Dec) of the coordinates used for tracking.
Valid formats is 'hh:mm:ss.sss' or 'ddd.ddd'
**Keyword:** STT_CTD2
"""
),
)
return elems