Skip to content

Commit

Permalink
Refactor and adding filters.
Browse files Browse the repository at this point in the history
astronomerritt committed Mar 8, 2024
1 parent 610d429 commit 9a900d9
Showing 7 changed files with 603 additions and 435 deletions.
8 changes: 6 additions & 2 deletions src/adler/adler.py
Original file line number Diff line number Diff line change
@@ -4,7 +4,7 @@


def runAdler(args):
planetoid = AdlerPlanetoid(args.ssoid)
planetoid = AdlerPlanetoid.construct_from_RSP(args.ssoid, args.filter_list)

planetoid.do_pretend_science()

@@ -13,12 +13,16 @@ def main():
parser = argparse.ArgumentParser(description="Runs Adler for a select planetoid and given user input.")

parser.add_argument("-s", "--ssoid", help="SSObject ID of planetoid.", type=str, required=True)
parser.add_argument(
"-f", "--filters", help="Comma-separated list of filters required.", type=str, default="u,g,r,i,z,y"
)

# can add arguments to specify a date range etc later
# alternatively we may start using a config file

args = parser.parse_args()

args.filter_list = args.filters.split(",")

runAdler(args)


207 changes: 154 additions & 53 deletions src/adler/dataclasses/AdlerPlanetoid.py
Original file line number Diff line number Diff line change
@@ -1,91 +1,192 @@
from adler.dataclasses.DataSchema import Observations, MPCORB, SSObject
from lsst.rsp import get_tap_service

from adler.dataclasses.Observations import Observations
from adler.dataclasses.MPCORB import MPCORB
from adler.dataclasses.SSObject import SSObject
from adler.dataclasses.AdlerData import AdlerData
from adler.dataclasses.dataclass_utilities import get_data_table
from adler.science.DummyScience import DummyScience


class AdlerPlanetoid:
"""AdlerPlanetoid class. Contains the Observations, MPCORB and SSObject objects."""
"""AdlerPlanetoid class. Contains the Observations, MPCORB and SSObject dataclass objects."""

def __init__(self, ssObjectId, population_location="RSP", sql_filename=None):
def __init__(self, ssObjectId, filter_list, observations_by_filter, mpcorb, ssobject, adler_data):
"""Initialises the AdlerPlanetoid object.
Parameters
Attributes
-----------
ssObjectId : str
ssObjectId of the object of interest.
population_location : str
String delineating source of data. Should be "RSP" for Rubin Science Platform or "SQL" for a SQL table.
sql_filename: str, optional
Location of local SQL database, if using.
filter_list : list of str
A comma-separated list of the filters of interest.
observations_by_filter : list of Observations objects
A list of Observations objects holding joined DIASource/SSSource observations of the planetoid specified by ssObjectId. Each item in the list holds observations of a different filter, in the order specified by filter_list.
mpcorb : MPCORB object
An MPCORB object, holding the MPCORB database information of the planetoid specified by ssObjectId.
ssobject : SSObject object
An SSObject object, holding the SSObject database information of the planetoid specified by ssObjectId.
adler_data : AdlerData object
An empty AdlerData object ready to store Adler-calculated values.
"""
self.ssObjectId = ssObjectId
self.population_location = population_location
self.sql_filename = sql_filename
# can also include date ranges at some point

# create empty AdlerData dataclass object.
self.AdlerData = AdlerData(["r"])

# this creates the AdlerPlanetoid.Observations, AdlerPlanetoid.MPCORB and
# AdlerPlanetoid.SSObject objects.
self.populate_observations()
self.populate_MPCORB()
self.populate_SSObject()

def populate_observations(self):
"""Populates the Observations object class attribute."""
observations_sql_query = f"""
SELECT
ssObject.ssObjectId, mag, magErr, band, midpointMjdTai, ra, dec, phaseAngle,
topocentricDist, heliocentricDist
FROM
dp03_catalogs_10yr.ssObject
JOIN dp03_catalogs_10yr.diaSource ON dp03_catalogs_10yr.ssObject.ssObjectId = dp03_catalogs_10yr.diaSource.ssObjectId
JOIN dp03_catalogs_10yr.ssSource ON dp03_catalogs_10yr.diaSource.diaSourceId = dp03_catalogs_10yr.ssSource.diaSourceId
WHERE
ssObject.ssObjectId = {self.ssObjectId} and band='r'
"""
self.filter_list = filter_list
self.observations_by_filter = observations_by_filter
self.MPCORB = mpcorb
self.SSObject = ssobject
self.AdlerData = adler_data

