API Reference

Core Modules

parser

`c2py23.parser`

Parser for c2py23 interface definitions.

Handles

Python dict format (native, no dependency)
C2PY_BEGIN blocks embedded in C source files
C function signature parsing
Expression parsing (for 'when' conditions and 'map' substitutions)
Building the ModuleDef data model

The entry point is load_c2py(path) which auto-detects the input format. For programmatic use, from_c2py_dict(raw_dict, path) accepts a Python dict directly.

`ModuleDef`

Bases: namedtuple('ModuleDef', ['name', 'sources', 'headers', 'functions', 'constants', 'timing', 'free_threading'])

constants is a dict of {name: int_value} for module-level integer constants. timing is a bool enabling per-function performance profiling. free_threading is a bool; when true, the module declares Py_MOD_GIL_NOT_USED on free-threaded Python builds (prevents GIL re-enablement).

Source code in c2py23/parser.py

class ModuleDef(
    namedtuple(
        "ModuleDef",
        [
            "name",
            "sources",
            "headers",
            "functions",
            "constants",
            "timing",
            "free_threading",
        ],
    )
):
    """constants is a dict of {name: int_value} for module-level integer constants.
    timing is a bool enabling per-function performance profiling.
    free_threading is a bool; when true, the module declares Py_MOD_GIL_NOT_USED
    on free-threaded Python builds (prevents GIL re-enablement)."""

    pass

`FuncDef`

Bases: namedtuple('FuncDef', ['name', 'py_params', 'return_type', 'checks', 'overloads', 'default_raise', 'doc', 'gil_release'])

A wrapped Python function definition.

params is an optional dict mapping parameter names to human-readable descriptions, parsed from the interface definition. Keys are validated against py_sig parameter names.

Source code in c2py23/parser.py

class FuncDef(
    namedtuple(
        "FuncDef",
        [
            "name",
            "py_params",
            "return_type",
            "checks",
            "overloads",
            "default_raise",
            "doc",
            "gil_release",
        ],
    )
):
    """A wrapped Python function definition.

    params is an optional dict mapping parameter names to human-readable
    descriptions, parsed from the interface definition. Keys are validated against
    py_sig parameter names.
    """

    def __new__(
        cls,
        name,
        py_params,
        return_type,
        checks,
        overloads,
        default_raise,
        doc,
        gil_release,
        params=None,
        acquire=None,
    ):
        self = super(FuncDef, cls).__new__(
            cls,
            name,
            py_params,
            return_type,
            checks,
            overloads,
            default_raise,
            doc,
            gil_release,
        )
        self.params = params or {}
        self.acquire = acquire
        return self

`COverload`

Bases: namedtuple('_COverload', ['sig_str', 'params', 'return_type', 'map_exprs', 'when_expr', 'name', 'group_name', 'variants', 'c_name'])

A C function overload alternative or a dispatch group.

For flat overloads (backward compatible): sig_str, params, return_type, map_exprs, when_expr are populated. name is optional (required if when_expr is static for rebind support). variants is None. c_name is the extracted C function name (no re-parsing needed).

For grouped dispatch

variants is a non-empty list of CVariant. sig_str, params, return_type are None. map_exprs is the shared argument map for all variants in the group. when_expr is the per-call group condition (e.g. data.format == 'f'). group_name is an optional label (for rebind qualifiers, docstrings). c_name is None (variants carry their own c_name).

outputs maps C parameter names to ctypes types (e.g. {'minval': 'double'}). Output params are auto-allocated as 1-element arrays and returned in the tuple.

doc is an optional per-overload or per-group description string.

Source code in c2py23/parser.py

class COverload(
    namedtuple(
        "_COverload",
        [
            "sig_str",
            "params",
            "return_type",
            "map_exprs",
            "when_expr",
            "name",
            "group_name",
            "variants",
            "c_name",
        ],
    )
):
    """A C function overload alternative or a dispatch group.

    For flat overloads (backward compatible):
        sig_str, params, return_type, map_exprs, when_expr are populated.
        name is optional (required if when_expr is static for rebind support).
        variants is None.
        c_name is the extracted C function name (no re-parsing needed).

    For grouped dispatch:
        variants is a non-empty list of CVariant.
        sig_str, params, return_type are None.
        map_exprs is the shared argument map for all variants in the group.
        when_expr is the per-call group condition (e.g. data.format == 'f').
        group_name is an optional label (for rebind qualifiers, docstrings).
        c_name is None (variants carry their own c_name).

    outputs maps C parameter names to ctypes types (e.g. {'minval': 'double'}).
    Output params are auto-allocated as 1-element arrays and returned in the tuple.

    doc is an optional per-overload or per-group description string.
    """

    def __new__(
        cls,
        sig_str,
        params,
        return_type,
        map_exprs,
        when_expr,
        name=None,
        group_name=None,
        variants=None,
        outputs=None,
        doc=None,
        c_name=None,
    ):
        self = super(COverload, cls).__new__(
            cls,
            sig_str,
            params,
            return_type,
            map_exprs,
            when_expr,
            name,
            group_name,
            variants,
            c_name,
        )
        self.outputs = outputs or {}
        self.doc = doc
        return self

`CVariant`

Bases: namedtuple('CVariant', ['name', 'sig_str', 'params', 'return_type', 'when_expr', 'outputs', 'c_name'])

A variant within a dispatch group. Inherits map_exprs from the parent group.

name is required for rebind, docstring, and timing identification. when_expr is the static (CPU feature) dispatch condition, or None for default. outputs is an optional dict for scalar output parameters (same format as COverload). doc is an optional per-variant description string. c_name is the extracted C function name (no re-parsing needed). default is True if the variant should be auto-selected at init. Set default: false to make the variant reachable only via rebind().

Source code in c2py23/parser.py

class CVariant(
    namedtuple(
        "CVariant",
        ["name", "sig_str", "params", "return_type", "when_expr", "outputs", "c_name"],
    )
):
    """A variant within a dispatch group. Inherits map_exprs from the parent group.

    name is required for rebind, docstring, and timing identification.
    when_expr is the static (CPU feature) dispatch condition, or None for default.
    outputs is an optional dict for scalar output parameters (same format as COverload).
    doc is an optional per-variant description string.
    c_name is the extracted C function name (no re-parsing needed).
    default is True if the variant should be auto-selected at init.
      Set default: false to make the variant reachable only via _rebind_<name>().
    """

    def __new__(
        cls,
        name,
        sig_str,
        params,
        return_type,
        when_expr,
        outputs=None,
        doc=None,
        c_name=None,
        default=True,
    ):
        self = super(CVariant, cls).__new__(cls, name, sig_str, params, return_type, when_expr, outputs, c_name)
        self.doc = doc
        self.default = default
        return self

