tessl/pypi-astroquery

Functions and classes to access online astronomical data resources

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Virtual Observatory Services

Name: tessl/pypi-astroquery
Author: tessl

Services implementing Virtual Observatory (VO) standards for interoperability, cross-matching, and standardized data access protocols. These services enable seamless integration between different astronomical databases and tools.

Capabilities

VO Cone Search - Virtual Observatory Cone Search

VO Cone Search provides standardized access to astronomical catalogs using the Simple Cone Search protocol, enabling uniform queries across participating data centers worldwide.

from astroquery.vo_conesearch import conesearch, list_catalogs
from astroquery.vo_conesearch.validator.validate import cs_validate

def conesearch(center: Union[SkyCoord, str], radius: Union[Quantity, str],
              catalog_db: str = None, pedantic: bool = None, 
              verbose: bool = False, cache: bool = True) -> Table:
    """
    Perform cone search on Virtual Observatory catalogs.
    
    Parameters:
    - center: Search center coordinates
    - radius: Search radius
    - catalog_db: Specific catalog database to query
    - pedantic: Strict VO compliance checking
    - verbose: Show detailed output
    - cache: Enable result caching
    
    Returns:
    Table with sources within the search cone
    """

def list_catalogs(**kwargs) -> List[str]:
    """
    List available VO cone search catalogs.
    
    Parameters:
    - **kwargs: Filtering criteria for catalogs
    
    Returns:
    List of available catalog identifiers
    """

def cs_validate(url: str, verbose: bool = True) -> Dict:
    """
    Validate a cone search service for VO compliance.
    
    Parameters:
    - url: Cone search service URL
    - verbose: Show detailed validation results
    
    Returns:
    Dictionary with validation results
    """

# Configuration
from astroquery.vo_conesearch import conf
conf.pedantic: bool = False   # Strict VO compliance
conf.timeout: int = 60        # Connection timeout

Usage examples:

from astroquery.vo_conesearch import conesearch, list_catalogs
from astropy.coordinates import SkyCoord
import astropy.units as u

# Basic cone search
coords = SkyCoord('12h30m43s', '+12d23m28s', frame='icrs')
result = conesearch(coords, radius=2*u.arcmin)
print(f"Found {len(result)} sources in cone search")
print(result['RAJ2000', 'DEJ2000', 'Vmag'])

# List available catalogs
catalogs = list_catalogs()
print(f"Available VO catalogs: {len(catalogs)}")

# Search specific catalog
specific_result = conesearch(coords, radius=1*u.arcmin, 
                           catalog_db='2MASS All-Sky Catalog of Point Sources')

# Validate cone search service
validation = cs_validate('http://example.com/conesearch')
print(f"Service valid: {validation['valid']}")

Cross-match Service - CDS X-Match

CDS X-Match service provides cross-matching capabilities between astronomical catalogs, enabling identification of the same sources across different surveys and databases.

from astroquery.xmatch import XMatch

def query(cat1: Union[Table, str], cat2: str, max_distance: Union[Quantity, str],
         colRA1: str = None, colDec1: str = None, colRA2: str = None, colDec2: str = None,
         area: str = 'allsky') -> Table:
    """
    Cross-match two catalogs.
    
    Parameters:
    - cat1: First catalog (Table object or VizieR catalog name)
    - cat2: Second catalog (VizieR catalog name)
    - max_distance: Maximum matching distance
    - colRA1: RA column name in first catalog
    - colDec1: Dec column name in first catalog  
    - colRA2: RA column name in second catalog
    - colDec2: Dec column name in second catalog
    - area: Sky area constraint ('allsky' or region specification)
    
    Returns:
    Table with cross-matched sources
    """

def query_async(cat1: Union[Table, str], cat2: str, max_distance: Union[Quantity, str],
               **kwargs) -> JobResults:
    """
    Submit asynchronous cross-match query.
    
    Parameters:
    - cat1: First catalog
    - cat2: Second catalog
    - max_distance: Maximum matching distance
    - **kwargs: Additional cross-match parameters
    
    Returns:
    JobResults object for retrieving results
    """

# Configuration
from astroquery.xmatch import conf
conf.server: str              # X-Match server URL
conf.timeout: int = 300       # Connection timeout

Usage examples:

from astroquery.xmatch import XMatch
from astroquery.vizier import Vizier
import astropy.units as u
from astropy.coordinates import SkyCoord

# Cross-match with VizieR catalogs
result = XMatch.query(cat1='vizier:2MASS-PSC', cat2='vizier:SDSS-DR12',
                     max_distance=2*u.arcsec, area='allsky')
print(f"Cross-matched {len(result)} sources between 2MASS and SDSS")

# Cross-match user catalog with reference catalog
# First, get a sample catalog
coords = SkyCoord('12h30m43s', '+12d23m28s', frame='icrs')
my_sources = Vizier.query_region(coords, radius=10*u.arcmin, 
                                catalog='2MASS-PSC')[0]

# Cross-match with Gaia
gaia_matches = XMatch.query(cat1=my_sources, cat2='vizier:I/350/gaiaedr3',
                           max_distance=1*u.arcsec,
                           colRA1='RAJ2000', colDec1='DEJ2000')
print(f"Found {len(gaia_matches)} Gaia matches")

# Large area cross-match (async)
job = XMatch.query_async(cat1='vizier:HIPPARCOS', cat2='vizier:Gaia-EDR3',
                        max_distance=3*u.arcsec, area='allsky')
large_xmatch = job.get_results()

Table Access Protocol (TAP) Services

