Skip to main content
`Ctrl`+`K`

# Source code for astropy.io.registry.interface

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

    import inspect
    import os
    import pydoc
    import re
    import sys

    from .base import IORegistryError

    __all__ = ["UnifiedReadWrite", "UnifiedReadWriteMethod"]

    # -----------------------------------------------------------------------------

    class UnifiedReadWrite:
        """Base class for the worker object used in unified read() or write() methods.

        This lightweight object is created for each `read()` or `write()` call
        via ``read`` / ``write`` descriptors on the data object class.  The key
        driver is to allow complete format-specific documentation of available
        method options via a ``help()`` method, e.g. ``Table.read.help('fits')``.

        Subclasses must define a ``__call__`` method which is what actually gets
        called when the data object ``read()`` or ``write()`` method is called.

        For the canonical example see the `~astropy.table.Table` class
        implementation (in particular the ``connect.py`` module there).

        Parameters
        ----------
        instance : object
            Descriptor calling instance or None if no instance
        cls : type
            Descriptor calling class (either owner class or instance class)
        method_name : str
            Method name, e.g. 'read' or 'write'
        registry : ``_UnifiedIORegistryBase`` or None, optional
            The IO registry.
        """

        def __init__(self, instance, cls, method_name, registry=None):
            if registry is None:
                from astropy.io.registry.compat import default_registry as registry

            self._registry = registry
            self._instance = instance
            self._cls = cls
            self._method_name = method_name  # 'read' or 'write'

        @property
        def registry(self):
            """Unified I/O registry instance."""
            return self._registry

        def help(self, format=None, out=None):
            """Output help documentation for the specified unified I/O ``format``.

            By default the help output is printed to the console via ``pydoc.pager``.
            Instead one can supplied a file handle object as ``out`` and the output
            will be written to that handle.

            Raises
            ------
            RuntimeError
                If called within a python runtime running with optimization level >= 2
                (-OO CLI flag).

            Parameters
            ----------
            format : str
                Unified I/O format name, e.g. 'fits' or 'ascii.ecsv'
            out : None or file-like
                Output destination (default is stdout via a pager)
            """
            if sys.flags.optimize >= 2:
                raise RuntimeError(
                    "The help method is not available under Python's optimized mode."
                )

            cls = self._cls
            method_name = self._method_name

            # Get reader or writer function associated with the registry
            get_func = (
                self._registry.get_reader
                if method_name == "read"
                else self._registry.get_writer
            )
            try:
                if format:
                    read_write_func = get_func(format, cls)
            except IORegistryError as err:
                reader_doc = "ERROR: " + str(err)
            else:
                if format:
                    # Format-specific
                    header = (
                        f"{cls.__name__}.{method_name}(format='{format}') documentation\n"
                    )
                    doc = read_write_func.__doc__
                else:
                    # General docs
                    header = f"{cls.__name__}.{method_name} general documentation\n"
                    doc = getattr(cls, method_name).__doc__

                reader_doc = re.sub(".", "=", header)
                reader_doc += header
                reader_doc += re.sub(".", "=", header)
                reader_doc += os.linesep
                if doc is not None:
                    reader_doc += inspect.cleandoc(doc)

            if out is None:
                pydoc.pager(reader_doc)
            else:
                out.write(reader_doc)

        def list_formats(self, out=None):
            """Print a list of available formats to console (or ``out`` filehandle).

            out : None or file handle object
                Output destination (default is stdout via a pager)
            """
            tbl = self._registry.get_formats(self._cls, self._method_name.capitalize())
            del tbl["Data class"]

            if out is None:
                tbl.pprint(max_lines=-1, max_width=-1)
            else:
                out.write("\n".join(tbl.pformat(max_lines=-1, max_width=-1)))

            return out

    # -----------------------------------------------------------------------------

    class UnifiedReadWriteMethod(property):
        """Descriptor class for creating read() and write() methods in unified I/O.

        The canonical example is in the ``Table`` class, where the ``connect.py``
        module creates subclasses of the ``UnifiedReadWrite`` class.  These have
        custom ``__call__`` methods that do the setup work related to calling the
        registry read() or write() functions.  With this, the ``Table`` class
        defines read and write methods as follows::

          read = UnifiedReadWriteMethod(TableRead)
          write = UnifiedReadWriteMethod(TableWrite)

        Parameters
        ----------
        func : `~astropy.io.registry.UnifiedReadWrite` subclass
            Class that defines read or write functionality

        """

        def __init__(self, readwritecls: UnifiedReadWrite, /) -> None:  # type: ignore[arg-type]
            super().__init__(fget=readwritecls)

        # We subclass property to ensure that __set__ is defined and that,
        # therefore, we are a data descriptor, which cannot be overridden.
        # This also means we automatically inherit the __doc__ of fget (which will
        # be a UnifiedReadWrite subclass), and that this docstring gets recognized
        # and properly typeset by sphinx (which was previously an issue; see
        # gh-11554).
        # We override __get__ to pass both instance and class to UnifiedReadWrite.
        def __get__(self, instance, owner_cls):
            return self.fget(instance, owner_cls)