`PyParam`

Bases: namedtuple('PyParam', ['name', 'pytype', 'default'])

pytype is one of 'buffer', 'int', 'float'. default is None for required params, or a numeric value for optional int/float params.

Source code in c2py23/parser.py

class PyParam(namedtuple("PyParam", ["name", "pytype", "default"])):
    """pytype is one of 'buffer', 'int', 'float'. default is None for
    required params, or a numeric value for optional int/float params."""

    pass

`CParam`

Bases: namedtuple('CParam', ['name', 'ctype', 'base_type', 'is_const', 'is_pointer', 'array_dims'])

ctype is the full C type string, base_type is the element type. array_dims is a list of dimension values (strings or None for []) from C array notation like gv[][3] (-> [None, '3']). None means the parameter was declared with plain * pointer notation.

Source code in c2py23/parser.py

class CParam(namedtuple("CParam", ["name", "ctype", "base_type", "is_const", "is_pointer", "array_dims"])):
    """ctype is the full C type string, base_type is the element type.
    array_dims is a list of dimension values (strings or None for [])
    from C array notation like gv[][3] (-> [None, '3']).
    None means the parameter was declared with plain * pointer notation."""

    def __new__(cls, name, ctype, base_type, is_const, is_pointer, array_dims=None):
        return super(CParam, cls).__new__(cls, name, ctype, base_type, is_const, is_pointer, array_dims)

`load_c2py(path)`

Load and parse an interface definition, returning a ModuleDef.

Supports two formats, auto-detected: 1. C source (.c, .h): C2PY_BEGIN..C2PY_END blocks embedded in comments (parsed via c2py23.harvester). 2. Python dict (.c2py, .c2py.py): a file containing a Python dict literal (parsed via ast.literal_eval). Lines starting with '#' are stripped as comments.

For .c and .h files, interface definitions are embedded as: / C2PY_BEGIN "module": "mymod", "source": ["mymod.c"], "functions": ... C2PY_END /

Source code in c2py23/parser.py

def load_c2py(path):
    """Load and parse an interface definition, returning a ModuleDef.

    Supports two formats, auto-detected:
      1. C source (.c, .h): C2PY_BEGIN..C2PY_END blocks embedded in
         comments (parsed via c2py23.harvester).
      2. Python dict (.c2py, .c2py.py): a file containing a Python
         dict literal (parsed via ast.literal_eval).
         Lines starting with '#' are stripped as comments.

    For .c and .h files, interface definitions are embedded as:
        /* C2PY_BEGIN
        "module": "mymod",
        "source": ["mymod.c"],
        "functions": ...
        C2PY_END */
    """
    # C source files -- extract C2PY_BEGIN blocks via harvester
    if path.endswith(".c") or path.endswith(".h"):
        from c2py23.harvester import extract_from_file

        raw = extract_from_file(path)
        if not isinstance(raw, dict) or "module" not in raw:
            raise ValueError("No C2PY_BEGIN block with 'module' key found in {}".format(path))
        mod = from_c2py_dict(raw, path)
        base_dir = os.path.dirname(os.path.abspath(path))
        _validate_module(mod, base_dir)
        return mod

    with open(path, "r") as f:
        text = f.read()

    # Try Python dict format first (safe, no dependencies).
    # Strip whole-line comments (#) before passing to literal_eval.
    try:
        stripped = re.sub(r"(?m)^\s*#.*$", "", text)
        raw = ast.literal_eval(stripped)
    except (ValueError, SyntaxError):
        raw = None

    if isinstance(raw, dict):
        mod = from_c2py_dict(raw, path)
        base_dir = os.path.dirname(os.path.abspath(path))
        _validate_module(mod, base_dir)
        return mod

    raise ValueError("Could not parse '{}': not a valid Python dict".format(path))

`from_c2py_dict(raw_dict, path='<dict>')`

Parse a Python dict (from dict format) into a ModuleDef.

Parameters:

Name	Type	Description	Default
`raw_dict`		A dict with keys: module, source, headers, functions, constants, timing, free_threading.	required
`path`		A label for error messages (file path or "").	`'<dict>'`

Returns:

Type	Description
	A ModuleDef namedtuple.

Source code in c2py23/parser.py

def from_c2py_dict(raw_dict, path="<dict>"):
    """Parse a Python dict (from dict format) into a ModuleDef.

    Args:
        raw_dict: A dict with keys: module, source, headers, functions,
                  constants, timing, free_threading.
        path: A label for error messages (file path or "<dict>").

    Returns:
        A ModuleDef namedtuple.
    """
    module_name = _get_required(raw_dict, "module", path)
    sources = raw_dict.get("source", [])
    if isinstance(sources, _STRING_TYPES):
        sources = [sources]
    headers = raw_dict.get("headers", [])
    if isinstance(headers, _STRING_TYPES):
        headers = [headers]

    funcs = []
    for f in raw_dict.get("functions", []):
        funcs.extend(_expand_func_template(f, path))

    constants = raw_dict.get("constants", {})
    if not isinstance(constants, dict):
        raise ValueError("'constants' must be a dict in {}".format(path))
    for k, v in constants.items():
        if not isinstance(v, int):
            raise ValueError("Constant '{}' in {} must be an integer, got {}".format(k, path, type(v)))

    timing = bool(raw_dict.get("timing", False))
    free_threading = bool(raw_dict.get("free_threading", False))

    mod = ModuleDef(module_name, sources, headers, funcs, constants, timing, free_threading)
    return mod

generator

`c2py23.generator`

C code generator for c2py23.

Transpiles a parsed ModuleDef AST into a compilable CPython C extension. Uses the CBuilder class to enforce structural invariants at emit time: buffer acquires are paired with releases, GIL saves with restores, output scalars with NULL checks and PyTuple_SetItem.

`CBuilder`

Stateful builder for generating C wrapper code.

Tracks buffer acquires, GIL depth, and output-object construction so that cleanup code is always structurally correct.

Source code in c2py23/generator.py

