Source code for psrdb.tables.pulsar_fold_result

from datetime import datetime

from psrdb.graphql_table import GraphQLTable
from psrdb.load_data import EXCLUDE_BADGES_CHOICES


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 PulsarFoldResult database object on the command line in different ways based on the sub-commands.")
    PulsarFoldResult.configure_parsers(parser)
    return parser


[docs] class PulsarFoldResult(GraphQLTable): """Class for interacting with the PulsarFoldResult 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 = "pulsar_fold_result" self.field_names = [ "id", "observation { utcStart }", "observation { band }", "observation { duration }", "pipelineRun { id }", "pipelineRun { dm }", "pipelineRun { dmErr }", "pipelineRun { dmEpoch }", "pipelineRun { dmEpoch }", "pipelineRun { dmChi2r }", "pipelineRun { dmTres }", "pipelineRun { sn }", "pipelineRun { flux }", "pipelineRun { rm }", "pipelineRun { rmErr }", "pipelineRun { percentRfiZapped }", ]
[docs] def list( self, pulsar=None, mainProject=None, utcStart=None, beam=None, utcs=None, utce=None, ): """Return a list of PulsarFoldResult information based on the `self.field_names` and filtered by the parameters. Parameters ---------- pulsar : str, optional Filter by the pulsar name, by default None mainProject : str, optional Filter by the main project name, by default None utcStart : str, optional Filter by the utcStart, by default None beam : int, optional Filter by the beam number, 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. """ filters = [ {"field": "pulsar", "value": pulsar}, {"field": "mainProject", "value": mainProject}, {"field": "utcStart", "value": utcStart}, {"field": "beam", "value": beam}, ] if utcs is not None: d = datetime.strptime(utcs, '%Y-%m-%d-%H:%M:%S') filters.append({"field": "utcStartGte", "value": f"{d.date()}T{d.time()}+00:00"}) if utce is not None: d = datetime.strptime(utce, '%Y-%m-%d-%H:%M:%S') filters.append({"field": "utcStartLte", "value": f"{d.date()}T{d.time()}+00:00"}) return GraphQLTable.list_graphql(self, self.table_name, filters, [], self.field_names)
def download( self, pulsar, mainProject=None, utcStart=None, beam=None, exclude_badges=None, utcs=None, utce=None, ): # Grab a dictionary of the pulsar_fold_results filters = [ {"field": "pulsar", "value": pulsar}, {"field": "mainProject", "value": mainProject}, {"field": "utcStart", "value": utcStart}, {"field": "beam", "value": beam}, ] if exclude_badges is not None: filters.append({"field": "excludeBadges", "value": exclude_badges}) if utcs is not None: d = datetime.strptime(utcs, '%Y-%m-%d-%H:%M:%S') filters.append({"field": "utcStartGte", "value": f"{d.date()}T{d.time()}+00:00"}) if utce is not None: d = datetime.strptime(utce, '%Y-%m-%d-%H:%M:%S') filters.append({"field": "utcStartLte", "value": f"{d.date()}T{d.time()}+00:00"}) self.get_dicts = True pulsar_fold_result_dicts = GraphQLTable.list_graphql(self, self.table_name, filters, [], self.field_names) # Create the output name output_name = f"pulsar_fold_result_{pulsar}" if mainProject is not None and mainProject: output_name += f"_{mainProject}" if utcStart is not None and utcStart: output_name += f"_{utcStart}" if beam is not None and beam: output_name += f"_beam{beam}" output_name += ".csv" # Loop over the pulsar_fold_results and dump them as a file with open(output_name, "w") as f: f.write("ID,UTC Start,Observing band,Duration (s),DM (pc cm^-3),DM error (pc cm^-3),DM epoch (MJD),DM chi2r,DM tres,SN,Flux (mJy),RM (rad m^-2),RM error (rad m^-2),RFI zapped (%)\n") for pulsar_fold_result_dict in pulsar_fold_result_dicts: data_line = [ str(pulsar_fold_result_dict["id"]), str(pulsar_fold_result_dict["observation"]["utcStart"]), str(pulsar_fold_result_dict["observation"]["band"]), str(pulsar_fold_result_dict["observation"]["duration"]), str(pulsar_fold_result_dict["pipelineRun"]["dm"]), str(pulsar_fold_result_dict["pipelineRun"]["dmErr"]), str(pulsar_fold_result_dict["pipelineRun"]["dmEpoch"]), str(pulsar_fold_result_dict["pipelineRun"]["dmChi2r"]), str(pulsar_fold_result_dict["pipelineRun"]["dmTres"]), str(pulsar_fold_result_dict["pipelineRun"]["sn"]), str(pulsar_fold_result_dict["pipelineRun"]["flux"]), str(pulsar_fold_result_dict["pipelineRun"]["rm"]), str(pulsar_fold_result_dict["pipelineRun"]["rmErr"]), str(pulsar_fold_result_dict["pipelineRun"]["percentRfiZapped"]), ] f.write(f"{','.join(data_line)}\n") return output_name def process(self, args): """Parse the arguments collected by the CLI.""" self.print_stdout = True if args.subcommand == "list": return self.list( args.pulsar, args.mainProject, args.utcStart, args.beam, ) elif args.subcommand == "download": return self.download( args.pulsar, args.mainProject, args.utcStart, args.beam, args.exclude_badges, args.utcs, args.utce, ) else: raise RuntimeError(f"{args.subcommand} command is not implemented") @classmethod def get_name(cls): return "pulsar_fold_result" @classmethod def get_description(cls): return "A pulsar pulsar_fold_result/standard" @classmethod def get_parsers(cls): """Returns the default parser for this model""" parser = GraphQLTable.get_default_parser("PulsarFoldResult model parser") cls.configure_parsers(parser) return parser @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("--pulsar", type=str, help="Name of the pulsar [str]") parser_list.add_argument("--mainProject", type=str, help="Name of the main project you want to filter pulsar_fold_results by [str]") parser_list.add_argument("--utcStart", type=str, help="UTC start time you want the results of [str]") parser_list.add_argument("--beam", type=int, help="Beam number you want to filter pulsar_fold_results by [int]") # create the parser for the "download" command parser_download = subs.add_parser("download", help="Download TOAs for a pulsar to a .tim file") parser_download.add_argument("pulsar", type=str, help="Name of the pulsar [str]") parser_download.add_argument("--mainProject", type=str, help="Name of the main project you want to filter pulsar_fold_results by [str]") parser_download.add_argument("--utcStart", type=str, help="UTC start time you want the results of [str]") parser_download.add_argument("--beam", type=int, help="Beam number you want to filter pulsar_fold_results by [int]") parser_download.add_argument( '--exclude_badges', nargs='*', choices=EXCLUDE_BADGES_CHOICES, help=f'List of observation badges/flags to exclude from download ToAs. The choices are: {EXCLUDE_BADGES_CHOICES}' ) parser_download.add_argument( "--utcs", type=str, help="Only use observations with utc_start greater than or equal to the timestamp [YYYY-MM-DD-HH:MM:SS]", ) parser_download.add_argument( "--utce", type=str, help="Only use observations with utc_start less than or equal to the timestamp [YYYY-MM-DD-HH:MM:SS]", )