@classmethod
def construct_from_SQL(cls):
# to-do
pass

@classmethod
def construct_from_RSP(cls, ssObjectId, filter_list=["u", "g", "r", "i", "z", "y"]):
"""Custom constructor which builds the AdlerPlanetoid object and the associated Observations, MPCORB and SSObject objects
from the RSP.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
filter_list : list of str
A comma-separated list of the filters of interest.
"""

service = get_tap_service("ssotap")
observations_by_filter = cls.populate_observations(cls, ssObjectId, filter_list, service=service)
mpcorb = cls.populate_MPCORB(cls, ssObjectId, service=service)
ssobject = cls.populate_SSObject(cls, ssObjectId, filter_list, service=service)

adler_data = AdlerData(filter_list)

return cls(ssObjectId, filter_list, observations_by_filter, mpcorb, ssobject, adler_data)

def populate_observations(self, ssObjectId, filter_list, service=None, sql_filename=None):
"""Populates the observations_by_filter class attribute. Can populate from either the RSP for a SQL database:
this behaviour is controlled by the service and sql_filename parameters, one of which must be supplied.
self.Observations = Observations(
self.ssObjectId, self.population_location, observations_sql_query, sql_filename=self.sql_filename
)
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
def populate_MPCORB(self):
"""Populates the MPCORB object class attribute."""
filter_list : list of str
A comma-separated list of the filters of interest.
service : pyvo.dal.tap.TAPService object
TAPService object linked to the RSP. Default=None.
sql_filename : str
Filepath to a SQL database. Default=None.
"""

observations_by_filter = []

for filter_name in filter_list:
observations_sql_query = f"""
SELECT
ssObject.ssObjectId, mag, magErr, band, midpointMjdTai, ra, dec, phaseAngle,
topocentricDist, heliocentricDist
FROM
dp03_catalogs_10yr.ssObject
JOIN dp03_catalogs_10yr.diaSource ON dp03_catalogs_10yr.ssObject.ssObjectId = dp03_catalogs_10yr.diaSource.ssObjectId
JOIN dp03_catalogs_10yr.ssSource ON dp03_catalogs_10yr.diaSource.diaSourceId = dp03_catalogs_10yr.ssSource.diaSourceId
WHERE
ssObject.ssObjectId = {ssObjectId} and band = '{filter_name}'
"""

data_table = get_data_table(observations_sql_query, service=service, sql_filename=sql_filename)

observations_by_filter.append(
Observations.construct_from_data_table(ssObjectId, filter, data_table)
)

return observations_by_filter

def populate_MPCORB(self, ssObjectId, service=None, sql_filename=None):
"""Populates the MPCORB object class attribute. Can populate from either the RSP for a SQL database:
this behaviour is controlled by the service and sql_filename parameters, one of which must be supplied.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
service : pyvo.dal.tap.TAPService object
TAPService object linked to the RSP. Default=None.
sql_filename : str
Filepath to a SQL database. Default=None.
"""
MPCORB_sql_query = f"""
SELECT
ssObjectId, mpcDesignation, mpcNumber, mpcH, mpcG, epoch, peri, node, incl, e, n, q,
uncertaintyParameter, flags
FROM
dp03_catalogs_10yr.MPCORB
WHERE
ssObjectId = {self.ssObjectId}
ssObjectId = {ssObjectId}
"""

self.MPCORB = MPCORB(
self.ssObjectId, self.population_location, MPCORB_sql_query, sql_filename=self.sql_filename
)
data_table = get_data_table(MPCORB_sql_query, service=service, sql_filename=sql_filename)

return MPCORB.construct_from_data_table(ssObjectId, data_table)