class CBuilder:
    """Stateful builder for generating C wrapper code.

    Tracks buffer acquires, GIL depth, and output-object construction so
    that cleanup code is always structurally correct.
    """

    def __init__(self):
        self.lines = []
        self._buf_names = []  # declared buffer names in order
        self._acq_names = set()  # declared acq flag names
        self._acquired = []  # buffers acquired (stack, in order)
        self._acq_order = []  # backend source order (C2PY_PIN_* values)
        self._gil_depth = 0
        self._in_wrapper = False
        self._in_cleanup = False
        self._impl_has_gil_release = False
        self._gil_restore_before_py = True  # invariant marker
        self._has_goto_cleanup = False

    # -- Low-level emit --

    def emit(self, line=""):
        self.lines.append(line)

    def emit_indent(self, indent, line=""):
        self.lines.append(indent + line)

    def emit_blank(self):
        self.lines.append("")

    def get_code(self):
        return "\n".join(self.lines) + "\n"

    def extend(self, other):
        """Merge another CBuilder's output into this one."""
        self.lines.extend(other.lines)

    # -- Buffer management in wrapper --

    def declare_buffer(self, name):
        self._buf_names.append(name)
        base = name.replace("info_", "", 1)
        self.emit("    c2py_buf_pin pin_{0};".format(base))
        self.emit("    c2py_ptr_info info_{0};".format(base))

    def emit_buf_memset(self, buf_var):
        base = buf_var.replace("info_", "", 1)
        self.emit("    memset(&pin_{0}.buf, 0, C2PY.pybuffer_size);".format(base))

    def acquire_buffer(self, buf_var, py_var, flags, func_name):
        """Emits c2py_pin with correct failure path.

        First buffer: return NULL on failure (nothing to clean up).
        Subsequent: goto cleanup on failure.
        """
        first = len(self._acquired) == 0
        base = buf_var.replace("info_", "", 1)
        self.emit(
            "    if (c2py_pin({0}, &pin_{1}, &info_{1}, {2}, _acqord_{3}, {4}) == -1)".format(
                py_var, base, flags, func_name, len(self._acq_order)
            )
        )
        if first:
            self.emit("        return NULL;")
        else:
            self.emit("        goto cleanup;")
            self._has_goto_cleanup = True
        self._acquired.append(buf_var)

    def enter_cleanup(self):
        self._in_cleanup = True
        if not self._acquired:
            return
        if self._has_goto_cleanup:
            self.emit("cleanup:")
        for buf_var in reversed(self._acquired):
            base = buf_var.replace("info_", "", 1)
            self.emit("    c2py_unpin_buffer(&pin_{0});".format(base))
        self._acquired = []
        self._has_goto_cleanup = False

    def _acq_name(self, buf_var):
        base = buf_var.replace("info_", "", 1)
        return "pin_{0}.acquired".format(base)

    # -- GIL management --

    def gil_save(self, cond_var, thread_state_var, indent="    "):
        self._gil_depth += 1
        self.emit(indent + "if ({0}) {1} = PyEval_SaveThread();".format(cond_var, thread_state_var))

    def gil_restore(self, cond_var, thread_state_var, indent="    "):
        if self._gil_depth <= 0:
            raise ValueError("gil_restore without matching gil_save")
        self._gil_depth -= 1
        self.emit(indent + "if ({0}) PyEval_RestoreThread({1});".format(cond_var, thread_state_var))

    def assert_gil_balanced(self, name):
        if self._gil_depth != 0:
            raise ValueError("Function '%s': unbalanced GIL save/restore (%d)" % (name, self._gil_depth))

    # -- Output scalar construction --

    def _check_output_obj(self, obj_var, indent="    "):
        """Emit NULL check + Py_DECREF cleanup for an output object."""
        self.emit(indent + "if ({0} == NULL) {{".format(obj_var))
        self.emit(indent + "    Py_DECREF(_c2py_tup);")
        self.emit(indent + "    return NULL;")
        self.emit(indent + "}")

    def emit_tuple_new(self, n, indent="    "):
        self.emit(indent + "PyObject *_c2py_tup = PyTuple_New({0});".format(n))
        self.emit(indent + "if (_c2py_tup == NULL) return NULL;")

    def emit_output_long(self, index, val, ctype="int", indent="    "):
        if ctype in ("int64_t",):
            self.emit(indent + "PyObject *_c2py_obj{0} = PyLong_FromLongLong((long long){1});".format(index, val))
        elif ctype in ("uint64_t",):
            self.emit(indent + "PyObject *_c2py_obj{0} = PyLong_FromUnsignedLongLong({1});".format(index, val))
        else:
            self.emit(indent + "PyObject *_c2py_obj{0} = PyLong_FromLong((long){1});".format(index, val))
        self._check_output_obj("_c2py_obj{0}".format(index), indent)
        self.emit(indent + "PyTuple_SetItem(_c2py_tup, {0}, _c2py_obj{0});".format(index))

    def emit_output_double(self, index, val, indent="    "):
        self.emit(indent + "PyObject *_c2py_obj{0} = PyFloat_FromDouble((double){1});".format(index, val))
        self._check_output_obj("_c2py_obj{0}".format(index), indent)
        self.emit("    PyTuple_SetItem(_c2py_tup, {0}, _c2py_obj{0});".format(index))

    # -- Restrict and contiguity checks --

    def emit_restrict_checks(self, buf_params, func):
        """Emit restrict alias checks between writable buffers."""
        writable = set()
        const_set = set()
        for p in buf_params:
            for ol in func.overloads:
                if ol.variants:
                    all_params = []
                    for v in ol.variants:
                        all_params.extend(v.params)
                    for cp in all_params:
                        if cp.is_pointer:
                            expr = ol.map_exprs.get(cp.name)
                            if expr is not None and _expr_refers_to(expr, p.name):
                                if cp.is_const:
                                    const_set.add(p.name)
                                else:
                                    writable.add(p.name)
                else:
                    for cp in ol.params:
                        if cp.is_pointer:
                            expr = ol.map_exprs.get(cp.name)
                            if expr is not None and _expr_refers_to(expr, p.name):
                                if cp.is_const:
                                    const_set.add(p.name)
                                else:
                                    writable.add(p.name)

        checked = set()
        for wn in writable:
            for other in list(writable | const_set):
                if other == wn:
                    continue
                pair = tuple(sorted([wn, other]))
                if pair in checked:
                    continue
                checked.add(pair)
                self.emit("    /* restrict check: {} vs {} */".format(wn, other))
                self.emit("    if ((char*)info_{0}.ptr >= (char*)info_{1}.ptr && ".format(wn, other))
                self.emit("        (char*)info_{0}.ptr < (char*)info_{1}.ptr + info_{1}.len) {{".format(wn, other))
                self.emit('        PyErr_SetString(PyExc_ValueError, "buffer aliasing forbidden");')
                self.emit("        goto cleanup;")
                self._has_goto_cleanup = True
                self.emit("    }")
                self.emit("    if ((char*)info_{0}.ptr >= (char*)info_{1}.ptr && ".format(other, wn))
                self.emit("        (char*)info_{0}.ptr < (char*)info_{1}.ptr + info_{1}.len) {{".format(other, wn))
                self.emit('        PyErr_SetString(PyExc_ValueError, "buffer aliasing forbidden");')
                self.emit("        goto cleanup;")
                self._has_goto_cleanup = True
                self.emit("    }")
                self.emit("")

    def emit_contiguity_checks(self, buf_params):
        """Emit contiguity validation for each buffer.

        Also emits _c2py_slow_axis_buf_{name} -- the index of the slowest
        varying axis (0 for C-contiguous, ndim-1 for F-contiguous).
        Set to -1 if the buffer is not C or F contiguous (rejected anyway).
        """
        if not buf_params:
            return

        for p in buf_params:
            name = p.name
            fmt = lambda s: s.format(name)
            self.emit("    int _c2py_slow_axis_info_{0} = -1;".format(name))
            self.emit("    int _c2py_fast_axis_info_{0} = -1;".format(name))
            self.emit("    (void)_c2py_slow_axis_info_{0};".format(name))
            self.emit("    (void)_c2py_fast_axis_info_{0};".format(name))
            self.emit("    /* contiguity check: {0} */".format(name))
            self.emit("    do {")
            self.emit("        int _ok = 1;")
            self.emit("        if (info_{0}->strides == NULL && info_{0}->ndim <= 1) {{".format(name))
            self.emit("            _c2py_slow_axis_info_{0} = 0;".format(name))
            self.emit("            _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1);".format(name))
            self.emit("            break;")
            self.emit("        }")
            self.emit(fmt("        if (info_{0}->len == 0) {{"))
            self.emit(fmt("            _c2py_slow_axis_info_{0} = 0;".format(name)))
            self.emit(fmt("            _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1);".format(name)))
            self.emit("            break;")
            self.emit("        }")
            self.emit(fmt("        if (info_{0}->ndim >= 1) {{"))
            self.emit(fmt("            Py_ssize_t _expected = info_{0}->itemsize;"))
            self.emit("            int _d;")
            self.emit("            /* check F-contiguous (column-major): first dim varies fastest */")
            self.emit(fmt("            for (_d = 0; _d < info_{0}->ndim; _d++) {{"))
            self.emit(fmt("                if (info_{0}->strides[_d] < 0) {{ _ok = 0; break; }}"))
            self.emit(fmt("                if (info_{0}->strides[_d] != _expected) {{ _ok = 0; break; }}"))
            self.emit(fmt("                _expected *= info_{0}->shape[_d];"))
            self.emit("            }")
            self.emit(
                "            if (_ok) {{ _c2py_slow_axis_info_{0} = (int)(info_{0}->ndim - 1); _c2py_fast_axis_info_{0} = 0; break; }}".format(
                    name
                )
            )
            self.emit("            /* check C-contiguous (row-major): last dim varies fastest */")
            self.emit("            _ok = 1;")
            self.emit(fmt("            _expected = info_{0}->itemsize;"))
            self.emit(fmt("            for (_d = info_{0}->ndim - 1; _d >= 0; _d--) {{"))
            self.emit(fmt("                if (info_{0}->strides[_d] < 0) {{ _ok = 0; break; }}"))
            self.emit(fmt("                if (info_{0}->strides[_d] != _expected) {{ _ok = 0; break; }}"))
            self.emit(fmt("                _expected *= info_{0}->shape[_d];"))
            self.emit("            }")
            self.emit(
                "            if (_ok) {{ _c2py_slow_axis_info_{0} = 0; _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1); }}".format(
                    name
                )
            )
            self.emit("        }")
            self.emit("        if (!_ok) {")
            self.emit("            PyErr_SetString(PyExc_ValueError,")
            self.emit('                "buffer not contiguous (C or Fortran contiguous required)");')
            self.emit("            return NULL;")
            self.emit("        }")
            self.emit("    } while(0);")
            self.emit("")

