Initial import

agentui/pipeline/storage.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List
+import json
+from agentui.pipeline.defaults import default_pipeline
+
+
+PIPELINE_FILE = Path("pipeline.json")
+PRESETS_DIR = Path("presets")
+
+
+def load_pipeline() -> Dict[str, Any]:
+    if PIPELINE_FILE.exists():
+        try:
+            return json.loads(PIPELINE_FILE.read_text(encoding="utf-8"))
+        except Exception:
+            pass
+    return default_pipeline()
+
+
+def save_pipeline(pipeline: Dict[str, Any]) -> None:
+    PIPELINE_FILE.write_text(json.dumps(pipeline, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def list_presets() -> List[str]:
+    PRESETS_DIR.mkdir(parents=True, exist_ok=True)
+    return sorted([p.stem for p in PRESETS_DIR.glob("*.json")])
+
+
+def load_preset(name: str) -> Dict[str, Any]:
+    PRESETS_DIR.mkdir(parents=True, exist_ok=True)
+    path = PRESETS_DIR / f"{name}.json"
+    if not path.exists():
+        raise FileNotFoundError(name)
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def save_preset(name: str, pipeline: Dict[str, Any]) -> None:
+    PRESETS_DIR.mkdir(parents=True, exist_ok=True)
+    path = PRESETS_DIR / f"{name}.json"
+    path.write_text(json.dumps(pipeline, ensure_ascii=False, indent=2), encoding="utf-8")
+
+