"""
Interface module for generating CSP configuration.
Handles parsing and validation of inputs and passes them on to the
internal configuration functions in :py:mod:`.config`.
"""
import copy
import json
from typing import Union
from .. import schema
from . import config, version
from .low_version import LOWCSP_CONFIGURE_PREFIX
[docs]
def make_mid_csp_config(
csp_interface_version: str,
sdp_interface_version: str,
scan_type: str,
csp_config_str: Union[str, dict],
sdp_receive_addrs_map_str: Union[str, dict],
) -> dict:
"""Generate CSP scan configuration for a scan using SDP receive addresses.
This should be used right before CSP is configured so that data
streams are sent to the right ingest nodes.
:param csp_interface_version: Version of CSP interface (URI)
:param sdp_interface_version: Version of SDP interface (URI)
:param scan_type: Type of scan to configure
:param csp_config_str: General CSP configuration
:param sdp_receive_addrs_map_str: Receive addresses map for scan types,
generated by SDP
:return: new CSP configuration. dict format.
:raise: `ValueError` when the input JSON configuration fails validation.
"""
# Parse JSON
if isinstance(csp_config_str, str):
csp_config = json.loads(csp_config_str)
else:
csp_config = copy.deepcopy(dict(csp_config_str))
if isinstance(sdp_receive_addrs_map_str, str):
sdp_receive_addrs = json.loads(sdp_receive_addrs_map_str)
else:
sdp_receive_addrs = dict(sdp_receive_addrs_map_str)
# Convert version to standard format
csp_interface_version = version.normalize_csp_config_version(
csp_interface_version, csp_config
)
# Valid?
version.check_csp_interface_version(
csp_interface_version,
[version.CSP_CONFIG_PREFIX, version.CSP_CONFIGSCAN_PREFIX],
)
schema.validate(csp_interface_version, csp_config)
# Get receive addresses for scan type
if scan_type not in sdp_receive_addrs:
raise ValueError(
f"No receive addresses found for scan type '{scan_type}'!"
)
scan_receive_addrs = sdp_receive_addrs[scan_type]
# Add receive addresses into CSP configuration
csp_config = config.add_receive_addresses(
scan_type,
csp_config,
scan_receive_addrs,
csp_interface_version,
sdp_interface_version,
telescope_branch="mid",
)
# Do schema check
schema.validate(csp_interface_version, csp_config)
return csp_config
[docs]
def make_low_csp_config(
csp_interface_version: str,
sdp_interface_version: str,
scan_type: str,
csp_config_str: Union[str, dict],
sdp_receive_addrs_map_str: Union[str, dict],
) -> dict:
"""
Generate CSP scan configuration for a scan using SDP receive addresses.
This should be used right before CSP is configured so that data
streams are sent to the right ingest nodes.
:param csp_interface_version: Version of CSP interface (URI)
:param sdp_interface_version: Version of SDP interface (URI)
:param scan_type: Type of scan to configure
:param csp_config_str: General CSP configuration
:param sdp_receive_addrs_map_str: Receive addresses map for scan types,
generated by SDP
:return: new CSP configuration. dict format.
:raise: `ValueError` when the input JSON configuration fails validation.
:raise: `ValueError` when the input interface version fails validation.
:raise: `ValueError` when the input telescope_branch fails validation.
:raise: `ValueError` when the input scan_type not in sdp_receive_addrs
"""
# Parse JSON
if isinstance(csp_config_str, str):
csp_config = json.loads(csp_config_str)
else:
csp_config = copy.deepcopy(dict(csp_config_str))
if isinstance(sdp_receive_addrs_map_str, str):
sdp_receive_addrs = json.loads(sdp_receive_addrs_map_str)
else:
sdp_receive_addrs = dict(sdp_receive_addrs_map_str)
# Valid?
version.check_csp_interface_version(
csp_interface_version,
LOWCSP_CONFIGURE_PREFIX,
)
# in low this line does not work because the sdo addresses
# (host and port) keys are mandatory in the TMC-CSP interface!
# schema.validate(csp_interface_version, csp_config)
# Get receive addresses for scan type
if scan_type not in sdp_receive_addrs:
raise ValueError(
f"No receive addresses found for scan type '{scan_type}'!"
)
scan_receive_addrs = sdp_receive_addrs[scan_type]
# Add receive addresses into CSP configuration
# Which is the structure of the information received from SDP?
csp_config = config.add_receive_addresses(
scan_type,
csp_config,
scan_receive_addrs,
csp_interface_version,
sdp_interface_version,
telescope_branch="low",
)
# Do schema check
# DO we need this line? Maybe only if strictness ==2 ???
schema.validate(csp_interface_version, csp_config, strictness=2)
return csp_config
[docs]
def make_csp_config(
csp_interface_version: str,
sdp_interface_version: str,
scan_type: str,
csp_config_str: Union[str, dict],
sdp_receive_addrs_map_str: Union[str, dict],
telescope_branch: str = "mid",
) -> str:
"""Generate CSP scan configuration for a scan using SDP receive addresses.
This should be used right before CSP is configured so that data
streams are sent to the right ingest nodes.
:param csp_interface_version: Version of CSP interface (URI)
:param sdp_interface_version: Version of SDP interface (URI)
:param scan_type: Type of scan to configure
:param csp_config_str: General CSP configuration
:param sdp_receive_addrs_map_str: Receive addresses map for scan types,
generated by SDP
:param telescope_branch: flag to identify if the function is called by
"mid" or "low" telescope
:return: A validated JSON string with CSP configuration.
:raise: `ValueError` when the input JSON configuration fails validation.
"""
csp_config = {}
if telescope_branch.lower() == "mid":
csp_config = make_mid_csp_config(
csp_interface_version,
sdp_interface_version,
scan_type,
csp_config_str,
sdp_receive_addrs_map_str,
)
elif telescope_branch.lower() == "low":
csp_config = make_low_csp_config(
csp_interface_version,
sdp_interface_version,
scan_type,
csp_config_str,
sdp_receive_addrs_map_str,
)
# Translate back into JSON
return json.dumps(csp_config, indent=2)