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

# TODO: Test FITS parsing

# STDLIB

import base64

import binascii

import codecs

import gzip

import io

import json

import os

import re

import urllib.request

# THIRD-PARTY

import numpy as np

from numpy import ma

# LOCAL

from astropy import __version__ as astropy_version

from astropy.io import fits

from astropy.utils import deprecated

from astropy.utils.collections import HomogeneousList

from astropy.utils.data import get_pkg_data_filename

from astropy.utils.xml import iterparser

from astropy.utils.xml.writer import XMLWriter

from . import converters, util, xmlutil

from . import ucd as ucd_mod

from .exceptions import (

E06,

E08,

E09,

E10,

E11,

E12,

E13,

E15,

E16,

E17,

E18,

E19,

E20,

E21,

E22,

E23,

E25,

E26,

W06,

W07,

W08,

W09,

W10,

W11,

W12,

W13,

W15,

W17,

W18,

W19,

W20,

W21,

W22,

W26,

W27,

W28,

W29,

W32,

W33,

W35,

W36,

W37,

W38,

W40,

W41,

W42,

W43,

W44,

W45,

W50,

W52,

W53,

W54,

W56,

W57,

vo_raise,

vo_reraise,

vo_warn,

warn_or_raise,

warn_unknown_attrs,

)

try:

from . import tablewriter

_has_c_tabledata_writer = True

except ImportError:

_has_c_tabledata_writer = False

__all__ = [

"CooSys",

"Element",

"Field",

"FieldRef",

"Group",

"Info",

"Link",

"MivotBlock",

"Param",

"ParamRef",

"Resource",

"TableElement",

"TimeSys",

"VOTableFile",

"Values",

]

# The default number of rows to read in each chunk before converting

# to an array.

DEFAULT_CHUNK_SIZE = 256

RESIZE_AMOUNT = 1.5

######################################################################

# FACTORY FUNCTIONS

class _MaskedArrayBuilder:

"""

Zeroed data and mask arrays for a table being read and grown in place.

Masked arrays can not be resized in place, but these plain arrays own

their memory, so ``ndarray.resize`` can grow and trim them without

holding a second copy of the table. Call `to_masked` once they are

filled.

`resize` can move the arrays to new memory, so they never leave this

class until `to_masked`, and nothing here keeps a view of them.

"""

def __init__(self, dtype, nrows):

self._values = np.zeros(nrows, dtype=dtype)

self._mask = np.zeros(nrows, dtype=ma.make_mask_descr(dtype))

def __len__(self):

return len(self._values)

def set_row(self, index, values, mask):

"""

Set one row.

Parameters

----------

index : int

The row to set.

values : tuple

One value per field of the dtype, in field order, as returned

by the column converters

mask : tuple

One mask flag per field, in the same order

"""

self._values[index] = values

self._mask[index] = mask

def set_rows(self, start, values, masks):

"""

Set consecutive rows, starting at row *start*.

Parameters

----------

start : int

The first row to set.

values : list of tuple

One tuple per row, each as the *values* of `set_row`.

masks : list of tuple

One tuple per row, each as the *mask* of `set_row`, and as

many as there are in *values*.

"""

stop = start + len(values)

self._values[start:stop] = values

self._mask[start:stop] = masks

def resize(self, nrows):

"""Grow or trim both arrays to *nrows* rows, keeping their contents."""

# refcheck=False: from Python 3.14, numpy's reference count check

# rejects a count of 2, which the attribute and this call already

# make. It is safe because no views of these arrays exist.

# Resizing while a view exists would leave the view pointing at

# freed memory.

self._values.resize(nrows, refcheck=False)

self._mask.resize(nrows, refcheck=False)

def to_masked(self):

"""

Wrap the arrays in a masked array without copying them.

The builder can not be used afterwards, since resizing would free

the memory the masked array uses.

"""

values, mask = self._values, self._mask

del self._values, self._mask

# keep_mask=False: for a structured dtype, MaskedArray starts from

# its own all-False mask and, with keep_mask=True, ORs this one into

# it, which allocates a second full-size mask. Unneeded here.

return ma.MaskedArray(values, mask=mask, copy=False, keep_mask=False)

def _lookup_by_attr_factory(attr, unique, iterator, element_name, doc):

