"""
Used for checking CSP configuration strings for conformance
"""
from inspect import cleandoc
from schema import And, Or
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:
"""Pulsar Timing 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(
"""
Pulsar Timing specific parameters.
To be borrowed from IICD
"""
),
)
if (major, minor) < (2, 2):
elems.add_opt_field("dummy_param", str)
else:
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)
)
return elems
[docs]def get_pst_beam_config_schema(version: str, strict: bool) -> TMSchema:
"""Pulsar Timing 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(
"""
Pulsar Timing specific beam configuration parameters.
As of version 2.3 this schema has no elements and is
deprecated
"""
),
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:
"""Pulsar Timing 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(
"""
Pulsar Timing specific parameters for channelization
stage configuration.
"""
),
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.
**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.
**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.
**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]).
**Keyword:** OVERSAMP_k
"""
),
)
return elems
[docs]def get_pst_scan_config_schema(
version: str, strict: bool, beamid_required=True
) -> TMSchema:
"""Pulsar Timing specific scan configuration
:param version: Interface Version URI
:param strict: Schema strictness
:return: the JSON Schema for the PST scan configuration.
"""
if_strict = mk_if(strict)
(major, minor) = split_interface_version(version)
elems = TMSchema.new(
"PST scan configuration",
version,
strict,
description="Pulsar Timing specific scan configuration parameters.",
as_reference=True,
)
elems.add_field(
"activation_time",
str,
description=cleandoc(
"""
Date and time when to start the PST reconfiguration.
**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_descr = """
Identifier assigned by LMC/TM used to identify the beam
configuraiton.
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_descr),
)
else:
elems.add_opt_field(
"timing_beam_id",
str,
description=cleandoc(pst_beam_id_descr),
)
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.
**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.
**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_NSMAP
**Range:** 4 (Low), 32 (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.
**Range:** 4 (Low), 32 (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.
**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
**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 to 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 observation mode used for the scan.
The value VOLTAGE_RECORDER is added for AA0.5, while the other
values will be needed for in the future for data processing.
**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(
"pointing_id",
str,
description=cleandoc(
"""
The ID for the sub-array pointing.
**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.
**Units:** metres
**Keyword:** ITRF
"""
),
)
elems.add_field(
"receiver_id",
str,
description=cleandoc(
"""
The receiver name or ID (instrument).
**Keyword:** FRONTEND
"""
),
)
elems.add_field(
"feed_polarization",
str,
check_strict=Or("LIN", "CIRC"),
description=cleandoc(
"""
The native polarization of feed.
**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.
**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 focuse receiver).
**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 cordinate mode is GALATIC, 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.
**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 (parallic angle = 0) or with respect
to the Galactic north for coordinate_mode = 'GALACTIC'.
**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]).
**Range:** 8/7 or 4/3
**Keyword:** OVERSAMP
"""
),
)
elems.add_field(
"coordinates",
_get_pst_equitorial_coordinates(version, strict),
description=cleandoc(
"""
The tied-array beam's tracking co-ordinates.
As of version 2.2 of the schema this only handles
equitorial 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.
**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.
"""
),
)
elems.add_opt_field(
"test_vector_id",
str,
description=cleandoc(
"""
Identifier for a test vectore that will be present in the
tied-array beam data stream beam CBF and PST.
**Keyword:** TEST_VECTOR
"""
),
)
elems.add_opt_field("pt", _get_pst_pt_mode_schema(version, strict))
elems.add_opt_field("ds", _get_pst_ds_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.
**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_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(
"""
Pulsar Timing 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/inchoerent 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 spectural kurtosis configuration.
"""
elems = TMSchema.new(
"PST spectral kurtosis configuration",
version,
strict,
description=cleandoc(
"""
Pulsar Timing 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_ds_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(
"""
Pulsar Timing 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 measture for coherent/inchoerent 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(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.
"""
if_strict = mk_if(strict)
(major, minor) = split_interface_version(version)
elems = TMSchema.new(
"PST 'FLOW_THROUGH' mode configuration",
version,
strict,
description=cleandoc(
"""
Pulsar Timing 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 16 and 32 bits per sample, by default the data will be scaled
such that the maximum fixed precision output value
(2^{num_bits_out-1}) corresponds to 6 times the standard deviation.
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_equitorial_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="""
Pulsar Timing 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