diff --git a/src/adler/adler.py b/src/adler/adler.py index 2d390a1..b8def5b 100644 --- a/src/adler/adler.py +++ b/src/adler/adler.py @@ -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) diff --git a/src/adler/dataclasses/AdlerPlanetoid.py b/src/adler/dataclasses/AdlerPlanetoid.py index b07262c..5b2166f 100644 --- a/src/adler/dataclasses/AdlerPlanetoid.py +++ b/src/adler/dataclasses/AdlerPlanetoid.py @@ -1,58 +1,135 @@ -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. + + 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. + + """ + + 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) - self.Observations = Observations( - self.ssObjectId, self.population_location, observations_sql_query, sql_filename=self.sql_filename - ) + observations_by_filter.append( + Observations.construct_from_data_table(ssObjectId, filter_name, data_table) + ) - def populate_MPCORB(self): - """Populates the MPCORB object class attribute.""" + 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, @@ -60,32 +137,78 @@ def populate_MPCORB(self): FROM dp03_catalogs_10yr.MPCORB WHERE - ssObjectId = {self.ssObjectId} + ssObjectId = {ssObjectId} + """ + + 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. + """ - self.MPCORB = MPCORB( - self.ssObjectId, self.population_location, MPCORB_sql_query, sql_filename=self.sql_filename - ) + 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} + """ + + 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 observations_in_filter(self, filter_name): + """User-friendly helper function. Returns the Observations object for a given filter. + + Parameters + ----------- + filter_name : str + The desired filter. + + Returns + ----------- + Observations object + The Observations object in self.observations_by_filter corresponding to the desired filter. + """ - self.SSObject = SSObject( - self.ssObjectId, - self.population_location, - sql_query=SSObject_sql_query, - sql_filename=self.sql_filename, - ) + try: + filter_index = self.filter_list.index(filter_name) + except ValueError: + raise ValueError("Filter {} is not in AdlerPlanetoid.filter_list.".format(filter_name)) + + return self.observations_by_filter[filter_index] def do_pretend_science(self): self.DummyScienceResult = DummyScience().science_result diff --git a/src/adler/dataclasses/DataSchema.py b/src/adler/dataclasses/DataSchema.py deleted file mode 100644 index b83bace..0000000 --- a/src/adler/dataclasses/DataSchema.py +++ /dev/null @@ -1,380 +0,0 @@ -import numpy as np -from dataclasses import dataclass, field -from lsst.rsp import get_tap_service - - -class DataSchema: - """Parent class for Observations (a join of DiaSource and SSSource), MPCORB - and SSObject data classes. Largely a collection of common methods. Should never be instantiated - by itself. - """ - - def populate(self, population_location, sql_query, sql_filename): - """Populates the DataSchema object, either from the RSP or a SQL table. Note that this calls the method - populate_from_table(), which must exist in the child classes. - - Parameters - ----------- - population_location : str - String delineating source of data. Should be "RSP" for Rubin Science Platform or "SQL" for a SQL table. - sql_query: str - SQL query to retrieve data from database. - - """ - - if population_location == "RSP": # pragma: no cover - data_table = self.get_RSP_table(sql_query) - elif population_location == "SQL": - data_table = self.get_SQL_table(sql_query, sql_filename) - else: - raise Exception( - "Population source not recognised. Please supply either 'RSP' or 'SQL' for population_location argument." - ) - - self.populate_from_table(data_table) - - def get_RSP_table(self, sql_query): # pragma: no cover - """Retrieves the table of data from the RSP. Populates the data_table class variable. - - Parameters - ----------- - sql_query : str - SQL query to be sent to the RSP tap service. - - """ - - self.service = get_tap_service("ssotap") - - return self.service.search(sql_query) - - def get_SQL_table(self, sql_query, sql_filename): - pass - - def get_from_table(self, 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: - sys.exit("Could not cast column name to type.") - - -@dataclass -class Observations(DataSchema): - """This is a SQL join of DiaSource and SSSource which contains all of the - observations of the object. Inherits from DataSchema. 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. - - mag: array_like - Magnitude. This is a placeholder and will be replaced by flux. - - magErr: array_like - Magnitude error. This is a placeholder and will be replaced by flux error. - - midpointMjdTai: array_like - Effective mid-visit time for this diaSource, expressed as Modified Julian Date, International Atomic Time. - - ra: array_like - Right ascension coordinate of the center of this diaSource. - - dec: array_like - Declination coordinate of the center of this diaSource. - - phaseAngle: array_like - Phase angle. - - topocentricDist: array_like - Topocentric distance. - - heliocentricDist: array_like - Heliocentric distance. - - reduced_mag: array_like - The reduced magnitude. - - num_obs : int - The number of observations contained in this structure. - - """ - - ssObjectId: 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 - - def __init__(self, ssObjectId, population_location, sql_query, sql_filename=None): - """Initialises the Observations object. - - Parameters - ----------- - ssObjectId : str - ssObjectId of the object of interest. - population_location : str - String delineating source of data. Should be "RSP" for Rubin Science Platform, "SQL" for a SQL table, - or "arguments" for arguments. - sql_query : str - SQL query to retrieve data from database. - sql_filename: str, optional - Location of local SQL database, if using. - - """ - - self.ssObjectId = ssObjectId - self.population_location = population_location - self.populate(self.population_location, sql_query, sql_filename) - self.num_obs = len(self.midpointMjdTai) - self.calculate_reduced_mag() - - def populate_from_table(self, data_table): - """Populates the Observations object from the data_table class variable created on initialisation.""" - - self.mag = self.get_from_table(data_table, "mag", "array") - self.magErr = self.get_from_table(data_table, "magErr", "array") - self.midpointMjdTai = self.get_from_table(data_table, "midpointMjdTai", "array") - self.ra = self.get_from_table(data_table, "ra", "array") - self.dec = self.get_from_table(data_table, "dec", "array") - self.phaseAngle = self.get_from_table(data_table, "phaseAngle", "array") - self.topocentricDist = self.get_from_table(data_table, "topocentricDist", "array") - self.heliocentricDist = self.get_from_table(data_table, "heliocentricDist", "array") - - def calculate_reduced_mag(self): - """ - Calculates the reduced magnitude column. - """ - self.reduced_mag = self.mag - 5 * np.log10(self.topocentricDist * self.heliocentricDist) - - -@dataclass -class MPCORB(DataSchema): - """Grabs 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 = "" - - def __init__(self, ssObjectId, population_location, sql_query, sql_filename): - """Initialises the MPCORB object. - - Parameters - ----------- - 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_query : str - SQL query to retrieve data from database. - sql_filename: str, optional - Location of local SQL database, if using. - """ - - self.ssObjectId = ssObjectId - self.population_location = population_location - self.populate(self.population_location, sql_query, sql_filename) - - def populate_from_table(self, data_table): - """Populates the MPCORB object from the data_table class variable created on initialisation.""" - - self.mpcDesignation = self.get_from_table(data_table, "mpcDesignation", "str") - self.mpcNumber = self.get_from_table(data_table, "mpcNumber", "int") - self.mpcH = self.get_from_table(data_table, "mpcH", "float") - self.mpcG = self.get_from_table(data_table, "mpcH", "float") - self.epoch = self.get_from_table(data_table, "epoch", "float") - self.peri = self.get_from_table(data_table, "peri", "float") - self.node = self.get_from_table(data_table, "node", "float") - self.incl = self.get_from_table(data_table, "incl", "float") - self.e = self.get_from_table(data_table, "e", "float") - self.n = self.get_from_table(data_table, "n", "float") - self.q = self.get_from_table(data_table, "q", "float") - self.uncertaintyParameter = self.get_from_table(data_table, "uncertaintyParameter", "str") - self.flags = self.get_from_table(data_table, "flags", "str") - - -@dataclass -class SSObject(DataSchema): - """Grabs information from SSObject. All attributes carry the same names as the column names from the SSObject table. - - Attributes: - ----------- - - ssObjectId: str - LSST unique identifier (if observed by LSST) - - 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 - - r_H: float - Best fit absolute magnitude (r band) - - r_G12: float - Best fit G12 slope parameter (r band) - - r_Herr: - Uncertainty of H (r band) - - r_G12Err: - Uncertainty of G12 (r band) - - r_nData: int - The number of data points used to fit the phase curve (r band) - - 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 = "" - discoverySubmissionDate: float = 0.0 - firstObservationDate: float = 0.0 - arc: float = 0.0 - numObs: int = 0 - r_H: float = 0.0 - r_G12: float = 0 - r_Herr: float = 0.0 - r_G12err: float = 0.0 - r_nData: int = 0 - maxExtendedness: float = 0.0 - minExtendedness: float = 0.0 - medianExtendedness: float = 0.0 - - def __init__(self, ssObjectId, population_location, sql_query, sql_filename): - """Initialises the SSObject object. - - Parameters - ----------- - 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_query : str - SQL query to retrieve data from database. - sql_filename: str, optional - Location of local SQL database, if using. - """ - - self.ssObjectId = ssObjectId - self.population_location = population_location - self.populate(self.population_location, sql_query, sql_filename) - - 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") diff --git a/src/adler/dataclasses/MPCORB.py b/src/adler/dataclasses/MPCORB.py new file mode 100644 index 0000000..76fe782 --- /dev/null +++ b/src/adler/dataclasses/MPCORB.py @@ -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, + ) diff --git a/src/adler/dataclasses/Observations.py b/src/adler/dataclasses/Observations.py new file mode 100644 index 0000000..2d5aa1e --- /dev/null +++ b/src/adler/dataclasses/Observations.py @@ -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) diff --git a/src/adler/dataclasses/SSObject.py b/src/adler/dataclasses/SSObject.py new file mode 100644 index 0000000..0b1e1fd --- /dev/null +++ b/src/adler/dataclasses/SSObject.py @@ -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") diff --git a/src/adler/dataclasses/dataclass_utilities.py b/src/adler/dataclasses/dataclass_utilities.py new file mode 100644 index 0000000..04372c3 --- /dev/null +++ b/src/adler/dataclasses/dataclass_utilities.py @@ -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.")