1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
# An example script that does a registry query (first argument to main,
# here "standard stars") and then runs cone searches (RA, DEC, SRC
# taken from the command line in that order) against them until the
# first returns a result. That result is then broadcast to all SAMP
# clients.
#
# Usage example: Start TOPCAT, then run
#
# python serialquery.py 314 27.1 2
#
# After a short while you should see a result table in TOPCAT.
#
# Disclaimer: Of course, this isn't how you should write a command line
# interface.
#
# This file has been put in the public domain by the authors,
# Markus Demleitner <msdemlei@ari.uni-heidelberg.de> and
# Stefan Becker <sbecker@ari.uni-heidelberg.de>.
import contextlib
import os
import sys
import tempfile
import time
import astropy
from astropy.vo.samp import SAMPIntegratedClient
from pyvo.dal import scs
from pyvo.registry import regtap
@contextlib.contextmanager
def samp_accessible(astropy_table):
"""a context manager making astroy_table available under a (file)
URL for the controlled section.
This is useful with uploads.
"""
handle, f_name = tempfile.mkstemp(suffix=".xml")
with os.fdopen(handle, "w") as f:
astropy_table.write(output=f,
format="votable")
try:
yield "file://"+f_name
finally:
os.unlink(f_name)
@contextlib.contextmanager
def SAMP_conn():
"""a context manager to give the controlled block a SAMP connection.
The program will disconnect as the controlled block is exited.
"""
client = SAMPIntegratedClient(name="serialquery",
description="A serial SCS querier.")
client.connect()
try:
yield client
finally:
client.disconnect()
def broadcast(astropy_table):
"""broadcasts an astropy table object to all SAMP clients on the local
bus.
"""
with SAMP_conn() as client:
with samp_accessible(astropy_table) as table_url:
client.notify_all(
{
"samp.mtype": "table.load.votable",
"samp.params": {
"url": table_url,
}})
time.sleep(2) # hack: give other clients time to pick our table up
def main(query_terms, ra, dec, sr):
for resource in regtap.search(
keywords=[query_terms], servicetype="image"
):
print(resource.res_title)
result = resource.service.search((ra, dec), sr)
print(len(result))
if result:
break
else:
sys.exit("No service has results for CIRCLE({0}, {1}, {2})".format(
ra, dec, sr))
broadcast(result.table)
if __name__=="__main__":
# serialquery.py RA DEC SR
main("standard stars", *[float(v) for v in sys.argv[1:]])
|
