"""
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)
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):
self.functions = Path('h5py', 'api_functions.txt').open('r')
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:
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()
if len(text) == 0 or text[0] == '#':
continue
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':
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':
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():
config = BuildConfig.from_env()
lp = LineProcessor(config)
lp.run()
if __name__ == '__main__':
run()