"""
Functions to build the Low CSP JSON schemas for Subarray commmands.
"""
from inspect import cleandoc
from ska_telmodel.lowcbf.schema import (
get_lowcbf_assignresources_schema,
get_lowcbf_configurescan_schema,
get_lowcbf_releaseresources_schema,
get_lowcbf_scan_schema,
)
from .._common import TMSchema, split_interface_version
from .common_schema import _get_common_config_schema_without_band
from .low_csp_schema_for_oso_tmc import update_csp_conf_schema_for_oso_tmc
from .low_version import get_csp_subsystem_version
from .mid_low_schema import (
get_csp_pss_assign_release_resources,
get_csp_pss_configure_schema,
get_csp_pst_assign_release_resources,
get_csp_pst_config_schema,
)
from .schema import get_subarray_config_schema
[docs]
def get_low_csp_assignresources_schema(version: str, strict: bool) -> TMSchema:
"""
Build the assignresources JSON schema for the TM-CSP interface.
:param version: Interface Version URI
:param strict: Schema strictness
:return: Schema for assignresources command.
"""
csp_schema = TMSchema.new("LOWCSP assign resources", version, strict)
(major, minor) = split_interface_version(version)
csp_schema.add_field(
"interface",
str,
description="URI of JSON schema for this command's JSON payload.",
)
if (major, minor) >= (3, 2):
csp_schema.add_opt_field(
"transaction_id",
str,
description="A transaction id specific to the command",
)
_, uri_cbf = get_csp_subsystem_version(version, "lowcbf")
cbf_schema = get_lowcbf_assignresources_schema(uri_cbf, strict)
cbf_schema = _update_lowcbf_cmd_schema_for_csp(cbf_schema)
common_schema = TMSchema.new(
"LOWCSP releaseresources description", version, strict
)
common_schema.add_field(
"subarray_id", int, description=cleandoc("subarray id")
)
desc = "LOWCSP subarray id arguments"
csp_schema.add_field("common", common_schema, description=cleandoc(desc))
csp_schema.add_field(
"lowcbf",
cbf_schema,
description="Low CBF resources",
),
# PST schema
get_csp_pst_assign_release_resources(version, csp_schema, strict)
# PSS schema
get_csp_pss_assign_release_resources(version, csp_schema, strict)
return csp_schema
[docs]
def get_low_csp_releaseresources_schema(
version: str, strict: bool
) -> TMSchema:
"""
Build the releaseresources JSON schema for the TM-CSP interface.
:param version: Interface version
:param strict: Strict mode - refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The JSON Schema for the CSP releaseresources command.
:raise: `ValueError` exception on mismatch major version or invalid JSON
Schema URI
"""
csp_schema = TMSchema.new("LOWCSP release resources", version, strict)
(major, minor) = split_interface_version(version)
csp_schema.add_field(
"interface",
str,
description=cleandoc(
"URI of JSON schema for this command's" "JSON payload.."
),
)
if (major, minor) >= (3, 2):
csp_schema.add_opt_field(
"transaction_id",
str,
description="A transaction id specific to the command",
)
common_schema = TMSchema.new(
"LOWCSP releaseresources description",
version,
strict,
as_reference=True,
)
common_schema.add_field(
"subarray_id", int, description=cleandoc("subarray id")
)
desc = "LOWCSP subarray id arguments"
csp_schema.add_field("common", common_schema, description=cleandoc(desc))
# PST schema
get_csp_pst_assign_release_resources(version, csp_schema, strict)
# PSS schema
get_csp_pss_assign_release_resources(version, csp_schema, strict)
_, cbf_uri = get_csp_subsystem_version(version, "lowcbf")
cbf_schema = get_lowcbf_releaseresources_schema(cbf_uri, strict)
cbf_schema = _update_lowcbf_cmd_schema_for_csp(cbf_schema)
csp_schema.add_field(
"lowcbf",
cbf_schema,
)
return csp_schema
[docs]
def get_low_csp_scan_schema(version: str, strict: bool) -> TMSchema:
"""
Build the scan JSON schema for the TMC-CSP interface.
:param version: Interface Version URI
:param strict: Schema strictness
:return: Schema for configure command.
:raise: `ValueError` exception on invalid JSON Schema URI.
"""
items = TMSchema.new("LOWCSP scan description", version, strict)
(major, minor) = split_interface_version(version)
items.add_field(
"interface", str, description=cleandoc("LOW CSP SCAN interface")
)
if (major, minor) >= (3, 2):
items.add_opt_field(
"transaction_id",
str,
description="A transaction id specific to the command",
)
common_schema = TMSchema.new("LOWCSP common section", version, strict)
common_schema.add_field(
"subarray_id", int, description=cleandoc("subarray id")
)
desc = "LOWCSP subarray id arguments"
items.add_field("common", common_schema, description=desc)
major, minor = split_interface_version(version)
if (major, minor) < (4, 0):
_, cbf_uri = get_csp_subsystem_version(version, "lowcbf")
cbf_schema = get_lowcbf_scan_schema(cbf_uri, strict)
cbf_schema = _update_lowcbf_cmd_schema_for_csp(cbf_schema)
items.add_field(
"lowcbf", cbf_schema, description="LOW CBF scan schema"
)
else:
items.add_field("scan_id", int, description="Scan ID")
return items
def _update_lowcbf_configurescan_for_csp(
cbf_schema: TMSchema, version: str
) -> TMSchema:
"""
Updated the configurescan JSON schema for the TM-CSP interface, removing
the schema part specific for lowcbf.
:param lowcbf: Low CBF schema
:param version: Interface version
:return: The Low CBF JSON Schema for the CSP configurescan command.
"""
lowcbf = _update_lowcbf_cmd_schema_for_csp(cbf_schema)
if "timing_beams" in lowcbf:
cbf_version = split_interface_version(version) == (0, 1)
for beam in lowcbf["timing_beams"]["beams"]:
if cbf_version:
beam.__delitem__("dest_ip")
beam.__delitem__("dest_chans")
else:
beam.__delitem__("destinations")
if "search_beams" in lowcbf:
cbf_version = split_interface_version(version)
if cbf_version in [(0, 1), (0, 2), (1, 0)]:
pass
else:
for beam in lowcbf["search_beams"]["beams"]:
beam.__delitem__("destinations")
return lowcbf
def _update_lowcbf_cmd_schema_for_csp(cbf_schema: TMSchema) -> TMSchema:
"""Update Low CBF commands schema for the CSP schema.
:param cbf_schema: Low CBF schema
:return: TMSchema
"""
cbf_schema.__delitem__("interface")
lowcbf = cbf_schema["lowcbf"]
return lowcbf