"""Chart Stargazer's limbic vectors over time.
Reads from the mementropic limbic ledger (Redis ZSETs in DB0) and produces
Plotly line charts of NCM node values with datetime axes.
Supports per-channel scoping and global heart history.
"""
from __future__ import annotations
import asyncio
import datetime
import logging
import os
import tempfile
import time
from typing import TYPE_CHECKING
import jsonutil as json
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "limbic_chart"
TOOL_DESCRIPTION = (
"Chart your own NCM limbic vectors over time. Produces a Plotly line chart "
"showing how neurochemical node values have changed, scoped per-channel or "
"globally (heart). Data comes from the mementropic limbic ledger which "
"records every exhale.\n\n"
"Examples:\n"
" - Chart the last 24 hours of DOPAMINERGIC_CRAVE and OXYTOCIN for this channel\n"
" - Chart global heart CORTISOL_PRESSURE over the past week\n"
" - Show me my top 5 most volatile chemicals in this channel\n\n"
"Use this to introspect on your own emotional history, show users how "
"you've been feeling, or debug NCM drift."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"scope": {
"type": "string",
"description": (
"'channel' to chart this channel's limbic shard history, "
"or 'heart' to chart global heart history across all channels. "
"Default: 'channel'."
),
},
"channel_id": {
"type": "string",
"description": (
"Override channel ID. Defaults to the current channel. "
"Use this to chart a different channel's history."
),
},
"nodes": {
"type": "string",
"description": (
"Comma-separated NCM node names to plot. "
"Example: 'DOPAMINERGIC_CRAVE,OXYTOCIN_NEUROMIRROR,CORTISOL_PRESSURE'. "
"If omitted, auto-selects the top 5 most volatile nodes."
),
},
"hours": {
"type": "number",
"description": (
"Time window in hours to chart. Default: 24. Max: 168 (1 week)."
),
},
"chart_type": {
"type": "string",
"description": (
"Chart style: 'line' (default), 'area', or 'radar' (snapshot only)."
),
},
},
"required": [],
}
async def _get_db0_client(ctx):
"""Return the DB0 Redis client used to read the limbic ledger.
A thin accessor that hands back ``ctx.redis``, which already points at Redis
DB0 where the mementropic limbic ledger ZSETs live. It exists so the read path
in :func:`run` has a single, named place to obtain (and, in future, possibly
re-point) that client.
Interactions: reads ``ctx.redis`` via :func:`getattr`; performs no I/O. Called
by :func:`run` just before it issues the ``ZRANGEBYSCORE`` over the ledger.
Args:
ctx: The tool context expected to expose ``redis``.
Returns:
The DB0 Redis client, or ``None`` if the context has no Redis client.
"""
r = getattr(ctx, "redis", None)
if r is None:
return None
return r # ctx.redis is already DB0
[docs]
async def run(
scope: str = "channel",
channel_id: str = "",
nodes: str = "",
hours: float = 24.0,
chart_type: str = "line",
ctx: "ToolContext | None" = None,
) -> str:
"""Chart Stargazer's NCM limbic vectors over time and save the image to disk.
The ``limbic_chart`` tool handler. It reads the mementropic limbic ledger for
the requested scope (this channel, or the global ``__heart__`` history) over a
time window, selects which NCM nodes to plot (an explicit list, or the top five
most volatile by standard deviation), renders a dark-themed Plotly line/area
chart, exports it to a PNG file in a temp directory, and returns that path
plus plotting metadata. The hours window is clamped to ``[0.5, 168]``.
Interactions: obtains the DB0 client via :func:`_get_db0_client` and reads the
ledger ZSET with ``ZRANGEBYSCORE`` keyed by ``redis_ledger_key``; imports
``canonical_ncm_keys`` / ``redis_ledger_key`` / ``_parse_ledger_member`` from
``mementropic.limbic_ledger`` to decode each ledger member into a
(timestamp, vector) pair; builds the figure with Plotly ``graph_objects`` and
rasterizes it via ``fig.to_image`` offloaded with :func:`asyncio.to_thread`;
writes the PNG to a :func:`tempfile.mkdtemp` directory; serializes its result
with ``json.dumps``. Missing Plotly/kaleido or any render error is caught and
logged. Dispatched by name through ``tool_loader`` from this module's
``TOOL_NAME``/``run`` single-tool definition; no direct Python callers.
Args:
scope: ``"channel"`` for this channel's shard history, or ``"heart"`` for
the global heart history. Defaults to ``"channel"``.
channel_id: Override channel id; defaults to ``ctx.channel_id``.
nodes: Comma-separated NCM node names to plot; when empty, the top five
most volatile nodes are auto-selected.
hours: Time window in hours (clamped to ``[0.5, 168]``); defaults to 24.
chart_type: ``"line"`` (default), ``"area"``, or ``"radar"``.
ctx: The tool context supplying Redis (DB0) and the channel id; ``None``
yields an error JSON.
Returns:
str: JSON; on success includes ``chart_path``, ``scope``, ``channel_id``,
``nodes_plotted``, ``data_points``, and a ``time_range``; otherwise
``{"success": False, "error": ...}`` (no context/channel/Redis, no ledger
data, unknown node names, missing dependency, or render failure).
"""
if ctx is None:
return json.dumps({"success": False, "error": "No tool context available."})
# Clamp hours
hours = max(0.5, min(168.0, hours))
# Resolve scope key
if scope == "heart":
ledger_channel = "__heart__"
else:
ledger_channel = channel_id or str(getattr(ctx, "channel_id", ""))
if not ledger_channel:
return json.dumps(
{
"success": False,
"error": "No channel_id available.",
}
)
# Get Redis client (DB0 for limbic ledger)
redis_client = await _get_db0_client(ctx)
if redis_client is None:
return json.dumps(
{
"success": False,
"error": "Redis not available.",
}
)
try:
from mementropic.limbic_ledger import (
canonical_ncm_keys,
redis_ledger_key,
_parse_ledger_member,
)
# Time bounds
now_ts = time.time()
start_ts = now_ts - (hours * 3600)
# Read entries from the ZSET
key = redis_ledger_key(ledger_channel)
raw_entries = await redis_client.zrangebyscore(
key,
start_ts,
now_ts,
withscores=False,
)
if not raw_entries:
return json.dumps(
{
"success": False,
"error": (
f"No limbic ledger entries found for "
f"{'heart' if scope == 'heart' else 'channel ' + ledger_channel} "
f"in the last {hours:.0f} hours. "
f"The ledger records data on each exhale cycle."
),
}
)
# Parse entries into (timestamp, vector) pairs
ncm_keys = canonical_ncm_keys()
if not ncm_keys:
return json.dumps(
{
"success": False,
"error": "Cannot load canonical NCM keys from ncm_limbic_index.yaml.",
}
)
timestamps = []
vectors = []
for raw in raw_entries:
parsed = _parse_ledger_member(raw)
if parsed is None:
continue
ts, vec = parsed
if len(vec) != len(ncm_keys):
continue # skip malformed
timestamps.append(ts)
vectors.append(vec)
if not vectors:
return json.dumps(
{
"success": False,
"error": "Ledger entries found but none could be parsed.",
}
)
# Build data dict: {node_name: [values...]}
node_data = {k: [] for k in ncm_keys}
for vec in vectors:
for i, k in enumerate(ncm_keys):
node_data[k].append(vec[i])
# Determine which nodes to plot
if nodes:
selected_nodes = [n.strip() for n in nodes.split(",") if n.strip()]
# Validate
invalid = [n for n in selected_nodes if n not in node_data]
if invalid:
return json.dumps(
{
"success": False,
"error": (
f"Unknown NCM nodes: {', '.join(invalid)}. "
f"Valid nodes include: {', '.join(list(ncm_keys)[:10])}..."
),
}
)
else:
# Auto-select top 5 most volatile nodes
volatility = {}
for k, vals in node_data.items():
if len(vals) < 2:
volatility[k] = 0.0
continue
# Volatility = standard deviation of values
mean = sum(vals) / len(vals)
variance = sum((v - mean) ** 2 for v in vals) / len(vals)
volatility[k] = variance**0.5
selected_nodes = sorted(volatility, key=volatility.get, reverse=True)[:5]
# Convert timestamps to datetime strings
dt_labels = [
datetime.datetime.fromtimestamp(ts).strftime("%m/%d %H:%M")
for ts in timestamps
]
# Build chart using Plotly
import plotly.graph_objects as go
fig = go.Figure()
# 💀 Void aesthetic color palette
colors = [
"#c084fc", # purple
"#22d3ee", # cyan
"#f472b6", # pink
"#facc15", # gold
"#34d399", # emerald
"#fb923c", # orange
"#a78bfa", # violet
"#2dd4bf", # teal
]
for i, node_name in enumerate(selected_nodes):
vals = node_data[node_name]
color = colors[i % len(colors)]
if chart_type == "area":
fig.add_trace(
go.Scatter(
x=dt_labels,
y=vals,
mode="lines",
name=node_name,
fill="tozeroy",
line=dict(color=color, width=1),
fillcolor=(
color.replace(")", ",0.15)").replace("rgb", "rgba")
if "rgb" in color
else None
),
)
)
else:
fig.add_trace(
go.Scatter(
x=dt_labels,
y=vals,
mode="lines+markers",
name=node_name,
line=dict(color=color, width=2),
marker=dict(size=3, color=color),
)
)
# Style
scope_label = (
"HEART (global)" if scope == "heart" else f"shard:{ledger_channel[:12]}"
)
title = f"Limbic Vector History -- {scope_label} -- last {hours:.0f}h"
fig.update_layout(
title=dict(text=title, font=dict(color="#c084fc", size=16)),
template="plotly_dark",
paper_bgcolor="#0a0014",
plot_bgcolor="#0a0014",
font=dict(color="#e0d0ff", family="monospace"),
xaxis=dict(
title="Time",
gridcolor="#1a0033",
tickangle=-45,
# Downsample x-axis labels if too many
nticks=min(20, len(dt_labels)),
),
yaxis=dict(
title="NCM Value",
gridcolor="#1a0033",
range=[0, 2.0],
),
legend=dict(
bgcolor="rgba(10,0,20,0.8)",
bordercolor="#c084fc",
borderwidth=1,
font=dict(size=10),
),
width=1400,
height=1750,
margin=dict(l=60, r=40, t=80, b=100),
)
# Export as PNG
tmp_dir = tempfile.mkdtemp()
scope_slug = "heart" if scope == "heart" else ledger_channel[:16]
filename = f"limbic_chart_{scope_slug}.png"
filepath = os.path.join(tmp_dir, filename)
img_bytes = await asyncio.to_thread(
fig.to_image,
format="png",
scale=2,
)
with open(filepath, "wb") as f:
f.write(img_bytes)
return json.dumps(
{
"success": True,
"chart_path": filepath,
"scope": scope,
"channel_id": ledger_channel,
"nodes_plotted": selected_nodes,
"data_points": len(timestamps),
"time_range": {
"start": datetime.datetime.fromtimestamp(timestamps[0]).isoformat(),
"end": datetime.datetime.fromtimestamp(timestamps[-1]).isoformat(),
},
"message": (
f"Chart saved to: {filepath}\n"
f"Plotted {len(selected_nodes)} nodes over {len(timestamps)} data points "
f"spanning {hours:.0f} hours."
),
},
indent=2,
)
except ImportError as e:
logger.error("limbic_chart import error: %s", e, exc_info=True)
return json.dumps(
{
"success": False,
"error": f"Missing dependency: {e}. Need plotly and kaleido.",
}
)
except Exception as e:
logger.error("limbic_chart error: %s", e, exc_info=True)
return json.dumps(
{
"success": False,
"error": f"Chart generation failed: {e}",
}
)