api_gen.py

"""
    Generate the lowest-level Cython bindings to HDF5.

    In order to translate HDF5 errors to exceptions, the raw HDF5 API is
    wrapped with Cython "error wrappers".  These are cdef functions with
    the same names and signatures as their HDF5 equivalents, but implemented
    in the h5py.defs extension module.

    The h5py.defs files (defs.pyx and defs.pxd), along with the "real" HDF5
    function definitions (_hdf5.pxd), are auto-generated by this script from
    api_functions.txt.  This file also contains annotations which indicate
    whether a function requires a certain minimum version of HDF5, an
    MPI-aware build of h5py, or special error handling.

    This script is called automatically by the h5py build system when the
    output files are missing, or api_functions.txt has been updated.

    See the Line class in this module for documentation of the format for
    api_functions.txt.

    h5py/_hdf5.pxd:     Cython "extern" definitions for HDF5 functions
    h5py/defs.pxd:      Cython definitions for error wrappers
    h5py/defs.pyx:      Cython implementations of error wrappers
"""

import re
import os
from hashlib import md5
from pathlib import Path
from setup_configure import BuildConfig


def replace_or_remove(new: Path) -> None:
    assert new.is_file()
    assert new.suffix == '.new'
    old = new.with_suffix('')
    def get_hash(p: Path) -> str:
        return md5(p.read_bytes()).hexdigest()

    if not old.exists() or get_hash(new) != get_hash(old):
        os.replace(new, old)
    else:
        os.remove(new)


class Line:

    """
        Represents one line from the api_functions.txt file.

        Exists to provide the following attributes:

        nogil:      String indicating if we should release the GIL to call this
                    function. Any Python callbacks it could trigger must
                    acquire the GIL (e.g. using 'with gil' in Cython).
        mpi:        Bool indicating if MPI required
        ros3:       Bool indicating if ROS3 required
        direct_vfd: Bool indicating if DIRECT_VFD required
        version:    None or a minimum-version tuple
        code:       String with function return type
        fname:      String with function name
        sig:        String with raw function signature
        args:       String with sequence of arguments to call function

        Example:    MPI 1.12.2 int foo(char* a, size_t b)

        .nogil:     ""
        .mpi:       True
        .ros3:      False
        .direct_vfd: False
        .version:   (1, 12, 2)
        .code:      "int"
        .fname:     "foo"
        .sig:       "char* a, size_t b"
        .args:      "a, b"
    """

    PATTERN = re.compile("""(?P<mpi>(MPI)[ ]+)?
                            (?P<ros3>(ROS3)[ ]+)?
                            (?P<direct_vfd>(DIRECT_VFD)[ ]+)?
                            (?P<min_version>([0-9]+\.[0-9]+\.[0-9]+))?
                            (-(?P<max_version>([0-9]+\.[0-9]+\.[0-9]+)))?
                            ([ ]+)?
                            (?P<code>(unsigned[ ]+)?[a-zA-Z_]+[a-zA-Z0-9_]*\**)[ ]+
                            (?P<fname>[a-zA-Z_]+[a-zA-Z0-9_]*)[ ]*
                            \((?P<sig>[a-zA-Z0-9_,* ]*)\)
                            ([ ]+)?
                            (?P<nogil>(nogil))?
                            """, re.VERBOSE)

    SIG_PATTERN = re.compile("""
                             (?:unsigned[ ]+)?
                             (?:[a-zA-Z_]+[a-zA-Z0-9_]*\**)
                             [ ]+[ *]*
                             (?P<param>[a-zA-Z_]+[a-zA-Z0-9_]*)
                             """, re.VERBOSE)

    def __init__(self, text):
        """ Break the line into pieces and populate object attributes.

        text: A valid function line, with leading/trailing whitespace stripped.
        """

        m = self.PATTERN.match(text)
        if m is None:
            raise ValueError("Invalid line encountered: {0}".format(text))

        parts = m.groupdict()
        self.nogil = "nogil" if parts['nogil'] else ""
        self.mpi = parts['mpi'] is not None
        self.ros3 = parts['ros3'] is not None
        self.direct_vfd = parts['direct_vfd'] is not None
        self.min_version = parts['min_version']
        if self.min_version is not None:
            self.min_version = tuple(int(x) for x in self.min_version.split('.'))
        self.max_version = parts['max_version']
        if self.max_version is not None:
            self.max_version = tuple(int(x) for x in self.max_version.split('.'))
        self.code = parts['code']
        self.fname = parts['fname']
        self.sig = parts['sig']

        sig_const_stripped = self.sig.replace('const', '')
        self.args = self.SIG_PATTERN.findall(sig_const_stripped)
        if self.args is None:
            raise ValueError("Invalid function signature: {0}".format(self.sig))
        self.args = ", ".join(self.args)

        # Figure out what test and return value to use with error reporting
        if '*' in self.code or self.code in ('H5T_conv_t',):
            self.err_condition = "==NULL"
            self.err_value = f"<{self.code}>NULL"
        elif self.code in ('int', 'herr_t', 'htri_t', 'hid_t', 'hssize_t', 'ssize_t') \
                or re.match(r'H5[A-Z]+_[a-zA-Z_]+_t', self.code):
            self.err_condition = "<0"
            self.err_value = f"<{self.code}>-1"
        elif self.code in ('unsigned int', 'haddr_t', 'hsize_t', 'size_t'):
            self.err_condition = "==0"
            self.err_value = f"<{self.code}>0"
        else:
            raise ValueError("Return code <<%s>> unknown" % self.code)


