Source code for ska_oso_pdm.entities.common.field_configuration

"""
The entities.common.field_configuration module defines a Python representation of the
target of the observation.
"""
import uuid
from abc import ABC, abstractmethod
from enum import Enum
from typing import List, Optional, Sequence, Tuple, Union

from astropy.coordinates import SkyCoord, get_body
from astropy.time import Time

__all__ = [
    "FieldConfiguration",
    "FieldConfigurationID",
    "SiderealTarget",
    "DriftScanTarget",
    "Planet",
    "PlanetName",
    "TargetID",
]

# aliases to str for entity IDs
TargetID = str
FieldConfigurationID = str


class Target(ABC):
    """
    Target represents the argument for SKA scheduling block.
    """

    def __init__(
        self, target_id: Optional[TargetID] = None, target_name: Optional[str] = None
    ):
        if target_id is None:
            target_id = str(uuid.uuid4())
        self.target_id = target_id

        if target_name is None:
            target_name = ""
        self.target_name = target_name

    @abstractmethod
    def coord(self):
        """
        abstract method : Target encapsulates source coordinates and source metadata.
        """
        raise NotImplementedError


[docs]class SiderealTarget(Target): """ SiderealTarget represents the argument for SKA scheduling block. """ OFFSET_MARGIN_IN_RAD = 6e-17 # Arbitrary small number # pylint: disable=too-many-arguments def __init__( self, ra: Union[float, str], dec: Union[float, str], target_name: Optional[str] = None, reference_frame: Optional[str] = "icrs", target_id: Optional[str] = None, unit: Union[str, Tuple[str, str]] = ("hourangle", "deg"), ): super().__init__(target_id, target_name) self._coord = SkyCoord(ra, dec, unit=unit, frame=reference_frame) @property def coord(self): # pylint: disable=invalid-overridden-method return self._coord def __eq__(self, other): if not isinstance(other, SiderealTarget): return False # Please replace this with a more elegant way of dealing with differences # due to floating point arithmetic when comparing targets # defined in different ways. sep = self.coord.separation(other.coord) return ( self.target_name == other.target_name and self.coord.frame.name == other.coord.frame.name and sep.radian < self.OFFSET_MARGIN_IN_RAD ) def __repr__(self): raw_ra = self.coord.ra.value raw_dec = self.coord.dec.value units = (self.coord.ra.unit.name, self.coord.dec.unit.name) reference_frame = self.coord.frame.name target_name = self.target_name return ( f"<SiderealTarget(ra={raw_ra}, dec={raw_dec}, name='{target_name}', " f"frame='{reference_frame}', unit={units})>" ) def __str__(self): return ( f"<SiderealTarget: '{self.target_name}' " f'({self.coord.to_string(style="hmsdms")} {self.coord.frame.name})>' )
[docs]class DriftScanTarget(Target): """ DriftScanTarget defines AltAz target for SKA scheduling block. """ OFFSET_MARGIN_IN_RAD = 6e-17 # Arbitrary small number # pylint: disable=too-many-arguments def __init__( self, az: float, el: float, target_name: Optional[str] = None, target_id: Optional[str] = None, unit: str = "deg", ): super().__init__(target_id, target_name) self._coord = SkyCoord(az=az, alt=el, unit=unit, frame="altaz") @property def coord(self): # pylint: disable=invalid-overridden-method return self._coord def __eq__(self, other): if not isinstance(other, DriftScanTarget): return False sep = self.coord.separation(other.coord) return ( self.target_name == other.target_name and self.coord.frame.name == other.coord.frame.name and sep.radian < self.OFFSET_MARGIN_IN_RAD ) def __repr__(self): raw_az = self.coord.az.value raw_alt = self.coord.alt.value units = (self.coord.az.unit.name, self.coord.alt.unit.name) reference_frame = self.coord.frame.name target_name = self.target_name return ( f"<DriftScanTarget(az={raw_az}, alt={raw_alt}, name='{target_name}', " f"frame='{reference_frame}', unit={units})>" ) def __str__(self): return ( f"<DriftScanTarget: '{self.target_name}' " f"({self.coord.to_string()} {self.coord.frame.name})>" )
[docs]class PlanetName(Enum): """ PlanetName represents name of the planet. """ MERCURY = "mercury" VENUS = "venus" MARS = "mars"
[docs]class Planet(Target): """ Planet represents the argument for SKA scheduling block. """ def __init__(self, target_name: PlanetName, target_id: Optional[str] = None): super().__init__(target_id, target_name.value) @property def coord(self): # pylint: disable=invalid-overridden-method return get_body(self.target_name, Time.now()) def __eq__(self, other): if not isinstance(other, Planet): return False return self.target_id == other.target_id and PlanetName( self.target_name ) == PlanetName(other.target_name) def __repr__(self): name_enum = PlanetName(self.target_name) return f"<Planet({name_enum})>" def __str__(self): return f"<Planet({self.target_name})>"
[docs]class FieldConfiguration: """ FieldConfiguration represents the argument for SKA scheduling block. """ def __init__( self, field_id: Optional[FieldConfigurationID] = None, targets: Optional[Sequence[Target]] = None, ): if field_id is None: field_id = str(uuid.uuid4()) self.field_id: str = field_id if targets is None: targets = [] self.targets: List[Target] = [] self.targets.extend(targets) def __eq__(self, other): if not isinstance(other, FieldConfiguration): return False return self.field_id == other.field_id and self.targets == other.targets def __repr__(self): return f"<FieldConfiguration(id={self.field_id}, targets={self.targets})>" def __str__(self): summary = sorted( [f"{target.target_id}={target.target_name}" for target in self.targets] ) return f'<FieldConfiguration({", ".join(summary)})>'