HUD Segments for Spokes

This guide shows how spokes can add custom segments to the Axium HUD with automatic name:value composition and wrapper normalization.

Overview

Spokes create HUD segments by subclassing HudSegment and registering them with the HUD registry. The registry automatically: - Composes segments as name:value (e.g., "creds:Y") - Applies wrapper normalization (e.g., [creds:Y] if configured) - Handles caching for expensive operations

HudSegment Classes

All HUD segments inherit from the HudSegment base class:

from axium.core.hud import register_hud_segment, get_registry
from axium.core.hud_segments import HudSegment

class MyStatusSegment(HudSegment):
    """Custom HUD segment for spoke."""

    name = "mystatus"
    priority = 100  # After core segments (5-20)
    cached = False  # Set to True for expensive operations

    def render(self, context: dict) -> str:
        """
        Render the segment VALUE only.

        The registry automatically composes this as "mystatus:{value}"
        and applies wrapper normalization.

        Args:
            context: Dict with keys:
                - state: Daemon state dict
                - pane_id: Optional tmux pane ID
                - env: Active environment name
                - started: Daemon start timestamp
                - wrapper: Dict with prefix/suffix for wrapping

        Returns:
            Value string (e.g., "ok") - registry composes to "mystatus:ok"
        """
        # Access current environment
        env = context.get("env", "-")

        # Compute status (fast operation)
        status = self.check_status(env)

        return status  # Returns "ok", registry composes "mystatus:ok"

    def check_status(self, env: str) -> str:
        """Check current status (called on every render)."""
        # Keep this fast! Called ~1-2 times per second in tmux
        return "ok" if env != "-" else "none"

# In spoke's register() function:
def register(app, events):
    # Register HUD segment
    register_hud_segment(MyStatusSegment())

Cached Segments (For Expensive Operations)

For segments that require expensive operations (network calls, file I/O, etc.), use the caching system:

from axium.core.hud import register_hud_segment, get_registry
from axium.core.hud_segments import HudSegment

class CredsSegment(HudSegment):
    """Credential validity HUD segment with caching."""

    name = "creds"
    priority = 100
    cached = True  # Expensive check, only update on events

    def render(self, context: dict) -> str:
        """
        Check credential validity. Returns just "Y" or "N".

        Registry automatically composes to "creds:Y" or "creds:N".
        """
        from axium.core import api
        from . import helpers

        env_name = context.get("env") or api.get_active_env()
        is_valid, _ = helpers.get_status(env_name)

        return "Y" if is_valid else "N"

def register(app, events):
    # Register cached segment
    creds_segment = CredsSegment()
    register_hud_segment(creds_segment)

    # Update cache on environment change
    def on_env_change(new_env: str, old_env: str, **kwargs):
        registry = get_registry()
        context = {"env": new_env}
        registry.update_cached_segments(context)  # Refresh cache

    # Update cache on spoke load
    def on_spoke_loaded(spoke_name: str):
        if spoke_name == "creds":
            from axium.core import api
            registry = get_registry()
            env_name = api.get_active_env()
            context = {"env": env_name}
            registry.update_cached_segments(context)  # Initial cache

    events.on("env_change", on_env_change)
    events.on("spoke_loaded", on_spoke_loaded)

How Caching Works

1. Cached segments (cached=True) store their rendered value and only update when update_cached_segments() is called
2. Uncached segments (cached=False) render fresh on every HUD update
3. Event handlers trigger cache updates, keeping expensive operations out of the render loop

Conditional Rendering

Segments can implement should_render() for conditional display:

class ConditionalSegment(HudSegment):
    name = "conditional"
    priority = 100

    def should_render(self, context: dict) -> bool:
        """Only show in specific environments."""
        env = context.get("env", "")
        return env in ["prod", "staging"]

    def render(self, context: dict) -> str:
        return "critical"  # Composes to "conditional:critical"

Wrapper Configuration

Users can configure wrappers in their ~/.config/axium/hud.yaml:

style:
  wrapper:
    prefix: '['
    suffix: ']'

