import copy
from inspect import cleandoc
from .._common import split_interface_version
from ..pss.example import get_pss_configure_example
from ..pst.examples import (
get_pst_config_example,
)
from .common_schema import MAX_CHANNELS_PER_STREAM, MAX_STREAMS_PER_FSP
from .version import (
CSP_ASSIGNRESOURCES_PREFIX,
CSP_CONFIG_PREFIX,
CSP_CONFIGSCAN_PREFIX,
CSP_DELAYMODEL_PREFIX,
CSP_ENDSCAN_PREFIX,
CSP_LOW_DELAYMODEL_PREFIX,
CSP_MID_DELAYMODEL_PREFIX,
CSP_RELEASERESOURCES_PREFIX,
CSP_SCAN_PREFIX,
check_csp_interface_version,
get_csp_config_subsystem_version,
)
def get_csp_config_scan_0_1_example(scan: str = None) -> dict:
csp_config_0_1 = {
"id": "sbi-mvp01-20200325-00001-science_A",
"frequencyBand": "1",
"fsp": [
{
"fspID": 1,
"functionMode": "CORR",
"frequencySliceID": 1,
"integrationTime": 1400,
"corrBandwidth": 0,
"channelAveragingMap": [[0, 2], [744, 0]],
"fspChannelOffset": 0,
"outputLinkMap": [[0, 0], [200, 1]],
},
{
"fspID": 2,
"functionMode": "CORR",
"frequencySliceID": 2,
"integrationTime": 1400,
"corrBandwidth": 0,
"channelAveragingMap": [[0, 2], [744, 0]],
"fspChannelOffset": 744,
"outputLinkMap": [[0, 4], [200, 5]],
},
],
}
if scan == "cal_a":
csp_config_0_1["fsp"][0]["outputHost"] = [[0, "192.168.1.1"]]
csp_config_0_1["fsp"][0]["outputPort"] = [[0, 9000, 1]]
csp_config_0_1["fsp"][1]["outputHost"] = [[0, "192.168.1.1"]]
csp_config_0_1["fsp"][1]["outputPort"] = [[0, 9744, 1]]
elif scan == "science_a":
csp_config_0_1["fsp"][0]["outputHost"] = [
[0, "192.168.0.1"],
[400, "192.168.0.2"],
]
csp_config_0_1["fsp"][0]["outputMac"] = [[0, "06-00-00-00-00-00"]]
csp_config_0_1["fsp"][0]["outputPort"] = [[0, 9000, 1], [400, 9000, 1]]
csp_config_0_1["fsp"][1]["outputHost"] = [
[0, "192.168.0.3"],
[400, "192.168.0.4"],
]
csp_config_0_1["fsp"][1]["outputMac"] = [[0, "06-00-00-00-00-01"]]
csp_config_0_1["fsp"][1]["outputPort"] = [[0, 9000, 1], [400, 9000, 1]]
return csp_config_0_1
def get_csp_config_scan_1_0_example(scan: str = None) -> dict:
csp_config_1_0 = {
"interface": "https://schema.skatelescope.org/ska-csp-configure/1.0",
"subarray": {"subarrayName": "science period 23"},
"common": {
"id": "sbi-mvp01-20200325-00001-science_A",
"frequencyBand": "1",
"subarrayID": 1,
},
"cbf": {
"fsp": [
{
"fspID": 1,
"functionMode": "CORR",
"frequencySliceID": 1,
"integrationTime": 1400,
"corrBandwidth": 0,
"channelAveragingMap": [[0, 2], [744, 0]],
"fspChannelOffset": 0,
"outputLinkMap": [[0, 0], [200, 1]],
},
{
"fspID": 2,
"functionMode": "CORR",
"frequencySliceID": 2,
"integrationTime": 1400,
"corrBandwidth": 0,
"channelAveragingMap": [[0, 2], [744, 0]],
"fspChannelOffset": 744,
"outputLinkMap": [[0, 4], [200, 5]],
},
],
"vlbi": {},
},
}
if scan == "cal_a":
csp_config_1_0["cbf"]["fsp"][0]["outputHost"] = [[0, "192.168.1.1"]]
csp_config_1_0["cbf"]["fsp"][0]["outputPort"] = [[0, 9000, 1]]
csp_config_1_0["cbf"]["fsp"][1]["outputHost"] = [[0, "192.168.1.1"]]
csp_config_1_0["cbf"]["fsp"][1]["outputPort"] = [[0, 9744, 1]]
elif scan == "science_a":
csp_config_1_0["cbf"]["fsp"][0]["outputHost"] = [
[0, "192.168.0.1"],
[400, "192.168.0.2"],
]
csp_config_1_0["cbf"]["fsp"][0]["outputMac"] = [
[0, "06-00-00-00-00-00"]
]
csp_config_1_0["cbf"]["fsp"][0]["outputPort"] = [
[0, 9000, 1],
[400, 9000, 1],
]
csp_config_1_0["cbf"]["fsp"][1]["outputHost"] = [
[0, "192.168.0.3"],
[400, "192.168.0.4"],
]
csp_config_1_0["cbf"]["fsp"][1]["outputMac"] = [
[0, "06-00-00-00-00-01"]
]
csp_config_1_0["cbf"]["fsp"][1]["outputPort"] = [
[0, 9000, 1],
[400, 9000, 1],
]
return csp_config_1_0
def get_csp_config_scan_2_0_example(scan: str = None) -> dict:
csp_config_2_0 = {
"interface": "https://schema.skao.int/ska-csp-configure/2.0",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"frequency_band": "1",
"subarray_id": 1,
},
"cbf": {
"fsp": [
{
"fsp_id": 1,
"function_mode": "CORR",
"frequency_slice_id": 1,
"integration_factor": 1,
"zoom_factor": 0,
"channel_averaging_map": [[0, 2], [744, 0]],
"channel_offset": 0,
"output_link_map": [[0, 0], [200, 1]],
},
{
"fsp_id": 2,
"function_mode": "CORR",
"frequency_slice_id": 2,
"integration_factor": 1,
"zoom_factor": 1,
"zoom_window_tuning": 650000,
"channel_averaging_map": [[0, 2], [744, 0]],
"channel_offset": 744,
"output_link_map": [[0, 4], [200, 5]],
},
],
"vlbi": {},
},
"pst": {},
}
if scan == "cal_a":
csp_config_2_0["cbf"]["fsp"][0]["output_host"] = [[0, "192.168.1.1"]]
csp_config_2_0["cbf"]["fsp"][0]["output_port"] = [[0, 9000, 1]]
csp_config_2_0["cbf"]["fsp"][1]["output_host"] = [[0, "192.168.1.1"]]
csp_config_2_0["cbf"]["fsp"][1]["output_port"] = [[0, 9744, 1]]
elif scan == "science_a":
csp_config_2_0["cbf"]["fsp"][0]["output_host"] = [
[0, "192.168.0.1"],
[400, "192.168.0.2"],
]
csp_config_2_0["cbf"]["fsp"][0]["output_mac"] = [
[0, "06-00-00-00-00-00"]
]
csp_config_2_0["cbf"]["fsp"][0]["output_port"] = [
[0, 9000, 1],
[400, 9000, 1],
]
csp_config_2_0["cbf"]["fsp"][1]["output_host"] = [
[0, "192.168.0.3"],
[400, "192.168.0.4"],
]
csp_config_2_0["cbf"]["fsp"][1]["output_mac"] = [
[0, "06-00-00-00-00-01"]
]
csp_config_2_0["cbf"]["fsp"][1]["output_port"] = [
[0, 9000, 1],
[400, 9000, 1],
]
return csp_config_2_0
def get_csp_config_scan_2_1_example(scan: str = None) -> dict:
csp_config_2_1 = get_csp_config_scan_2_0_example(scan)
csp_config_2_1[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.1"
if scan == "pss":
del csp_config_2_1["pst"]
csp_config_2_1["cbf"] = {
"fsp": [
{
"fsp_id": 1,
"function_mode": "PSS-BF",
"frequency_slice_id": 1,
"integration_factor": 1,
"zoom_factor": 0,
"output_link_map": [[0, 1], [200, 2]],
},
{
"fsp_id": 2,
"function_mode": "CORR",
"frequency_slice_id": 1,
"integration_factor": 1,
"zoom_factor": 1,
"zoom_window_tuning": 650000,
"output_link_map": [[0, 4], [200, 5]],
},
],
"search_window": [
{
"search_window_id": 0,
"search_window_tuning": 1000,
"tdc_enable": True,
},
],
}
csp_config_2_1["pss"] = get_pss_configure_example(
"https://schema.skao.int/ska-pss-configure/0.1"
)
return csp_config_2_1
def get_csp_config_scan_2_2_example(scan: str = None) -> dict:
csp_config_2_2 = get_csp_config_scan_2_1_example(scan)
csp_config_2_2[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.2"
if scan == "pst_beam":
csp_config_2_2["pst"] = {
"beam": {
"activation_time": "2022-01-19T23:07:45Z",
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
},
}
elif scan == "pst_scan_pt":
csp_config_2_2["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "beam1",
"capability": "capability1",
"scan_id": 1,
"bits_per_sample": 24,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.81480,
"observation_mode": "PULSAR_TIMING",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"subarray_id": "subarray42",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 10000.5,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"pt": {
"dispersion_measure": 100.0,
"rotation_measure": 0.0,
"ephemeris": "",
"pulsar_phase_predictor": "",
"output_frequency_channels": 1,
"output_phase_bins": 64,
"num_sk_config": 1,
"sk_config": [
{
"sk_range": [0.8, 0.9],
"sk_integration_limit": 100,
"sk_excision_limit": 25.0,
}
],
"target_snr": 0.0,
},
},
}
elif scan == "pst_scan_ds":
csp_config_2_2["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "beam1",
"capability": "capability1",
"scan_id": 1,
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "DYNAMIC_SPECTRUM",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"subarray_id": "subarray42",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 13000.2,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"ds": {
"dispersion_measure": 100.0,
"output_frequency_channels": 1,
"stokes_parameters": "Q",
"num_bits_out": 16,
"time_decimation_factor": 10,
"frequency_decimation_factor": 4,
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
},
},
}
elif scan == "pst_scan_ft":
csp_config_2_2["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "beam1",
"capability": "capability1",
"scan_id": 1,
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "FLOW_THROUGH",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"subarray_id": "subarray42",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"ft": {
"num_bits_out": 32,
"num_channels": 1,
"channels": [1],
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
},
},
}
return csp_config_2_2
def get_csp_config_scan_2_3_example(scan: str = None) -> dict:
csp_config_2_3 = get_csp_config_scan_2_2_example(scan)
csp_config_2_3[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.3"
if scan == "pst_beam":
csp_config_2_3["pst"] = {
"beam": {},
}
elif scan == "pst_scan_pt":
del csp_config_2_3["pst"]["scan"]["timing_beam_id"]
del csp_config_2_3["pst"]["scan"]["capability"]
del csp_config_2_3["pst"]["scan"]["scan_id"]
csp_config_2_3["pst"]["scan"]["bits_per_sample"] = 32
del csp_config_2_3["pst"]["scan"]["subarray_id"]
csp_config_2_3["pst"]["scan"]["num_channelization_stages"] = 1
csp_config_2_3["pst"]["scan"]["channelization_stages"] = [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
]
elif scan == "pst_scan_ds":
del csp_config_2_3["pst"]["scan"]["timing_beam_id"]
del csp_config_2_3["pst"]["scan"]["capability"]
del csp_config_2_3["pst"]["scan"]["scan_id"]
del csp_config_2_3["pst"]["scan"]["subarray_id"]
csp_config_2_3["pst"]["scan"]["num_channelization_stages"] = 1
csp_config_2_3["pst"]["scan"]["channelization_stages"] = [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
]
elif scan == "pst_scan_ft":
del csp_config_2_3["pst"]["scan"]["timing_beam_id"]
del csp_config_2_3["pst"]["scan"]["capability"]
del csp_config_2_3["pst"]["scan"]["scan_id"]
del csp_config_2_3["pst"]["scan"]["subarray_id"]
csp_config_2_3["pst"]["scan"]["num_channelization_stages"] = 1
csp_config_2_3["pst"]["scan"]["channelization_stages"] = [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
]
elif scan == "pst_scan_vr":
csp_config_2_3["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "VOLTAGE_RECORDER",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "LIN",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
},
}
return csp_config_2_3
def get_csp_config_scan_2_4_example(scan: str = None) -> dict:
csp_config_2_4 = get_csp_config_scan_2_3_example(scan)
csp_config_2_4[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.4"
csp_config_2_4["common"]["eb_id"] = "eb-m001-20230712-56789"
if scan == "pst_scan_pt":
csp_config_2_4["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "1",
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.81480,
"observation_mode": "PULSAR_TIMING",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 10000.5,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
"pt": {
"dispersion_measure": 100.0,
"rotation_measure": 0.0,
"ephemeris": "",
"pulsar_phase_predictor": "",
"output_frequency_channels": 1,
"output_phase_bins": 64,
"num_sk_config": 1,
"sk_config": [
{
"sk_range": [0.8, 0.9],
"sk_integration_limit": 100,
"sk_excision_limit": 25.0,
}
],
"target_snr": 0.0,
},
},
}
elif scan == "pst_scan_ds":
csp_config_2_4["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "1",
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "DYNAMIC_SPECTRUM",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 13000.2,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
"ds": {
"dispersion_measure": 100.0,
"output_frequency_channels": 1,
"stokes_parameters": "Q",
"num_bits_out": 16,
"time_decimation_factor": 10,
"frequency_decimation_factor": 4,
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
},
},
}
elif scan == "pst_scan_ft":
csp_config_2_4["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "1",
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "FLOW_THROUGH",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "CIRC",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_rfi_frequency_masks": 1,
"rfi_frequency_masks": [[1.0, 1.1]],
"destination_address": ["192.168.178.26", 9021],
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
"ft": {
"num_bits_out": 32,
"num_channels": 1,
"channels": [1],
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
},
},
}
elif scan == "pst_scan_vr":
csp_config_2_4["common"]["frequency_band"] = "low"
csp_config_2_4["pst"] = {
"scan": {
"activation_time": "2022-01-19T23:07:45Z",
"timing_beam_id": "1",
"bits_per_sample": 32,
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 361689.8148,
"observation_mode": "VOLTAGE_RECORDER",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "LIN",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21.884",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_channelization_stages": 1,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 10,
"oversampling_ratio": [8, 7],
}
],
},
}
return csp_config_2_4
def get_csp_config_scan_2_5_example(scan: str = None) -> dict:
csp_config_2_5 = get_csp_config_scan_2_4_example(scan)
csp_config_2_5[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.5"
if scan == "pst_scan_ft":
csp_config_2_5["pst"]["scan"]["ft"] = {
"num_bits_out": 4,
"channels": [0, 24299],
"polarizations": "Both",
"requantisation_scale": 1.0,
"requantisation_init_time": 1.0,
}
return csp_config_2_5
def get_csp_config_scan_2_6_example(scan: str = None) -> dict:
csp_config_2_6 = get_csp_config_scan_2_5_example(scan)
csp_config_2_6[
"interface"
] = "https://schema.skao.int/ska-csp-configure/2.6"
csp_config_2_6["transaction_id"] = "txn-....-00001"
return csp_config_2_6
def get_csp_config_scan_3_0_example(scan: str = None) -> dict:
csp_config_3_0 = get_csp_config_scan_2_6_example(scan)
csp_config_3_0[
"interface"
] = "https://schema.skao.int/ska-csp-configurescan/3.0"
csp_config_3_0["cbf"][
"delay_model_subscription_point"
] = "ska_mid/tm_leaf_node/csp_subarray_01/delayModel"
del csp_config_3_0["subarray"]
del csp_config_3_0["cbf"]["fsp"][0]["zoom_factor"]
del csp_config_3_0["cbf"]["fsp"][1]["zoom_factor"]
del csp_config_3_0["cbf"]["fsp"][1]["zoom_window_tuning"]
csp_config_3_0["cbf"]["fsp"][0]["output_link_map"] = [[0, 1]]
csp_config_3_0["cbf"]["fsp"][1]["output_link_map"] = [[0, 1]]
csp_config_3_0["cbf"]["fsp"][1]["channel_offset"] = 740
if scan == "cal_a":
csp_config_3_0["cbf"]["fsp"][0]["output_port"] = expand_output_port(
[[0, 9000]], 740
)
csp_config_3_0["cbf"]["fsp"][1]["output_port"] = expand_output_port(
[[0, 9037]], 14140
)
csp_config_3_0["cbf"]["fsp"][1]["output_host"] = [[0, "192.168.1.1"]]
elif scan == "science_a":
del csp_config_3_0["cbf"]["fsp"][0]["output_mac"]
csp_config_3_0["cbf"]["fsp"][0]["output_host"] = [
[0, "192.168.0.1"],
[400, "192.168.0.2"],
]
csp_config_3_0["cbf"]["fsp"][0]["output_port"] = expand_output_port(
[[0, 9000], [400, 9000]], 740
)
del csp_config_3_0["cbf"]["fsp"][1]["output_mac"]
csp_config_3_0["cbf"]["fsp"][1]["output_host"] = [
[0, "192.168.0.3"],
[400, "192.168.0.4"],
]
csp_config_3_0["cbf"]["fsp"][1]["output_port"] = expand_output_port(
[[0, 9000], [400, 9000]], 14140
)
if scan == "pss":
csp_config_3_0["cbf"]["search_window"] = [
{
"search_window_id": 0,
"search_window_tuning": 1000,
},
]
csp_config_3_0["cbf"]["fsp"][1]["output_link_map"] = [[0, 1]]
elif scan == "pst_scan_vr":
csp_config_3_0["common"]["frequency_band"] = "1"
return csp_config_3_0
def get_csp_config_scan_4_0_example(scan: str = None) -> dict:
csp_config_4_0 = get_csp_config_scan_3_0_example(scan)
csp_config_4_0[
"interface"
] = "https://schema.skao.int/ska-csp-configurescan/4.0"
csp_config_4_0["midcbf"] = csp_config_4_0.pop("cbf")
del csp_config_4_0["midcbf"]["fsp"]
csp_config_4_0["midcbf"]["correlation"] = {
"processing_regions": [
{
"fsp_ids": [1],
"start_freq": 350000000,
"channel_width": 13440,
"channel_count": 740,
"sdp_start_channel_id": 0,
"integration_factor": 1,
"output_link_map": [[0, 1]],
},
{
"fsp_ids": [4],
"start_freq": 750000000,
"channel_width": 13440,
"channel_count": 14140,
"sdp_start_channel_id": 740,
"integration_factor": 1,
"output_link_map": [[740, 1]],
},
]
}
if scan in [
"cal_a",
"pss",
"pst_beam",
"pst_scan_pt",
"pst_scan_ds",
"pst_scan_ft",
"pst_scan_vr",
]:
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][0][
"output_port"
] = expand_output_port([[0, 9000]], 740)
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][1][
"output_port"
] = expand_output_port([[740, 9037]], 14880)
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][0][
"output_host"
] = [[0, "192.168.1.1"]]
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][1][
"output_host"
] = [[740, "192.168.1.1"]]
elif scan == "science_a":
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][0][
"output_port"
] = expand_output_port([[0, 9000], [400, 9000]], 740)
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][1][
"output_port"
] = expand_output_port([[740, 9000], [1140, 9000]], 14880)
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][0][
"output_host"
] = [
[0, "192.168.0.1"],
[400, "192.168.0.2"],
]
csp_config_4_0["midcbf"]["correlation"]["processing_regions"][1][
"output_host"
] = [
[740, "192.168.0.3"],
[1140, "192.168.0.4"],
]
return csp_config_4_0
def get_csp_config_scan_4_1_example(scan: str = None) -> dict:
csp_config_4_1 = get_csp_config_scan_4_0_example(scan)
csp_config_4_1[
"interface"
] = "https://schema.skao.int/ska-csp-configurescan/4.1"
return csp_config_4_1
def get_csp_config_scan_5_0_example(scan: str = None) -> dict:
csp_config_5_0 = get_csp_config_scan_4_1_example(scan)
csp_config_5_0[
"interface"
] = "https://schema.skao.int/ska-csp-configurescan/5.0"
if scan == "pst_bf":
csp_config_5_0["midcbf"]["pst_bf"] = {
"processing_regions": [
{
"fsp_ids": [5],
"start_freq": 495075840,
"channel_count": 3700,
"pst_start_channel_id": 0,
"timing_beams": [
{
"timing_beam_id": 1,
"receptors": ["SKA063", "SKA001"],
"output_link_map": [[0, 1]],
"output_host": [[0, "192.168.178.26"]],
"output_port": [[0, 20000]],
}
],
},
],
}
return csp_config_5_0
def get_csp_config_scan_6_0_example(scan: str = None) -> dict:
csp_config_6_0 = get_csp_config_scan_5_0_example(scan)
csp_config_6_0[
"interface"
] = "https://schema.skao.int/ska-csp-configurescan/6.0"
if scan == "pss":
csp_config_6_0["pss"] = get_pss_configure_example(
"https://schema.skao.int/ska-pss-configure/1.0"
)
return csp_config_6_0
def get_csp_config_scan_6_1_example(version: str, scan: str = None) -> dict:
csp_config_6_1 = get_csp_config_scan_6_0_example(scan)
csp_config_6_1["interface"] = version
if scan == "pss":
_, pss_uri = get_csp_config_subsystem_version(version, "pss")
csp_config_6_1["pss"] = get_pss_configure_example(pss_uri)
return csp_config_6_1
def get_csp_config_scan_7_0_example(version: str, scan: str = None) -> dict:
csp_config_7_0 = get_csp_config_scan_6_1_example(version, scan)
csp_config_7_0["interface"] = version
if scan and scan not in ["pst_bf", "cal_a", "science_a", "pss"]:
scan_type = "mid_" + scan
_, pst_uri = get_csp_config_subsystem_version(version, "pst")
# pst_version = PST_CONFIG_VER3_0
csp_config_7_0["midcbf"]["pst_bf"] = {
"processing_regions": [
{
"fsp_ids": [5],
"start_freq": 495075840,
"channel_count": 3700,
"pst_start_channel_id": 0,
"timing_beams": [
{
"timing_beam_id": 1,
"receptors": ["SKA063", "SKA001"],
"output_link_map": [[0, 1]],
"output_host": [[0, "192.168.178.26"]],
"output_port": [[0, 20000]],
}
],
},
],
}
pst_scan = get_pst_config_example(pst_uri, scan_type)
csp_config_7_0["pst"] = {"beams": [{"beam_id": 1, "scan": pst_scan}]}
delete_config_pst_keys(csp_config_7_0)
else:
if "pst" in csp_config_7_0:
del csp_config_7_0["pst"]
return csp_config_7_0
def get_csp_config_scan_8_0_example(version: str, scan: str = None) -> dict:
csp_config_8_0 = get_csp_config_scan_7_0_example(version, scan)
csp_config_8_0["interface"] = version
if "pss" in csp_config_8_0:
update_pss_config_example(csp_config_8_0)
return csp_config_8_0
[docs]
def delete_config_pst_keys(json_example):
"""Remove PST schema attributes that are not needed in the csp schema."""
for beam in json_example["pst"]["beams"]:
beam["scan"] = beam["scan"]["pst"]["scan"]
del beam["scan"]["timing_beam_id"]
[docs]
def update_pss_config_example(json_example):
"""Remove PSS example attributes that are not needed in the csp example."""
del json_example["pss"]["common"]
del json_example["pss"]["interface"]
[docs]
def get_csp_config_example(version: str, scan: str = None) -> dict:
"""Generate examples for CSP configuration strings
:param version: Version URI of configuration format
:param scan: Includes SDP receive addresses for a scan? `None`
means that this is "template" configuration as passed to
TMC. Valid parameters: cal_a, science_a
"""
version_number = check_csp_interface_version(
version, [CSP_CONFIG_PREFIX, CSP_CONFIGSCAN_PREFIX]
)
(major, minor) = split_interface_version(version)
scans_and_min_versions = {
None: (0, 0),
"cal_a": (0, 0),
"science_a": (0, 0),
"pss": (2, 1),
"pst_beam": (2, 2),
"pst_scan_pt": (2, 2),
"pst_scan_ds": (2, 2),
"pst_scan_ft": (2, 2),
"pst_scan_vr": (2, 3),
"pst_bf": (5, 0),
"pst_scan_df": (7, 0),
}
if scan not in scans_and_min_versions:
raise ValueError(f"Invalid scan name: {scan}!")
if scans_and_min_versions[scan] > (major, minor):
raise ValueError(
cleandoc(
f"""
Version: {version_number} for scan type: {scan} is too old.
Scan type is not supported in this earlier version number.
"""
)
)
if version_number == "0.0":
example = get_csp_config_scan_0_1_example(scan)
elif version_number == "0.1":
example = get_csp_config_scan_0_1_example(scan)
elif version_number == "1.0":
example = get_csp_config_scan_1_0_example(scan)
elif version_number == "2.0":
example = get_csp_config_scan_2_0_example(scan)
elif version_number == "2.1":
example = get_csp_config_scan_2_1_example(scan)
elif version_number == "2.2":
example = get_csp_config_scan_2_2_example(scan)
elif version_number == "2.3":
example = get_csp_config_scan_2_3_example(scan)
elif version_number == "2.4":
example = get_csp_config_scan_2_4_example(scan)
elif version_number == "2.5":
example = get_csp_config_scan_2_5_example(scan)
elif version_number == "2.6":
example = get_csp_config_scan_2_6_example(scan)
elif version_number == "3.0":
example = get_csp_config_scan_3_0_example(scan)
elif version_number == "4.0":
example = get_csp_config_scan_4_0_example(scan)
elif version_number == "4.1":
example = get_csp_config_scan_4_1_example(scan)
elif version_number == "5.0":
example = get_csp_config_scan_5_0_example(scan)
elif version_number == "6.0":
example = get_csp_config_scan_6_0_example(scan)
elif version_number == "6.1":
example = get_csp_config_scan_6_1_example(version, scan)
elif version_number == "7.0":
example = get_csp_config_scan_7_0_example(version, scan)
elif version_number == "8.0":
example = get_csp_config_scan_8_0_example(version, scan)
else:
raise ValueError(f"Could not generate example for schema {version}!")
return copy.deepcopy(example)
def get_csp_scan_2_2_example() -> dict:
csp_scan_2_2 = {
"interface": "https://schema.skao.int/ska-csp-scan/2.2",
"scan_id": 7,
}
return csp_scan_2_2
def get_csp_scan_2_3_example() -> dict:
csp_scan_2_3 = get_csp_scan_2_2_example()
csp_scan_2_3["interface"] = "https://schema.skao.int/ska-csp-scan/2.3"
csp_scan_2_3["transaction_id"] = "txn-....-00001"
return csp_scan_2_3
[docs]
def get_csp_scan_example(version: str) -> dict:
"""Generate example of CSP scan argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(version, CSP_SCAN_PREFIX)
if version_number == "2.2":
scan_example = get_csp_scan_2_2_example()
elif version_number == "2.3":
scan_example = get_csp_scan_2_3_example()
else:
raise ValueError(f"Could not generate example for schema {version}!")
return scan_example
def get_csp_assignresources_2_2_example() -> dict:
csp_assignresources_2_2 = {
"interface": "https://schema.skao.int/ska-csp-assignresources/2.2",
"subarray_id": 1,
"dish": {
"receptor_ids": [
"SKA001",
"SKA036",
]
},
}
return csp_assignresources_2_2
def get_csp_assignresources_2_3_example() -> dict:
csp_assignresources_2_3 = get_csp_assignresources_2_2_example()
csp_assignresources_2_3[
"interface"
] = "https://schema.skao.int/ska-csp-assignresources/2.3"
csp_assignresources_2_3["transaction_id"] = "txn-....-00001"
return csp_assignresources_2_3
def get_csp_assignresources_3_0_example() -> dict:
csp_assignresources_3_0 = get_csp_assignresources_2_3_example()
csp_assignresources_3_0[
"interface"
] = "https://schema.skao.int/ska-csp-assignresources/3.0"
return csp_assignresources_3_0
def get_csp_assignresources_3_1_example() -> dict:
csp_assignresources_3_1 = get_csp_assignresources_3_0_example()
csp_assignresources_3_1[
"interface"
] = "https://schema.skao.int/ska-csp-assignresources/3.1"
csp_assignresources_3_1["pst"] = {"pst_beam_ids": [1]}
return csp_assignresources_3_1
def get_csp_assignresources_3_2_example() -> dict:
csp_assignresources_3_2 = get_csp_assignresources_3_1_example()
csp_assignresources_3_2[
"interface"
] = "https://schema.skao.int/ska-csp-assignresources/3.2"
csp_assignresources_3_2["pss"] = {"pss_beam_ids": [1]}
return csp_assignresources_3_2
[docs]
def get_csp_assignresources_example(version: str) -> dict:
"""Generate example of CSP assignresources argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(
version, CSP_ASSIGNRESOURCES_PREFIX
)
if version_number == "2.2":
assign_resource_example = get_csp_assignresources_2_2_example()
elif version_number == "2.3":
assign_resource_example = get_csp_assignresources_2_3_example()
elif version_number == "3.0":
assign_resource_example = get_csp_assignresources_3_0_example()
elif version_number == "3.1":
assign_resource_example = get_csp_assignresources_3_1_example()
elif version_number == "3.2":
assign_resource_example = get_csp_assignresources_3_2_example()
else:
raise ValueError(f"Could not generate example for schema {version}!")
return assign_resource_example
def get_csp_endscan_2_2_example() -> dict:
csp_endscan_2_2 = {
"interface": "https://schema.skao.int/ska-csp-endscan/2.2",
"scan_id": 15,
}
return csp_endscan_2_2
def get_csp_endscan_2_3_example() -> dict:
csp_endscan_2_3 = get_csp_endscan_2_2_example()
csp_endscan_2_3[
"interface"
] = "https://schema.skao.int/ska-csp-endscan/2.3"
csp_endscan_2_3["transaction_id"] = "txn-....-00001"
return csp_endscan_2_3
[docs]
def get_csp_endscan_example(version: str) -> dict:
"""Generate example of CSP endscan argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(version, CSP_ENDSCAN_PREFIX)
if version_number == "2.2":
endscan_example = get_csp_endscan_2_2_example()
elif version_number == "2.3":
endscan_example = get_csp_endscan_2_3_example()
else:
raise ValueError(f"Could not generate example for schema {version}!")
return endscan_example
def get_csp_releaseresources_2_2_example_1() -> dict:
csp_releaseresources_2_2_example_1 = {
"interface": "https://schema.skao.int/ska-csp-releaseresources/2.2",
"subarray_id": 1,
"release_all": True,
}
return csp_releaseresources_2_2_example_1
def get_csp_releaseresources_2_2_example_2() -> dict:
csp_releaseresources_2_2_example_2 = {
"interface": "https://schema.skao.int/ska-csp-releaseresources/2.2",
"subarray_id": 2,
"receptor_ids": [
"SKA001",
"SKA036",
],
}
return csp_releaseresources_2_2_example_2
def get_csp_releaseresources_2_3_example_1() -> dict:
csp_releaseresources_2_3_example_1 = (
get_csp_releaseresources_2_2_example_1()
)
csp_releaseresources_2_3_example_1[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/2.3"
csp_releaseresources_2_3_example_1["transaction_id"] = "txn-....-00001"
return csp_releaseresources_2_3_example_1
def get_csp_releaseresources_2_3_example_2() -> dict:
csp_releaseresources_2_3_example_2 = (
get_csp_releaseresources_2_2_example_2()
)
csp_releaseresources_2_3_example_2[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/2.3"
csp_releaseresources_2_3_example_2["transaction_id"] = "txn-....-00001"
return csp_releaseresources_2_3_example_2
def get_csp_releaseresources_3_0_example_1() -> dict:
csp_releaseresources_3_0_example_1 = (
get_csp_releaseresources_2_3_example_1()
)
csp_releaseresources_3_0_example_1[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.0"
return csp_releaseresources_3_0_example_1
def get_csp_releaseresources_3_0_example_2() -> dict:
csp_releaseresources_3_0_example_2 = (
get_csp_releaseresources_2_3_example_2()
)
csp_releaseresources_3_0_example_2[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.0"
return csp_releaseresources_3_0_example_2
def get_csp_releaseresources_3_1_example_2() -> dict:
csp_releaseresources_3_1_example_2 = (
get_csp_releaseresources_2_3_example_2()
)
csp_releaseresources_3_1_example_2["pst"] = {"pst_beam_ids": [1]}
csp_releaseresources_3_1_example_2[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.1"
return csp_releaseresources_3_1_example_2
def get_csp_releaseresources_3_1_example_1() -> dict:
csp_releaseresources_3_1_example_1 = (
get_csp_releaseresources_2_3_example_1()
)
csp_releaseresources_3_1_example_1[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.1"
return csp_releaseresources_3_1_example_1
def get_csp_releaseresources_3_2_example_2() -> dict:
csp_releaseresources_3_2_example_2 = (
get_csp_releaseresources_3_1_example_2()
)
csp_releaseresources_3_2_example_2["pss"] = {"pss_beam_ids": [1]}
csp_releaseresources_3_2_example_2[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.2"
return csp_releaseresources_3_2_example_2
def get_csp_releaseresources_3_2_example_1() -> dict:
csp_releaseresources_3_2_example_1 = (
get_csp_releaseresources_3_1_example_1()
)
csp_releaseresources_3_2_example_1[
"interface"
] = "https://schema.skao.int/ska-csp-releaseresources/3.2"
return csp_releaseresources_3_2_example_1
[docs]
def get_csp_releaseresources_example(
version: str, example: str = None
) -> dict:
"""Generate example of CSP releaseresources argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(
version, CSP_RELEASERESOURCES_PREFIX
)
if version_number == "2.2" and example == "2":
releaseresources_example = get_csp_releaseresources_2_2_example_2()
elif version_number == "2.2":
releaseresources_example = get_csp_releaseresources_2_2_example_1()
elif version_number == "2.3" and example == "2":
releaseresources_example = get_csp_releaseresources_2_2_example_2()
elif version_number == "2.3":
releaseresources_example = get_csp_releaseresources_2_2_example_1()
elif version_number == "3.0" and example == "2":
releaseresources_example = get_csp_releaseresources_3_0_example_2()
elif version_number == "3.0":
releaseresources_example = get_csp_releaseresources_3_0_example_1()
elif version_number == "3.1" and example == "2":
releaseresources_example = get_csp_releaseresources_3_1_example_2()
elif version_number == "3.1":
releaseresources_example = get_csp_releaseresources_3_1_example_1()
elif version_number == "3.2" and example == "2":
releaseresources_example = get_csp_releaseresources_3_2_example_2()
elif version_number == "3.2":
releaseresources_example = get_csp_releaseresources_3_2_example_1()
else:
raise ValueError(f"Could not generate example for schema {version}!")
return releaseresources_example
def get_csp_delay_model_2_2_example() -> dict:
csp_delaymodel_2_2 = {
"interface": "https://schema.skao.int/ska-csp-delaymodel/2.2",
"epoch": 12345678.123456,
"validity_period": 10.0,
"delay_details": [
{
"receptor": "SKA001",
"poly_info": [
{
"polarization": "X",
"coeffs": [1.01, 1.02, 1.03, 1.04, 1.05, 1.06],
},
{
"polarization": "Y",
"coeffs": [1.1, 1.2, 1.3, 1.4, 1.5, 1.6],
},
],
},
{
"receptor": "SKA100",
"poly_info": [
{
"polarization": "X",
"coeffs": [1.101, 1.102, 1.103, 1.104, 1.105, 1.106],
},
{
"polarization": "Y",
"coeffs": [1.11, 1.12, 1.13, 1.14, 1.15, 1.16],
},
],
},
],
}
return csp_delaymodel_2_2
def get_csp_mid_delaymodel_3_0_example() -> dict:
csp_mid_delaymodel_3_0 = {
"interface": "https://schema.skao.int/ska-mid-csp-delaymodel/3.0",
"start_validity_sec": 748656000.0,
"cadence_sec": 10.0,
"validity_period_sec": 30.0,
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray": 2,
"receptor_delays": [
{
"receptor": "SKA001",
"xypol_coeffs_ns": [
7.5e2,
4.6e-3,
-2.0e-6,
-4.1e-12,
9.0e-16,
-1.9e-19,
],
"ypol_offset_ns": -0.1,
},
{
"receptor": "SKA002",
"xypol_coeffs_ns": [
7.5e2,
4.6e-3,
-2.0e-6,
-4.1e-12,
9.0e-16,
-1.9e-19,
],
"ypol_offset_ns": -0.1,
},
],
}
return csp_mid_delaymodel_3_0
[docs]
def get_csp_delaymodel_example(version: str) -> dict:
"""Generate example of CSP delay model argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(
version, [CSP_DELAYMODEL_PREFIX, CSP_MID_DELAYMODEL_PREFIX]
)
if version_number == "2.2":
return get_csp_delay_model_2_2_example()
elif version_number == "3.0":
return get_csp_mid_delaymodel_3_0_example()
else:
raise ValueError(f"Could not generate example for schema {version}!")
def get_csp_low_delaymodel_1_0_example() -> dict:
csp_low_delaymodel_1_0 = {
"interface": "https://schema.skao.int/ska-low-csp-delaymodel/1.0",
"start_validity_sec": 748656000.0,
"cadence_sec": 10.0,
"validity_period_sec": 600.0,
"config_id": "sbi-mvp02-20200325-00001-science_A",
"station_beam": 5,
"subarray": 2,
"station_beam_delays": [
{
"station_id": 512,
"substation_id": 3,
"xypol_coeffs_ns": [
7.5e2,
4.6e-3,
-2.0e-6,
-4.1e-12,
9.0e-16,
-1.9e-19,
],
"ypol_offset_ns": -0.1,
},
{
"station_id": 1,
"substation_id": 0,
"xypol_coeffs_ns": [
7.5e2,
4.6e-3,
-2.0e-6,
-4.1e-12,
9.0e-16,
-1.9e-19,
],
"ypol_offset_ns": 0.5,
},
],
}
return csp_low_delaymodel_1_0
[docs]
def get_csp_low_delaymodel_example(version: str) -> dict:
"""Generate example of CSP low delay model argument
:param version: Version URI of configuration format
"""
version_number = check_csp_interface_version(
version, CSP_LOW_DELAYMODEL_PREFIX
)
if version_number == "1.0":
delaymodel_example = get_csp_low_delaymodel_1_0_example()
else:
raise ValueError(f"Could not generate example for schema {version}!")
return delaymodel_example
[docs]
def expand_output_port(
input_mapping: list[list[int]],
max_channels: int = MAX_CHANNELS_PER_STREAM * MAX_STREAMS_PER_FSP,
) -> list[list[int]]:
"""
Expand output_port mapping to give concrete ports for each stream of 20
channels. Port value is set from initial port and incremented by the
port_increment value if given, or 1 if the increment is not present.
:param input_mapping: output_port mapping to convert from
"""
output_mapping = []
for index, entry in enumerate(input_mapping):
start_channel = entry[0]
port = entry[1]
port_increment = entry[2] if len(entry) > 2 else 1
# Expand entries up until the next concrete output_port entry
# if we're at the last entry, then we expand until the last possible
# channel
if (index + 1) == len(input_mapping):
end_channel = max_channels
else:
end_channel = input_mapping[index + 1][0]
for channel in range(
start_channel, end_channel, MAX_CHANNELS_PER_STREAM
):
new_entry = [channel, port]
output_mapping.append(new_entry)
port = port + port_increment
return output_mapping