Source code for astroquery.esa.xmm_newton.core

# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""

@author: Elena Colomo
@contact: ecolomo@esa.int

European Space Astronomy Centre (ESAC)
European Space Agency (ESA)

Created on 3 Sept 2019


"""
import re
from ...utils.tap.core import TapPlus
from ...query import BaseQuery
import shutil
import cgi
from pathlib import Path

from . import conf
from astropy import log


__all__ = ['XMMNewton', 'XMMNewtonClass']


[docs]class XMMNewtonClass(BaseQuery): data_url = conf.DATA_ACTION data_aio_url = conf.DATA_ACTION_AIO metadata_url = conf.METADATA_ACTION TIMEOUT = conf.TIMEOUT def __init__(self, tap_handler=None): super(XMMNewtonClass, self).__init__() if tap_handler is None: self._tap = TapPlus(url="http://nxsa.esac.esa.int" "/tap-server/tap/") else: self._tap = tap_handler
[docs] def download_data(self, observation_id, *, filename=None, verbose=False, **kwargs): """ Download data from XMM-Newton Parameters ---------- observation_id : string id of the observation to be downloaded, mandatory The identifier of the observation we want to retrieve, 10 digits example: 0144090201 filename : string file name to be used to store the file verbose : bool optional, default 'False' flag to display information about the process level : string level to download, optional, by default everything is downloaded values: ODF, PPS instname : string instrument name, optional, two characters, by default everything values: OM, R1, R2, M1, M2, PN instmode : string instrument mode, optional examples: Fast, FlatFieldLow, Image, PrimeFullWindow filter : string filter, optional examples: Closed, Open, Thick, UVM2, UVW1, UVW2, V expflag : string exposure flag, optional, by default everything values: S, U, X(not applicable) expno : integer exposure number with 3 digits, by default all exposures examples: 001, 003 name : string product type, optional, 6 characters, by default all product types examples: 3COLIM, ATTTSR, EVENLI, SBSPEC, EXPMAP, SRCARF datasubsetno : character data subset number, optional, by default all examples: 0, 1 sourceno : hex value source number, optional, by default all sources example: 00A, 021, 001 extension : string file format, optional, by default all formats values: ASC, ASZ, FTZ, HTM, IND, PDF, PNG Returns ------- None if not verbose. It downloads the observation indicated If verbose returns the filename """ link = self.data_aio_url + "obsno=" + observation_id link = link + "".join("&{0}={1}".format(key, val) for key, val in kwargs.items()) if verbose: log.info(link) response = self._request('GET', link, save=False, cache=True) # Get original extension _, params = cgi.parse_header(response.headers['Content-Disposition']) r_filename = params["filename"] suffixes = Path(r_filename).suffixes if filename is None: filename = observation_id filename += "".join(suffixes) log.info("Copying file to {0}...".format(filename)) with open(filename, 'wb') as f: f.write(response.content) if verbose: log.info("Wrote {0} to {1}".format(link, filename))
[docs] def get_postcard(self, observation_id, *, image_type="OBS_EPIC", filename=None, verbose=False): """ Download postcards from XSA Parameters ---------- observation_id : string id of the observation for which download the postcard, mandatory The identifier of the observation we want to retrieve, regardless of whether it is simple or composite. image_type : string image type, optional, default 'OBS_EPIC' The image_type to be returned. It can be: OBS_EPIC, OBS_RGS_FLUXED, OBS_RGS_FLUXED_2, OBS_RGS_FLUXED_3, OBS_EPIC_MT, OBS_RGS_FLUXED_MT, OBS_OM_V, OBS_OM_B, OBS_OM_U, OBS_OM_L, OBS_OM_M, OBS_OM_S, OBS_OM_W filename : string file name to be used to store the postcard, optional, default None verbose : bool optional, default 'False' Flag to display information about the process Returns ------- None. It downloads the observation postcard indicated """ params = {'RETRIEVAL_TYPE': 'POSTCARD', 'OBSERVATION_ID': observation_id, 'OBS_IMAGE_TYPE': image_type, 'PROTOCOL': 'HTTP'} link = self.data_url + "".join("&{0}={1}".format(key, val) for key, val in params.items()) if verbose: log.info(link) local_filepath = self._request('GET', link, params, cache=True, save=True) if filename is None: response = self._request('HEAD', link) response.raise_for_status() filename = re.findall('filename="(.+)"', response.headers[ "Content-Disposition"])[0] else: filename = observation_id + ".png" log.info("Copying file to {0}...".format(filename)) shutil.move(local_filepath, filename) if verbose: log.info("Wrote {0} to {1}".format(link, filename)) return filename
[docs] def query_xsa_tap(self, query, *, output_file=None, output_format="votable", verbose=False): """Launches a synchronous job to query the XSA tap Parameters ---------- query : str, mandatory query (adql) to be executed output_file : str, optional, default None file name where the results are saved if dumpToFile is True. If this parameter is not provided, the jobid is used instead output_format : str, optional, default 'votable' possible values 'votable' or 'csv' verbose : bool, optional, default 'False' flag to display information about the process Returns ------- A table object """ job = self._tap.launch_job(query=query, output_file=output_file, output_format=output_format, verbose=verbose, dump_to_file=output_file is not None) table = job.get_results() return table
[docs] def get_tables(self, *, only_names=True, verbose=False): """Get the available table in XSA TAP service Parameters ---------- only_names : bool, TAP+ only, optional, default 'True' True to load table names only verbose : bool, optional, default 'False' flag to display information about the process Returns ------- A list of tables """ tables = self._tap.load_tables(only_names=only_names, include_shared_tables=False, verbose=verbose) if only_names: return [t.name for t in tables] else: return tables
[docs] def get_columns(self, table_name, *, only_names=True, verbose=False): """Get the available columns for a table in XSA TAP service Parameters ---------- table_name : string, mandatory, default None table name of which, columns will be returned only_names : bool, TAP+ only, optional, default 'True' True to load table names only verbose : bool, optional, default 'False' flag to display information about the process Returns ------- A list of columns """ tables = self._tap.load_tables(only_names=False, include_shared_tables=False, verbose=verbose) columns = None for table in tables: if str(table.name) == str(table_name): columns = table.columns break if columns is None: raise ValueError("table name specified is not found in " "XSA TAP service") if only_names: return [c.name for c in columns] else: return columns
XMMNewton = XMMNewtonClass()