`acquire_buffer(buf_var, py_var, flags, func_name)`

Emits c2py_pin with correct failure path.

First buffer: return NULL on failure (nothing to clean up). Subsequent: goto cleanup on failure.

Source code in c2py23/generator.py

def acquire_buffer(self, buf_var, py_var, flags, func_name):
    """Emits c2py_pin with correct failure path.

    First buffer: return NULL on failure (nothing to clean up).
    Subsequent: goto cleanup on failure.
    """
    first = len(self._acquired) == 0
    base = buf_var.replace("info_", "", 1)
    self.emit(
        "    if (c2py_pin({0}, &pin_{1}, &info_{1}, {2}, _acqord_{3}, {4}) == -1)".format(
            py_var, base, flags, func_name, len(self._acq_order)
        )
    )
    if first:
        self.emit("        return NULL;")
    else:
        self.emit("        goto cleanup;")
        self._has_goto_cleanup = True
    self._acquired.append(buf_var)

`emit_contiguity_checks(buf_params)`

Emit contiguity validation for each buffer.

Also emits c2py_slow_axis_buf{name} -- the index of the slowest varying axis (0 for C-contiguous, ndim-1 for F-contiguous). Set to -1 if the buffer is not C or F contiguous (rejected anyway).

Source code in c2py23/generator.py

