Installation of pyvo
In order to interact with the TAP interface of cars.aip.de
you only require
python 3+
and pyvo 1+
.
pip install pyvo>=1.0
Importing PyVo and checking the version
It is useful to always print the version of pyvo you are using. Most of non-working scripts fail because of an old version of pyvo
.
from pkg_resources import parse_version
import pyvo
# Verify the version of pyvo
if parse_version(pyvo.__version__) < parse_version('1.0'):
raise ImportError('pyvo version must be at least than 1.0')
print('\npyvo version %s \n' % (pyvo.__version__,))
Authentication
After registration you can access your API Token by clicking on your user name in the right side of the menu bar. Then select API Token
.
You will see a long alphanumerical word. Just copy it where ever you see <your-token>
; in the following examples.
The
API Token
identifies you and provides access to the results tables of your queries.
The connection to the TAP service can be done that way:
import requests
import pyvo
# Setup tap_service connection
service_name = "CARS@AIP"
url = "https://cars.aip.de/tap"
token = 'Token <your-token>; '
print('TAP service %s \n' % (service_name,))
# Setup authorization
tap_session = requests.Session()
tap_session.headers['Authorization'] = token
tap_service = pyvo.dal.TAPService(url, session=tap_session)
Short queries
Many queries last less than a few seconds, we call them short queries. The latter can be executed with synchronized jobs. You will retrieve the results interactively.
lang = "PostgreSQL"
query = '''
-- Spectral Energy Distribution (SED) for target HE0040-1105
SELECT phot_band, phot_centwave, phot_flux, phot_flux_error, phot_flux_limit,
log(phot_centwave) as "log_centwave",
log(phot_flux) as "log_phot_flux"
FROM cars_dr1.photometry
WHERE hes_name = 'HE0040-1105'
'''
tap_result = tap_service.run_sync(query, language=lang)
Remark: the
lang
parameter can take two values eitherPostgreSQL
orADQL
this allows to access some features present in the one or the other language for more details about the difference between both please refer : Documentation or to IOVA docs
The result tap_result
is a so called TAPResults
that is essentially a wrapper around an Astropy votable.Table
. For standard conversion see Convert to various python types.
The entire script can be found on github: cars-sync-query.py
Asynchronous jobs
For slightly longer queries, typically counting or larger selections (>10000 entries) a synchronized job will fail because of timeouts (from http protocol or server settings). This is why we provide the possibility to submit asynchronous jobs. These type of jobs will run on the server side, store the results such that you can retrieve them at a later time. They come in 2 flavors:
- 30 second queue
- 5 min queue
The 30 seconds queue
Most of the asynchronous queries will require less than 30 seconds, basically all queries without JOIN
, or CONE SEARCH
. Therefore this queue is the default and should be preferred.
#
# Submit the query as an async job
#
query_name = "BH-mass-vs-ENLR-max-size"
lang = 'PostgreSQL' # ADQL or PostgreSQL
query = '''
-- BH mass in comparison to ENLR maximum size
SELECT agn.target_id,
agn.hes_name,
agn.log_bh_mass,
host.enlr_max_kpc,
log(host.enlr_max_kpc) as "log_enlr_max_kpc"
FROM cars_dr1.agn_parameters as agn,
cars_dr1.host_properties as host
WHERE agn.target_id = host.target_id
AND host.enlr_max_kpc > 0.0;
'''
job = tap_service.submit_job(query, language=lang, runid=query_name, queue="30s")
job.run()
#
# Wait to be completed (or an error occurs)
#
job.wait(phases=["COMPLETED", "ERROR", "ABORTED"], timeout=30.0)
print('JOB %s: %s' % (job.job.runid, job.phase))
#
# Fetch the results
#
job.raise_if_error()
print('\nfetching the results...')
tap_results = job.fetch_result()
print('...DONE\n')
As for sync jobs, the result is a TAPResults
object.
The entire script can be found at: cars-async-30s.py
The 5 minutes queue
If you want to extract information on specific targets from various tables you have to JOIN
tables. Your query may need more than a few seconds. For that, the 5 minutes queue provide a good balance. It should be noticed that for such a queue the wait method should not be used to prevent an overload of the server at peak usage. Therefore using the script with the sleep()
method is recommended.
#
# Submit the query as an async job
#
lang = 'PostgreSQL'
query_name = "O3-wind-velo-disp"
query = '''
-- [OIII] wing velocity dispersion selection >300km/s
SELECT agn.hes_name,
agn.wing_sigma, ifu.observed, ifu.maps_full
FROM cars_dr1.agn_parameters as agn,
cars_dr1.ifu_observations as ifu
WHERE agn.target_id = ifu.target_id
AND agn.wing_sigma > 300;
'''
job = tap_service.submit_job(query, language=lang, runid=query_name, queue="5m")
job.run()
print('JOB %s: SUBMITTED' % (job.job.runid,))
#
# Wait for the query to finish
#
while job.phase not in ("COMPLETED", "ERROR", "ABORTED"):
print('WAITING...')
time.sleep(120.0) # do nothing for some time
print('JOB ' + (job.phase))
#
# Fetch the results
#
job.raise_if_error()
print('\nfetching the results...')
results = job.fetch_result()
print('...DONE\n')
The entire script can be found on github: cars-async-5m.py
Submitting multiple queries
Some time it is needed to submit several queries at one time. Either because the entire query may last longer than 5 minutes and you need to cut it in smaller parts, or because you need non JOIN
-able information from various tables.
List of file queries
Sometimes it is useful to just send all .sql
queries present in a directory. For such purpose you can use comments to provide the proper parameters.
Let us consider the file BPT-line.sql
-- BPT line construction for CARS target ID 12
-- LANGUAGE = PostgreSQL
-- QUEUE = 30s
SELECT log(oiii5007_flux/hbeta_flux) as "log(oiii_hb)",
log(nii6583_flux/halpha_flux) as "log(nii_ha)",
distance
FROM cars_dr1.ifu_datatable_unbinned
WHERE target_id = 12
AND (oiii5007_flux/oiii5007_flux_err) > 3
AND (nii6583_flux/nii6583_flux_err) > 3
AND (hbeta_flux/hbeta_flux_err) > 3
AND (halpha_flux/halpha_flux_err) > 3
AND (gas_vel_err < 20)
AND (gas_fwhm_err < 30);
The language
and queue
are prescibed as comments. The query can then be submitted in a script like the following:
import glob
#
# Submit the query as an Asynchrone job
#
# find all .sql files in current directory
queries_filename = sorted(glob.glob('./*.sql'))
print('Sending %d examples' % (len(queries_filename),))
# initialize test results
jobs = []
failed = []
# send all queries
for query_filename in queries_filename:
# read the .SQL file
with open(query_filename, 'r') as fd:
query = ' '.join(fd.readlines())
# Set language from comments (default: PostgreSQL)
if 'LANGUAGE = ADQL' in query:
lang = 'ADQL'
else:
lang = 'PostgreSQL'
# Set queue from comments (default: 30s)
if 'QUEUE = 5m' in query:
queue = '5m'
else:
queue = '30s'
# Set the runid from sql filename
base = os.path.basename(query_filename)
runid = os.path.splitext(base)[0]
print('\n> Query : %s\n%s\n' % (runid, query))
The rest of the submission process and retrieval can be done in any manner. An example can be found here: cars-tutorial-from-files.py
Downloading files from path results
Some queries do not return data but the urls of the files where the data are stored; examples are images or spectral data cubes. When your query returns a few file-paths it is possible to download them by hand, however it is usually more practical to download them automaticaly. Here is an example using python:
import os
import urlib
#
# Query the fits files
#
lang = "PostgreSQL"
query = '''
-- [OIII] wing velocity dispersion selection >300km/s
SELECT ifu.observed as "file_urls"
FROM cars_dr1.agn_parameters as agn,
cars_dr1.ifu_observations as ifu
WHERE agn.target_id = ifu.target_id
AND agn.wing_sigma > 300
'''
# Submit the query as Synchronous job (can be also done asynchronously)
tap_result = tap_service.run_sync(query, language=lang)
files_url = tap_result.to_table()
#
# Download the fits files into local directory
#
target_directory = './fits/'
files_base_url = 'https://cars.aip.de/files/'
for file_url in files_url:
# extract name of the fits
filename = os.path.basename(file_url[0])
# set the target local file
full_filename = os.path.join(target_directory, filename)
# build the url pointing to the file
full_file_url = os.path.join(files_base_url, file_url[0])
# download and save into target file
print("Downloaded {filename} into {target}".format(filename=filename, target=full_filename))
urllib.request.urlretrieve(full_file_url, full_filename)
print('\nDone')
Convert result to various python types
The results obtained via the fetch_results()
method returns a so called TAPResults
object. The latter is essencially a votable
. In case you are not familiar with votables
here are a few tricks to get back to some more general pythonic types.
-
Print the data:
python tap_results.to_table().pprint(max_lines=10)
It is important to notice themax_lines
keyword, printing too many lines may crash a low-memory machine. -
Show as html (in a browser):
python tap_results.to_table().show_in_browser(max_lines=10)
It is important to notice themax_lines
keyword, printing too many lines may crash a low-memory machine. -
Show in a notebook (ipython, jupyter or jupyterlab):
python tap_results.to_table().show_in_notebook(display_length=10)
It is important to notice thedisplay_length
keyword, printing too many lines may crash a low-memory machine. -
Get a numpy array:
python np_array = tap_results.to_table().as_array()
-
Get a Panda's DataFrame
python df = tap_results.to_table().to_pandas()
- Get the header of DataFrame:
python df.head()
- Get the header of DataFrame:
Archiving your jobs
If you submit several large queries you may go over quota: set to 10 GB. In order to avoid to get over quota you may consider archiving your jobs. Archiving removes the data from the server side but keeps the SQL query. This allows to resubmit a query at a later time.
Deleting (Archiving) a job with pyvo
can be simply done that way:
job.delete()
Archiving all COMPLETED
jobs
A nice feature of the TAP service
is to retrieve all jobs that are marked as COMPLETED
and archive them at ones. This can be done as follows:
#
# Archiving all COMPLETED jobs
#
# obtain the list of completed job_descriptions
completed_job_descriptions = tap_service.get_job_list(phases='COMPLETED')
# Archiving each of them
for job_description in completed_job_descriptions:
# get the jobid
jobid = job_description.jobid
# recreate the url by hand
job_url = tap_service.baseurl + '/async/' + jobid
# recreate the job
job = AsyncTAPJob(job_url, session=tap_session)
print('Archiving: {url}'.format(url=job_url))
job.delete() # archive job
Rerunning ARCHIVED
jobs
Rerunning and retrieving results from a job that have been archived previously, can be achieved that way:
#
# Rerunning Archived jobs
#
# obtain the list of the two last ARCHIVED job_descriptions
archived_job_descriptions = tap_service.get_job_list(phases='ARCHIVED', last=2)
# rerunning the two last Archived jobs
for job_description in archived_job_descriptions:
# get jobid
jobid = job_description.jobid
# recreate the url by hand
job_url = tap_service.baseurl + '/async/' + jobid
# recreate the archived job
archived_job = AsyncTAPJob(job_url, session=tap_session)
# extract the query
query = archived_job.query
# resubmit the query with corresponding parameters
job = tap_service.submit_job(query, language='PostgreSQL', runid='rerun', queue='30s')
print('%(url)s:\n%(query)s\n' % {"url": job_url, "query": query})
# start the job
job.run()
Retrieving the results is done alike explained above.
If you prefer you can also filter for a given runid
.
#
# Filtering by runid
#
target_runid = 'O3-wind-velo-disp'
# obtain the list of completed job_descriptions
archived_job_descriptions = tap_service.get_job_list(phases='ARCHIVED')
for job_description in archived_job_descriptions:
# select the job with runid fitting target_runid
if job_description.runid == target_runid:
# get jobid
jobid = job_description.jobid
# recreate the url by hand
job_url = tap_service.baseurl + '/async/' + jobid
# recreate the archived job
archived_job = AsyncTAPJob(job_url, session=tap_session)
# extract the query
query = archived_job.query
# resubmit the query with corresponding parameters
job = tap_service.submit_job(query, language='PostgreSQL', runid='rerun', queue='30s')
print('%(url)s:\n%(query)s\n' % {"url": job_url, "query": query})
# start the job
job.run()