# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
This file contains a contains the high-level functions to read the various
VOSI Endpoints.
"""
from astropy.utils.xml import iterparser
from astropy.utils.collections import HomogeneousList
from astropy.io.votable.exceptions import vo_raise, vo_warn
from astropy.io.votable.util import version_compare
from ...utils.xml.elements import xmlattribute, xmlelement, Element
from . import voresource as vr
from . import vodataservice as vs
from . import availability as av
from .exceptions import W15, W16, E07, E10
__all__ = [
"parse_tables", "parse_capabilities", "parse_availability",
"TablesFile", "CapabilitiesFile", "AvailabilityFile"]
def _pedantic_settings(pedantic):
"""
Controls the pedantic parser settings. Based on the bool
passed in to pedantic, create a config to be passed to
astropy parsing to raise exceptions or ignore them on
pedantic errors.
Parameters
----------
pedantic : bool
When `True`, raise an error when the file violates the spec,
otherwise issue a warning. Warnings may be controlled using
the standard Python mechanisms. See the `warnings`
module in the Python standard library for more information.
Returns
-------
A dict containing 'verify' configuration settings.
"""
if pedantic:
return {'verify': 'exception'}
else:
return {'verify': 'warn'}
[docs]def parse_tables(source, *, pedantic=None, filename=None,
_debug_python_based_parser=False):
"""
Parses a tableset xml file (or file-like object), and returns a
`~pyvo.io.vosi.endpoint.TablesFile` object.
Parameters
----------
source : str or readable file-like object
Path or file object containing a tableset xml file.
pedantic : bool, optional
When `True`, raise an error when the file violates the spec,
otherwise issue a warning. Warnings may be controlled using
the standard Python mechanisms. See the `warnings`
module in the Python standard library for more information.
Defaults to False.
filename : str, optional
A filename, URL or other identifier to use in error messages.
If *filename* is None and *source* is a string (i.e. a path),
then *source* will be used as a filename for error messages.
Therefore, *filename* is only required when source is a
file-like object.
Returns
-------
tables_file : `~pyvo.io.vosi.endpoint.TablesFile` object
See also
--------
pyvo.io.vosi.exceptions : The exceptions this function may raise.
"""
config = _pedantic_settings(pedantic)
if filename is None and isinstance(source, str):
config['filename'] = source
else:
config['filename'] = filename
with iterparser.get_xml_iterator(
source,
_debug_python_based_parser=_debug_python_based_parser
) as iterator:
return TablesFile(
config=config, pos=(1, 1)).parse(iterator, config)
[docs]def parse_capabilities(source, *, pedantic=None, filename=None,
_debug_python_based_parser=False):
"""
Parses a capabilities xml file (or file-like object), and returns a
`~pyvo.io.vosi.endpoint.CapabilitiesFile` object.
Parameters
----------
source : str or readable file-like object
Path or file object containing a capabilities xml file.
pedantic : bool, optional
When `True`, raise an error when the file violates the spec,
otherwise issue a warning. Warnings may be controlled using
the standard Python mechanisms. See the `warnings`
module in the Python standard library for more information.
Defaults to False.
filename : str, optional
A filename, URL or other identifier to use in error messages.
If *filename* is None and *source* is a string (i.e. a path),
then *source* will be used as a filename for error messages.
Therefore, *filename* is only required when source is a
file-like object.
Returns
-------
capabilities_file : `~pyvo.io.vosi.endpoint.CapabilitiesFile` object
See also
--------
pyvo.io.vosi.exceptions : The exceptions this function may raise.
"""
config = _pedantic_settings(pedantic)
if filename is None and isinstance(source, str):
config['filename'] = source
else:
config['filename'] = filename
with iterparser.get_xml_iterator(
source,
_debug_python_based_parser=_debug_python_based_parser
) as iterator:
return CapabilitiesFile(
config=config, pos=(1, 1)).parse(iterator, config)
[docs]def parse_availability(source, *, pedantic=None, filename=None,
_debug_python_based_parser=False):
"""
Parses a availability xml file (or file-like object), and returns a
`~pyvo.io.vosi.endpoint.AvailabilityFile` object.
Parameters
----------
source : str or readable file-like object
Path or file object containing a availability xml file.
pedantic : bool, optional
When `True`, raise an error when the file violates the spec,
otherwise issue a warning. Warnings may be controlled using
the standard Python mechanisms. See the `warnings`
module in the Python standard library for more information.
Defaults to False.
filename : str, optional
A filename, URL or other identifier to use in error messages.
If *filename* is None and *source* is a string (i.e. a path),
then *source* will be used as a filename for error messages.
Therefore, *filename* is only required when source is a
file-like object.
Returns
-------
availability_file : `~pyvo.io.vosi.endpoint.AvailabilityFile` object
See also
--------
pyvo.io.vosi.exceptions : The exceptions this function may raise.
"""
config = _pedantic_settings(pedantic)
if filename is None and isinstance(source, str):
config['filename'] = source
else:
config['filename'] = filename
with iterparser.get_xml_iterator(
source,
_debug_python_based_parser=_debug_python_based_parser
) as iterator:
return AvailabilityFile(
config=config, pos=(1, 1)).parse(iterator, config)
[docs]class TablesFile(Element):
"""
TABLESET/TABLE element: represents an entire file.
The keyword arguments correspond to setting members of the same
name, documented below.
"""
def __init__(self, *, config=None, pos=None, version="1.1"):
Element.__init__(self, config, pos)
self._tableset = None
self._table = None
self._ntables = None
version = str(version)
if version not in ("1.0", "1.1"):
raise ValueError("'version' should be one of '1.0' or '1.1'")
config['version'] = version
self._version = version
def __repr__(self):
if self.table:
return repr(self.table)
elif self.tableset:
return repr(self.tableset)
else:
return super().__repr__()
@xmlattribute
def version(self):
"""
The version of the TableSet specification that the file uses.
"""
return self._version
@version.setter
def version(self, version):
version = str(version)
if version not in ('1.0', '1.1'):
raise ValueError(
"pyvo.io.vosi.tables only supports VOSI versions 1.0 and 1.1")
self._version = version
@xmlelement
def tableset(self):
"""
The tableset. Must be a `TableSet` object.
"""
return self._tableset
@tableset.setter
def tableset(self, tableset):
self._tableset = tableset
@tableset.adder
def tableset(self, iterator, tag, data, config, pos):
tableset = vs.TableSet(config, pos, 'tableset', **data)
tableset.parse(iterator, config)
self._tableset = tableset
@xmlelement(cls=vs.VODataServiceTable)
def table(self):
"""
The `VODataServiceTable` root element if present.
"""
return self._table
@table.setter
def table(self, table):
self._table = table
@property
def ntables(self):
"""
The number of tables in the file.
"""
return self._ntables
[docs] def parse(self, iterator, config):
super().parse(iterator, config)
if self.tableset is None and self.table is None:
vo_raise(E07, config=config, pos=self._pos)
self._version = config['version']
if config['version'] not in ('1.0', '1.1'):
vo_warn(W15, config=config, pos=self._pos)
if self.table:
if version_compare(config['version'], '1.1') < 0:
vo_warn(W16, config=config, pos=self._pos)
self._ntables = 1
else:
self._ntables = sum(
len(schema.tables) for schema in self.tableset.schemas)
return self
[docs] def iter_tables(self):
"""
Iterates over all tables in the VOSITables file in a "flat" way,
ignoring the schemas.
"""
if self.table:
yield self.table
else:
for schema in self.tableset.schemas:
yield from schema.tables
[docs] def get_first_table(self):
"""
When you parse table metadata for a single table here is only one table
in the file, and that's all you need.
This method returns that first table.
"""
for table in self.iter_tables():
return table
raise IndexError("No table found in VOSITables file.")
[docs] def get_table_by_name(self, name):
"""
Looks up a table element by the given name.
"""
for table in self.iter_tables():
if table.name == name:
return table
raise KeyError("No table with name {} found".format(name))
[docs]class CapabilitiesFile(Element, HomogeneousList):
"""
capabilities element: represents an entire file.
The keyword arguments correspond to setting members of the same
name, documented below.
"""
def __init__(self, *, config=None, pos=None, _name='capabilities', **kwargs):
Element.__init__(self, config=config, pos=pos, **kwargs)
HomogeneousList.__init__(self, vr.Capability)
@xmlelement(name='capability')
def capabilities(self):
"""List of `~pyvo.io.vosi.voresource.Capability` objects"""
return self
@capabilities.adder
def capabilities(self, iterator, tag, data, config, pos):
capability = vr.Capability(config, pos, 'capability', **data)
capability.parse(iterator, config)
self.append(capability)
[docs] def parse(self, iterator, config):
for start, tag, data, pos in iterator:
if start:
if tag == "xml":
pass
elif tag == "capabilities":
break
else:
vo_raise(E10, config=config, pos=pos)
super().parse(iterator, config)
return self
[docs]class AvailabilityFile(av.Availability):
"""
availability element: represents an entire file.
The keyword arguments correspond to setting members of the same
name, documented below.
"""
[docs] def parse(self, iterator, config):
for start, tag, data, pos in iterator:
if start:
if tag == 'xml':
pass
elif tag == 'availability':
break
super().parse(iterator, config)
return self