def emit_contiguity_checks(self, buf_params):
    """Emit contiguity validation for each buffer.

    Also emits _c2py_slow_axis_buf_{name} -- the index of the slowest
    varying axis (0 for C-contiguous, ndim-1 for F-contiguous).
    Set to -1 if the buffer is not C or F contiguous (rejected anyway).
    """
    if not buf_params:
        return

    for p in buf_params:
        name = p.name
        fmt = lambda s: s.format(name)
        self.emit("    int _c2py_slow_axis_info_{0} = -1;".format(name))
        self.emit("    int _c2py_fast_axis_info_{0} = -1;".format(name))
        self.emit("    (void)_c2py_slow_axis_info_{0};".format(name))
        self.emit("    (void)_c2py_fast_axis_info_{0};".format(name))
        self.emit("    /* contiguity check: {0} */".format(name))
        self.emit("    do {")
        self.emit("        int _ok = 1;")
        self.emit("        if (info_{0}->strides == NULL && info_{0}->ndim <= 1) {{".format(name))
        self.emit("            _c2py_slow_axis_info_{0} = 0;".format(name))
        self.emit("            _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1);".format(name))
        self.emit("            break;")
        self.emit("        }")
        self.emit(fmt("        if (info_{0}->len == 0) {{"))
        self.emit(fmt("            _c2py_slow_axis_info_{0} = 0;".format(name)))
        self.emit(fmt("            _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1);".format(name)))
        self.emit("            break;")
        self.emit("        }")
        self.emit(fmt("        if (info_{0}->ndim >= 1) {{"))
        self.emit(fmt("            Py_ssize_t _expected = info_{0}->itemsize;"))
        self.emit("            int _d;")
        self.emit("            /* check F-contiguous (column-major): first dim varies fastest */")
        self.emit(fmt("            for (_d = 0; _d < info_{0}->ndim; _d++) {{"))
        self.emit(fmt("                if (info_{0}->strides[_d] < 0) {{ _ok = 0; break; }}"))
        self.emit(fmt("                if (info_{0}->strides[_d] != _expected) {{ _ok = 0; break; }}"))
        self.emit(fmt("                _expected *= info_{0}->shape[_d];"))
        self.emit("            }")
        self.emit(
            "            if (_ok) {{ _c2py_slow_axis_info_{0} = (int)(info_{0}->ndim - 1); _c2py_fast_axis_info_{0} = 0; break; }}".format(
                name
            )
        )
        self.emit("            /* check C-contiguous (row-major): last dim varies fastest */")
        self.emit("            _ok = 1;")
        self.emit(fmt("            _expected = info_{0}->itemsize;"))
        self.emit(fmt("            for (_d = info_{0}->ndim - 1; _d >= 0; _d--) {{"))
        self.emit(fmt("                if (info_{0}->strides[_d] < 0) {{ _ok = 0; break; }}"))
        self.emit(fmt("                if (info_{0}->strides[_d] != _expected) {{ _ok = 0; break; }}"))
        self.emit(fmt("                _expected *= info_{0}->shape[_d];"))
        self.emit("            }")
        self.emit(
            "            if (_ok) {{ _c2py_slow_axis_info_{0} = 0; _c2py_fast_axis_info_{0} = (int)(info_{0}->ndim - 1); }}".format(
                name
            )
        )
        self.emit("        }")
        self.emit("        if (!_ok) {")
        self.emit("            PyErr_SetString(PyExc_ValueError,")
        self.emit('                "buffer not contiguous (C or Fortran contiguous required)");')
        self.emit("            return NULL;")
        self.emit("        }")
        self.emit("    } while(0);")
        self.emit("")

`emit_restrict_checks(buf_params, func)`

Emit restrict alias checks between writable buffers.

Source code in c2py23/generator.py

def emit_restrict_checks(self, buf_params, func):
    """Emit restrict alias checks between writable buffers."""
    writable = set()
    const_set = set()
    for p in buf_params:
        for ol in func.overloads:
            if ol.variants:
                all_params = []
                for v in ol.variants:
                    all_params.extend(v.params)
                for cp in all_params:
                    if cp.is_pointer:
                        expr = ol.map_exprs.get(cp.name)
                        if expr is not None and _expr_refers_to(expr, p.name):
                            if cp.is_const:
                                const_set.add(p.name)
                            else:
                                writable.add(p.name)
            else:
                for cp in ol.params:
                    if cp.is_pointer:
                        expr = ol.map_exprs.get(cp.name)
                        if expr is not None and _expr_refers_to(expr, p.name):
                            if cp.is_const:
                                const_set.add(p.name)
                            else:
                                writable.add(p.name)

    checked = set()
    for wn in writable:
        for other in list(writable | const_set):
            if other == wn:
                continue
            pair = tuple(sorted([wn, other]))
            if pair in checked:
                continue
            checked.add(pair)
            self.emit("    /* restrict check: {} vs {} */".format(wn, other))
            self.emit("    if ((char*)info_{0}.ptr >= (char*)info_{1}.ptr && ".format(wn, other))
            self.emit("        (char*)info_{0}.ptr < (char*)info_{1}.ptr + info_{1}.len) {{".format(wn, other))
            self.emit('        PyErr_SetString(PyExc_ValueError, "buffer aliasing forbidden");')
            self.emit("        goto cleanup;")
            self._has_goto_cleanup = True
            self.emit("    }")
            self.emit("    if ((char*)info_{0}.ptr >= (char*)info_{1}.ptr && ".format(other, wn))
            self.emit("        (char*)info_{0}.ptr < (char*)info_{1}.ptr + info_{1}.len) {{".format(other, wn))
            self.emit('        PyErr_SetString(PyExc_ValueError, "buffer aliasing forbidden");')
            self.emit("        goto cleanup;")
            self._has_goto_cleanup = True
            self.emit("    }")
            self.emit("")

`extend(other)`

Merge another CBuilder's output into this one.

Source code in c2py23/generator.py

def extend(self, other):
    """Merge another CBuilder's output into this one."""
    self.lines.extend(other.lines)

`generate(module_def, use_single_header=False)`

Generate C wrapper source for a module using CBuilder pattern.

Parameters:

Name	Type	Description	Default
`module_def`		Parsed ModuleDef AST.	required
`use_single_header`		If True, emit #define C2PY_IMPLEMENTATION + include "c2py.h" instead of #include "c2py_runtime.h".	`False`

Returns C source string.

Source code in c2py23/generator.py

def generate(module_def, use_single_header=False):
    """Generate C wrapper source for a module using CBuilder pattern.

    Args:
        module_def: Parsed ModuleDef AST.
        use_single_header: If True, emit #define C2PY_IMPLEMENTATION +
            #include "c2py.h" instead of #include "c2py_runtime.h".

    Returns C source string.
    """
    b = CBuilder()

    # Header
    b.emit("/* Generated by c2py23 - do not edit by hand */")
    b.emit("#include <stdio.h>")
    if use_single_header:
        b.emit("#define C2PY_IMPLEMENTATION")
        b.emit('#include "c2py.h"')
    else:
        b.emit('#include "c2py_runtime.h"')
    for h in module_def.headers:
        b.emit('#include "{0}"'.format(h))
    b.emit_blank()

    has_gil_release = any(f.gil_release for f in module_def.functions)
    has_free_threading = module_def.free_threading

    seen = set()
    for func in module_def.functions:
        for ol in func.overloads:
            if ol.variants:
                for v in ol.variants:
                    cn = v.c_name
                    if cn is None:
                        cn = v.sig_str.split("(")[0].strip().split()[-1]
                    if cn not in seen:
                        seen.add(cn)
                        b.emit(_make_decl_string(v.return_type, cn, v.params))
            else:
                cn = ol.c_name
                if cn is None:
                    cn = ol.sig_str.split("(")[0].strip().split()[-1]
                if cn not in seen:
                    seen.add(cn)
                    b.emit(_make_decl_string(ol.return_type, cn, ol.params))
    b.emit_blank()

    # Timing declarations
    if module_def.timing:
        _emit_timing_decls(b, module_def)

    # GIL release declarations
    if has_gil_release:
        b.emit("/* ---- GIL release ---- */")
        b.emit("static int _c2py_gil_release_enabled = 1;")
        for func in module_def.functions:
            if func.gil_release:
                b.emit("static int _gil_release_{0} = 1;".format(func.name))
        b.emit_blank()

    # Per-function emission (each function gets its own CBuilder to
    # prevent state leaks -- _has_goto_cleanup, _acquired, _gil_depth
    # all start fresh for each function).
    for func in module_def.functions:
        fb = CBuilder()
        _emit_function(fb, func, module_def.name, module_def.timing, has_gil_release)
        b.extend(fb)

    # Module init
    _emit_module_init(b, module_def, has_free_threading, has_gil_release)

    code = b.get_code()
    verify_c_invariants(code)
    return code

