"""
Contains the common methods between Mid and Low schema
"""
from inspect import cleandoc
from schema import Schema
from .._common import TMSchema, split_interface_version
from ..pss.schema import get_pss_configure_schema
from ..pst.schema import get_pst_config_schema
from .low_version import MAX_LOW_PSS_BEAMS
from .version import MAX_MID_PSS_BEAMS
[docs]
def get_csp_pst_config_schema(
csp_version: str,
pst_version: str,
strict: bool,
) -> TMSchema:
"""Mid and Low Common method that aims to generate the PST configure
schema for the CSP input file
:param pst_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 PST JSON Schema for the CSP configure.
"""
major, minor = split_interface_version(csp_version)
if "low" in csp_version or (
"low" not in csp_version and (major, minor) > (6, 1)
):
pst_schema = TMSchema.new(
"PST configure",
pst_version,
strict,
description=cleandoc(
"""
Main configuration for the
CSP Pulsar timing sub-system
"""
),
as_reference=True,
)
# PST schema
pst_beam_schema = TMSchema.new(
"CSP configurescan pst beams",
pst_version,
strict,
description=cleandoc("Parameters to configure the PST sub-system"),
# as_reference=True
)
pst_beam_schema.add_field(
"beam_id",
int,
description=cleandoc("Configuration for a PST beam ID in low csp"),
)
# Retrieve the PST configuration schema for scan and beam and
# add then to the CSP schema.
pst_beam_schema.update(get_pst_configure_schema(pst_version, strict))
# PST beams
pst_schema.add_field(
"beams",
[pst_beam_schema],
description=cleandoc("List of PST Beams IDs to configure"),
)
else:
pst_schema = get_pst_configure_schema(pst_version, strict)
return pst_schema
[docs]
def get_csp_pst_assign_release_resources(
csp_version: str, schema: TMSchema, strict: bool
) -> Schema:
"""
Add the PST assign/release resources schema to the CSP input schema
(assignresources/releaseresources).
:param csp_version: Interface version URI
:param schema: CSP assignresources/releaseresources schema
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The updated JSON Schema for the CSP assign/release resources
with PST element.
"""
(major, minor) = split_interface_version(csp_version)
update_beams_id_name = False
if "low" in csp_version:
# low schema rename attribute from version 6.0
update_beams_id_name = True if (major, minor) >= (6, 0) else False
else:
# mid schema rename attribute from version 3.2
update_beams_id_name = True if (major, minor) >= (3, 1) else False
pst_schema = TMSchema.new("CSP PST beams", csp_version, strict)
pst_schema.add_field(
"pst_beam_ids" if update_beams_id_name else "beams_id",
[int],
check_strict=lambda lst: all(1 <= beam_id <= 16 for beam_id in lst)
and len(lst) <= 16,
description="list of PST beam IDs",
)
if "low" in csp_version or (
"low" not in csp_version and (major, minor) >= (3, 1)
):
schema.add_opt_field(
"pst",
pst_schema,
description="PST section for the assign/release resources",
)
[docs]
def get_csp_pss_assign_release_resources(
csp_version: str, schema: TMSchema, strict: bool
) -> Schema:
"""
Add the PSS assign/release resources schema to the CSP input schema
(assignresources/releaseresources).
:param csp_version: Interface version URI
:param schema: CSP assignresources/releaseresources schema
:param strict: Strict mode. If true, refuse even harmless schema
violations (like extra keys). DO NOT USE FOR INPUT VALIDATION!
:return: The updated JSON Schema for the CSP assign/release resources
with PSS element.
"""
(major, minor) = split_interface_version(csp_version)
max_beams = 0
update_beams_id_name = False
if "low" in csp_version:
max_beams = MAX_LOW_PSS_BEAMS
update_beams_id_name = True if (major, minor) >= (6, 0) else False
else:
# Mid schema
max_beams = MAX_MID_PSS_BEAMS
update_beams_id_name = True if (major, minor) >= (3, 2) else False
pss_schema = TMSchema.new("CSP PSS beams", csp_version, strict)
pss_schema.add_field(
"pss_beam_ids" if update_beams_id_name else "beams_id",
[int],
check_strict=lambda lst: all(
1 <= beam_id <= max_beams for beam_id in lst
)
and len(lst) <= max_beams,
description="list of PSS beam IDs",
)
if "low" in csp_version or (
"low" not in csp_version and (major, minor) >= (3, 2)
):
schema.add_opt_field(
"pss",
pss_schema,
description="PST section for the assign/release resources",
)