"""

Creates a function useful for looking up an element by a given

attribute.

Parameters

----------

attr : str

The attribute name

unique : bool

Should be `True` if the attribute is unique and therefore this

should return only one value. Otherwise, returns a list of

values.

iterator : generator

A generator that iterates over some arbitrary set of elements

element_name : str

The XML element name of the elements being iterated over (used

for error messages only).

doc : str

A docstring to apply to the generated function.

Returns

-------

factory : function

A function that looks up an element by the given attribute.

"""

def lookup_by_attr(self, ref, before=None):

"""

Given a string *ref*, finds the first element in the iterator

where the given attribute == *ref*. If *before* is provided,

will stop searching at the object *before*. This is

important, since "forward references" are not allowed in the

VOTABLE format.

"""

for element in getattr(self, iterator)():

if element is before:

if getattr(element, attr, None) == ref:

vo_raise(

KeyError,

f"{element_name} references itself",

element._config,

element._pos,

)

break

if getattr(element, attr, None) == ref:

yield element

def lookup_by_attr_unique(self, ref, before=None):

for element in lookup_by_attr(self, ref, before=before):

return element

raise KeyError(

f"No {element_name} with {attr} '{ref}' found before the referencing "

f"{element_name}"

)

if unique:

lookup_by_attr_unique.__doc__ = doc

return lookup_by_attr_unique

else:

lookup_by_attr.__doc__ = doc

return lookup_by_attr

def _lookup_by_id_or_name_factory(iterator, element_name, doc):

"""

Like `_lookup_by_attr_factory`, but looks in both the "ID" and

"name" attributes.

"""

def lookup_by_id_or_name(self, ref, before=None):

"""

Given a key *ref*, finds the first element in the iterator

with the attribute ID == *ref* or name == *ref*. If *before*

is provided, will stop searching at the object *before*. This

is important, since "forward references" are not allowed in

the VOTABLE format.

"""

for element in getattr(self, iterator)():

if element is before:

if ref in (element.ID, element.name):

vo_raise(

KeyError,

f"{element_name} references itself",

element._config,

element._pos,

)

break

if ref in (element.ID, element.name):

return element

raise KeyError(

f"No {element_name} with ID or name '{ref}' found before the referencing "

f"{element_name}"

)

lookup_by_id_or_name.__doc__ = doc

return lookup_by_id_or_name

def _get_default_unit_format(config):

"""

Get the default unit format as specified in the VOTable spec.

"""

# The unit format changed between VOTable versions 1.3 and 1.4,

# see issue #10791.

if config["version_1_4_or_later"]:

return "vounit"

else:

return "cds"

def _get_unit_format(config):

"""

Get the unit format based on the configuration.

"""

if config.get("unit_format") is None:

format = _get_default_unit_format(config)

else:

format = config["unit_format"]

return format

def _attach_default_config(votable, config) -> dict:

"""Create default config if none given.

Parameters

----------

votable

Object that might potentially have a config attached to it.

config : `dict` or `None`

Config to be used or defined.

Returns

-------

config : `dict`

If config was `None` this will be a new `dict`. If the ``votable``

supported version checks then version information will be stored in

the returned `dict`. If it was defined this will be the given

parameter unchanged.

"""

if config is None:

if hasattr(votable, "_get_version_checks"):

config = votable._get_version_checks()

else:

config = {}

return config

######################################################################

# ATTRIBUTE CHECKERS

def check_astroyear(year, field, config=None, pos=None):

"""

Raises a `~astropy.io.votable.exceptions.VOTableSpecError` if

*year* is not a valid astronomical year as defined by the VOTABLE

standard.

Parameters

----------

year : str

An astronomical year string

field : str

The name of the field this year was found in (used for error

message)

config, pos : optional

Information about the source of the value

"""

if year is not None and re.match(r"^[JB]?[0-9]+([.][0-9]*)?$", year) is None:

warn_or_raise(W07, W07, (field, year), config, pos)

return False

return True

def check_string(string, attr_name, config=None, pos=None):

"""

Raises a `~astropy.io.votable.exceptions.VOTableSpecError` if

*string* is not a string or Unicode string.

Parameters

----------

string : str

An astronomical year string

attr_name : str

The name of the field this year was found in (used for error

message)

config, pos : optional

Information about the source of the value

"""

if string is not None and not isinstance(string, str):

warn_or_raise(W08, W08, attr_name, config, pos)

return False

return True

def resolve_id(ID, id, config=None, pos=None):

if ID is None and id is not None:

warn_or_raise(W09, W09, (), config, pos)

return id

