from dataclasses import dataclass, field
import numpy as np
from adler.objectdata.objectdata_utilities import get_from_table, get_from_dictionary
[docs]
OBSERVATIONS_KEYS = {
"diaSourceId": np.ndarray,
"mag": np.ndarray,
"magErr": np.ndarray,
"midPointMjdTai": np.ndarray,
"ra": np.ndarray,
"dec": np.ndarray,
"phaseAngle": np.ndarray,
"topocentricDist": np.ndarray,
"heliocentricDist": np.ndarray,
"heliocentricX": np.ndarray,
"heliocentricY": np.ndarray,
"heliocentricZ": np.ndarray,
"topocentricX": np.ndarray,
"topocentricY": np.ndarray,
"topocentricZ": np.ndarray,
"eclipticLambda": np.ndarray,
"eclipticBeta": np.ndarray,
}
@dataclass
[docs]
class Observations:
"""A SQL join of DiaSource and SSSource which contains all of the
observations of the object in a select filter. All attributes carry
the same names as the column names from the DiaSource and SSSource tables.
Attributes:
-----------
ssObjectId: str
Id of the ssObject this source was associated with, if any. If not, it is set to NULL.
filter_name : str
Filter of the observations.
diaSourceId: array_like of ints
Unique identifier of the observation
mag: array_like of floats
Magnitude. This is a placeholder and will be replaced by flux.
magErr: array_like of floats
Magnitude error. This is a placeholder and will be replaced by flux error.
midPointMjdTai: array_like of floats
Effective mid-visit time for this diaSource, expressed as Modified Julian Date, International Atomic Time.
ra: array_like of floats
Right ascension coordinate of the center of this diaSource.
dec: array_like of floats
Declination coordinate of the center of this diaSource.
phaseAngle: array_like of floats
Phase angle.
topocentricDist: array_like of floats
Topocentric distance.
heliocentricDist: array_like of floats
Heliocentric distance.
heliocentricX: array_like of floats
x-axis component of the heliocentric distance.
heliocentricY: array_like of floats
y-axis component of the heliocentric distance.
heliocentricZ: array_like of floats
z-axis component of the heliocentric distance.
topocentricX: array_like of floats
x-axis component of the topocentric distance.
topocentricY: array_like of floats
y-axis component of the topocentric distance.
topocentricZ: array_like of floats
z-axis component of the topocentric distance.
eclipticLambda: array_like of floats
The ecliptic longitude.
eclipticBeta: array_like of floats
The ecliptic latitude.
reduced_mag: array_like of floats
The reduced magnitude.
num_obs : int
The number of observations contained in this structure.
"""
[docs]
diaSourceId: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
mag: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
magErr: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
midPointMjdTai: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
ra: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
dec: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
phaseAngle: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
topocentricDist: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
heliocentricDist: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
heliocentricX: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
heliocentricY: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
heliocentricZ: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
topocentricX: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
topocentricY: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
topocentricZ: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
eclipticLambda: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
eclipticBeta: np.ndarray = field(default_factory=lambda: np.zeros(0))
[docs]
reduced_mag: np.ndarray = field(default_factory=lambda: np.zeros(0))
@classmethod
[docs]
def construct_from_data_table(cls, ssObjectId, filter_name, data_table, cassandra=False):
"""Initialises the Observations object from a table of data.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
filter_name : str
String of the filter the observations are taken in,
data_table : table-like object
Table of data from which attributes shoud be populated.
Returns
-----------
Observations object
Observations object with class attributes populated from data_table.
"""
obs_dict = {"ssObjectId": ssObjectId, "filter_name": filter_name, "num_obs": len(data_table)}
for obs_key, obs_type in OBSERVATIONS_KEYS.items():
try:
obs_dict[obs_key] = get_from_table(data_table, obs_key, obs_type, "SSSource/DIASource")
except KeyError: # sometimes we have case issues...
# Specific handling for this one as the case of the P was changed between DP0.3 and DP1
if obs_key == "midPointMjdTai":
obs_dict[obs_key] = get_from_table(
data_table, "midpointMjdTai", obs_type, "SSSource/DIASource"
)
else:
obs_dict[obs_key] = get_from_table(
data_table, obs_key.casefold(), obs_type, "SSSource/DIASource"
)
obs_dict["reduced_mag"] = cls.calculate_reduced_mag(
cls, obs_dict["mag"], obs_dict["topocentricDist"], obs_dict["heliocentricDist"]
)
return cls(**obs_dict)
@classmethod
[docs]
def construct_from_dictionary(cls, ssObjectId, filter_name, data_dict):
"""Initialises the Observations object from a dictionary of data.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
filter_name : str
String of the filter the observations are taken in,
data_dict : dict or dict-like object
Dictionary of data from which attributes shoud be populated.
Returns
-----------
Observations object
Observations object with class attributes populated from data_dict.
"""
obs_dict = {"ssObjectId": ssObjectId, "filter_name": filter_name, "num_obs": 1}
for obs_key, obs_type in OBSERVATIONS_KEYS.items():
obs_dict[obs_key] = get_from_dictionary(data_dict, obs_key, obs_type, "SSSource/DIASource")
obs_dict["reduced_mag"] = cls.calculate_reduced_mag(
cls, obs_dict["mag"], obs_dict["topocentricDist"], obs_dict["heliocentricDist"]
)
return cls(**obs_dict)
@classmethod
def construct_from_dictionary(cls, ssObjectId, filter_name, data_dict):
obs_dict = {"ssObjectId": ssObjectId, "filter_name": filter_name, "num_obs": 1}
for obs_key, obs_type in OBSERVATIONS_KEYS.items():
obs_dict[obs_key] = get_from_dictionary(data_dict, obs_key, obs_type, "SSSource/DIASource")
obs_dict["reduced_mag"] = cls.calculate_reduced_mag(
cls, obs_dict["mag"], obs_dict["topocentricDist"], obs_dict["heliocentricDist"]
)
return cls(**obs_dict)
[docs]
def calculate_reduced_mag(self, mag, topocentric_dist, heliocentric_dist):
"""
Calculates the reduced magnitude column.
Parameters
-----------
mag : array_like of floats
Magnitude.
topocentric_dist : array_like of floats
Topocentric distance.
heliocentric_dist: array_like of floats
Heliocentric distance.
Returns
-----------
array_like of floats
The reduced magnitude.
"""
return mag - 5 * np.log10(topocentric_dist * heliocentric_dist)