The registry automatically: 1. Composes name:value (e.g., "creds:Y") 2. Detects and removes existing wrappers 3. Applies configured wrapper (e.g., "[creds:Y]")

Supported wrappers: [], {}, (), <>, ||, "", ''

Segment Types Comparison

Type	cached flag	Best For	Pros	Cons
Uncached	False	Fast computations	Simple, real-time	Must be fast
Cached	True	Expensive checks	Doesn't slow HUD	More complex setup

Performance Guidelines

HUD renders frequently in tmux (~1-2 times per second)

Uncached Segments (cached=False)

• ✅ DO: Keep render() fast (<5ms)
• ✅ DO: Use context data (already available)
• ✅ DO: Read environment variables
• ❌ DON'T: Make network calls in render()
• ❌ DON'T: Read files in render()
• ❌ DON'T: Run subprocesses in render()

Cached Segments (cached=True)

• ✅ DO: Use for network calls, file I/O, subprocesses
• ✅ DO: Update cache in event handlers
• ✅ DO: Call registry.update_cached_segments() after state changes
• ❌ DON'T: Forget to update cache on relevant events

Priority Guidelines

• 0-4: Reserved (future use)
• 5-9: Pane ID (PaneSegment)
• 10-19: Core segments (Environment, Uptime)
• 20-99: System extensions
• 100+: Spoke segments (recommended)

Testing Segments

def test_my_segment():
    """Test HUD segment rendering."""
    segment = MyStatusSegment()

    context = {
        "env": "prod",
        "pane_id": "%1",
        "started": "2024-01-01T00:00:00Z",
    }

    result = segment.render(context)
    assert result == "ok"  # Returns value only, not "mystatus:ok"

Examples

K8s Context (Uncached)

class K8sContextSegment(HudSegment):
    name = "k8s"
    priority = 100
    cached = False

    def render(self, context: dict) -> str:
        # Fast: read from env var set by kubectl
        import os
        ctx = os.getenv("KUBECONFIG_CONTEXT", "none")
        return ctx  # Registry composes to "k8s:prod-cluster"

Git Branch (Uncached with Subprocess)

class GitBranchSegment(HudSegment):
    name = "git"
    priority = 110
    cached = False

    def should_render(self, context: dict) -> bool:
        # Only show in development envs
        return context.get("env") in ["dev", "local"]

    def render(self, context: dict) -> str:
        import subprocess
        try:
            # Fast subprocess with timeout
            branch = subprocess.check_output(
                ["git", "branch", "--show-current"],
                cwd=os.getcwd(),
                stderr=subprocess.DEVNULL,
                timeout=0.05  # 50ms timeout
            ).decode().strip()
            return branch  # Registry composes to "git:main"
        except:
            return ""  # Empty string hides segment

Database Status (Cached)

class DbStatusSegment(HudSegment):
    name = "db"
    priority = 105
    cached = True  # Expensive connection check

    def render(self, context: dict) -> str:
        """Check database connectivity (cached)."""
        import psycopg2
        try:
            conn = psycopg2.connect(
                host="localhost",
                database="mydb",
                user="user",
                password="pass",
                connect_timeout=2
            )
            conn.close()
            return "UP"  # Registry composes to "db:UP"
        except:
            return "DOWN"  # Registry composes to "db:DOWN"

def register(app, events):
    db_segment = DbStatusSegment()
    register_hud_segment(db_segment)

    # Update cache every 5 minutes
    import threading
    import time

    def periodic_update():
        while True:
            time.sleep(300)  # 5 minutes
            registry = get_registry()
            context = {"env": api.get_active_env()}
            registry.update_cached_segments(context)

    threading.Thread(target=periodic_update, daemon=True).start()

    # Also update on environment change
    def on_env_change(new_env: str, old_env: str, **kwargs):
        registry = get_registry()
        context = {"env": new_env}
        registry.update_cached_segments(context)

    events.on("env_change", on_env_change)

See Also

• Event System Reference
• Spoke Development Guide
• HUD Configuration