Source code for psrdb.tables.ephemeris

import json
import hashlib

from psrdb.graphql_table import GraphQLTable


def get_parsers():
    """Returns the default parser for this model"""
    parser = GraphQLTable.get_default_parser("The following options will allow you to interact with the Ephemeris database object on the command line in different ways based on the sub-commands.")
    Ephemeris.configure_parsers(parser)
    return parser


[docs] class Ephemeris(GraphQLTable): """Class for interacting with the Ephemeris database object. Parameters ---------- client : GraphQLClient GraphQLClient class instance with the URL and Token already set. """ def __init__(self, client): GraphQLTable.__init__(self, client) self.table_name = "ephemeris" self.field_names = [ "id", "pulsar {name}", "createdAt", "createdBy {email}", "ephemerisHash", "p0", "dm", "comment", "validFrom", "validTo", ] self.literal_field_names = [ "id", "pulsar {id}", "createdAt", "createdBy {email}", "ephemerisHash", "p0", "dm", "comment", "validFrom", "validTo", ]
[docs] def list(self, id=None, pulsar_id=None, p0=None, dm=None, eph=None): """Return a list of Ephemeris information based on the `self.field_names` and filtered by the parameters. Parameters ---------- id : int, optional Filter by the database ID, by default None. pulsar_id : int, optional Filter by the pulsar ID, by default None. p0 : float, optional Filter by the pulsar period, by default None. dm : float, optional Filter by the pulsar DM, by default None. eph : str, optional Filter by the ephemeris hash, by default None. Returns ------- list of dicts If `self.get_dicts` is `True`, a list of dictionaries containing the results. client_response: Else a client response object. """ # P0 is stored with a maximum of 8 decimal places only m = 10 ** 8 if p0 is None: p0_filtered = None else: p0_filtered = round(p0 * m) / m eph_hash = None if eph: # convert string to json dict, to ensure the hash matches eph_json = json.loads(eph) eph_hash = hashlib.md5(json.dumps(eph_json, sort_keys=True, indent=2).encode("utf-8")).hexdigest() filters = [ {"field": "id", "value": id}, {"field": "pulsar_Id", "value": pulsar_id}, {"field": "p0", "value": p0_filtered}, {"field": "dm", "value": dm}, {"field": "ephemerisHash", "value": eph_hash}, ] return GraphQLTable.list_graphql(self, self.table_name, filters, [], self.field_names)
[docs] def create( self, pulsar, ephemeris, project_code=None, project_short=None, comment=None, ): """Create a new Ephemeris database object. Parameters ---------- pulsar : str The pulsar name. ephemeris : str The ephemeris text as a single string (includes new line characters). project_code : str, optional The project code, by default None project_short : str, optional The project short name (e.g PTA), by default None comment : str, optional A comment about the ephemeris, by default None Returns ------- client_response: A client response object. """ self.mutation_name = "createEphemeris" self.mutation = """ mutation ( $pulsar: String!, $ephemerisText: String!, $projectCode: String, $projectShort: String, $comment: String, ) { createEphemeris (input: { pulsarName: $pulsar, ephemerisText: $ephemerisText, projectCode: $projectCode, projectShort: $projectShort, comment: $comment, }) { ephemeris { id } } } """ # Read ephemeris file with open(ephemeris, "r") as f: ephemeris_str = f.read() self.variables = { "pulsar": pulsar, "ephemerisText": ephemeris_str, "projectCode": project_code, "projectShort": project_short, "comment": comment, } return self.mutation_graphql()
[docs] def update(self, id, pulsar, created_at, created_by, ephemeris, p0, dm, rm, comment, valid_from, valid_to): """Update a Ephemeris database object. Parameters ---------- id : int The database ID pulsar : str The pulsar name. ephemeris : str The ephemeris text as a single string (includes new line characters). project_code : str, optional The project code, by default None project_short : str, optional The project short name (e.g PTA), by default None comment : str, optional A comment about the ephemeris, by default None Returns ------- client_response: A client response object. """ self.mutation_name = "updateEphemeris" self.mutation = """ mutation ($id: Int!, $pulsar: Int!, $created_at: DateTime!, $created_by: String!, $ephemeris: JSONString!, $p0: Decimal!, $dm: Float!, $rm: Float!, $comment: String!, $valid_from: DateTime!, $valid_to: DateTime!) { updateEphemeris (id: $id, input: { pulsar_id: $pulsar, created_at: $created_at, created_by: $created_by, ephemeris: $ephemeris, p0: $p0, dm: $dm, rm: $rm, comment: $comment, valid_from: $valid_from, valid_to: $valid_to }) { ephemeris { id, pulsar {id}, createdAt, createdBy, ephemeris, p0, dm, rm, comment, validFrom, validTo } } } """ self.variables = { "id": id, "pulsar": pulsar, "created_at": created_at, "created_by": created_by, "ephemeris": ephemeris, "p0": p0, "dm": dm, "comment": comment, "valid_from": valid_from, "valid_to": valid_to, } return self.mutation_graphql()
[docs] def delete(self, id): """Delete a Ephemeris database object. Parameters ---------- id : int The database ID Returns ------- client_response: A client response object. """ self.mutation_name = "deleteEphemeris" self.mutation = """ mutation ($id: Int!) { deleteEphemeris(id: $id) { ok } } """ self.variables = { "id": id, } return self.mutation_graphql()
def process(self, args): """Parse the arguments collected by the CLI.""" self.print_stdout = True if args.subcommand == "create": return self.create( args.pulsar, args.ephemeris_loc, args.project_code, args.comment, ) elif args.subcommand == "update": return self.update( args.id, args.pulsar, args.created_at, args.created_by, args.ephemeris, args.p0, args.dm, args.rm, args.comment, args.valid_from, args.valid_to, ) elif args.subcommand == "list": return self.list(args.id, args.pulsar, args.p0, args.dm, args.rm, args.eph) elif args.subcommand == "delete": return self.delete(args.id) else: raise RuntimeError(f"{args.subcommand} command is not implemented") @classmethod def get_name(cls): return "ephemeris" @classmethod def get_description(cls): return "A pulsar ephemeris" @classmethod def configure_parsers(cls, parser): """Add sub-parsers for each of the valid commands.""" # create the parser for the "list" command parser.set_defaults(command=cls.get_name()) subs = parser.add_subparsers(dest="subcommand") subs.required = True parser_list = subs.add_parser("list", help="list existing ephemerides") parser_list.add_argument("--id", metavar="ID", type=int, help="list ephemeris matching the id [int]") parser_list.add_argument( "--pulsar", metavar="PSR", type=int, help="list ephemeris matching the pulsar id [int]" ) parser_list.add_argument( "--p0", metavar="P0", type=float, help="list ephemeris matching the pulsar P0 [float]" ) parser_list.add_argument( "--dm", metavar="DM", type=float, help="list ephemeris matching the pulsar DM [float]" ) parser_list.add_argument( "--rm", metavar="RM", type=float, help="list ephemeris matching the pulsar RM [float]" ) parser_list.add_argument("--eph", metavar="EPH", type=str, help="list ephemeris matching the ephemeris [JSON]")