cli

`c2py23.cli`

CLI entry point for c2py23 -- generate C wrapper from .c2py interface.

Usage

c2py23 file.c2py -o wrapper.c # generate wrapper to file c2py23 file.c2py # generate to stdout c2py23 --version # print version

`generate(c2py_path, output_path=None, use_single_header=False)`

Parse a .c2py file and generate the wrapper C file.

Parameters:

Name	Description	Default
`c2py_path`	Path to .c2py interface file.	required
`output_path`	Optional output .c path. If None, returns code.	`None`
`use_single_header`	If True, emit #include "c2py.h" (with C2PY_IMPLEMENTATION) instead of #include "c2py_runtime.h".	`False`

Returns:

Type	Description
	Generated C code as string.

Source code in c2py23/cli.py

def generate(c2py_path, output_path=None, use_single_header=False):
    """Parse a .c2py file and generate the wrapper C file.

    Args:
        c2py_path: Path to .c2py interface file.
        output_path: Optional output .c path.  If None, returns code.
        use_single_header: If True, emit #include "c2py.h" (with C2PY_IMPLEMENTATION)
            instead of #include "c2py_runtime.h".

    Returns:
        Generated C code as string.
    """
    if not os.path.exists(c2py_path):
        print("ERROR: file not found: {}".format(c2py_path), file=sys.stderr)
        sys.exit(1)

    print("Parsing {}...".format(c2py_path))
    module_def = load_c2py(c2py_path)

    from c2py23.generator import generate as _gen

    c_code = _gen(module_def, use_single_header=use_single_header)

    if output_path:
        try:
            with open(output_path, "w") as f:
                f.write(c_code)
        except IOError as e:
            sys.exit("Error writing {}: {}".format(output_path, e))
        print("Wrapper written to: {}".format(output_path))

    return c_code

c2py_loader

`c2py23.c2py_loader`

c2py_loader - Load a c2py23-native .so by explicit filename.

Convention: .c2py23-_.so

Example files

_mymodule.c2py23-linux_x86_64.so _mymodule.c2py23-linux_ppc64le.so _mymodule.c2py23-linux_aarch64.so _mymodule.c2py23-win_amd64.pyd _mymodule.c2py23-darwin_arm64.so

No monkeypatching of EXTENSION_SUFFIXES. No sys.path hacking. Loads the .so by full path via ExtensionFileLoader (Python 3.x) or imp.load_dynamic (Python 2.7).

Usage in your package's init.py:

from c2py23.c2py_loader import load_native
import os as _os
_mod = load_native(_os.path.dirname(_os.path.abspath(__file__)),
                   '_mymodule')
# Re-export public names (skip dunders, keep single-underscore API)
for _k, _v in _mod.__dict__.items():
    if _k.startswith('__') and _k.endswith('__'):
        continue
    globals()[_k] = _v

Users who do not want a c2py23 runtime dependency can copy the function body into their own init.py. The code is intentionally small and self-contained.

`load_native(package_dir, module_name='_native', tag='c2py23')`

Load .-.so from package_dir.

package_dir: Absolute path to the package directory containing the .so files. module_name: Base module name (default '_native'). Must match the c2py23 'module:' field. Use a unique name per package (e.g. '_mymodule') to avoid collisions in sys.modules. tag: Tag string inserted in the filename (default 'c2py23').

Returns:

Type	Description
	The loaded module object. The caller should re-export public
	names from it.

Example

_mod = load_native(os.path.dirname(file), '_mymodule')

Source code in c2py23/c2py_loader.py

def load_native(package_dir, module_name="_native", tag="c2py23"):
    """Load <module_name>.<tag>-<platform_key>.so from package_dir.

    Args:
    package_dir: Absolute path to the package directory containing
                 the .so files.
    module_name: Base module name (default '_native').
                 Must match the c2py23 'module:' field.
                 Use a unique name per package (e.g. '_mymodule')
                 to avoid collisions in sys.modules.
        tag:         Tag string inserted in the filename (default 'c2py23').

    Returns:
        The loaded module object.  The caller should re-export public
        names from it.

    Example:
        _mod = load_native(os.path.dirname(__file__), '_mymodule')
    """
    _key = _platform_key()
    if os.name == "nt":
        _ext = ".pyd"
    else:
        _ext = ".so"
    _filename = "%s.%s-%s%s" % (module_name, tag, _key, _ext)
    _path = os.path.join(package_dir, _filename)

    _trace = os.environ.get("C2PY_TRACE")
    if _trace:
        print("[c2py_loader] platform=%s file=%s" % (_key, _filename), file=sys.stderr)

    if not os.path.isfile(_path):
        _alternatives = [f for f in os.listdir(package_dir) if f.startswith(module_name + ".")]
        _hint = ", ".join(sorted(_alternatives)) if _alternatives else "none"
        raise ImportError(
            "c2py23: native module not found for platform '%s'\n"
            "  Expected: %s\n"
            "  Available: %s\n"
            "  Build with: gcc -shared -fPIC ... -o %s" % (_key, _filename, _hint, _path)
        )

    if _trace:
        print("[c2py_loader] loading %s -> %s" % (module_name, _path), file=sys.stderr)

    if sys.version_info[0] >= 3:
        import importlib.machinery
        import importlib.util

        loader = importlib.machinery.ExtensionFileLoader(module_name, _path)
        spec = importlib.util.spec_from_file_location(module_name, _path, loader=loader)
        mod = importlib.util.module_from_spec(spec)
        # Warn if overwriting an existing module in sys.modules.
        # Using distinct module names per package (e.g. '_mymodule'
        # instead of '_native') avoids collisions.
        if module_name in sys.modules:
            import warnings as _w

            _w.warn(
                "c2py_loader: overwriting existing module '%s' "
                "in sys.modules. Use a unique module name." % module_name
            )
        sys.modules[module_name] = mod
        loader.exec_module(mod)
        return mod
    else:
        import imp

        return imp.load_dynamic(module_name, _path)