raw_preamble = """\
# cython: language_level=3
#
# Warning: this file is auto-generated from api_gen.py. DO NOT EDIT!
#

include "config.pxi"
from .api_types_hdf5 cimport *
from .api_types_ext cimport *

"""

def_preamble = """\
# cython: language_level=3
#
# Warning: this file is auto-generated from api_gen.py. DO NOT EDIT!
#

include "config.pxi"

from .api_types_hdf5 cimport *
from .api_types_ext cimport *

"""

imp_preamble = """\
# cython: language_level=3
#
# Warning: this file is auto-generated from api_gen.py. DO NOT EDIT!
#

include "config.pxi"
from .api_types_ext cimport *
from .api_types_hdf5 cimport *

from . cimport _hdf5

from ._errors cimport set_exception, set_default_error_handler

"""


class LineProcessor:

    def __init__(self, config) -> None:
        self.config = config

    def run(self):

        # Function definitions file
        self.functions = Path('h5py', 'api_functions.txt').open('r')

        # Create output files
        path_raw_defs = Path('h5py', '_hdf5.pxd.new')
        path_cython_defs = Path('h5py', 'defs.pxd.new')
        path_cython_imp = Path('h5py', 'defs.pyx.new')
        self.raw_defs = path_raw_defs.open('w')
        self.cython_defs = path_cython_defs.open('w')
        self.cython_imp = path_cython_imp.open('w')

        self.raw_defs.write(raw_preamble)
        self.cython_defs.write(def_preamble)
        self.cython_imp.write(imp_preamble)

        for text in self.functions:

            # Directive specifying a header file
            if not text.startswith(' ') and not text.startswith('#') and \
            len(text.strip()) > 0:
                inc = text.split(':')[0]
                self.raw_defs.write('cdef extern from "%s.h":\n' % inc)
                continue

            text = text.strip()

            # Whitespace or comment line
            if len(text) == 0 or text[0] == '#':
                continue

            # Valid function line
            self.line = Line(text)
            self.write_raw_sig()
            self.write_cython_sig()
            self.write_cython_imp()

        self.functions.close()
        self.cython_imp.close()
        self.cython_defs.close()
        self.raw_defs.close()

        replace_or_remove(path_cython_imp)
        replace_or_remove(path_cython_defs)
        replace_or_remove(path_raw_defs)

    def check_settings(self) -> bool:
        """ Return True if and only if the code block should be compiled within given settings """
        if (
            (self.line.mpi and not self.config.mpi)
            or (self.line.ros3 and not self.config.ros3)
            or (self.line.direct_vfd and not self.config.direct_vfd)
            or (self.line.min_version is not None and self.config.hdf5_version < self.line.min_version)
            or (self.line.max_version is not None and self.config.hdf5_version > self.line.max_version)
        ):
            return False
        else:
            return True

    def write_raw_sig(self):
        """ Write out "cdef extern"-style definition for an HDF5 function """
        if not self.check_settings():
            return
        raw_sig = "{0.code} {0.fname}({0.sig}) {0.nogil}\n".format(self.line)
        raw_sig = "\n".join(("    " + x if x.strip() else x) for x in raw_sig.split("\n"))
        self.raw_defs.write(raw_sig)

    def write_cython_sig(self):
        """ Write out Cython signature for wrapper function """
        if not self.check_settings():
            return
        if self.line.fname == 'H5Dget_storage_size':
            # Special case: https://github.com/h5py/h5py/issues/1475
            cython_sig = "cdef {0.code} {0.fname}({0.sig}) except? {0.err_value}\n".format(self.line)
        else:
            cython_sig = "cdef {0.code} {0.fname}({0.sig}) except {0.err_value}\n".format(self.line)
        self.cython_defs.write(cython_sig)

    def write_cython_imp(self):
        """ Write out Cython wrapper implementation """
        if not self.check_settings():
            return
        if self.line.nogil:
            imp = """\
cdef {0.code} {0.fname}({0.sig}) except {0.err_value}:
    cdef {0.code} r
    with nogil:
        set_default_error_handler()
        r = _hdf5.{0.fname}({0.args})
    if r{0.err_condition}:
        if set_exception():
            return {0.err_value}
        else:
            raise RuntimeError("Unspecified error in {0.fname} (return value {0.err_condition})")
    return r

"""
        else:
            if self.line.fname == 'H5Dget_storage_size':
                # Special case: https://github.com/h5py/h5py/issues/1475
                imp = """\
cdef {0.code} {0.fname}({0.sig}) except? {0.err_value}:
    cdef {0.code} r
    set_default_error_handler()
    r = _hdf5.{0.fname}({0.args})
    if r{0.err_condition}:
        if set_exception():
            return {0.err_value}
    return r

"""
            else:
                imp = """\
cdef {0.code} {0.fname}({0.sig}) except {0.err_value}:
    cdef {0.code} r
    set_default_error_handler()
    r = _hdf5.{0.fname}({0.args})
    if r{0.err_condition}:
        if set_exception():
            return {0.err_value}
        else:
            raise RuntimeError("Unspecified error in {0.fname} (return value {0.err_condition})")
    return r

"""
        imp = imp.format(self.line)
        self.cython_imp.write(imp)


def run():
    # Get configuration from environment variables
    config = BuildConfig.from_env()
    lp = LineProcessor(config)
    lp.run()


if __name__ == '__main__':
    run()