return ID

def check_ucd(ucd, config=None, pos=None):

"""

Warns or raises a

`~astropy.io.votable.exceptions.VOTableSpecError` if *ucd* is not

a valid `unified content descriptor`_ string as defined by the

VOTABLE standard.

Parameters

----------

ucd : str

A UCD string.

config, pos : optional

Information about the source of the value

"""

if config is None:

config = {}

if config.get("version_1_1_or_later"):

try:

ucd_mod.parse_ucd(

ucd,

check_controlled_vocabulary=config.get("version_1_2_or_later", False),

has_colon=config.get("version_1_2_or_later", False),

)

except ValueError as e:

# This weird construction is for Python 3 compatibility

if config.get("verify", "ignore") == "exception":

vo_raise(W06, (ucd, str(e)), config, pos)

elif config.get("verify", "ignore") == "warn":

vo_warn(W06, (ucd, str(e)), config, pos)

return False

else:

return False

return True

######################################################################

# PROPERTY MIXINS

class _IDProperty:

@property

def ID(self):

"""

The XML ID_ of the element. May be `None` or a string

conforming to XML ID_ syntax.

"""

return self._ID

@ID.setter

def ID(self, ID):

xmlutil.check_id(ID, "ID", self._config, self._pos)

self._ID = ID

@ID.deleter

def ID(self):

self._ID = None

class _NameProperty:

@property

def name(self):

"""An optional name for the element."""

return self._name

@name.setter

def name(self, name):

xmlutil.check_token(name, "name", self._config, self._pos)

self._name = name

@name.deleter

def name(self):

self._name = None

class _XtypeProperty:

@property

def xtype(self):

"""Extended data type information."""

return self._xtype

@xtype.setter

def xtype(self, xtype):

if xtype is not None and not self._config.get("version_1_2_or_later"):

warn_or_raise(

W28, W28, ("xtype", self._element_name, "1.2"), self._config, self._pos

)

check_string(xtype, "xtype", self._config, self._pos)

self._xtype = xtype

@xtype.deleter

def xtype(self):

self._xtype = None

class _UtypeProperty:

_utype_in_v1_2 = False

@property

def utype(self):

"""The usage-specific or `unique type`_ of the element."""

return self._utype

@utype.setter

def utype(self, utype):

if (

self._utype_in_v1_2

and utype is not None

and not self._config.get("version_1_2_or_later")

):

warn_or_raise(

W28, W28, ("utype", self._element_name, "1.2"), self._config, self._pos

)

check_string(utype, "utype", self._config, self._pos)

self._utype = utype

@utype.deleter

def utype(self):

self._utype = None

class _UcdProperty:

_ucd_in_v1_2 = False

@property

def ucd(self):

"""The `unified content descriptor`_ for the element."""

return self._ucd

@ucd.setter

def ucd(self, ucd):

if ucd is not None and ucd.strip() == "":

ucd = None

if ucd is not None:

if self._ucd_in_v1_2 and not self._config.get("version_1_2_or_later"):

warn_or_raise(

W28,

W28,

("ucd", self._element_name, "1.2"),

self._config,

self._pos,

)

check_ucd(ucd, self._config, self._pos)

self._ucd = ucd

@ucd.deleter

def ucd(self):

self._ucd = None

class _DescriptionProperty:

@property

def description(self):

"""

An optional string describing the element. Corresponds to the

DESCRIPTION_ element.

"""

return self._description

@description.setter

def description(self, description):

self._description = description

@description.deleter

def description(self):

self._description = None

######################################################################

# ELEMENT CLASSES

class Element:

"""

A base class for all classes that represent XML elements in the

VOTABLE file.

"""

_element_name = ""

_attr_list = []

def _add_unknown_tag(self, iterator, tag, data, config, pos):

warn_or_raise(W10, W10, tag, config, pos)

def _ignore_add(self, iterator, tag, data, config, pos):

warn_unknown_attrs(tag, data.keys(), config, pos)

def _add_definitions(self, iterator, tag, data, config, pos):

if config.get("version_1_1_or_later"):

warn_or_raise(W22, W22, (), config, pos)

warn_unknown_attrs(tag, data.keys(), config, pos)

def parse(self, iterator, config):

"""

For internal use. Parse the XML content of the children of the

element.

Parameters

----------

iterator : xml iterable

An iterator over XML elements as returned by

`~astropy.utils.xml.iterparser.get_xml_iterator`.

config : dict

The configuration dictionary that affects how certain

elements are read.

Returns

-------

self : `~astropy.io.votable.tree.Element`

Returns self as a convenience.

"""