perf

`c2py23.perf`

c2py23 performance data decoder.

Reads c2py_perf_t data via module-level C accessor functions -- no ctypes required. Modules built with timing: true expose _c2py_perf_read, _c2py_perf_meta, _c2py_perf_reset and pointer attributes on the module so that read_perf() can locate the right perf counter from the wrapper function object alone.

Usage::

from c2py23.perf import read_perf, reset_perf

import my_timed_module
stats = read_perf(my_timed_module.wsum)
print(stats)

# Switch to CPU cycle counter at runtime (rdtsc/CNTVCT_EL0/mftb):
my_timed_module._c2py_set_tick_source("cycle")
stats = read_perf(my_timed_module.wsum)

# Switch back to wall clock:
my_timed_module._c2py_set_tick_source("clock")

# Read the currently-selected variant's timing:
stats = read_perf(my_timed_module.poly, variant=True)

# Reset counters between benchmark batches:
reset_perf(my_timed_module.wsum)

`read_enabled(func)`

Return 1 if per-call timing is enabled, 0 otherwise.

Parameters

func : callable Any wrapper function from a timing-enabled module. The module is found via func.__self__ (Python 3) or func.__module__ (Python 2.7).

Source code in c2py23/perf.py

def read_enabled(func):
    """Return 1 if per-call timing is enabled, 0 otherwise.

    Parameters
    ----------
    func : callable
        Any wrapper function from a timing-enabled module.  The module
        is found via ``func.__self__`` (Python 3) or ``func.__module__``
        (Python 2.7).
    """
    mod = _get_mod(func)
    return mod._c2py_perf_get_enabled()

`read_perf(func, freq_hz=None, variant=None)`

Decode a c2py_perf_t counter.

Parameters

func : callable Wrapper function on a timing-enabled module (e.g. mod.wsum). The module is found via func.__self__ and the counter pointer is looked up by func.__name__ on the module. freq_hz : int or None, optional Tick source frequency in Hz. When None or 0, the raw tick values are returned under the _ns keys (correct for the default clock_gettime source which returns nanoseconds). When provided and not equal to 1e9, the values are converted and the raw tick values are additionally returned under _cycles keys. variant : None, True, or str, optional Which perf counter to read:

- ``None`` (default) -- wrapper-overhead counter
- ``True`` -- the currently-selected variant (functions with
  variant groups only; falls back to wrapper overhead otherwise)
- ``str`` -- a named overload, e.g. ``"weighted_sum"`` for the
  counter ``_c2py_ol_ptr_wsum__weighted_sum``

Returns

dict with keys: call_count, t_enter, t_pre_c, t_post_c, t_exit, c_dur_ns, wrap_dur_ns, c_min_ns, c_max_ns, c_mean_ns, wrap_min_ns, wrap_max_ns, wrap_mean_ns, variant, group_idx, variant_name. Additional keys when freq_hz is provided and != 1e9: c_dur_cycles, wrap_dur_cycles, c_min_cycles, c_max_cycles, c_mean_cycles, wrap_min_cycles, wrap_max_cycles, wrap_mean_cycles.

Source code in c2py23/perf.py

def read_perf(func, freq_hz=None, variant=None):
    """Decode a c2py_perf_t counter.

    Parameters
    ----------
    func : callable
        Wrapper function on a timing-enabled module (e.g. ``mod.wsum``).
        The module is found via ``func.__self__`` and the counter pointer
        is looked up by ``func.__name__`` on the module.
    freq_hz : int or None, optional
        Tick source frequency in Hz.  When None or 0, the raw tick values
        are returned under the ``_ns`` keys (correct for the default
        ``clock_gettime`` source which returns nanoseconds).  When provided
        and not equal to 1e9, the values are converted and the raw tick
        values are additionally returned under ``_cycles`` keys.
    variant : None, True, or str, optional
        Which perf counter to read:

        - ``None`` (default) -- wrapper-overhead counter
        - ``True`` -- the currently-selected variant (functions with
          variant groups only; falls back to wrapper overhead otherwise)
        - ``str`` -- a named overload, e.g. ``"weighted_sum"`` for the
          counter ``_c2py_ol_ptr_wsum__weighted_sum``

    Returns
    -------
    dict with keys:
        call_count, t_enter, t_pre_c, t_post_c, t_exit,
        c_dur_ns, wrap_dur_ns,
        c_min_ns, c_max_ns, c_mean_ns,
        wrap_min_ns, wrap_max_ns, wrap_mean_ns,
        variant, group_idx, variant_name.
      Additional keys when freq_hz is provided and != 1e9:
        c_dur_cycles, wrap_dur_cycles,
        c_min_cycles, c_max_cycles, c_mean_cycles,
        wrap_min_cycles, wrap_max_cycles, wrap_mean_cycles.
    """
    ptr = _get_perf_ptr(func, variant)
    if ptr == 0:
        return {"call_count": 0}

    mod = _get_mod(func)

    buf = _make_perf_buf()
    mod._c2py_perf_read(ptr, buf)
    variant_val, group_idx, vname = mod._c2py_perf_meta(ptr)

    n = _read_buf(buf, _I_CALL_COUNT)
    t_enter = _read_buf(buf, _I_T_ENTER)
    t_pre_c = _read_buf(buf, _I_T_PRE_C)
    t_post_c = _read_buf(buf, _I_T_POST_C)
    t_exit = _read_buf(buf, _I_T_EXIT)
    t_c_min = _read_buf(buf, _I_T_C_MIN)
    t_c_max = _read_buf(buf, _I_T_C_MAX)
    t_c_total = _read_buf(buf, _I_T_C_TOTAL)
    t_w_min = _read_buf(buf, _I_T_WRAP_MIN)
    t_w_max = _read_buf(buf, _I_T_WRAP_MAX)
    t_w_total = _read_buf(buf, _I_T_WRAP_TOTAL)

    c_dur = t_post_c - t_pre_c
    wrap_dur = (t_pre_c - t_enter) + (t_exit - t_post_c) if t_enter or t_exit else 0

    result = {
        "call_count": n,
        "t_enter": t_enter,
        "t_pre_c": t_pre_c,
        "t_post_c": t_post_c,
        "t_exit": t_exit,
        "c_dur_ns": _to_ns(c_dur, freq_hz),
        "wrap_dur_ns": _to_ns(wrap_dur, freq_hz),
        "c_min_ns": _to_ns(t_c_min, freq_hz),
        "c_max_ns": _to_ns(t_c_max, freq_hz),
        "c_mean_ns": _to_ns(t_c_total / float(n), freq_hz) if n else 0,
        "wrap_min_ns": _to_ns(t_w_min, freq_hz),
        "wrap_max_ns": _to_ns(t_w_max, freq_hz),
        "wrap_mean_ns": _to_ns(t_w_total / float(n), freq_hz) if n else 0,
        "variant": variant_val,
        "group_idx": group_idx,
        "variant_name": vname,
    }

    if freq_hz is not None and freq_hz != 0 and freq_hz != 1000000000:
        result["c_dur_cycles"] = c_dur
        result["wrap_dur_cycles"] = wrap_dur
        result["c_min_cycles"] = t_c_min
        result["c_max_cycles"] = t_c_max
        result["c_mean_cycles"] = t_c_total / float(n) if n else 0
        result["wrap_min_cycles"] = t_w_min
        result["wrap_max_cycles"] = t_w_max
        result["wrap_mean_cycles"] = t_w_total / float(n) if n else 0

    return result

`reset_perf(func, variant=None)`

Reset a c2py_perf_t counter to its initial state.

Parameters

func : callable Wrapper function on a timing-enabled module (e.g. mod.wsum). variant : None, True, or str, optional Selects which counter to reset. See :func:read_perf.

Source code in c2py23/perf.py

def reset_perf(func, variant=None):
    """Reset a c2py_perf_t counter to its initial state.

    Parameters
    ----------
    func : callable
        Wrapper function on a timing-enabled module (e.g. ``mod.wsum``).
    variant : None, True, or str, optional
        Selects which counter to reset.  See :func:`read_perf`.
    """
    ptr = _get_perf_ptr(func, variant)
    if ptr == 0:
        return
    mod = _get_mod(func)
    mod._c2py_perf_reset(ptr)

`set_enabled(func, value)`

Enable (value=1) or disable (value=0) per-call timing.

Parameters

func : callable See :func:get_enabled. value : int 1 to enable, 0 to disable.

Source code in c2py23/perf.py

def set_enabled(func, value):
    """Enable (value=1) or disable (value=0) per-call timing.

    Parameters
    ----------
    func : callable
        See :func:`get_enabled`.
    value : int
        1 to enable, 0 to disable.
    """
    mod = _get_mod(func)
    mod._c2py_perf_set_enabled(int(value))

invariant_checker

`c2py23.invariant_checker`

Invariant checker for generated C code.

Scans generated C source and verifies structural properties

Buffer acquire/release pairs in wrapper functions
GIL save/restore pairs in impl functions
Output scalar NULL checks + PyTuple_SetItem
Balanced braces

Raises ValueError with line number on first violation.

`verify_c_invariants(code)`

Check generated C for structural errors before returning.

Scans the generated C and verifies

Buffer acquire/release pairs in wrapper functions
GIL save/restore pairs in impl functions
Output scalar NULL checks + PyTuple_SetItem
Balanced braces

Raises ValueError with line number on first violation.

Source code in c2py23/invariant_checker.py

def verify_c_invariants(code):
    """Check generated C for structural errors before returning.

    Scans the generated C and verifies:
      - Buffer acquire/release pairs in wrapper functions
      - GIL save/restore pairs in impl functions
      - Output scalar NULL checks + PyTuple_SetItem
      - Balanced braces

    Raises ValueError with line number on first violation.
    """
    lines = code.split("\n")
    _check_balanced_braces(lines)
    _check_buffer_invariants(lines)
    _check_output_scalar_invariants(lines)

Tools

convert_c2py_to_dict

`tools.convert_c2py_to_dict`

Convert YAML .c2py files to Python dict format sidecar (.c2py.py).

The .c2py.py file is written alongside the original .c2py file and contains the equivalent interface as a Python dict literal. It is auto-detected by load_c2py() and can be loaded in place of the YAML.

Usage

python3 tools/convert_c2py_to_dict.py # tests/cases/ python3 tools/convert_c2py_to_dict.py --all # tests/cases/ + examples/ python3 tools/convert_c2py_to_dict.py path/file.c2py # single file python3 tools/convert_c2py_to_dict.py --check # validate symmetry only

`convert_file(c2py_path, check_only=False)`

Read YAML, write .c2py.py sidecar. Returns True on success.

Source code in tools/convert_c2py_to_dict.py

def convert_file(c2py_path, check_only=False):
    """Read YAML, write .c2py.py sidecar.  Returns True on success."""
    try:
        import yaml
    except ImportError:
        print("PyYAML required for conversion: pip install PyYAML", file=sys.stderr)
        return False

    with open(c2py_path) as f:
        data = yaml.safe_load(f)
    if not isinstance(data, dict):
        return False

    dict_path = c2py_path + '.py'
    result = py_repr(data, 0)
    if not check_only:
        with open(dict_path, 'w') as f:
            f.write('# Python dict format equivalent of %s\n' % os.path.basename(c2py_path))
            f.write(result)
            f.write('\n')
    return True

`py_repr(obj, indent=0)`

Render a Python object as a pretty-printed Python dict literal.

Source code in tools/convert_c2py_to_dict.py

def py_repr(obj, indent=0):
    """Render a Python object as a pretty-printed Python dict literal."""
    sp = "    "
    if isinstance(obj, dict):
        if not obj:
            return "{}"
        items = []
        for k, v in obj.items():
            kr = json.dumps(k)
            vr = py_repr(v, indent + 1)
            items.append(sp * (indent + 1) + kr + ": " + vr)
        inner = ",\n".join(items)
        return "{\n" + inner + ",\n" + sp * indent + "}"
    elif isinstance(obj, list):
        if not obj:
            return "[]"
        if len(obj) == 1 and not isinstance(obj[0], (dict, list)):
            return "[" + py_repr(obj[0], indent) + "]"
        items = []
        for v in obj:
            items.append(sp * (indent + 1) + py_repr(v, indent + 1))
        inner = ",\n".join(items)
        return "[\n" + inner + ",\n" + sp * indent + "]"
    elif isinstance(obj, bool):
        return "True" if obj else "False"
    elif obj is None:
        return "None"
    elif isinstance(obj, str):
        return json.dumps(obj)
    elif isinstance(obj, (int, float)):
        return json.dumps(obj)
    else:
        return json.dumps(obj)

Converts .c2py files from legacy YAML to Python dict format. YAML is no longer supported by c2py23 with default dependencies. This script is only needed for migration. Run with:

python3 -m tools.convert_c2py_to_dict path/to/file.c2py