def populate_SSObject(self, ssObjectId, filter_list, service=None, sql_filename=None):
"""Populates the SSObject class attribute. Can populate from either the RSP for a SQL database:
this behaviour is controlled by the service and sql_filename parameters, one of which must be supplied.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
filter_list : list of str
A comma-separated list of the filters of interest.
service : pyvo.dal.tap.TAPService object
TAPService object linked to the RSP. Default=None.
sql_filename : str
Filepath to a SQL database. Default=None.
"""

filter_dependent_columns = ""

for filter_name in filter_list:
filter_string = "{}_H, {}_G12, {}_Herr, {}_G12err, {}_nData, ".format(
filter_name, filter_name, filter_name, filter_name, filter_name
)

filter_dependent_columns += filter_string

def populate_SSObject(self):
"""Populates the SSObject class attribute."""
SSObject_sql_query = f"""
SELECT
discoverySubmissionDate, firstObservationDate, arc, numObs,
r_H, r_G12, r_Herr, r_G12err, r_nData,
{filter_dependent_columns}
maxExtendedness, minExtendedness, medianExtendedness
FROM
dp03_catalogs_10yr.SSObject
WHERE
ssObjectId = {self.ssObjectId}
ssObjectId = {ssObjectId}
"""

self.SSObject = SSObject(
self.ssObjectId,
self.population_location,
sql_query=SSObject_sql_query,
sql_filename=self.sql_filename,
)
data_table = get_data_table(SSObject_sql_query, service=service, sql_filename=sql_filename)

return SSObject.construct_from_data_table(ssObjectId, filter_list, data_table)

def do_pretend_science(self):
self.DummyScienceResult = DummyScience().science_result
380 changes: 0 additions & 380 deletions src/adler/dataclasses/DataSchema.py

This file was deleted.

119 changes: 119 additions & 0 deletions src/adler/dataclasses/MPCORB.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
from dataclasses import dataclass

from adler.dataclasses.dataclass_utilities import get_from_table


@dataclass
class MPCORB:
"""Object information from MPCORB. All attributes carry the same names as the column names from the MPCORB table.
Attributes:
-----------
ssObjectId: str
LSST unique identifier (if observed by LSST)
mpcDesignation: str
Number or provisional designation (in packed form)
mpcNumber: int
MPC number (if the asteroid has been numbered; NULL otherwise). Provided for convenience.
mpcH: float
Absolute magnitude, H
mpcG: float
Slope parameter, G
epoch: float
Epoch (in MJD, .0 TT)
peri: float
Argument of perihelion, J2000.0 (degrees)
node: float
Longitude of the ascending node, J2000.0 (degrees)
incl: float
Inclination to the ecliptic, J2000.0 (degrees)
e: float
Orbital eccentricity
n: float
Mean daily motion (degrees per day)
q: float
Perihelion distance (AU)
uncertaintyParameter: str
Uncertainty parameter, U
flags: str
4-hexdigit flags. See https://minorplanetcenter.net//iau/info/MPOrbitFormat.html for details
"""

ssObjectId: str = ""
mpcDesignation: str = ""
mpcNumber: int = 0
mpcH: float = 0.0
mpcG: float = 0.0
epoch: float = 0.0
peri: float = 0.0
node: float = 0.0
incl: float = 0.0
e: float = 0.0
n: float = 0.0
q: float = 0.0
uncertaintyParameter: str = ""
flags: str = ""

@classmethod
def construct_from_data_table(cls, ssObjectId, data_table):
"""Initialises the MPCORB object from a table of data.
Parameters
-----------
ssObjectId : str
ssObjectId of the object of interest.
data_table : table-like object
Table of data from which attributes shoud be populated.
Returns
-----------
MPCORB object
MPCORB object with class attributes populated from data_table.
"""

mpcDesignation = get_from_table(data_table, "mpcDesignation", "str")
mpcNumber = get_from_table(data_table, "mpcNumber", "int")
mpcH = get_from_table(data_table, "mpcH", "float")
mpcG = get_from_table(data_table, "mpcH", "float")
epoch = get_from_table(data_table, "epoch", "float")
peri = get_from_table(data_table, "peri", "float")
node = get_from_table(data_table, "node", "float")
incl = get_from_table(data_table, "incl", "float")
e = get_from_table(data_table, "e", "float")
n = get_from_table(data_table, "n", "float")
q = get_from_table(data_table, "q", "float")
uncertaintyParameter = get_from_table(data_table, "uncertaintyParameter", "str")
flags = get_from_table(data_table, "flags", "str")

