summaryrefslogtreecommitdiffstats
path: root/scripts/gen_bindings.py
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/gen_bindings.py')
-rw-r--r--scripts/gen_bindings.py274
1 files changed, 274 insertions, 0 deletions
diff --git a/scripts/gen_bindings.py b/scripts/gen_bindings.py
new file mode 100644
index 0000000..845c8a5
--- /dev/null
+++ b/scripts/gen_bindings.py
@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+"""Generate python/ts/__init__.py from C source.
+
+Produces a compact module with a single _dispatch() function and a
+FUNC_META lookup table instead of 350+ individual wrappers.
+"""
+
+import importlib.util
+import os
+import py_compile
+import sys
+
+_spec = importlib.util.spec_from_file_location(
+ "ts_parser",
+ os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"),
+)
+if _spec and _spec.loader:
+ _ts_parser = importlib.util.module_from_spec(_spec)
+ _spec.loader.exec_module(_ts_parser)
+ parse_all = _ts_parser.parse_all
+else:
+ print("Error: could not load ts_parser", file=sys.stderr)
+ sys.exit(1)
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+OUT = os.path.join(ROOT, "python", "ts", "__init__.py")
+
+DEFAULT_PARAMS = {
+ "ichimoku": {"tenkan": 9, "kijun": 26, "senkou": 52},
+}
+
+MIN_PARAM_DEFAULTS = {
+ "hull_ma": {"k": 2},
+ "variable_index_dynamic_average": {"period": 2},
+ "logarithm": {"b": 2.0},
+ "choppiness_index": {"k": 2},
+ "performance_ratio": {"k": 2},
+ "standard_error": {"k": 2},
+}
+
+
+def _py_default(pytype, py_fname, name):
+ if py_fname in DEFAULT_PARAMS.get(name, {}):
+ return DEFAULT_PARAMS[name][py_fname]
+ if py_fname in MIN_PARAM_DEFAULTS.get(name, {}):
+ return MIN_PARAM_DEFAULTS[name][py_fname]
+ return 1 if "uint64" in pytype else 1.0
+
+
+def main():
+ funcs = parse_all(detect_x_access=True)
+ if not funcs:
+ print("Warning: no ts_parameter declarations found.", file=sys.stderr)
+ sys.exit(1)
+
+ os.makedirs(os.path.dirname(OUT), exist_ok=True)
+
+ with open(OUT, "w", encoding="utf-8") as f:
+ # --- Header ---
+ f.write('"""Python bindings for ts (time series) C library.\n')
+ f.write("\nAuto-generated. Uses a single dispatch function backed by\n")
+ f.write('a FUNC_META lookup table.\n"""\n\n')
+
+ f.write("import ctypes\n")
+ f.write("import ctypes.util\n")
+ f.write("import os\n")
+ f.write("import numpy as np\n")
+ f.write("import threading\n\n")
+
+ f.write("_tls = threading.local()\n\n")
+
+ # --- Library loader ---
+ f.write("def _find_lib():\n")
+ f.write(' """Try system path, pkg-config, then build dir."""\n')
+ f.write(" try:\n")
+ f.write(' return ctypes.CDLL("libts.so", ctypes.RTLD_GLOBAL)\n')
+ f.write(" except OSError:\n")
+ f.write(" pass\n")
+ f.write(" try:\n")
+ f.write(" import subprocess\n")
+ f.write(" libdir = subprocess.check_output(\n")
+ f.write(' ["pkg-config", "--variable=libdir", "ts"],\n')
+ f.write(" stderr=subprocess.DEVNULL,\n")
+ f.write(" ).decode().strip()\n")
+ f.write(
+ ' return ctypes.CDLL(os.path.join(libdir, "libts.so"), ctypes.RTLD_GLOBAL)\n'
+ )
+ f.write(" except (OSError, subprocess.CalledProcessError):\n")
+ f.write(" pass\n")
+ f.write(" build_dir = os.path.join(\n")
+ f.write(
+ " os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),\n"
+ )
+ f.write(' "build",\n')
+ f.write(" )\n")
+ f.write(' lib_path = os.path.join(build_dir, "libts.so")\n')
+ f.write(" if os.path.exists(lib_path):\n")
+ f.write(" return ctypes.CDLL(lib_path, ctypes.RTLD_GLOBAL)\n")
+ f.write(
+ ' raise OSError("Cannot find libts.so. Run make install or build from source.")\n\n\n'
+ )
+
+ f.write("__version__ = '0.1.0'\n\n")
+ f.write("_lib = _find_lib()\n\n")
+
+ # --- Helpers ---
+ f.write("_CD = ctypes.c_double\n\n\n")
+
+ f.write("def _cp(arr):\n")
+ f.write(' """Copy numpy array to ctypes buffer (tracked in TLS list)."""\n')
+ f.write(" buf = (_CD * len(arr))(*arr)\n")
+ f.write(" if not hasattr(_tls, 'bufs'):\n")
+ f.write(" _tls.bufs = []\n")
+ f.write(" _tls.bufs.append(buf)\n")
+ f.write(" return buf\n\n\n")
+
+ f.write("def _clr():\n")
+ f.write(' """Clear tracked ctypes buffers."""\n')
+ f.write(" try:\n")
+ f.write(" if hasattr(_tls, 'bufs'):\n")
+ f.write(" _tls.bufs.clear()\n")
+ f.write(" except Exception:\n")
+ f.write(" pass\n\n\n")
+
+ # --- Parameter structs ---
+ f.write("# --- Parameter structs ---\n")
+ seen_structs = set()
+ for fm in sorted(funcs, key=lambda x: x.name):
+ if fm.struct_fields and fm.name not in seen_structs:
+ cls_name = f"ts_{fm.name}_params_t"
+ f.write(f"class {cls_name}(ctypes.Structure):\n")
+ f.write(" _fields_ = [\n")
+ for c_fname, py_fname, pytype in fm.struct_fields:
+ f.write(f' ("{py_fname}", {pytype}),\n')
+ f.write(" ]\n\n")
+ seen_structs.add(fm.name)
+
+ # --- FUNC_META table ---
+ f.write("# --- Function metadata table ---\n")
+ f.write(
+ "# Each entry: (name, x_n, y_n, struct_class_or_None, [(param_name, default), ...])\n"
+ )
+ f.write("FUNC_META = [\n")
+ for fm in sorted(funcs, key=lambda x: x.name):
+ struct_cls = None
+ params = []
+ if fm.struct_fields:
+ struct_cls = f"ts_{fm.name}_params_t"
+ for c_fname, py_fname, pytype in fm.struct_fields:
+ d = _py_default(pytype, py_fname, fm.name)
+ params.append(f" ('{py_fname}', {d!r}),")
+ f.write(
+ f" ({fm.name!r}, {fm.x_n}, {fm.y_n}, {struct_cls}, [{chr(10).join(params)}]),\n"
+ )
+ f.write("]\n\n")
+
+ # --- Single dispatch function ---
+ f.write("# --- Dispatch ---\n")
+ f.write("def _dispatch(meta, *x_arrays):\n")
+ f.write(' """Call a ts indicator via ctypes, given a FUNC_META entry."""\n')
+ f.write(" name, x_n, y_n, struct_cls, param_defaults = meta\n")
+ f.write(" if len(x_arrays) != x_n:\n")
+ f.write(
+ ' raise TypeError(f"Expected {x_n} input arrays, got {len(x_arrays)}")\n'
+ )
+ f.write(" if x_n > 0:\n")
+ f.write(" n = x_arrays[0].shape[0]\n")
+ f.write(" for a in x_arrays:\n")
+ f.write(" if a.shape[0] != n:\n")
+ f.write(
+ ' raise ValueError(f"Array length mismatch: expected {n}, got {a.shape[0]}")\n'
+ )
+ f.write(" else:\n")
+ f.write(" n = 0\n")
+ f.write(" y = [np.zeros(n, dtype=np.float64) for _ in range(y_n)]\n")
+ f.write(" try:\n")
+ f.write(" x_bufs = [_cp(x_arrays[i]) for i in range(x_n)]\n")
+ f.write(
+ " y_ptrs = [y[i].ctypes.data_as(ctypes.POINTER(ctypes.c_double)) for i in range(y_n)]\n"
+ )
+ f.write(" c_func = getattr(_lib, f'ts_{name}_flat')\n")
+ f.write(" if struct_cls is not None:\n")
+ f.write(" _p = struct_cls()\n")
+ f.write(" for k, v in param_defaults:\n")
+ f.write(" setattr(_p, k, v)\n")
+ f.write(
+ " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), ctypes.byref(_p)]))\n"
+ )
+ f.write(" else:\n")
+ f.write(
+ " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), None]))\n"
+ )
+ f.write(" if ret != 0:\n")
+ f.write(
+ ' raise RuntimeError(f"ts_{name} failed with return code {ret}")\n'
+ )
+ f.write(" return y[0] if y_n == 1 else tuple(y)\n")
+ f.write(" finally:\n")
+ f.write(" _clr()\n\n\n")
+
+ # --- Helper for param override ---
+ f.write(
+ "def _dispatch_with_params(name, x_n, y_n, struct_cls, _p, x_arrays):\n"
+ )
+ f.write(' """Call indicator with a pre-built params struct."""\n')
+ f.write(" if len(x_arrays) != x_n:\n")
+ f.write(
+ ' raise TypeError(f"Expected {x_n} input arrays, got {len(x_arrays)}")\n'
+ )
+ f.write(" n = x_arrays[0].shape[0] if x_n > 0 else 0\n")
+ f.write(" y = [np.zeros(n, dtype=np.float64) for _ in range(y_n)]\n")
+ f.write(" try:\n")
+ f.write(" x_bufs = [_cp(x_arrays[i]) for i in range(x_n)]\n")
+ f.write(
+ " y_ptrs = [y[i].ctypes.data_as(ctypes.POINTER(ctypes.c_double)) for i in range(y_n)]\n"
+ )
+ f.write(" c_func = getattr(_lib, f'ts_{name}_flat')\n")
+ f.write(
+ " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), ctypes.byref(_p)]))\n"
+ )
+ f.write(" if ret != 0:\n")
+ f.write(
+ ' raise RuntimeError(f"ts_{name} failed with return code {ret}")\n'
+ )
+ f.write(" return y[0] if y_n == 1 else tuple(y)\n")
+ f.write(" finally:\n")
+ f.write(" _clr()\n\n\n")
+
+ # --- Dynamic attribute access ---
+ f.write("# --- Dynamic attribute access ---\n")
+ f.write("_FUNC_MAP = {m[0]: m for m in FUNC_META}\n\n\n")
+
+ f.write("def __getattr__(name):\n")
+ f.write(' """Return a callable wrapper for indicator `name`."""\n')
+ f.write(" if name.startswith('_'):\n")
+ f.write(
+ ' raise AttributeError(f"module {__name__!r} has no attribute {name!r}")\n'
+ )
+ f.write(" meta = _FUNC_MAP.get(name)\n")
+ f.write(" if meta is None:\n")
+ f.write(
+ ' raise AttributeError(f"{name!r} is not a known ts indicator")\n'
+ )
+ f.write(" name_s, x_n, y_n, struct_cls, param_defaults = meta\n")
+ f.write(" def wrapper(*x_arrays, **kw):\n")
+ f.write(" for a in x_arrays:\n")
+ f.write(" if not isinstance(a, np.ndarray):\n")
+ f.write(
+ ' raise TypeError(f"Expected numpy.ndarray, got {type(a).__name__}")\n'
+ )
+ f.write(" if struct_cls is not None and kw:\n")
+ f.write(" _p = struct_cls()\n")
+ f.write(" for k, v in kw.items():\n")
+ f.write(" if hasattr(_p, k):\n")
+ f.write(" setattr(_p, k, v)\n")
+ f.write(" else:\n")
+ f.write(
+ ' raise TypeError(f"Unknown parameter {k!r} for {name_s}")\n'
+ )
+ f.write(
+ " return _dispatch_with_params(name_s, x_n, y_n, struct_cls, _p, x_arrays)\n"
+ )
+ f.write(" return _dispatch(meta, *x_arrays)\n")
+ f.write(" return wrapper\n\n\n")
+
+ # --- __all__ ---
+ f.write("__all__ = [m[0] for m in FUNC_META]\n")
+
+ print(f"Generated {len(funcs)} entries -> {OUT}")
+ py_compile.compile(OUT, doraise=True)
+
+
+if __name__ == "__main__":
+ main()
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-12 16:58:46 +0000