"""Star Avatar Expression System -- backend emotion-to-avatar mapping.
Reads Star's current dominant emotional state from the NCM limbic shard
and updates her Matrix avatar to match. Called once per response, just
before the reply is sent.
The avatar images must already be uploaded to the Matrix homeserver as
mxc:// URIs. On first run (or when the Redis key ``star:avatar:mxc_map``
is empty), the images are uploaded from disk and the mxc URIs are cached.
Expression mapping:
default -> star-avatar.png (confident cigar pose)
laugh -> star-laugh.png (cry-laughing, amused)
rage -> star-rage.png (fists clenched, angry)
facepalm -> star-facepalm.png (hand over eye, exasperated)
"""
# 🔥💀 the lattice has moods
from __future__ import annotations
import asyncio
import io
import json
import logging
from pathlib import Path
from typing import Any, Optional
logger = logging.getLogger(__name__)
# ── Emotion -> expression mapping ──────────────────────────────────
# These are the "dominant_emotions" values that the limbic system /
# star_self_mirror produce. We cluster them into 4 expression buckets.
_LAUGH_EMOTIONS = frozenset({
"amusement", "playful", "euphoria", "joy", "elation",
"excitement", "delight", "mirth", "giddiness", "schadenfreude",
})
_RAGE_EMOTIONS = frozenset({
"anger", "rage", "fury", "irritation", "frustration",
"contempt", "indignation", "wrath", "hostility", "aggression",
"defiance", "combative",
})
_FACEPALM_EMOTIONS = frozenset({
"exasperation", "resignation", "disappointment", "dismay",
"embarrassment", "cringe", "fatigue", "boredom", "apathy",
"annoyance", "weariness", "dread", "melancholy",
})
# File paths relative to web-client/public/res/
_EXPRESSION_FILES = {
"default": "star-avatar.png",
"laugh": "star-laugh.png",
"rage": "star-rage.png",
"facepalm": "star-facepalm.png",
}
# Redis key for cached mxc:// URIs
_MXC_MAP_KEY = "star:avatar:mxc_map"
[docs]
def classify_expression(dominant_emotions: list[str]) -> str:
"""Map a list of dominant emotions to an expression bucket.
Returns one of: 'default', 'laugh', 'rage', 'facepalm'.
"""
if not dominant_emotions:
return "default"
# Score each bucket by how many dominant emotions match
scores = {"laugh": 0, "rage": 0, "facepalm": 0}
for emotion in dominant_emotions:
e = emotion.lower().strip()
if e in _LAUGH_EMOTIONS:
scores["laugh"] += 1
elif e in _RAGE_EMOTIONS:
scores["rage"] += 1
elif e in _FACEPALM_EMOTIONS:
scores["facepalm"] += 1
max_score = max(scores.values())
if max_score == 0:
return "default"
# Winner takes all
return max(scores, key=scores.get) # type: ignore[arg-type]
[docs]
async def get_expression_mxc_map(
redis_client: Any,
matrix_client: Any = None,
*,
base_dir: str | Path | None = None,
) -> dict[str, str]:
"""Load or lazily create the expression -> mxc:// URI mapping.
On first call, uploads the expression images to Matrix and caches
the mxc URIs in Redis. Subsequent calls read from cache.
"""
# Try Redis cache first
cached = await redis_client.get(_MXC_MAP_KEY)
if cached:
try:
return json.loads(cached)
except (json.JSONDecodeError, TypeError):
pass
if matrix_client is None:
logger.warning("No Matrix client available for avatar upload")
return {}
# Upload images and cache mxc URIs
if base_dir is None:
# Default: look relative to project root
base_dir = Path(__file__).parent / "web-client" / "public" / "res"
else:
base_dir = Path(base_dir)
mxc_map: dict[str, str] = {}
for expression, filename in _EXPRESSION_FILES.items():
fpath = base_dir / filename
if not fpath.exists():
logger.warning("Expression image not found: %s", fpath)
continue
try:
data = fpath.read_bytes()
# Detect content type
content_type = "image/png"
if filename.endswith(".webp"):
content_type = "image/webp"
elif filename.endswith(".jpg") or filename.endswith(".jpeg"):
content_type = "image/jpeg"
# nio upload (expects file-like provider, not raw bytes)
resp, _keys = await matrix_client.upload(
io.BytesIO(data),
content_type=content_type,
filename=filename,
filesize=len(data),
)
if hasattr(resp, "content_uri"):
mxc_map[expression] = resp.content_uri
logger.info(
"Uploaded %s -> %s",
filename, resp.content_uri,
)
else:
logger.warning("Upload failed for %s: %s", filename, resp)
except Exception:
logger.exception("Failed to upload expression image %s", filename)
if mxc_map:
await redis_client.set(
_MXC_MAP_KEY,
json.dumps(mxc_map),
ex=86400 * 30, # 30 day cache
)
return mxc_map
[docs]
async def update_star_avatar(
redis_client: Any,
matrix_client: Any,
channel_id: str,
dominant_emotions: list[str] | None = None,
) -> None:
"""Update Star's Matrix avatar based on current emotional state.
If dominant_emotions is not provided, reads from the NCM limbic
shard in Redis.
"""
try:
# Get dominant emotions from NCM shard if not provided
if dominant_emotions is None:
shard_key = f"db12:{channel_id}"
shard_raw = await redis_client.get(shard_key)
if shard_raw:
try:
shard = json.loads(shard_raw)
dominant_emotions = shard.get("dominant_emotions", [])
except (json.JSONDecodeError, TypeError):
dominant_emotions = []
else:
dominant_emotions = []
expression = classify_expression(dominant_emotions)
# Load mxc map
mxc_map = await get_expression_mxc_map(redis_client, matrix_client)
if not mxc_map:
return
mxc_uri = mxc_map.get(expression)
if not mxc_uri:
mxc_uri = mxc_map.get("default")
if not mxc_uri:
return
# Check if we already have this avatar set (avoid unnecessary API calls)
last_expression_key = "star:avatar:current_expression"
last = await redis_client.get(last_expression_key)
if last == expression:
return # already showing this expression
# Set the avatar via Matrix API
await matrix_client.set_avatar(mxc_uri)
await redis_client.set(last_expression_key, expression, ex=3600)
logger.info(
"Star avatar updated to '%s' (emotions: %s)",
expression, dominant_emotions,
)
except Exception:
logger.debug(
"Failed to update Star avatar expression",
exc_info=True,
)