return cls(
ssObjectId,
mpcDesignation,
mpcNumber,
mpcH,
mpcG,
epoch,
peri,
node,
incl,
e,
n,
q,
uncertaintyParameter,
flags,
)
131 changes: 131 additions & 0 deletions src/adler/dataclasses/Observations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
from dataclasses import dataclass, field
import numpy as np

from adler.dataclasses.dataclass_utilities import get_from_table


@dataclass
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.
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.
reduced_mag: array_like of floats
The reduced magnitude.
num_obs : int
The number of observations contained in this structure.
"""

ssObjectId: str = ""
filter_name: str = ""
mag: np.ndarray = field(default_factory=np.zeros(0))
magErr: np.ndarray = field(default_factory=np.zeros(0))
midpointMjdTai: np.ndarray = field(default_factory=np.zeros(0))
ra: np.ndarray = field(default_factory=np.zeros(0))
dec: np.ndarray = field(default_factory=np.zeros(0))
phaseAngle: np.ndarray = field(default_factory=np.zeros(0))
topocentricDist: np.ndarray = field(default_factory=np.zeros(0))
heliocentricDist: np.ndarray = field(default_factory=np.zeros(0))
reduced_mag: np.ndarray = field(default_factory=np.zeros(0))
num_obs: int = 0

@classmethod
def construct_from_data_table(cls, ssObjectId, filter_name, data_table):
"""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.
"""

reduced_mag = cls.calculate_reduced_mag(
cls, data_table["mag"], data_table["topocentricDist"], data_table["heliocentricDist"]
)

return cls(
ssObjectId,
filter_name,
data_table["mag"],
data_table["magErr"],
data_table["midpointMjdTai"],
data_table["ra"],
data_table["dec"],
data_table["phaseAngle"],
data_table["topocentricDist"],
data_table["heliocentricDist"],
reduced_mag,
len(data_table),
)

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)
128 changes: 128 additions & 0 deletions src/adler/dataclasses/SSObject.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
from dataclasses import dataclass, field
import numpy as np

from adler.dataclasses.dataclass_utilities import get_from_table


@dataclass
class SSObject:
"""Object information from SSObject. All attributes carry the same names as the column names from the SSObject table.
Attributes:
-----------
ssObjectId: str
LSST unique identifier
filter_list : list of str
A comma-separated list of the filters of interest.
discoverySubmissionDate : float
The date the LSST first linked and submitted the discovery observations to the MPC. May be NULL if not an LSST discovery. The date format will follow general LSST conventions (MJD TAI, at the moment).
firstObservationDate: float
The time of the first LSST observation of this object (could be precovered)
arc: float
Arc of LSST observations
numObs: int
Number of LSST observations of this object
H: array_like of floats
Best fit absolute magnitudes per-filter, in same order as filter_list.
G12: array_like of floats
Best fit G12 slope parameters per-filter, in same order as filter_list.
Herr : array_like of floats
Uncertainty of H per-filter, in same order as filter_list.
G12Err : array_like of floats
Uncertainty of G12 per-filter, in same order as filter_list.
nData: array_like of ints
The number of data points used to fit the phase curve per-filter, in same order as filter_list.
maxExtendedness: float
maximum `extendedness` value from the DIASource
minExtendedness: float
minimum `extendedness` value from the DIASource
medianExtendedness: float
median `extendedness` value from the DIASource
"""

ssObjectId: str = ""
filter_list: list = field(default_factory=list)
discoverySubmissionDate: float = 0.0
firstObservationDate: float = 0.0
arc: float = 0.0
numObs: int = 0
H: np.ndarray = field(default_factory=np.zeros(0))
G12: np.ndarray = field(default_factory=np.zeros(0))
Herr: np.ndarray = field(default_factory=np.zeros(0))
G12err: np.ndarray = field(default_factory=np.zeros(0))
nData: np.ndarray = field(default_factory=np.zeros(0))
maxExtendedness: float = 0.0
minExtendedness: float = 0.0
medianExtendedness: float = 0.0

@classmethod
def construct_from_data_table(cls, ssObjectId, filter_list, data_table):
discoverySubmissionDate = get_from_table(data_table, "discoverySubmissionDate", "float")
firstObservationDate = get_from_table(data_table, "firstObservationDate", "float")
arc = get_from_table(data_table, "arc", "float")
numObs = get_from_table(data_table, "numObs", "int")

H = np.zeros(len(filter_list))
G12 = np.zeros(len(filter_list))
Herr = np.zeros(len(filter_list))
G12err = np.zeros(len(filter_list))
nData = np.zeros(len(filter_list))

for i, filter in enumerate(filter_list):
H[i] = get_from_table(data_table, filter + "_H", "float")
G12[i] = get_from_table(data_table, filter + "_G12", "float")
Herr[i] = get_from_table(data_table, filter + "_Herr", "float")
G12err[i] = get_from_table(data_table, filter + "_G12err", "float")
nData[i] = get_from_table(data_table, filter + "_nData", "int")

maxExtendedness = get_from_table(data_table, "maxExtendedness", "float")
minExtendedness = get_from_table(data_table, "minExtendedness", "float")
medianExtendedness = get_from_table(data_table, "medianExtendedness", "float")

return cls(
ssObjectId,
filter_list,
discoverySubmissionDate,
firstObservationDate,
arc,
numObs,
H,
G12,
Herr,
G12err,
nData,
maxExtendedness,
minExtendedness,
medianExtendedness,
)

def populate_from_table(self, data_table):
"""Populates the SSObject object from the data_table class variable created on initialisation."""

self.discoverySubmissionDate = self.get_from_table(data_table, "discoverySubmissionDate", "float")
self.firstObservationDate = self.get_from_table(data_table, "firstObservationDate", "float")
self.arc = self.get_from_table(data_table, "arc", "float")
self.numObs = self.get_from_table(data_table, "numObs", "int")
self.r_H = self.get_from_table(data_table, "r_H", "float")
self.r_G12 = self.get_from_table(data_table, "r_G12", "float")
self.r_Herr = self.get_from_table(data_table, "r_Herr", "float")
self.r_G12Err = self.get_from_table(data_table, "r_G12err", "float")
self.r_nData = self.get_from_table(data_table, "r_nData", "int")
self.maxExtendedness = self.get_from_table(data_table, "maxExtendedness", "float")
self.minExtendedness = self.get_from_table(data_table, "minExtendedness", "float")
self.medianExtendedness = self.get_from_table(data_table, "medianExtendedness", "float")
65 changes: 65 additions & 0 deletions src/adler/dataclasses/dataclass_utilities.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import numpy as np


def get_data_table(sql_query, service=None, sql_filename=None):
"""Gets a table of data based on a SQL query. Table is pulled from either the RSP or a local SQL database:
this behaviour is controlled by the service and sql_filename parameters, one of which must be supplied.
Parameters
-----------
sql_query : str
The SQL query made to the RSP or SQL database.
service : pyvo.dal.tap.TAPService object
TAPService object linked to the RSP. Default=None.
sql_filename : str
Filepath to a SQL database. Default=None.
Returns
-----------
data_table : DALResultsTable
Data table containing the results of the SQL query.
"""

if service:
data_table = service.search(sql_query)
elif sql_filename:
# to-do
pass

return data_table


def get_from_table(data_table, column_name, data_type):
"""Retrieves information from the data_table class variable and forces it to be a specified type.
Parameters
-----------
column_name : str
Column name under which the data of interest is stored.
type : str
String delineating data type. Should be "str", "float", "int" or "array".
Returns
-----------
data : any type
The data requested from the table cast to the type required.
"""
try:
if data_type == "str":
return str(data_table[column_name][0])
elif data_type == "float":
return float(data_table[column_name][0])
elif data_type == "int":
return int(data_table[column_name][0])
elif data_type == "array":
return np.array(data_table[column_name])
else:
print("Type not recognised.")
except ValueError:
raise ValueError("Could not cast column name to type.")

0 comments on commit 9a900d9

Please sign in to comment.