star_avatar

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) desire -> star-desire.png (placeholder asset) fear -> star-fear.png (placeholder asset) tender -> star-tender.png (placeholder asset) smug -> star-smug.png (placeholder asset)

When the first dominant emotion in list order that matches the winning bucket is labelled (intense) or (overwhelming), the classifier may return {bucket}_intense (e.g. rage_intense). Avatar update falls back to the base bucket key if that variant is absent from the mxc map.

star_avatar.classify_expression(dominant_emotions)

Map a list of dominant emotions to an expression bucket.

Returns one of the base keys in _EXPRESSION_FILES (including '{bucket}_intense' when the first list-order emotion that matches the winning bucket is labelled (intense) or (overwhelming)), or 'default' when nothing matches a bucket.

Base buckets: default, laugh, rage, facepalm, desire, fear, tender, smug. Intense variants: laugh_intense, rage_intense, facepalm_intense, desire_intense, fear_intense, tender_intense, smug_intense.

Handles plain names ("RAGE") and intensity-labelled entries ("RAGE (intense)") from format_context_injection.

Return type:

str

Parameters:

dominant_emotions (list[str])

async star_avatar.get_expression_mxc_map(redis_client, matrix_client=None, *, base_dir=None)

Load or lazily create the expression -> mxc:// URI mapping.

Returns the dict that maps each expression bucket key (e.g. rage, laugh_intense) to the mxc:// URI of its avatar image on the Matrix homeserver. The mapping is cached in Redis under star:avatar:mxc_map so the (slow) upload only happens once; this lets the avatar swap on a hot path stay a pure dictionary lookup.

Reads the Redis cache first via redis_client.get; on a miss (or a corrupt JSON value) and when a matrix_client is supplied, it reads each PNG from base_dir on disk, uploads the bytes through the nio matrix_client.upload HTTP call, collects the returned content_uri values, and writes the assembled map back to Redis with a 30-day TTL. Missing image files and failed uploads are logged and skipped rather than raising. Called by update_star_avatar() (its only in-repo caller).

Parameters:

• redis_client (Any) – Async Redis client used to read and write the cached star:avatar:mxc_map value.

• matrix_client (Any) – nio Matrix client used to upload images and obtain their mxc:// URIs. When None and the cache is empty, the function logs a warning and returns an empty dict (no upload possible).

• base_dir (str | Path | None) – Optional directory holding the star-*.png expression assets. Defaults to the server-side egregore asset directory /home/star/large_files/assets/egregores/stargazer.

Return type:

dict[str, str]

Returns:

A mapping of expression key to mxc:// URI. Empty when nothing is cached and no Matrix client is available, or when no image uploaded.

async star_avatar.update_star_avatar(redis_client, matrix_client, channel_id, dominant_emotions=None)

Update Star’s Matrix avatar based on current emotional state.

Picks the expression image that matches Star’s dominant emotions and, if it differs from the one currently displayed, pushes it to the Matrix profile. This is the once-per-response side effect that makes the avatar visibly track her mood instead of staying static.

When dominant_emotions is not supplied it reads them from the NCM limbic shard at Redis key db12:shard:{channel_id}. It then runs classify_expression() to pick a bucket, resolves that bucket to an mxc:// URI via get_expression_mxc_map() (falling back from an _intense variant to its base key, then to default), short-circuits when Redis key star:avatar:current_expression already records that expression, and otherwise calls matrix_client.set_avatar over the Matrix API and stores the new expression in Redis with a 1-hour TTL. The whole body is wrapped so any failure is logged at debug and swallowed – a cosmetic avatar update must never break reply delivery. Called by message_processor/generate_and_send.py just before the reply is sent.

Parameters:

• redis_client (Any) – Async Redis client used to read the limbic shard and the cached expression map, and to read/write the current-expression key.

• matrix_client (Any) – nio Matrix client whose set_avatar is invoked and which is also used to lazily upload images via get_expression_mxc_map().

• channel_id (str) – Channel identifier used to locate the NCM limbic shard key when dominant_emotions is not passed in.

• dominant_emotions (list[str] | None) – Optional pre-computed list of dominant emotion labels. When None, read from the limbic shard in Redis.

Return type:

None

Returns:

None. Acts only through its Matrix and Redis side effects.

class star_avatar.AsyncDebouncer(delay_seconds=5.0)

Bases: object

Delays and coalesces rapid update requests, running only the final one.

A trailing-edge debouncer: each trigger() resets a delay window and cancels the previously pending task, so a burst of calls collapses to a single execution once the requests stop – useful for throttling something like an avatar refresh that would otherwise fire on every rapid update.

Holds the delay and a single in-flight asyncio.Task slot (self._timer); the actual waiting and execution happen in _wait_and_execute(). State lives entirely in memory with no Redis, network, or filesystem interaction. In this repo it is exercised only by tests/test_limbic_concurrency.py; no production call sites were found.

Parameters:

delay_seconds (float)

__init__(delay_seconds=5.0)

Initialise the debouncer with a coalescing delay window.

Stores the delay and a slot for the single in-flight timer task; no task is scheduled until trigger() is first called.

Parameters:

delay_seconds (float) – Seconds to wait after the most recent trigger() call before the coalesced task runs. Defaults to 5.0.

Return type:

None

trigger(task_func)

Schedule task_func to run after the delay, superseding any pending call.

Cancels the previously scheduled timer (if any) and starts a fresh asyncio.Task running _wait_and_execute(), so that when triggers arrive in quick succession only the final one’s coroutine actually executes once the window elapses.

Calls asyncio.create_task() to launch the deferred task and cancels self._timer to drop the prior pending execution; it must be called from within a running event loop. In the codebase its only observed caller is tests/test_limbic_concurrency.py (test_async_debouncer), which fires three rapid triggers and asserts only the last runs; no production call sites were found.

Parameters:

task_func (Any) – A zero-argument callable returning an awaitable (e.g. a coroutine function or lambda) to be awaited after the delay.

Return type:

None

Returns:

None.