"""PSS schema definitions."""
import copy
from inspect import cleandoc
from schema import Regex
from .._common import (
TMSchema,
get_unique_id_schema,
lookup_schema,
split_interface_version,
)
from . import cheetah_schemas
[docs]
def get_pss_cheetah_config_beam_schema(version: str, strict: int) -> TMSchema:
"""Method that aims to get the PSS Cheetah Beam schema.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The Cheetah Beam configure JSON Schema.
"""
return lookup_schema(
cheetah_schemas, version, strict, "get_pss_beams_config_schema"
)
def _get_common_config_schema(version: str, strict: int) -> TMSchema:
"""Method that aims to generate the Common elements for the PSS configure
schema.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The Common JSON Schema for the PSS configure.
"""
common_schema = TMSchema.new(
"Common configuration schema",
version,
strict,
description=cleandoc(
"""
Common section, containing the parameters and the sections
belonging to all CSP subsystems. This section is forwarded to
all sub-elements.
"""
),
as_reference=True,
)
common_schema.add_field("config_id", str)
common_schema.add_field(
"subarray_id",
int,
check_strict=lambda n: n >= 1 and n <= 16,
description=cleandoc(
"""
The Subarray ID that the list of receptors will be
assigned to.
For Mid, there are a maximum of 16 subarrays.
Range: Integer from 1-16 inclusive
"""
),
)
common_schema.add_opt_field(
"eb_id",
get_unique_id_schema(strict, r"eb"),
description=cleandoc(
"""
Execution block ID to associate scan configs to an observation.
This ID is used for associating generated data, especially
data products, for a given observation. Multiple scans can
be linked to one observation and this ID is used as metadata
to associate the data products from all scans of the same
observation.
This ID does not have to be unique for a scan configuration but
should be unique for different observations.
For example, all the data and weights files will have an
EB_ID header value populated with the value supplied in this
field.
"""
),
)
common_schema.add_field(
"frequency_band",
(Regex(r"^(1|2|3|4|5(a|b)|low)$") if strict else str),
description=cleandoc(
"""
Frequency band applies for all the receptors (VCCs)
that belong to the sub-array.
The value of 'low' is used to only within SKA Low.
As this field is a mandatory field but bands 1, 2,
3, 4, 5a and 5b only make sense for SKA Mid.
"""
),
)
return common_schema
[docs]
def update_pss_config_cheetah_beams_schema(
cheetah_version: str,
strict: bool,
) -> TMSchema:
"""Mid and Low Common method that aims to update the PSS cheetah schema
for the PSS subarray.
:param version: Interface version URI
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The Cheetah JSON Schema for the PSS configure.
"""
cheetah_schema = TMSchema.new(
"PSS cheetah beams configure",
cheetah_version,
strict,
description="Configuration for the Cheetah beams",
as_reference=True,
)
cheetah_schema.add_field(
"cheetah_id",
int,
check_strict=lambda x: x > 0,
description="Pipeline ID.",
)
beam_schema = get_pss_cheetah_config_beam_schema(cheetah_version, strict)
_update_pss_cheetah_configure_schema(beam_schema)
cheetah_schema.add_field(
"beams",
[beam_schema],
check_strict=lambda x: len(x) <= 3,
)
return cheetah_schema
def _update_pss_cheetah_configure_schema(beam_schema: TMSchema) -> None:
"""Updated the cheetah beams JSON schema.
:param beam_schema: Beam part of the Cheetah schema
:return: The PSS JSON Schema with the Beams attribute updated
for the subarray level configurescan command.
"""
sink_el = beam_schema["beam"]["sinks"]["channels"]["sps_events"]
sink_el.__delitem__("sink")
sink_el.add_field(
"sink",
[{"sink_id": str}],
description="activate the channel.",
)
# rename spccl_files id (remove id)
spccl_sigproc_el = beam_schema["beam"]["sinks"]["sink_configs"][
"spccl_files"
]
spccl_sigproc_el.__delitem__("id")
spccl_sigproc_el.add_field(
"sink_id",
str,
description="Sink ID.",
)
# rename sigproc id (remove id)
spccl_sigproc_el = beam_schema["beam"]["sinks"]["sink_configs"][
"spccl_sigproc_files"
]
spccl_sigproc_el.__delitem__("id")
spccl_sigproc_el.add_field(
"sink_id",
str,
description="Sink ID.",
)
# rename beam id (remove id)
beam_schema["beam"].__delitem__("id")
beam_schema["beam"].add_field(
"beam_id",
int,
description="Beam ID",
)
# add flag to determine the beam source
beam_schema["beam"]["source"]["sigproc"].add_field(
"active",
bool,
description=(
"activate the sigproc data source. Only one source between"
"'sigproc' and 'udp_low' should be active at the same time"
),
)
beam_schema["beam"]["source"]["udp_low"].add_field(
"active",
bool,
description=(
"activate the udp_low data source. Only one source between"
"'sigproc' and 'udp_low' should be active at the same time"
),
)
# remove listen attribute from UDP_LOW source
beam_schema["beam"]["source"]["udp_low"].__delitem__("listen")