otko/tools/gidopensees_import/codegen.py
smillmorel 612936a00b
Some checks failed
CI / lint (push) Has been cancelled
CI / test (macos-latest, 3.10) (push) Has been cancelled
CI / test (macos-latest, 3.11) (push) Has been cancelled
CI / test (macos-latest, 3.12) (push) Has been cancelled
CI / test (ubuntu-latest, 3.10) (push) Has been cancelled
CI / test (ubuntu-latest, 3.11) (push) Has been cancelled
CI / test (ubuntu-latest, 3.12) (push) Has been cancelled
CI / test (windows-latest, 3.10) (push) Has been cancelled
CI / test (windows-latest, 3.11) (push) Has been cancelled
CI / test (windows-latest, 3.12) (push) Has been cancelled
feat: initial otko import
2026-09-08 02:12:15 -04:00

481 lines
16 KiB
Python

"""
Codegen: gidopensees schemas.json -> Pydantic v2 model stubs.
CLI:
python -m tools.gidopensees_import.codegen \\
--schemas tools/gidopensees_import/schemas.json \\
--out src/otko/core/catalog/generated/
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ---------------------------------------------------------------------------
# Name helpers
# ---------------------------------------------------------------------------
def _to_snake(name: str) -> str:
"""Convert a GiD name to a valid Python snake_case identifier."""
s = re.sub(r"[^a-zA-Z0-9]", "_", name)
s = re.sub(r"_+", "_", s).strip("_")
return s.lower() if s else "field"
def _to_class_name(name: str) -> str:
"""Convert a GiD entry name to PascalCaseSpec."""
s = re.sub(r"[^a-zA-Z0-9]", "_", name)
return "".join(p.capitalize() for p in s.split("_") if p) + "Spec"
def _to_module_name(name: str) -> str:
"""Convert a GiD entry name to a Python module filename stem."""
s = re.sub(r"[^a-zA-Z0-9]", "_", name)
s = re.sub(r"_+", "_", s).strip("_")
return s.lower()
def _to_book_var(book_name: str) -> str:
"""Convert 'Standard_Uniaxial_Materials' -> 'StandardUniaxialMaterials'."""
s = re.sub(r"[^a-zA-Z0-9]", "_", book_name)
return "".join(p.capitalize() for p in s.split("_") if p)
def _dedupe(names: list[str]) -> list[str]:
"""Add _N suffix to duplicate identifiers, in order."""
seen: dict[str, int] = {}
result: list[str] = []
for name in names:
count = seen.get(name, 0)
result.append(name if count == 0 else f"{name}_{count}")
seen[name] = count + 1
return result
# ---------------------------------------------------------------------------
# Type-inference helpers
# ---------------------------------------------------------------------------
def _scalar_type(default: str) -> tuple[str, str]:
"""Return (python_type_str, repr_default) for a SCALAR field."""
if not default:
return "str", '""'
try:
int(default)
return "int", default
except ValueError:
pass
try:
float(default)
return "float", default
except ValueError:
pass
return "str", repr(default)
# ---------------------------------------------------------------------------
# Python-literal renderer (for model_config json_schema_extra)
# ---------------------------------------------------------------------------
def _py_lit(obj: Any, depth: int = 0) -> str:
"""Render a Python value as a valid Python literal expression."""
pad = " " * depth
inner = " " * (depth + 1)
if isinstance(obj, dict):
if not obj:
return "{}"
items = [f"{inner}{k!r}: {_py_lit(v, depth + 1)}" for k, v in obj.items()]
return "{\n" + ",\n".join(items) + f",\n{pad}}}"
if isinstance(obj, list):
if not obj:
return "[]"
items = [f"{inner}{_py_lit(v, depth + 1)}" for v in obj]
return "[\n" + ",\n".join(items) + f",\n{pad}]"
return repr(obj)
# ---------------------------------------------------------------------------
# File header template
# ---------------------------------------------------------------------------
_HEADER = """\
# ruff: noqa
# AUTOGENERATED - do not edit.
# File: {module_name}.py
# Generated: {timestamp}
# Source: {source_file} -- Book: {book} / {entry_type}: {entry_name}
#
# Schema data from gidopensees (https://github.com/rclab-auth/gidopensees)
# Copyright (C) Reinforced Concrete Laboratory,
# Aristotle University of Thessaloniki (AUTh)
"""
# ---------------------------------------------------------------------------
# Per-entry source generator
# ---------------------------------------------------------------------------
def _entry_source(
entry: dict[str, Any],
module_name: str,
class_name: str,
source_file: str,
timestamp: str,
) -> str:
lines: list[str] = []
# Header block
lines.append(
_HEADER.format(
module_name=module_name,
timestamp=timestamp,
source_file=source_file,
book=entry["book"],
entry_type=entry["source_type"],
entry_name=entry["name"],
)
)
# Visible fields
visible = [f for f in entry["fields"] if f.get("state") != "HIDDEN"]
# Detect which imports are needed
has_literal = any(f["widget_type"] == "CB" and f.get("options") for f in visible)
tuple_fields = [f for f in visible if f["widget_type"] == "TUPLE" and f.get("options")]
# Inner row classes for TUPLE fields
inner_classes: list[tuple[str, list[str]]] = []
seen_inner: set[str] = set()
for tf in tuple_fields:
inner_cls = (
"".join(p.capitalize() for p in _to_snake(tf["name"]).split("_") if p) + "Row"
)
if inner_cls not in seen_inner:
seen_inner.add(inner_cls)
rows: list[str] = [f"class {inner_cls}(BaseModel):"]
for sub in tf.get("options") or []:
rows.append(f" {_to_snake(sub)}: float = 0.0")
inner_classes.append((inner_cls, rows))
# Imports
lines.append("from __future__ import annotations\n\n")
if has_literal:
lines.append("from typing import Literal\n\n")
lines.append("from pydantic import BaseModel, ConfigDict, Field\n\n\n")
# Inner models
for _, cls_rows in inner_classes:
for row in cls_rows:
lines.append(row + "\n")
lines.append("\n\n")
# Collect per-field dependency metadata for model_config
deps: list[dict[str, Any]] = []
for f in visible:
if f.get("dependencies"):
deps.append(
{
"field": _to_snake(f["name"]),
"gid_name": f["name"],
"rules": f["dependencies"],
}
)
# Build json_schema_extra dict
extra: dict[str, Any] = {
"x-gid-name": entry["name"],
"x-book": entry["book"],
}
if entry.get("image"):
extra["x-icon"] = entry["image"]
if entry.get("tkwidgets"):
extra["x-tkwidgets"] = entry["tkwidgets"]
if deps:
extra["dependencies"] = deps
# Class header + model_config
lines.append(f"class {class_name}(BaseModel):\n")
extra_lit = _py_lit(extra, depth=2)
lines.append(f" model_config = ConfigDict(\n json_schema_extra={extra_lit}\n )\n\n")
# Deduplicated Python attribute names
py_names = _dedupe([_to_snake(f["name"]) for f in visible])
for py_name, field in zip(py_names, visible, strict=True):
wtype: str = field["widget_type"]
options: list[str] = field.get("options") or []
default: str = field.get("default") or ""
help_text: str = field.get("help_text") or ""
desc_arg = f", description={help_text!r}" if help_text else ""
if wtype == "CB":
if options:
opts_str = ", ".join(repr(o) for o in options)
py_type = f"Literal[{opts_str}]"
py_def = repr(default) if default in options else repr(options[0])
else:
py_type, py_def = "str", repr(default)
if desc_arg:
lines.append(f" {py_name}: {py_type} = Field({py_def}{desc_arg})\n")
else:
lines.append(f" {py_name}: {py_type} = {py_def}\n")
elif wtype == "UNITS":
py_def = repr(default)
if desc_arg:
lines.append(
f" {py_name}: str = Field({py_def}{desc_arg})"
" # TODO: unit-aware type\n"
)
else:
lines.append(f" {py_name}: str = {py_def} # TODO: unit-aware type\n")
elif wtype == "MAT":
py_def = repr(default)
if desc_arg:
lines.append(f" {py_name}: str = Field({py_def}{desc_arg})\n")
else:
lines.append(f" {py_name}: str = {py_def}\n")
elif wtype == "SCALAR":
py_type, py_def = _scalar_type(default)
if desc_arg:
lines.append(f" {py_name}: {py_type} = Field({py_def}{desc_arg})\n")
else:
lines.append(f" {py_name}: {py_type} = {py_def}\n")
elif wtype == "TUPLE":
inner_cls = (
"".join(p.capitalize() for p in _to_snake(field["name"]).split("_") if p)
+ "Row"
)
lines.append(
f" {py_name}: list[{inner_cls}] = Field(default_factory=list{desc_arg})\n"
)
else:
lines.append(f" {py_name}: str = {default!r}\n")
lines.append("\n")
return "".join(lines)
# ---------------------------------------------------------------------------
# generated/__init__.py builder
# ---------------------------------------------------------------------------
def _generated_init_source(
books: list[dict[str, Any]],
source: str,
module_prefix: str,
timestamp: str,
) -> str:
lines: list[str] = []
lines.append("# ruff: noqa\n")
lines.append(f"# AUTOGENERATED - do not edit. Generated: {timestamp}\n")
lines.append(
"# Schema data from gidopensees.\n\n"
)
lines.append("from __future__ import annotations\n\n")
lines.append("from typing import Union\n\n")
# Collect (module_stem, class_name) per book
book_entries: list[tuple[str, list[tuple[str, str]]]] = []
for book in books:
pairs: list[tuple[str, str]] = []
for entry in book["entries"]:
stem = _to_module_name(entry["name"])
cls = _to_class_name(entry["name"])
pairs.append((stem, cls))
book_entries.append((book["name"], pairs))
# Import all
for _, pairs in book_entries:
for stem, cls in pairs:
lines.append(f"from {module_prefix}.{stem} import {cls}\n")
lines.append("\n\n")
# One Union per book
for book_name, pairs in book_entries:
var_name = _to_book_var(book_name)
union_members = ", ".join(cls for _, cls in pairs)
if len(pairs) == 1:
lines.append(f"{var_name} = {pairs[0][1]}\n")
else:
lines.append(f"{var_name} = Union[{union_members}]\n")
lines.append("\n")
# __all__
all_names = [_to_book_var(b["name"]) for b in books]
all_cls = [cls for _, pairs in book_entries for _, cls in pairs]
all_items = ", ".join(repr(n) for n in all_cls + all_names)
lines.append(f"__all__ = [{all_items}]\n")
return "".join(lines)
# ---------------------------------------------------------------------------
# core/catalog/__init__.py builder
# ---------------------------------------------------------------------------
def _catalog_init_source(
mat_books: list[dict[str, Any]],
timestamp: str,
) -> str:
lines: list[str] = []
lines.append("# ruff: noqa\n")
lines.append(f"# AUTOGENERATED - do not edit. Generated: {timestamp}\n")
lines.append("# Flat CATALOG mapping gidopensees material name -> Pydantic Spec class.\n\n")
lines.append("from __future__ import annotations\n\n")
lines.append("from pydantic import BaseModel\n\n")
base_pkg = "otko.core.catalog.generated"
entries: list[tuple[str, str, str]] = [] # (gid_name, stem, class_name)
for book in mat_books:
for entry in book["entries"]:
gid_name = entry["name"]
stem = _to_module_name(gid_name)
cls = _to_class_name(gid_name)
entries.append((gid_name, stem, cls))
for _gid_name, stem, cls in entries:
lines.append(f"from {base_pkg}.{stem} import {cls}\n")
lines.append("\n\n")
lines.append("CATALOG: dict[str, type[BaseModel]] = {\n")
for gid_name, _, cls in entries:
lines.append(f" {gid_name!r}: {cls},\n")
lines.append("}\n\n")
all_items = ", ".join(repr(cls) for _, _, cls in entries)
lines.append(f'__all__ = ["CATALOG", {all_items}]\n')
return "".join(lines)
# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------
def run_codegen(schemas_path: Path, out_dir: Path) -> None:
data: dict[str, Any] = json.loads(schemas_path.read_text(encoding="utf-8"))
mat_books: list[dict[str, Any]] = data["mat_books"]
cnd_books: list[dict[str, Any]] = data["cnd_books"]
timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
source_file = schemas_path.name
# Ensure output directories
cnd_dir = out_dir / "conditions"
out_dir.mkdir(parents=True, exist_ok=True)
cnd_dir.mkdir(parents=True, exist_ok=True)
mat_count = 0
cnd_count = 0
# --- Material stubs ---
for book in mat_books:
for entry in book["entries"]:
stem = _to_module_name(entry["name"])
cls = _to_class_name(entry["name"])
src = _entry_source(
entry=entry,
module_name=stem,
class_name=cls,
source_file=source_file,
timestamp=timestamp,
)
(out_dir / f"{stem}.py").write_text(src, encoding="utf-8")
mat_count += 1
# --- Condition stubs ---
for book in cnd_books:
for entry in book["entries"]:
stem = _to_module_name(entry["name"])
cls = _to_class_name(entry["name"])
src = _entry_source(
entry=entry,
module_name=stem,
class_name=cls,
source_file=source_file,
timestamp=timestamp,
)
(cnd_dir / f"{stem}.py").write_text(src, encoding="utf-8")
cnd_count += 1
# --- generated/__init__.py ---
mat_pkg = "otko.core.catalog.generated"
gen_init = _generated_init_source(mat_books, "mat", mat_pkg, timestamp)
(out_dir / "__init__.py").write_text(gen_init, encoding="utf-8")
# --- generated/conditions/__init__.py ---
cnd_pkg = "otko.core.catalog.generated.conditions"
cnd_init = _generated_init_source(cnd_books, "cnd", cnd_pkg, timestamp)
(cnd_dir / "__init__.py").write_text(cnd_init, encoding="utf-8")
# --- core/catalog/__init__.py (parent of generated/) ---
catalog_dir = out_dir.parent
cat_init = _catalog_init_source(mat_books, timestamp)
(catalog_dir / "__init__.py").write_text(cat_init, encoding="utf-8")
# --- core/catalog/curated/__init__.py (empty placeholder) ---
curated_dir = catalog_dir / "curated"
curated_dir.mkdir(exist_ok=True)
curated_init = curated_dir / "__init__.py"
if not curated_init.exists():
curated_init.write_text("", encoding="utf-8")
print(f"Generated {mat_count} material stubs -> {out_dir}")
print(f"Generated {cnd_count} condition stubs -> {cnd_dir}")
print(f"Written -> {out_dir / '__init__.py'}")
print(f"Written -> {cnd_dir / '__init__.py'}")
print(f"Written -> {catalog_dir / '__init__.py'}")
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _cli(argv: list[str] | None = None) -> None:
parser = argparse.ArgumentParser(
description="Generate Pydantic v2 stubs from gidopensees schemas.json"
)
parser.add_argument(
"--schemas",
required=True,
type=Path,
help="Path to schemas.json (e.g. tools/gidopensees_import/schemas.json)",
)
parser.add_argument(
"--out",
required=True,
type=Path,
help="Output directory for generated stubs (e.g. src/otko/core/catalog/generated/)",
)
args = parser.parse_args(argv)
schemas_path: Path = args.schemas
out_dir: Path = args.out
if not schemas_path.exists():
print(f"error: schemas file not found: {schemas_path}", file=sys.stderr)
sys.exit(1)
run_codegen(schemas_path, out_dir)
if __name__ == "__main__":
_cli()