1a0f9b5f86
write_jupyter_launcher ahora exporta IPYTHONDIR al directorio local .ipython/ para que el kernel cargue el startup correcto cuando se ejecuta desde projects/. write_jupyter_registry_kernel usa descubrimiento inteligente de FN_REGISTRY_ROOT: prioriza env var, luego path hardcoded, luego sube desde CWD buscando registry.db. Esto permite que analyses dentro de projects/ encuentren el registry automaticamente.
129 lines
4.7 KiB
Bash
129 lines
4.7 KiB
Bash
# write_jupyter_registry_kernel
|
|
# -------------------------------
|
|
# Genera un script de startup de IPython que autoconfigura el acceso
|
|
# al fn_registry en cada notebook: FN_REGISTRY_ROOT, sys.path a
|
|
# python/functions, y un helper fn_query() para consultar registry.db.
|
|
#
|
|
# USO (sourced):
|
|
# source write_jupyter_registry_kernel.sh
|
|
# write_jupyter_registry_kernel /path/to/project
|
|
|
|
write_jupyter_registry_kernel() {
|
|
local project_dir="${1:-.}"
|
|
local startup_dir="${project_dir}/.ipython/profile_default/startup"
|
|
local registry_root
|
|
registry_root="$(cd "$project_dir" && cd "$(git -C "$project_dir" rev-parse --show-toplevel 2>/dev/null || echo "../..")" && pwd)"
|
|
|
|
# Fallback: si no es git, buscar registry.db subiendo directorios
|
|
if [ ! -f "$registry_root/registry.db" ] && [ -f "$project_dir/../../registry.db" ]; then
|
|
registry_root="$(cd "$project_dir/../.." && pwd)"
|
|
fi
|
|
|
|
mkdir -p "$startup_dir"
|
|
|
|
cat > "${startup_dir}/00_fn_registry.py" << STARTUP
|
|
"""
|
|
fn_registry kernel startup
|
|
Autoconfigura acceso al registry en cada notebook.
|
|
Generado por write_jupyter_registry_kernel (fn_registry).
|
|
"""
|
|
import os
|
|
import sys
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
# ── FN_REGISTRY_ROOT ────────────────────────────────────────
|
|
# Prioridad: env var > path hardcoded > descubrimiento automatico
|
|
def _discover_registry_root():
|
|
if os.environ.get("FN_REGISTRY_ROOT"):
|
|
return Path(os.environ["FN_REGISTRY_ROOT"]).resolve()
|
|
hardcoded = Path("${registry_root}")
|
|
if (hardcoded / "registry.db").exists():
|
|
return hardcoded
|
|
# Subir desde CWD hasta encontrar registry.db
|
|
p = Path.cwd()
|
|
for _ in range(10):
|
|
if (p / "registry.db").exists():
|
|
return p
|
|
if p.parent == p:
|
|
break
|
|
p = p.parent
|
|
return hardcoded
|
|
|
|
FN_REGISTRY_ROOT = _discover_registry_root()
|
|
os.environ["FN_REGISTRY_ROOT"] = str(FN_REGISTRY_ROOT)
|
|
|
|
# ── sys.path: importar funciones Python del registry ────────
|
|
_python_functions = FN_REGISTRY_ROOT / "python" / "functions"
|
|
for _domain in sorted(_python_functions.iterdir()) if _python_functions.exists() else []:
|
|
if _domain.is_dir() and not _domain.name.startswith("_"):
|
|
_path = str(_domain)
|
|
if _path not in sys.path:
|
|
sys.path.insert(0, _path)
|
|
|
|
# Tambien el directorio padre para imports por dominio: from core import filter_list
|
|
_pf = str(_python_functions)
|
|
if _pf not in sys.path:
|
|
sys.path.insert(0, _pf)
|
|
|
|
# ── fn_query: consultar registry.db desde el notebook ───────
|
|
_REGISTRY_DB = FN_REGISTRY_ROOT / "registry.db"
|
|
|
|
def fn_query(sql, params=()):
|
|
"""Ejecuta una consulta SQL sobre registry.db y retorna las filas.
|
|
|
|
Ejemplos:
|
|
fn_query("SELECT id, description FROM functions WHERE domain = ?", ("finance",))
|
|
fn_query("SELECT id FROM functions_fts WHERE functions_fts MATCH ?", ("slice*",))
|
|
"""
|
|
if not _REGISTRY_DB.exists():
|
|
raise FileNotFoundError(f"registry.db no encontrado en {_REGISTRY_DB}")
|
|
con = sqlite3.connect(str(_REGISTRY_DB))
|
|
con.row_factory = sqlite3.Row
|
|
try:
|
|
rows = con.execute(sql, params).fetchall()
|
|
return [dict(r) for r in rows]
|
|
finally:
|
|
con.close()
|
|
|
|
def fn_search(term):
|
|
"""Busca funciones y tipos en el registry por nombre o descripcion.
|
|
|
|
Ejemplo:
|
|
fn_search("slice")
|
|
fn_search("finance")
|
|
"""
|
|
fts_term = f"name:{term}* OR description:{term}*"
|
|
functions = fn_query(
|
|
"SELECT id, kind, purity, lang, description FROM functions "
|
|
"WHERE id IN (SELECT id FROM functions_fts WHERE functions_fts MATCH ?) "
|
|
"ORDER BY name", (fts_term,)
|
|
)
|
|
types = fn_query(
|
|
"SELECT id, algebraic, lang, description FROM types "
|
|
"WHERE id IN (SELECT id FROM types_fts WHERE types_fts MATCH ?) "
|
|
"ORDER BY name", (fts_term,)
|
|
)
|
|
return {"functions": functions, "types": types}
|
|
|
|
def fn_code(function_id):
|
|
"""Retorna el codigo fuente de una funcion del registry.
|
|
|
|
Ejemplo:
|
|
print(fn_code("filter_list_py_core"))
|
|
"""
|
|
rows = fn_query("SELECT code FROM functions WHERE id = ?", (function_id,))
|
|
if not rows:
|
|
raise KeyError(f"Funcion no encontrada: {function_id}")
|
|
return rows[0]["code"]
|
|
|
|
# ── Mensaje de bienvenida ───────────────────────────────────
|
|
print(f"fn_registry conectado: {FN_REGISTRY_ROOT}")
|
|
print(f" registry.db: {'OK' if _REGISTRY_DB.exists() else 'NO ENCONTRADO'}")
|
|
print(f" Python functions: {_pf}")
|
|
print(f" Helpers: fn_query(), fn_search(), fn_code()")
|
|
STARTUP
|
|
|
|
echo "${startup_dir}/00_fn_registry.py"
|
|
}
|