raise NotImplementedError()

def to_xml(self, w, **kwargs):

"""

For internal use. Output the element to XML.

Parameters

----------

w : astropy.utils.xml.writer.XMLWriter object

An XML writer to write to.

**kwargs : dict

Any configuration parameters to control the output.

"""

raise NotImplementedError()

class SimpleElement(Element):

"""

A base class for simple elements, such as FIELD, PARAM and INFO

that don't require any special parsing or outputting machinery.

"""

def __init__(self):

Element.__init__(self)

def __repr__(self):

buff = io.StringIO()

SimpleElement.to_xml(self, XMLWriter(buff))

return buff.getvalue().strip()

def parse(self, iterator, config):

for start, tag, data, pos in iterator:

if start and tag != self._element_name:

self._add_unknown_tag(iterator, tag, data, config, pos)

elif tag == self._element_name:

break

return self

def to_xml(self, w, **kwargs):

w.element(self._element_name, attrib=w.object_attrs(self, self._attr_list))

class SimpleElementWithContent(SimpleElement):

"""

A base class for simple elements, such as FIELD, PARAM and INFO

that don't require any special parsing or outputting machinery.

"""

def __init__(self):

SimpleElement.__init__(self)

self._content = None

def parse(self, iterator, config):

for start, tag, data, pos in iterator:

if start and tag != self._element_name:

self._add_unknown_tag(iterator, tag, data, config, pos)

elif tag == self._element_name:

if data:

self.content = data

break

return self

def to_xml(self, w, **kwargs):

w.element(

self._element_name,

self._content,

attrib=w.object_attrs(self, self._attr_list),

)

@property

def content(self):

"""The content of the element."""

return self._content

@content.setter

def content(self, content):

check_string(content, "content", self._config, self._pos)

self._content = content

@content.deleter

def content(self):

self._content = None

class Link(SimpleElement, _IDProperty):

"""

LINK_ elements: used to reference external documents and servers through a URI.

The keyword arguments correspond to setting members of the same

name, documented below.

"""

_attr_list = [

"ID",

"content_role",

"content_type",

"title",

"value",

"href",

"action",

]

_element_name = "LINK"

def __init__( # ruff: ignore[PLR0917]

self,

ID=None,

title=None,

value=None,

href=None,

action=None,

id=None,

config=None,

pos=None,

**kwargs,

):

if config is None:

config = {}

self._config = config

self._pos = pos

SimpleElement.__init__(self)

content_role = kwargs.get("content-role") or kwargs.get("content_role")

content_type = kwargs.get("content-type") or kwargs.get("content_type")

if "gref" in kwargs:

warn_or_raise(W11, W11, (), config, pos)

self.ID = resolve_id(ID, id, config, pos)

self.content_role = content_role

self.content_type = content_type

self.title = title

self.value = value

self.href = href

self.action = action

warn_unknown_attrs(

"LINK",

kwargs.keys(),

config,

pos,

["content-role", "content_role", "content-type", "content_type", "gref"],

)

@property

def content_role(self):

"""Defines the MIME role of the referenced object.

Must be one of:

None, 'query', 'hints', 'doc', 'location' or 'type'

"""

return self._content_role

@content_role.setter

def content_role(self, content_role):

if (

content_role == "type" and not self._config.get("version_1_3_or_later")

) or content_role not in (None, "query", "hints", "doc", "location"):

vo_warn(W45, (content_role,), self._config, self._pos)

self._content_role = content_role

@content_role.deleter

def content_role(self):

self._content_role = None

@property

def content_type(self):

"""Defines the MIME content type of the referenced object."""

return self._content_type

@content_type.setter

def content_type(self, content_type):

xmlutil.check_mime_content_type(content_type, self._config, self._pos)

self._content_type = content_type

@content_type.deleter

def content_type(self):

self._content_type = None

@property

def href(self):

"""

A URI to an arbitrary protocol. The vo package only supports

http and anonymous ftp.

"""

return self._href

@href.setter

def href(self, href):

xmlutil.check_anyuri(href, self._config, self._pos)

self._href = href

@href.deleter

def href(self):

self._href = None

def to_table_column(self, column):

meta = {}

for key in self._attr_list:

val = getattr(self, key, None)

if val is not None:

meta[key] = val

column.meta.setdefault("links", [])

column.meta["links"].append(meta)

@classmethod

def from_table_column(cls, d):

return cls(**d)

class Info(SimpleElementWithContent, _IDProperty, _XtypeProperty, _UtypeProperty):

"""

INFO_ elements: arbitrary key-value pairs for extensions to the standard.

The keyword arguments correspond to setting members of the same

name, documented below.

"""

_element_name = "INFO"

_attr_list_11 = ["ID", "name", "value"]

_attr_list_12 = _attr_list_11 + ["xtype", "ref", "unit", "ucd", "utype"]

_utype_in_v1_2 = True

def __init__( # ruff: ignore[PLR0917]

self,

ID=None,

name=None,

value=None,

id=None,

xtype=None,

ref=None,

unit=None,

ucd=None,

utype=None,

config=None,

pos=None,

**extra,

):

if config is None:

config = {}

self._config = config

self._pos = pos

SimpleElementWithContent.__init__(self)

self.ID = resolve_id(ID, id, config, pos) or xmlutil.fix_id(name, config, pos)

self.name = name

self.value = value

self.xtype = xtype

self.ref = ref

self.unit = unit

self.ucd = ucd

self.utype = utype

if config.get("version_1_2_or_later"):

self._attr_list = self._attr_list_12

else:

self._attr_list = self._attr_list_11

if xtype is not None:

warn_unknown_attrs("INFO", ["xtype"], config, pos)

if ref is not None:

warn_unknown_attrs("INFO", ["ref"], config, pos)

if unit is not None:

warn_unknown_attrs("INFO", ["unit"], config, pos)

if ucd is not None:

warn_unknown_attrs("INFO", ["ucd"], config, pos)

if utype is not None:

warn_unknown_attrs("INFO", ["utype"], config, pos)

warn_unknown_attrs("INFO", extra.keys(), config, pos)

@property

def name(self):

"""[*required*] The key of the key-value pair."""

return self._name

@name.setter

def name(self, name):

if name is None:

warn_or_raise(W35, W35, "name", self._config, self._pos)

xmlutil.check_token(name, "name", self._config, self._pos)

self._name = name

@property

def value(self):

"""

[*required*] The value of the key-value pair. (Always stored

as a string or unicode string).

"""

return self._value

@value.setter

def value(self, value):

if value is None:

warn_or_raise(W35, W35, "value", self._config, self._pos)

check_string(value, "value", self._config, self._pos)

self._value = value

@property

def content(self):

"""The content inside the INFO element."""

return self._content

@content.setter

def content(self, content):

check_string(content, "content", self._config, self._pos)

self._content = content

@content.deleter

def content(self):

self._content = None

@property

def ref(self):

"""

Refer to another INFO_ element by ID_, defined previously in

the document.

"""

return self._ref

@ref.setter

def ref(self, ref):

if ref is not None and not self._config.get("version_1_2_or_later"):

warn_or_raise(W28, W28, ("ref", "INFO", "1.2"), self._config, self._pos)

xmlutil.check_id(ref, "ref", self._config, self._pos)

# TODO: actually apply the reference

# if ref is not None:

# try:

# other = self._votable.get_values_by_id(ref, before=self)

# except KeyError:

# vo_raise(

# "VALUES ref='%s', which has not already been defined." %

# self.ref, self._config, self._pos, KeyError)

# self.null = other.null

# self.type = other.type

# self.min = other.min

# self.min_inclusive = other.min_inclusive

# self.max = other.max

# self.max_inclusive = other.max_inclusive

# self._options[:] = other.options

self._ref = ref

@ref.deleter

def ref(self):

self._ref = None

@property

def unit(self):

"""A string specifying the units_ for the INFO_."""

return self._unit

@unit.setter

def unit(self, unit):

if unit is None:

self._unit = None

return

from astropy import units as u

if not self._config.get("version_1_2_or_later"):

warn_or_raise(W28, W28, ("unit", "INFO", "1.2"), self._config, self._pos)

# First, parse the unit in the default way, so that we can

# still emit a warning if the unit is not to spec.

default_format = _get_default_unit_format(self._config)

unit_obj = u.Unit(unit, format=default_format, parse_strict="silent")

if isinstance(unit_obj, u.UnrecognizedUnit):

warn_or_raise(W50, W50, (unit,), self._config, self._pos)

format = _get_unit_format(self._config)

if format != default_format: