migrated core_nest to mainroom

hub/dataloader/__init__.py

@@ -0,0 +1,186 @@
+"""
+Soleprint Data Loader
+
+Loads JSON data files and provides typed access via Pydantic models.
+JSON files live in data/ directory (content only, no code).
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import List, Optional
+
+# When symlinked, __file__ resolves to actual location (hub/dataloader)
+# but we need to find models/ in the runtime directory (gen/)
+# Use cwd as the base since we always run from gen/
+_runtime_dir = Path.cwd()
+_file_parent = Path(__file__).resolve().parent.parent
+
+# Try runtime dir first (gen/), then fall back to file's parent (hub/)
+if (_runtime_dir / "models").exists():
+    sys.path.insert(0, str(_runtime_dir))
+else:
+    sys.path.insert(0, str(_file_parent))
+
+from models.pydantic import (
+    Book,
+    BookCollection,
+    Depot,
+    DepotCollection,
+    Desk,
+    DeskCollection,
+    Pulse,
+    PulseCollection,
+    Room,
+    RoomCollection,
+    Status,
+    Template,
+    TemplateCollection,
+    Tool,
+    ToolCollection,
+    Vein,
+    VeinCollection,
+)
+
+# Data directory - try runtime dir first, then file's parent
+_default_data = (
+    _runtime_dir / "data" if (_runtime_dir / "data").exists() else _file_parent / "data"
+)
+DATA_DIR = Path(os.getenv("SOLEPRINT_DATA_DIR", _default_data)).resolve()
+
+
+def _load_json(filename: str) -> dict:
+    """Load a JSON file from the data directory."""
+    filepath = DATA_DIR / filename
+    if filepath.exists():
+        with open(filepath) as f:
+            return json.load(f)
+    return {"items": []}
+
+
+def _save_json(filename: str, data: dict):
+    """Save data to a JSON file in the data directory."""
+    filepath = DATA_DIR / filename
+    with open(filepath, "w") as f:
+        json.dump(data, f, indent=2)
+
+
+# === Collection Loaders ===
+
+
+def get_veins() -> List[Vein]:
+    data = _load_json("veins.json")
+    return VeinCollection(**data).items
+
+
+def get_rooms() -> List[Room]:
+    data = _load_json("rooms.json")
+    return RoomCollection(**data).items
+
+
+def get_depots() -> List[Depot]:
+    data = _load_json("depots.json")
+    return DepotCollection(**data).items
+
+
+def get_templates() -> List[Template]:
+    data = _load_json("templates.json")
+    return TemplateCollection(**data).items
+
+
+def get_tools() -> List[Tool]:
+    data = _load_json("tools.json")
+    return ToolCollection(**data).items
+
+
+def get_cabinets() -> list:
+    """Load cabinets (simple dict for now)."""
+    data = _load_json("cabinets.json")
+    return data.get("items", [])
+
+
+def get_monitors() -> list:
+    """Load monitors (simple dict for now)."""
+    data = _load_json("monitors.json")
+    return data.get("items", [])
+
+
+def get_pulses() -> List[Pulse]:
+    data = _load_json("pulses.json")
+    return PulseCollection(**data).items
+
+
+def get_books() -> List[Book]:
+    data = _load_json("books.json")
+    return BookCollection(**data).items
+
+
+def get_desks() -> List[Desk]:
+    data = _load_json("desks.json")
+    return DeskCollection(**data).items
+
+
+# === Single Item Helpers ===
+
+
+def get_vein(name: str) -> Optional[Vein]:
+    for v in get_veins():
+        if v.name == name:
+            return v
+    return None
+
+
+def get_room(name: str) -> Optional[Room]:
+    for r in get_rooms():
+        if r.name == name:
+            return r
+    return None
+
+
+def get_depot(name: str) -> Optional[Depot]:
+    for d in get_depots():
+        if d.name == name:
+            return d
+    return None
+
+
+def get_tool(name: str) -> Optional[Tool]:
+    for t in get_tools():
+        if t.name == name:
+            return t
+    return None
+
+
+# === System Data (for frontend rendering) ===
+
+
+def get_artery_data() -> dict:
+    """Data for artery frontend."""
+    return {
+        "veins": [v.model_dump() for v in get_veins()],
+        "rooms": [r.model_dump() for r in get_rooms()],
+        "depots": [d.model_dump() for d in get_depots()],
+        "pulses": [p.model_dump() for p in get_pulses()],
+    }
+
+
+def get_atlas_data() -> dict:
+    """Data for atlas frontend."""
+    return {
+        "templates": [t.model_dump() for t in get_templates()],
+        "depots": [d.model_dump() for d in get_depots()],
+        "books": [b.model_dump() for b in get_books()],
+    }
+
+
+def get_station_data() -> dict:
+    """Data for station frontend."""
+    return {
+        "tools": [t.model_dump() for t in get_tools()],
+        "monitors": get_monitors(),
+        "cabinets": get_cabinets(),
+        "rooms": [r.model_dump() for r in get_rooms()],
+        "depots": [d.model_dump() for d in get_depots()],
+        "desks": [d.model_dump() for d in get_desks()],
+    }