"""
Python model of the Dish configure JSON schema
"https://schema.skao.int/ska-dish-configure/1.0".
Descriptions are copied from the schema implementation. For validation
consistent with ska-telmodel, validation is not coded in the Pydantic model but
enforced at serialisation time via the ska-telmodel schema.
"""
from pydantic import Field, field_serializer, field_validator
from ska_tmc_cdm import CdmObject
from ska_tmc_cdm.messages.subarray_node.configure.core import ReceiverBand
[docs]
class PseudoRandomNoiseGenerator(CdmObject):
"""
PseudoRandomNoiseGenerator maps the 'PseudoRandom' noise diode subschema of
Dish configure v1.0.
"""
binary_polynomial: int | None = Field(
None,
description="Defines which bits of the shift register are fed back to "
"generate the next bit sequence.",
)
seed: int | None = Field(
None,
description="Initial state of the shift register in the PRBS "
"generator.",
)
dwell: int | None = Field(
None,
description="Time interval (in nanoseconds) between transitions to the"
" next bit.",
)
[docs]
class PeriodicNoiseGenerator(CdmObject):
"""
PeriodicNoiseGenerator maps the 'Periodic' subschema of Dish configure
v1.0.
"""
period: int | None = Field(
None, description="Number of nanoseconds for the noise diode period."
)
duty_cycle: int | None = Field(
None,
description="Number of nanoseconds for which the noise diode should be"
" ON. If duty_cycle > period, then the noise diode will be on 100% of "
"the time.",
)
phase_shift: int | None = Field(
None,
description="Phase shift of the periodic signal in number of "
"nanoseconds.",
)
[docs]
class SPFRxProcessingParameters(CdmObject):
dishes: list[str] | None = Field(
None,
description="List of Dish IDs to apply the processing parameters. If "
'"all" is specified then the same parameters will be applied to all '
"dishes.",
)
sync_pps: bool | None = Field(
None,
description="Synchronize the SPFRx to the WR-1PPS clock. The system "
"assumes a default value of sync_pps is true if not specified.",
)
attenuation_pol_x: float | None = Field(
None,
description="The requested aggregate Pol X attenuation of both "
"attenuators. Must be in range 0 - max value for that particular "
"band.",
)
attenuation_1_pol_x: float | None = Field(
None,
description="The requested Pol X attenuation of attenuator 1. Must be "
"in range 0-31.75, and the sum of attenuator 1 + attenuator 2 must not"
" exceed the maximum for that band.",
)
attenuation_2_pol_x: float | None = Field(
None,
description="The requested Pol Y attenuation of attenuator 2. Must be "
"in range 0-31.75, and the sum of attenuator 1 + attenuator 2 must not"
" exceed the maximum for that band.",
)
attenuation_pol_y: float | None = Field(
None,
description="The requested aggregate Pol Y attenuation of both "
"attenuators. Must be in range 0 - max value for that particular "
"band.",
)
attenuation_1_pol_y: float | None = Field(
None,
description="The requested Pol Y attenuation of attenuator 1. Must be "
"in range 0-31.75, and the sum of attenuator 1 + attenuator 2 must not"
" exceed the maximum for that band.",
)
attenuation_2_pol_y: float | None = Field(
None,
description="The requested Pol Y attenuation of attenuator 2. Must be "
"in range 0-31.75, and the sum of attenuator 1 + attenuator 2 must not"
" exceed the maximum for that band.",
)
saturation_threshold: float | None = Field(
None,
description="Requested saturation threshold defined by SNR. Must be in"
" range 0-1.0.",
)
noise_diode: PseudoRandomNoiseGenerator | PeriodicNoiseGenerator | None = (
Field(None, description="Noise diode parameters")
)
@field_validator("noise_diode", mode="before")
@classmethod
def _parse_noise_diode(cls, v):
"""
Removes the intermediate 'noise_diode' wrapper from the JSON input.
"""
# Accept wrapper-shaped dicts from JSON input and unwrap
if v is None:
return v
if isinstance(v, dict):
if "pseudo_random" in v and v["pseudo_random"] is not None:
return PseudoRandomNoiseGenerator.model_validate(
v["pseudo_random"]
)
if "periodic" in v and v["periodic"] is not None:
return PeriodicNoiseGenerator.model_validate(v["periodic"])
return v
@field_serializer("noise_diode", when_used="always")
def _serialize_noise_diode(self, v):
"""
Adds an intermediate object with noise generator type to the JSON
output for noise_diode. For instance, for generator type 'foo', the
output should be
"foo": {
... // foo properties here
}
"""
if v is None:
return None
if isinstance(v, PseudoRandomNoiseGenerator):
return {"pseudo_random": v.model_dump(exclude_none=True)}
if isinstance(v, PeriodicNoiseGenerator):
return {"periodic": v.model_dump(exclude_none=True)}
return v
[docs]
class DishConfiguration(CdmObject):
"""
DishConfiguration specifies how SKA MID dishes in a sub-array should be
configured.
"""
interface: str = Field(
"https://schema.skao.int/ska-dish-configure/1.0",
description="URI of JSON schema for this command's JSON payload",
)
receiver_band: ReceiverBand
band5_downconversion_subband: str | None = Field(
None,
description="The requested band5_downconversion_subband. Allowed "
'values are "1", "2", and "3". Only allowed if receiver_band specified'
' is "5b".',
)
spfrx_processing_parameters: list[
SPFRxProcessingParameters
] | None = Field(
None,
description="List of requested SPFRx processing parameters per dish.",
)