"""Family Brain — live network visualization of the Beelink data constellation."""
import os, sqlite3, json, logging, html
from datetime import datetime, timezone
from fastapi import APIRouter
from fastapi.responses import HTMLResponse, Response
from jinja2 import Environment, FileSystemLoader
logger = logging.getLogger(name)
brain_router = APIRouter(prefix="/api/system/brain", tags=["brain"])
Template directory — absolute path to hoffdesk blog templates
TEMPLATE_DIR = "/home/hoffmann_admin/hoffdesk/blog/templates"
jinja_env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
def _escape(text):
"""Escape text for safe use in XML/SVG content."""
return html.escape(str(text), quote=False)
Color palette by type
COLORS = {
"hub": "#00e5ff",
"calendar": "#6366f1",
"blog": "#7c3aed",
"memory": "#ff00e5",
"ontology": "#ef4444",
"recipes": "#00e5ff",
"llm": "#f59e0b",
}
Orbital positions: 6 satellites around center (400, 310) at radius 130
plus inner positions for smaller nodes
ORBITS = [
# (id, x, y, priority)
("calendar", 270, 180, 1),
("memory", 530, 180, 2),
("blog", 270, 440, 3),
("llm", 530, 440, 4),
("ontology", 400, 440, 5),
("recipes", 400, 490, 6),
]
def _get_db_stats():
"""Query live stats from every data store."""
stats = {
"blog": {"published": 0, "total": 0, "last_activity": None},
"ontology": {"entities": 0, "relations": 0, "last_activity": None},
"memory": {"count": 0, "size_kb": 0},
"calendar": {"status": "unknown"},
"llm": {"models": 0},
"recipes": {"count": 0},
}
# Blog
try:
conn = sqlite3.connect("/home/hoffmann_admin/.openclaw/data/blog/blog.db")
cur = conn.execute("SELECT COUNT(*) FROM posts WHERE status='published'")
stats["blog"]["published"] = cur.fetchone()[0]
cur = conn.execute("SELECT COUNT(*) FROM posts")
stats["blog"]["total"] = cur.fetchone()[0]
cur = conn.execute("SELECT MAX(updated_at) FROM posts")
last = cur.fetchone()[0]
stats["blog"]["last_activity"] = last
conn.close()
except Exception as e:
logger.warning(f"Brain: blog db query failed: {e}")
# Ontology
try:
conn = sqlite3.connect("/home/hoffmann_admin/.openclaw/data/ontology.db")
cur = conn.execute("SELECT COUNT(*) FROM ontology_entities")
stats["ontology"]["entities"] = cur.fetchone()[0]
cur = conn.execute("SELECT COUNT(*) FROM ontology_relations")
stats["ontology"]["relations"] = cur.fetchone()[0]
cur = conn.execute("SELECT MAX(created_at) FROM ontology_entities")
last = cur.fetchone()[0]
if last:
stats["ontology"]["last_activity"] = last
conn.close()
except Exception as e:
logger.warning(f"Brain: ontology db query failed: {e}")
# Memory stores
total_size = 0
total_chunks = 0
for agent in ["main", "daedalus", "socrates"]:
path = f"/home/hoffmann_admin/.openclaw/memory/{agent}.sqlite"
if os.path.exists(path):
total_size += os.path.getsize(path)
try:
conn = sqlite3.connect(path)
cur = conn.execute("SELECT COUNT(*) FROM chunks")
total_chunks += cur.fetchone()[0]
conn.close()
except:
pass
stats["memory"]["count"] = total_chunks
stats["memory"]["size_kb"] = total_size // 1024
# Calendar
try:
import subprocess
r = subprocess.run(["systemctl", "is-active", "radicale"], capture_output=True, text=True, timeout=3)
stats["calendar"]["status"] = r.stdout.strip()
except:
stats["calendar"]["status"] = "unknown"
# LLM
try:
import urllib.request
req = urllib.request.Request("http://localhost:11434/api/tags")
resp = urllib.request.urlopen(req, timeout=3)
d = json.load(resp)
stats["llm"]["models"] = len(d.get("models", []))
except:
pass
# Recipes — stored in ontology with type recipe
try:
conn = sqlite3.connect("/home/hoffmann_admin/.openclaw/data/ontology.db")
cur = conn.execute("SELECT COUNT(*) FROM ontology_entities WHERE type='recipe'")
stats["recipes"]["count"] = cur.fetchone()[0]
conn.close()
except:
pass
return stats
def _build_nodes(stats):
"""Build node dicts with layout positions and visual parameters."""
nodes = []
# Hub
nodes.append({
"id": "beelink",
"x": 400, "y": 310, "r": 58,
"label": "BEELINK",
"subtitle": "titanium-butler",
"detail": "FastAPI · uvicorn",
"color": "#00e5ff",
"label_size": 13,
"text_opacity": 0.9,
"opacity": 0.9,
"pulse_dur": 4,
})
# Build satellite positions
occupied = list(ORBITS)
idx = 0
# Calendar
id_, x, y, _ = occupied[idx]; idx += 1
nodes.append({
"id": "calendar",
"x": x, "y": y, "r": 42,
"label": "CALENDAR",
"subtitle": "Radicale",
"detail": stats["calendar"]["status"],
"color": COLORS["calendar"],
"label_size": 11,
"text_opacity": 0.95,
"pulse_dur": 3.0,
})
# Memory
id_, x, y, _ = occupied[idx]; idx += 1
mem_detail = f'{stats["memory"]["size_kb"]}KB · {stats["memory"]["count"]} chunks'
nodes.append({
"id": "memory",
"x": x, "y": y, "r": 44,
"label": "MEMORY",
"subtitle": "ChromaDB",
"detail": mem_detail,
"color": COLORS["memory"],
"label_size": 11,
"text_opacity": 0.95,
"pulse_dur": 3.5,
})
# Blog
id_, x, y, _ = occupied[idx]; idx += 1
blog_detail = f'{stats["blog"]["published"]} pub · {stats["blog"]["total"]} total'
# Scale radius with post count
blog_r = min(48, max(36, 36 + stats["blog"]["total"]))
nodes.append({
"id": "blog",
"x": x, "y": y, "r": blog_r,
"label": "BLOG",
"subtitle": "SQLite",
"detail": blog_detail,
"color": COLORS["blog"],
"label_size": 11,
"text_opacity": 0.95,
"pulse_dur": 4.5,
})
# LLM
id_, x, y, _ = occupied[idx]; idx += 1
llm_detail = f'{stats["llm"]["models"]} models · local'
llm_r = min(48, max(36, 30 + stats["llm"]["models"]))
nodes.append({
"id": "llm",
"x": x, "y": y, "r": llm_r,
"label": "LLM",
"subtitle": "Ollama",
"detail": llm_detail,
"color": COLORS["llm"],
"label_size": 11,
"text_opacity": 0.95,
"pulse_dur": 4.2,
})
# Ontology
id_, x, y, _ = occupied[idx]; idx += 1
onto_detail = f'{stats["ontology"]["entities"]} entities · {stats["ontology"]["relations"]} rels'
onto_r = min(42, max(34, 30 + stats["ontology"]["entities"]))
nodes.append({
"id": "ontology",
"x": x, "y": y, "r": onto_r,
"label": "ONTOLOGY",
"subtitle": "Entities & Relations",
"detail": onto_detail,
"color": COLORS["ontology"],
"label_size": 10,
"text_opacity": 0.95,
"pulse_dur": 3.8,
})
# Recipes
id_, x, y, _ = occupied[idx]; idx += 1
r_detail = f'{stats["recipes"]["count"]} saved'
nodes.append({
"id": "recipes",
"x": x, "y": y, "r": 36,
"label": "RECIPES",
"subtitle": "Grocery List",
"detail": r_detail,
"color": COLORS["recipes"],
"label_size": 10,
"text_opacity": 0.95,
"pulse_dur": 3.2,
})
return nodes
def _build_connections(nodes):
"""Build connection lines from each satellite back to the hub."""
conns = []
hub = next(n for n in nodes if n["id"] == "beelink")
for node in nodes:
if node["id"] == "beelink":
continue
# Weight determines width and opacity
w = max(0.4, min(1.2, node["r"] / 40))
op = max(0.06, min(0.14, node["r"] / 200))
conns.append({
"x1": hub["x"], "y1": hub["y"],
"x2": node["x"], "y2": node["y"],
"color": node["color"],
"width": f"{w:.1f}",
"opacity": f"{op:.3f}",
})
return conns
def _build_data_packets(nodes):
"""Create animated data packet dots along each connection."""
packets = []
hub = next(n for n in nodes if n["id"] == "beelink")
for node in nodes:
if node["id"] == "beelink":
continue
dur = max(3.0, min(6.0, 6.0 - node["r"] / 15))
packets.append({
"x": hub["x"], "y": hub["y"],
"cx_from": str(node["x"]), "cx_to": str(hub["x"]),
"cy_from": str(node["y"]), "cy_to": str(hub["y"]),
"r": 1.5,
"color": node["color"],
"max_op": "0.6",
"dur": f"{dur:.1f}",
})
return packets
def _build_subtitle(stats):
parts = []
total_stores = 6
parts.append(f"{total_stores} DATA STORES")
parts.append(f'{stats["memory"]["size_kb"]}KB VECTORS')
parts.append(f'{stats["blog"]["published"]} BLOG POSTS')
parts.append(f'{stats["llm"]["models"]} LLM MODELS')
return " · ".join(parts)
@brain_router.get("/stats")
async def brain_stats():
"""Return raw brain stats as JSON."""
stats = _get_db_stats()
return stats
@brain_router.get("/svg")
async def brain_svg():
"""Return live Family Brain SVG."""
stats = _get_db_stats()
nodes = _build_nodes(stats)
connections = _build_connections(nodes)
data_packets = _build_data_packets(nodes)
subtitle = _build_subtitle(stats)
now = datetime.now(timezone.utc)
footer = (
f"ONE SERVER · {len(nodes)-1} DATA STORES · ZERO CLOUD DEPENDENCIES"
f" · UPDATED {now.strftime('%H:%M UTC')}"
)
try:
template = jinja_env.get_template("brain_svg.svg.j2")
svg = template.render(
nodes=nodes,
connections=connections,
data_packets=data_packets,
subtitle=subtitle,
footer=footer,
generated_at=now.isoformat(),
)
return Response(content=svg, media_type="image/svg+xml")
except Exception as e:
logger.error(f"Brain SVG render failed: {e}")
return Response(
content=f"<!-- Brain SVG render error: {e} -->",
media_type="image/svg+xml",
status_code=500,
)