TAP services provide standardized SQL-like query access to astronomical databases using ADQL (Astronomical Data Query Language).

from astroquery.utils.tap import TapPlus

def launch_job(query: str, name: str = None, output_file: str = None,
              output_format: str = 'votable', verbose: bool = False,
              dump_to_file: bool = False, background: bool = False) -> JobResults:
    """
    Execute ADQL query on TAP service.
    
    Parameters:
    - query: ADQL query string
    - name: Job name for identification
    - output_file: File to save results
    - output_format: Output format ('votable', 'csv', 'fits')
    - verbose: Show detailed output
    - dump_to_file: Save query results to file
    - background: Submit as background job
    
    Returns:
    JobResults with query results
    """

def launch_job_async(query: str, **kwargs) -> JobResults:
    """
    Submit asynchronous ADQL query.
    
    Parameters:
    - query: ADQL query string
    - **kwargs: Additional job parameters
    
    Returns:
    JobResults object for monitoring and retrieving results
    """

def get_tables(only_names: bool = False, verbose: bool = False) -> List:
    """
    Get available tables in TAP service.
    
    Parameters:
    - only_names: Return only table names
    - verbose: Show detailed table information
    
    Returns:
    List of table information or names
    """

def get_columns(table_name: str) -> Table:
    """
    Get column information for a specific table.
    
    Parameters:
    - table_name: Name of the table
    
    Returns:
    Table with column descriptions
    """

# Configuration for TAP services
conf.timeout: int = 600       # Query timeout

Usage examples:

from astroquery.utils.tap import TapPlus
from astroquery.gaia import Gaia

# Direct TAP query to Gaia
gaia_tap = TapPlus(url="https://gea.esac.esa.int/tap-server/tap")

# Get bright stars near galactic center
query = """
SELECT TOP 1000 source_id, ra, dec, phot_g_mean_mag, bp_rp
FROM gaiadr3.gaia_source 
WHERE CONTAINS(POINT('ICRS', ra, dec), 
               CIRCLE('ICRS', 266.417, -29.008, 1.0)) = 1
AND phot_g_mean_mag < 15
ORDER BY phot_g_mean_mag
"""
job = gaia_tap.launch_job(query)
bright_stars = job.get_results()
print(f"Found {len(bright_stars)} bright stars")

# Asynchronous complex query
complex_query = """
SELECT g.source_id, g.ra, g.dec, g.phot_g_mean_mag, 
       g.pmra, g.pmdec, g.parallax,
       rv.radial_velocity, rv.radial_velocity_error
FROM gaiadr3.gaia_source AS g
JOIN gaiadr3.rvs_mean_spectrum AS rv ON g.source_id = rv.source_id
WHERE g.parallax > 10 AND g.parallax_error/g.parallax < 0.1
AND g.phot_g_mean_mag < 12
"""
async_job = gaia_tap.launch_job_async(complex_query, name="nearby_stars_rv")
nearby_rv = async_job.get_results()

# Get table information
tables = gaia_tap.get_tables(only_names=True)
print(f"Available tables: {len(tables)}")

# Get column info
columns = gaia_tap.get_columns('gaiadr3.gaia_source')
print(columns['column_name', 'datatype', 'description'][:10])

Registry Services

VO Registry services for discovering available astronomical data services and their capabilities.

from astroquery.vo_conesearch.vos_catalog import get_remote_catalog_db

def get_remote_catalog_db(cache: bool = True, verbose: bool = False) -> Dict:
    """
    Get remote catalog database from VO registry.
    
    Parameters:
    - cache: Enable catalog database caching
    - verbose: Show detailed output
    
    Returns:
    Dictionary with available VO services
    """

def search_registry(keywords: List[str], service_type: str = None) -> List:
    """
    Search VO registry for services matching criteria.
    
    Parameters:
    - keywords: Keywords to search for
    - service_type: Type of service ('conesearch', 'tap', 'sia')
    
    Returns:
    List of matching services
    """

# Configuration
conf.registry_timeout: int = 60  # Registry query timeout

Usage examples:

from astroquery.vo_conesearch.vos_catalog import get_remote_catalog_db

# Get available VO services
catalog_db = get_remote_catalog_db(verbose=True)
print(f"Found {len(catalog_db)} VO cone search services")

# Filter for specific type of catalogs  
photometry_services = [svc for svc in catalog_db 
                      if 'photometry' in svc.lower()]
print(f"Photometry services: {len(photometry_services)}")

Common Types

from astropy.table import Table
from astropy.coordinates import SkyCoord
from astropy.units import Quantity
from typing import Union, List, Dict, Optional

# Input types
center: Union[SkyCoord, str]                # Search center
coordinates: Union[SkyCoord, str]           # Target coordinates
radius: Union[Quantity, str]                # Search radius  
max_distance: Union[Quantity, str]          # Cross-match distance

# Catalog specifications
cat1: Union[Table, str]                     # First catalog
cat2: str                                   # Second catalog (VizieR name)
catalog_db: str                             # Cone search catalog
table_name: str                             # TAP table name

# Query parameters
query: str                                  # ADQL query string
colRA1: str                                 # RA column name
colDec1: str                                # Dec column name
area: str = 'allsky'                        # Sky area constraint
pedantic: bool = False                      # VO compliance checking

# Service parameters
output_format: str = 'votable'              # Output format
background: bool = False                    # Background execution
verbose: bool = False                       # Detailed output

# Return types  
Table                                       # Query results
JobResults                                  # Asynchronous job results
List[str]                                   # Service/table names
Dict                                        # Service metadata/validation

Install with Tessl CLI