Commit 546d58d9 authored by Mattia Mancini's avatar Mattia Mancini

first commit

parents
import pyvo as vo
from argparse import ArgumentParser
from pyvo.dal import TAPService, TAPResults
from pyvo.dal.tap import TAPRecord
import numpy
import json
def list_tables(service: TAPService, args):
for key, value in service.tables.items():
print(key)
def _decode_bytestrings(value):
value = value.decode() if isinstance(value, bytes) else repr(value)
return value
class TAPRecordEncode(json.JSONEncoder):
def default(self, value):
if isinstance(value, numpy.integer):
return int(value)
elif isinstance(value, numpy.floating):
return float(value)
elif isinstance(value, numpy.complex):
return complex(value)
elif isinstance(value, bytes):
return value.decode()
elif isinstance(value, TAPResults):
return list(value)
elif isinstance(value, TAPRecord):
return dict(value)
else:
print(type(value))
super().default(value)
def _print_table_row(row_id, row):
print('ROW #{:3d}'.format(row_id + 1) + '-' * 10)
for key, value in row.items():
value = value.decode() if isinstance(value, bytes) else value
print('{:20s}:\t {}'.format(key, value))
def _print_result_set(result_set):
for row_n, result in enumerate(result_set):
_print_table_row(row_n, result)
def _save_result_set_to_file(result_set, path):
with open(path, 'w') as f_stream:
json.dump(result_set, f_stream, cls=TAPRecordEncode)
def print_table(service: TAPService, args):
table_name = args.table_name
limit = args.limit
where_statement = args.where
save = args.save
result_set = service.search('select top {} * from {} where {}'.format(limit, table_name,
where_statement))
if save:
_save_result_set_to_file(result_set, save)
else:
_print_result_set(result_set)
def parse_arguments():
parser = ArgumentParser(description='Inspect a VO registry')
parser.add_argument('--url', default='https://vo.astron.nl/__system__/tap/run/tap')
parser.set_defaults(func=lambda x, y: parser.print_help())
subparser = parser.add_subparsers()
list_table_parser = subparser.add_parser('list_tables')
list_table_parser.set_defaults(func=list_tables)
print_table_parser = subparser.add_parser('print_table')
print_table_parser.set_defaults(func=print_table)
print_table_parser.add_argument('table_name', help='name of the table')
print_table_parser.add_argument('--save', help='path of file where to save result as json', default=None)
print_table_parser.add_argument('--limit', help='limit number of results', default=10)
print_table_parser.add_argument('--where', help='where statement', default='')
parsed_arguments = parser.parse_args()
return parsed_arguments
def main():
arguments = parse_arguments()
service = vo.dal.TAPService('https://vo.astron.nl/__system__/tap/run/tap')
arguments.func(service, arguments)
if __name__ == '__main__':
main()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment