summaryrefslogtreecommitdiffstats
path: root/scripts
diff options
context:
space:
mode:
Diffstat (limited to 'scripts')
-rw-r--r--scripts/gen_all.py135
-rw-r--r--scripts/gen_bindings.py274
-rw-r--r--scripts/gen_flat.py96
-rw-r--r--scripts/gen_fuzz.py74
-rw-r--r--scripts/gen_registry.py103
-rw-r--r--scripts/ts_parser.py159
6 files changed, 841 insertions, 0 deletions
diff --git a/scripts/gen_all.py b/scripts/gen_all.py
new file mode 100644
index 0000000..dd9d395
--- /dev/null
+++ b/scripts/gen_all.py
@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Generate ts_generated.h with all indicator declarations and test registry.
+
+Parses lib/*.c for ts_parameter_t declarations and emits:
+- extern ts_parameter_t const ts_<name>_p
+- uint_t ts_<name>_offset(...)
+- int ts_<name>(...)
+- bool ts_test_<name>(void) [inside #ifdef TS_IMPLEMENTATION]
+"""
+
+import importlib.util
+import os
+import re
+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__)))
+TEST_DIR = os.path.join(ROOT, "test")
+OUT_H = os.path.join(ROOT, "ts_generated.h")
+OUT_TEST = os.path.join(TEST_DIR, "main_generated.c")
+
+
+def generate_header(funcs):
+ """Generate ts_generated.h."""
+ lines = []
+ lines.append("/* Auto-generated indicator declarations. Do not edit. */")
+ lines.append("")
+ lines.append("#ifndef TS_GENERATED_H")
+ lines.append("#define TS_GENERATED_H")
+ lines.append("")
+
+ for fm in funcs:
+ lines.append("extern ts_parameter_t const ts_" + fm.name + "_p;")
+ lines.append("uint_t ts_" + fm.name + "_offset( ts_constant_t const * );")
+ lines.append(
+ "int ts_"
+ + fm.name
+ + "( fp_t * const *y, uint_t y_n, fp_t const * const *x, "
+ "uint_t x_n, uint_t n, ts_constant_t const *c );"
+ )
+
+ lines.append("")
+ lines.append("#ifdef TS_IMPLEMENTATION")
+ lines.append("")
+ for fm in funcs:
+ lines.append("bool ts_test_" + fm.name + "( void );")
+ lines.append("")
+ lines.append("#endif /* TS_IMPLEMENTATION */")
+ lines.append("")
+ lines.append("#endif /* TS_GENERATED_H */")
+ lines.append("")
+
+ return "\n".join(lines)
+
+
+def generate_test_main(funcs):
+ """Generate test/main_generated.c with extern declarations and test array."""
+ lines = []
+ lines.append("/* Auto-generated test registry. Do not edit. */")
+ lines.append("")
+ lines.append("#define EXTERN_TEST(name) extern bool ts_test_##name(void)")
+ for fm in funcs:
+ lines.append("EXTERN_TEST(" + fm.name + ");")
+ lines.append("#undef EXTERN_TEST")
+ lines.append("")
+ lines.append("typedef struct {")
+ lines.append(" const char *name;")
+ lines.append(" bool (*func)(void);")
+ lines.append("} test_case_t;")
+ lines.append("")
+ lines.append("#define TEST_ENTRY(name) { #name, ts_test_##name }")
+ lines.append("")
+ lines.append("static test_case_t tests[] = {")
+ for fm in funcs:
+ lines.append(" TEST_ENTRY(" + fm.name + "),")
+ lines.append("};")
+ lines.append("")
+ lines.append("#define NUM_TESTS (sizeof(tests) / sizeof(tests[0]))")
+ lines.append("")
+ lines.append("#undef TEST_ENTRY")
+ lines.append("")
+
+ return "\n".join(lines)
+
+
+def _is_valid_c_identifier(name):
+ return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", name))
+
+
+def main():
+ funcs = parse_all(detect_x_access=False)
+
+ if not funcs:
+ print("Warning: no ts_parameter declarations found.", file=sys.stderr)
+ return 1
+
+ for fm in funcs:
+ if not _is_valid_c_identifier(fm.name):
+ print(f"Error: invalid C identifier '{fm.name}'", file=sys.stderr)
+ return 1
+
+ os.makedirs(TEST_DIR, exist_ok=True)
+
+ try:
+ with open(OUT_H, "w", encoding="utf-8") as f:
+ f.write(generate_header(funcs))
+ except OSError as e:
+ print(f"Error writing {OUT_H}: {e}", file=sys.stderr)
+ return 1
+
+ try:
+ with open(OUT_TEST, "w", encoding="utf-8") as f:
+ f.write(generate_test_main(funcs))
+ except OSError as e:
+ print(f"Error writing {OUT_TEST}: {e}", file=sys.stderr)
+ return 1
+
+ print(f"Generated {len(funcs)} declarations -> {OUT_H}")
+ print(f"Generated test registry -> {OUT_TEST}")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
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()
diff --git a/scripts/gen_flat.py b/scripts/gen_flat.py
new file mode 100644
index 0000000..c659432
--- /dev/null
+++ b/scripts/gen_flat.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Generate flat-array C wrappers for all ts functions.
+
+Each wrapper takes flat input/output arrays instead of pointer-of-pointers,
+eliminating ctypes pointer marshaling overhead.
+"""
+
+import argparse
+import importlib.util
+import os
+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, "lib", "ts_flat_generated.c")
+
+
+def generate_wrappers(funcs, dry_run=False):
+ """Generate flat-array C wrapper functions."""
+ if dry_run:
+ print(f"[dry-run] Would generate {len(funcs)} flat wrappers -> {OUT}")
+ return
+
+ with open(OUT, "w", encoding="utf-8") as f:
+ f.write("/* Auto-generated flat-array wrappers. Do not edit. */\n\n")
+ f.write("#include <string.h>\n")
+ f.write('#include "ts.h"\n\n')
+
+ for fm in funcs:
+ safe_x_n = max(fm.x_n, 1)
+ safe_y_n = max(fm.y_n, 1)
+
+ f.write(f"/* {fm.name}: x_n={fm.x_n}, y_n={fm.y_n}, c_n={fm.c_n} */\n")
+ f.write(f"int ts_{fm.name}_flat(\n")
+
+ for i in range(fm.y_n):
+ f.write(f" fp_t *y{i},\n")
+
+ for i in range(fm.x_n):
+ f.write(f" fp_t const *x{i},\n")
+
+ f.write(f" uint_t n,\n")
+ f.write(f" ts_constant_t const *params\n")
+ f.write(f")\n{{\n")
+
+ f.write(f" fp_t * const _y[{safe_y_n}] = {{")
+ f.write(", ".join(f"y{i}" for i in range(fm.y_n)))
+ f.write("};\n")
+
+ f.write(f" fp_t const * const _x[{safe_x_n}] = {{")
+ f.write(", ".join(f"x{i}" for i in range(fm.x_n)))
+ f.write("};\n")
+
+ for i in range(fm.y_n):
+ f.write(f" memset(y{i}, 0, n * sizeof(fp_t));\n")
+
+ f.write(
+ f" return ts_{fm.name}(_y, {fm.y_n}, _x, {fm.x_n}, n, params);\n"
+ )
+ f.write("}\n\n")
+
+
+def main():
+ parser = argparse.ArgumentParser(description="Generate flat-array C wrappers")
+ parser.add_argument(
+ "--dry-run",
+ action="store_true",
+ help="Print what would be generated without writing",
+ )
+ args = parser.parse_args()
+
+ funcs = parse_all(detect_x_access=True)
+
+ if not funcs:
+ print("Warning: no ts_parameter declarations found.", file=sys.stderr)
+ sys.exit(1)
+
+ generate_wrappers(funcs, dry_run=args.dry_run)
+
+ if not args.dry_run:
+ print(f"Generated {len(funcs)} flat wrappers -> {OUT}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/gen_fuzz.py b/scripts/gen_fuzz.py
new file mode 100644
index 0000000..a1d20e3
--- /dev/null
+++ b/scripts/gen_fuzz.py
@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Generate fuzz table entries from ts indicator metadata.
+
+Parses lib/*.c via ts_parser to extract function names and input/output
+counts, then generates a C source file with the complete fuzz table.
+"""
+
+import argparse
+import importlib.util
+import os
+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)
+
+OUT = os.path.join(
+ os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ "test",
+ "fuzz_table_generated.c",
+)
+
+
+def generate_table(funcs, dry_run=False):
+ """Generate the fuzz table C source file."""
+ if dry_run:
+ print(f"[dry-run] Would generate {len(funcs)} fuzz entries -> {OUT}")
+ return
+
+ with open(OUT, "w", encoding="utf-8") as f:
+ f.write("/* Auto-generated fuzz table. Do not edit. */\n\n")
+ f.write('#include "ts.h"\n\n')
+ f.write("static const fuzz_entry_t _fuzz_table_generated[] = {\n")
+
+ for fm in funcs:
+ f.write(f" {{ts_{fm.name}, {fm.x_n}, {fm.y_n}}},\n")
+
+ f.write("};\n\n")
+ f.write(
+ "static const uint_t _fuzz_generated_n = sizeof(_fuzz_table_generated) / sizeof(_fuzz_table_generated[0]);\n"
+ )
+
+
+def main():
+ parser = argparse.ArgumentParser(description="Generate fuzz table from registry")
+ parser.add_argument(
+ "--dry-run",
+ action="store_true",
+ help="Print what would be generated without writing",
+ )
+ args = parser.parse_args()
+
+ funcs = parse_all(detect_x_access=True)
+
+ if not funcs:
+ print("Warning: no ts_parameter declarations found.", file=sys.stderr)
+ sys.exit(1)
+
+ generate_table(funcs, dry_run=args.dry_run)
+
+ if not args.dry_run:
+ print(f"Generated {len(funcs)} fuzz entries -> {OUT}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/gen_registry.py b/scripts/gen_registry.py
new file mode 100644
index 0000000..e80759e
--- /dev/null
+++ b/scripts/gen_registry.py
@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""Generate ts_registry_generated.c with registry array and getter functions."""
+
+import importlib.util
+import os
+import re
+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, "lib", "ts_registry_generated.c")
+
+
+def _resolve_params_size(params_size, name):
+ """Resolve params_size to a compile-time constant.
+
+ If params_size is '0', keep it. If it's sizeof(ts_X_params_t), check
+ if the struct is visible in ts.h. If not, compute size from definition.
+ """
+ if params_size == "0":
+ return "0"
+
+ header_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "ts.h")
+ try:
+ with open(header_path, encoding="utf-8") as hf:
+ hcontent = hf.read()
+ if re.search(r"ts_" + re.escape(name) + r"_params_t", hcontent):
+ return params_size
+ except FileNotFoundError:
+ pass
+ # Struct not in header, compute size from definition
+ import glob
+
+ lib_dir = os.path.join(ROOT, "lib")
+ for cfile in glob.glob(os.path.join(lib_dir, "*.c")):
+ try:
+ with open(cfile, encoding="utf-8") as f:
+ content = f.read()
+ except OSError:
+ continue
+ struct_m = re.search(
+ r"typedef\s+struct\s*\{([^}]+)\}\s+ts_" + re.escape(name) + r"_params_t",
+ content,
+ )
+ if struct_m:
+ body = struct_m.group(1)
+ n_fp = len(re.findall(r"fp_t\s+", body))
+ n_uint = len(re.findall(r"uint_t\s+", body))
+ return str(n_fp * 8 + n_uint * 8)
+ return "0"
+
+
+def main():
+ funcs = parse_all(detect_x_access=False)
+
+ if not funcs:
+ print("Warning: no ts_parameter declarations found.", file=sys.stderr)
+ sys.exit(1)
+
+ with open(OUT, "w", encoding="utf-8") as f:
+ f.write("/* Auto-generated function registry. Do not edit. */\n\n")
+ f.write('#include "ts.h"\n\n')
+ f.write(f"static const uint_t _ts_fn_count = {len(funcs)};\n\n")
+ f.write("static const ts_fn_info_t _ts_registry[] = {\n")
+ for fm in funcs:
+ ps = _resolve_params_size(fm.params_size, fm.name)
+ f.write(
+ f" {{ .n_inputs = {fm.x_n}, .n_outputs = {fm.y_n}, "
+ f".n_constants = {fm.c_n}, .params_size = {ps} }},\n"
+ )
+ f.write("};\n\n")
+ f.write("static const char * const _ts_fn_names[] = {\n")
+ for fm in funcs:
+ f.write(f' "{fm.name}",\n')
+ f.write("};\n\n")
+
+ # Getter functions
+ f.write("uint_t ts_fn_count( void ) { return _ts_fn_count; }\n\n")
+ f.write("const ts_fn_info_t* ts_registry_get( uint_t i )\n")
+ f.write("{\n")
+ f.write(" return i < _ts_fn_count ? &_ts_registry[i] : NULL;\n")
+ f.write("}\n\n")
+ f.write("const char* ts_fn_name_get( uint_t i )\n")
+ f.write("{\n")
+ f.write(" return i < _ts_fn_count ? _ts_fn_names[i] : NULL;\n")
+ f.write("}\n")
+
+ print(f"Generated {len(funcs)} entries -> {OUT}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/ts_parser.py b/scripts/ts_parser.py
new file mode 100644
index 0000000..2ad3629
--- /dev/null
+++ b/scripts/ts_parser.py
@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""Shared parser for ts indicator metadata.
+
+Parses lib/*.c for ts_parameter_t declarations and extracts:
+- name, n_inputs, n_outputs, n_constants
+- Actual x[N] access count (may exceed declared n_inputs)
+- Parameter struct fields
+
+All generators should import this instead of duplicating regex logic.
+"""
+
+import glob
+import os
+import re
+import warnings
+
+LIB_DIR = os.path.join(
+ os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "lib"
+)
+
+SKIP_PREFIXES = ("ts_flat", "ts_registry", "gen_", "update_")
+SKIP_NAMES = {"test"}
+
+PY_RESERVED_NAMES = {"lambda": "lambda_", "type": "type_", "id": "id_"}
+
+VOLUME_INDEX = 4
+OHLCV_COUNT = 5
+
+
+class FuncMeta:
+ """Metadata for a single ts indicator function."""
+
+ __slots__ = ("name", "x_n", "y_n", "c_n", "params_size", "struct_fields")
+
+ def __init__(self, name, x_n, y_n, c_n, params_size="0", struct_fields=None):
+ self.name = name
+ self.x_n = x_n
+ self.y_n = y_n
+ self.c_n = c_n
+ self.params_size = params_size
+ self.struct_fields = struct_fields or []
+
+
+def _detect_actual_x_n(content, match_end, name):
+ """Detect actual x[N] access in function body."""
+ remaining = content[match_end:]
+ func_def = re.search(rf"\bts_{name}\s*\([^)]*\)\s*{{", remaining)
+ if not func_def:
+ return 0
+ brace_pos = remaining.find("{", func_def.start())
+ depth = 0
+ func_body = ""
+ for i in range(brace_pos, len(remaining)):
+ if remaining[i] == "{":
+ depth += 1
+ elif remaining[i] == "}":
+ depth -= 1
+ if depth == 0:
+ func_body = remaining[brace_pos : i + 1]
+ break
+ indices = [
+ int(xm.group(1)) for xm in re.finditer(r"x\s*\[\s*(\d+)\s*\]", func_body)
+ ]
+ return max(indices) + 1 if indices else 0
+
+
+def _parse_struct_fields(content, name):
+ """Parse parameter struct fields for a function."""
+ struct_m = re.search(
+ r"typedef\s+struct\s*\{([^}]+)\}\s+ts_" + re.escape(name) + r"_params_t",
+ content,
+ )
+ if not struct_m:
+ return []
+ fields_text = struct_m.group(1)
+ fields_text = re.sub(r"/\*.*?\*/", "", fields_text, flags=re.DOTALL)
+ fields_text = re.sub(r"//.*", "", fields_text)
+ fields = []
+ for fm in re.finditer(r"^\s*(\w+)\s+(\w+)\s*;", fields_text, re.MULTILINE):
+ ctype, c_fname = fm.group(1), fm.group(2)
+ if ctype == "uint_t":
+ pytype = "ctypes.c_uint64"
+ elif ctype == "fp_t":
+ pytype = "ctypes.c_double"
+ else:
+ warnings.warn(
+ f"Unknown type '{ctype}' in struct ts_{name}_params_t, defaulting to c_double"
+ )
+ pytype = "ctypes.c_double"
+ py_fname = PY_RESERVED_NAMES.get(c_fname, c_fname)
+ fields.append((c_fname, py_fname, pytype))
+ return fields
+
+
+def parse_all(detect_x_access=True):
+ """Parse all ts indicator functions from lib/*.c.
+
+ Args:
+ detect_x_access: If True, detect actual x[N] access in function body
+ and adjust x_n accordingly.
+
+ Returns:
+ List of FuncMeta objects.
+ """
+ funcs = []
+ for cfile in sorted(glob.glob(os.path.join(LIB_DIR, "*.c"))):
+ basename = os.path.basename(cfile)
+ stem = basename.replace(".c", "")
+ if any(stem.startswith(p) for p in SKIP_PREFIXES) or stem in SKIP_NAMES:
+ continue
+ try:
+ with open(cfile, encoding="utf-8") as f:
+ content = f.read()
+ except OSError as e:
+ warnings.warn(f"Could not read {cfile}: {e}")
+ continue
+
+ for m in re.finditer(
+ r"ts_parameter_t\s+const\s+ts_(\w+)_p\s*=\s*\{\s*"
+ r"\.n_inputs\s*=\s*(\d+)\s*,\s*"
+ r"\.n_outputs\s*=\s*(\d+)\s*,\s*"
+ r"\.n_constants\s*=\s*(\d+)"
+ r"(?:\s*,\s*\.params_size\s*=\s*(\S+?))?\s*\}",
+ content,
+ ):
+ name = m.group(1)
+ x_n = int(m.group(2))
+ y_n = int(m.group(3))
+ c_n = int(m.group(4))
+ params_size = m.group(5) or "0"
+
+ if detect_x_access:
+ actual = _detect_actual_x_n(content, m.end(), name)
+ if actual > x_n:
+ x_n = actual
+ if x_n < OHLCV_COUNT and re.search(
+ r"\bx\s*\[\s*" + str(VOLUME_INDEX) + r"\s*\]",
+ content[m.end() :],
+ ):
+ x_n = OHLCV_COUNT
+
+ struct_fields = _parse_struct_fields(content, name) if c_n > 0 else []
+
+ funcs.append(FuncMeta(name, x_n, y_n, c_n, params_size, struct_fields))
+
+ return funcs
+
+
+def parse_names_only():
+ """Parse function names and basic metadata (no x access detection).
+
+ Returns:
+ Dict mapping name -> {"x_n": int, "y_n": int}.
+ """
+ meta = {}
+ for fm in parse_all(detect_x_access=True):
+ if fm.name not in meta:
+ meta[fm.name] = {"x_n": fm.x_n, "y_n": fm.y_n}
+ return meta
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-12 23:18:28 +0000