"""
The schemas.common.field_configuration defines Marshmallow schema that map the
field_configurations section of an SKA scheduling block to/from a JSON
representation.
"""
import collections
from marshmallow import Schema, fields, post_load, pre_dump
from marshmallow.validate import OneOf
from marshmallow_oneofschema import OneOfSchema
from ska_oso_pdm.schemas import shared
from ...entities.common.field_configuration import (
DriftScanTarget,
FieldConfiguration,
Planet,
PlanetName,
SiderealTarget,
)
__all__ = ["FieldConfigurationSchema"]
class PlanetSchema(Schema):
"""
The planet section of an SKA scheduling block
"""
target_id = fields.String(required=True)
# For Planet the target_name is an enum, however in the parent
# Target class (which the schema uses) it is a string
# For this reason the target_name is not handled with the
# marshmallow_enum.EnumField like other enums
target_name = fields.String(
data_key="target_name",
validate=OneOf([e.value for e in PlanetName]),
required=True,
)
@post_load
def make_planet(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a planet object.
:param data: dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: Planet instance populated to match JSON
"""
name_enum = PlanetName(data["target_name"])
obj = Planet(target_id=data["target_id"], target_name=name_enum)
return obj
JsonTarget = collections.namedtuple(
"JsonTarget", "target_id ra dec reference_frame target_name"
)
DriftScanJsonTarget = collections.namedtuple(
"DriftScanJsonTarget", "target_id az el reference_frame target_name"
)
class SiderealTargetSchema(Schema):
"""
The siderealtarget section of an SKA scheduling block
"""
target_id = fields.String(required=True)
ra = fields.String(data_key="ra")
dec = fields.String(data_key="dec")
reference_frame = shared.UpperCasedField(data_key="reference_frame")
target_name = fields.String(data_key="target_name")
@pre_dump
def convert_to_icrs(
self, target: SiderealTarget, **_
): # pylint: disable=no-self-use
"""
Process SiderealTarget co-ordinates by converting them to ICRS frame
before the JSON marshalling process begins.
:param target: SiderealTarget instance to process
:param _: kwargs passed by Marshallow
:return: target struct with ICRS ra/dec expressed in hms/dms
"""
# All pointing coordinates are in ICRS
icrs_coord = target.coord.transform_to("icrs")
hms, dms = icrs_coord.to_string("hmsdms", sep=":").split(" ")
sexagesimal = JsonTarget(
target_id=target.target_id,
ra=hms,
dec=dms,
reference_frame=icrs_coord.frame.name,
target_name=target.target_name,
)
return sexagesimal
@post_load
def make_siderealtarget(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a SiderealTarget object.
:param data: dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: SiderealTarget instance populated to match JSON
"""
target_name = data["target_name"]
hms = data["ra"]
dms = data["dec"]
reference_frame = data["reference_frame"]
target = SiderealTarget(
hms,
dms,
reference_frame=reference_frame,
target_name=target_name,
unit=("hourangle", "deg"),
target_id=data["target_id"],
)
return target
class DriftScanTargetSchema(Schema):
"""
The DriftScan target section of an SKA scheduling block
"""
target_id = fields.String(required=True)
az = fields.Float()
el = fields.Float()
target_name = fields.String(data_key="target_name")
reference_frame = shared.UpperCasedField(data_key="reference_frame")
@pre_dump
def convert_to_altaz(
self, target: DriftScanTarget, **_
): # pylint: disable=no-self-use
"""
Process DriftScanTarget co-ordinates by converting them to AltAz frame
before the JSON marshalling process begins.
:param target: DriftScanTarget instance to process
:param _: kwargs passed by Marshallow
:return: target struct with ICRS ra/dec expressed in hms/dms
"""
coord = target.coord
sexagesimal = DriftScanJsonTarget(
target_id=target.target_id,
az=coord.az.value,
el=coord.alt.value,
target_name=target.target_name,
reference_frame="horizon",
)
return sexagesimal
@post_load
def make_driftscantarget(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a DriftScanTarget object.
:param data: dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: DriftScanTarget instance populated to match JSON
"""
target_name = data["target_name"]
azimuth = data["az"]
elevation = data["el"]
target = DriftScanTarget(
az=azimuth,
el=elevation,
target_name=target_name,
unit="deg",
target_id=data["target_id"],
)
return target
class TargetSchema(OneOfSchema):
"""
The target section of an SKA scheduling block
"""
type_field = "kind"
type_schemas = {
"planet": PlanetSchema,
"sidereal": SiderealTargetSchema,
"driftscan": DriftScanTargetSchema,
}
def get_obj_type(self, obj):
if isinstance(obj, Planet):
return "planet"
if isinstance(obj, SiderealTarget):
return "sidereal"
if isinstance(obj, DriftScanTarget):
return "driftscan"
raise Exception(f"Unknown object type: {obj.__class__.__name__}")
[docs]class FieldConfigurationSchema(Schema): # pylint: disable=too-few-public-methods
"""
The Field Configuration section of an SKA scheduling block
"""
field_id = fields.String(required=True)
targets = fields.List(fields.Nested(TargetSchema))
[docs] @post_load
def make_fieldconfiguration(self, data, **_): # pylint: disable=no-self-use
"""
Convert parsed JSON back into a FieldConfiguration object.
:param data: dict containing parsed JSON values
:param _: kwargs passed by Marshmallow
:return: FieldConfiguration instance populated to match JSON
"""
#
# TODO investigate entity registry
#
# When the ID is plucked by a different schema, the non-ID fields are
# not present in the JSON. Ideally, in the entity objects we'd have
# the referenced entities and not just the ID, so we'd need something
# like the OT entity registry and have the @post_load functions source
# the complete entities from the registry.
#
targets = data.get("targets")
return FieldConfiguration(field_id=data["field_id"], targets=targets)