unaiverse.stats

What this module does 🔴

Provides the statistics and telemetry subsystem: UIPlot for Plotly panels, DefaultBaseDash for dashboard layout, and Stats for SQLite-backed metric recording and rendering.

stats

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

THEMES module-attribute

THEMES = {'dark': {'bg_paper': 'rgba(0,0,0,0)', 'bg_plot': 'rgba(0,0,0,0)', 'text_main': '#C8CDD3', 'text_light': '#677385', 'grid': '#222A36', 'edge': '#495464', 'node_border': '#F5F6F8', 'main': '#1A5CFF', 'main_light': '#4D7FFF', 'table': {'header_bg': '#16171C', 'header_txt': '#F5F6F8', 'cell_bg': '#0E0F14', 'cell_txt': '#C8CDD3', 'line': '#222A36'}, 'peers': ['#1A5CFF', '#FF3B30', '#00D4AA', '#FFB347', '#4D7FFF', '#00B391', '#FF6692', '#33EABD', '#FFD080', '#6B9BFF']}, 'light': {'bg_paper': 'rgba(255,255,255,0)', 'bg_plot': 'rgba(255,255,255,0)', 'text_main': '#0A1628', 'text_light': '#677385', 'grid': '#DBE1EB', 'edge': '#909CB0', 'node_border': '#0A1628', 'main': '#1A5CFF', 'main_light': '#4D7FFF', 'table': {'header_bg': '#F6F8FA', 'header_txt': '#0A1628', 'cell_bg': '#FFFFFF', 'cell_txt': '#0A1628', 'line': '#C0C8D6'}, 'peers': ['#1A5CFF', '#FF3B30', '#00D4AA', '#FFB347', '#4D7FFF', '#00B391', '#FF6692', '#33EABD', '#FFD080', '#6B9BFF']}}

THEME module-attribute

THEME = THEMES['dark']

UIPlot

UIPlot(title: str = '', height: int = 400)

A Python abstraction for a single Plotly chart panel.

Wraps a Plotly data/layout pair behind a fluent builder API so that callers can construct charts with typed method calls rather than hand-crafting JSON dictionaries. All traces are accumulated in an internal list; call to_json to produce the JSON string expected by the frontend.

The default layout follows the dark theme defined by the module-level THEME constant. Layout properties can be overridden with set_layout_opt or with the purpose-built helpers (set_y_range, set_legend).

Attributes:

Name	Type	Description
_data	list[dict[str, Any]]	Ordered list of raw Plotly trace dictionaries.
_layout	dict[str, Any]	Plotly layout dictionary controlling axes, size, and appearance.

Initialize a UIPlot with an empty trace list and a dark-themed Plotly layout.

The layout is pre-configured with the module-level dark THEME: transparent background, dotted grid lines, mirrored axis borders, and matching font colors. Both axes default to the titles 'Time' (x-axis) and 'Value' (y-axis); override them after construction with set_layout_opt.

Parameters:

Name	Type	Description	Default
title	str	Title text shown at the top of the plot. Defaults to an empty string.	''
height	int	Height of the plot in pixels. Defaults to 400.	400

Source code in unaiverse/stats.py

def __init__(self, title: str = '', height: int = 400):
    """Initialize a UIPlot with an empty trace list and a dark-themed Plotly layout.

    The layout is pre-configured with the module-level dark ``THEME``: transparent
    background, dotted grid lines, mirrored axis borders, and matching font colors.
    Both axes default to the titles ``'Time'`` (x-axis) and ``'Value'`` (y-axis);
    override them after construction with ``set_layout_opt``.

    Args:
        title: Title text shown at the top of the plot. Defaults to an empty string.
        height: Height of the plot in pixels. Defaults to 400.
    """
    self._data: list[dict[str, Any]] = []

    # Define the standard axis style for a "boxed" look
    axis_style = {
        'gridcolor': THEME['grid'],
        'gridwidth': 1,
        'griddash': 'dot',
        'color': THEME['text_light'],
        'showline': True,  # Draw the axis line
        'mirror': True,  # Mirror it on top/right (creates the box)
        'linewidth': 2,  # Width of the box border
        'linecolor': THEME['grid'],  # Color of the box border
        'zeroline': False,  # Prevents double-thick borderlines at 0
        'layer': 'below traces'  # Key fix: puts grid BEHIND the box border
    }

    self._layout: dict[str, Any] = {
        'title': title,
        'height': height,
        'xaxis': {**axis_style, 'title': 'Time'},
        'yaxis': {**axis_style, 'title': 'Value'},
        'margin': {'l': 50, 'r': 50, 'b': 50, 't': 50},
        # Default dark theme friendly styling
        'paper_bgcolor': THEME['bg_paper'],
        'plot_bgcolor': THEME['bg_plot'],
        'font': {'color': THEME['text_main']}
    }

add_line

add_line(x: list[Any], y: list[Any], name: str, color: str | None = None, legend_group: str | None = None, show_legend: bool = True)

Add a scatter line trace with markers to the plot.

Creates a Plotly scatter trace in 'lines+markers' mode and appends it to the internal trace list. Multiple calls accumulate traces on the same axes.

Parameters:

Name	Type	Description	Default
x	list[Any]	Sequence of x-axis values (e.g., millisecond timestamps or category strings).	required
y	list[Any]	Sequence of y-axis values corresponding to each x entry.	required
name	str	Label shown in the legend for this trace.	required
color	str | None	CSS color string (hex, rgba(), or named color) applied to the line and markers. Defaults to the primary theme color THEME['main'].	None
legend_group	str | None	Plotly legend-group name. Traces sharing the same group are toggled together when the legend entry is clicked. Defaults to None.	None
show_legend	bool	Whether to display this trace in the legend. Defaults to True.	True

Source code in unaiverse/stats.py

def add_line(self, x: list[Any], y: list[Any], name: str, color: str | None = None,
             legend_group: str | None = None, show_legend: bool = True):
    """Add a scatter line trace with markers to the plot.

    Creates a Plotly ``scatter`` trace in ``'lines+markers'`` mode and appends it to
    the internal trace list. Multiple calls accumulate traces on the same axes.

    Args:
        x: Sequence of x-axis values (e.g., millisecond timestamps or category strings).
        y: Sequence of y-axis values corresponding to each x entry.
        name: Label shown in the legend for this trace.
        color: CSS color string (hex, ``rgba()``, or named color) applied to the line
            and markers. Defaults to the primary theme color ``THEME['main']``.
        legend_group: Plotly legend-group name. Traces sharing the same group are
            toggled together when the legend entry is clicked. Defaults to None.
        show_legend: Whether to display this trace in the legend. Defaults to True.
    """
    if color is None:
        color = THEME['main']
    trace = {
        'x': x, 'y': y,
        'name': name,
        'type': 'scatter',
        'mode': 'lines+markers',
        'line': {'color': color},
        "legendgroup": legend_group,
        "showlegend": show_legend
    }
    self._data.append(trace)

add_area

add_area(x: list[Any], y: list[Any], name: str, color: str | None = None)

Add a filled area trace (fill: tozeroy) to the plot.

Creates a Plotly scatter trace filled down to zero on the y-axis, suitable for visualizing cumulative or bounded quantities.

Parameters:

Name	Type	Description	Default
x	list[Any]	Sequence of x-axis values.	required
y	list[Any]	Sequence of y-axis values defining the upper boundary of the filled area.	required
name	str	Label shown in the legend for this trace.	required
color	str | None	CSS color string applied to both the line and the fill. Defaults to the primary theme color THEME['main'].	None

Source code in unaiverse/stats.py

def add_area(self, x: list[Any], y: list[Any], name: str, color: str | None = None):
    """Add a filled area trace (``fill: tozeroy``) to the plot.

    Creates a Plotly ``scatter`` trace filled down to zero on the y-axis, suitable
    for visualizing cumulative or bounded quantities.

    Args:
        x: Sequence of x-axis values.
        y: Sequence of y-axis values defining the upper boundary of the filled area.
        name: Label shown in the legend for this trace.
        color: CSS color string applied to both the line and the fill. Defaults to
            the primary theme color ``THEME['main']``.
    """
    if color is None:
        color = THEME['main']
    trace = {
        'x': x, 'y': y, 'name': name,
        'type': 'scatter', 'fill': 'tozeroy',
        'line': {'color': color}
    }
    self._data.append(trace)

add_indicator

add_indicator(value: Any, title: str)

Add a big-number indicator trace to the plot.

Creates a Plotly indicator trace in 'number' mode and appends it to the trace list. The layout height is also reduced to 300 pixels because indicators require less vertical space than full chart panels.

Parameters:

Name	Type	Description	Default
value	Any	Numeric (or any JSON-serializable) value to display prominently in the center of the panel.	required
title	str	Subtitle text rendered below the number.	required

Source code in unaiverse/stats.py

def add_indicator(self, value: Any, title: str):
    """Add a big-number indicator trace to the plot.

    Creates a Plotly ``indicator`` trace in ``'number'`` mode and appends it to the
    trace list. The layout height is also reduced to 300 pixels because indicators
    require less vertical space than full chart panels.

    Args:
        value: Numeric (or any JSON-serializable) value to display prominently in the
            center of the panel.
        title: Subtitle text rendered below the number.
    """
    self._data.append({
        'type': 'indicator',
        'mode': 'number',
        'value': value,
        'title': {'text': title}
    })
    self._layout['height'] = 300  # Indicators usually need less height

add_table

add_table(headers: list[str] | None, columns: list[list[Any]])

Add a Plotly table trace to the plot.

Renders a formatted data table using the dark-theme colors defined in THEME. When headers is None, a transparent zero-height header row is used so that no column labels appear, but the table structure is otherwise identical.

Parameters:

Name	Type	Description	Default
headers	list[str] | None	Column header labels shown at the top of the table, or None to suppress the header row entirely.	required
columns	list[list[Any]]	Column data as a list of per-column value lists. Each inner list corresponds to one column; all inner lists should have the same length.	required

Source code in unaiverse/stats.py

def add_table(self, headers: list[str] | None, columns: list[list[Any]]):
    """Add a Plotly table trace to the plot.

    Renders a formatted data table using the dark-theme colors defined in ``THEME``.
    When ``headers`` is ``None``, a transparent zero-height header row is used so
    that no column labels appear, but the table structure is otherwise identical.

    Args:
        headers: Column header labels shown at the top of the table, or ``None`` to
            suppress the header row entirely.
        columns: Column data as a list of per-column value lists. Each inner list
            corresponds to one column; all inner lists should have the same length.
    """
    num_columns = len(columns) if columns else 0
    if headers:
        header_cfg = {
            'values': headers,
            'fill': {'color': THEME['table']['header_bg']},
            'font': {'color': THEME['table']['header_txt']},
            'line': {'color': THEME['table']['line']}
        }
    else:
        header_cfg = {
            'values': [''] * num_columns,
            'height': 0,  # Hide it
            'fill': {'color': 'rgba(0,0,0,0)'},  # Transparent just in case
            'line': {'width': 0}  # No border
        }

    trace = {
        'type': 'table',
        'header': header_cfg,
        'cells': {
            'values': columns,
            'fill': {'color': THEME['table']['cell_bg']},
            'font': {'color': THEME['table']['cell_txt']},
            'line': {'color': THEME['table']['line']}
        }
    }
    self._data.append(trace)

add_bar

add_bar(xs: list[Any], ys: list[Any], names: list[str], colors: list[str] | str | None = None)

Add a bar chart trace with automatic text labels to the plot.

Creates a Plotly bar trace with text annotations positioned automatically on top of each bar. The y-axis title is reset to 'Value' so that it always reflects this trace.

Parameters:

Name	Type	Description	Default
xs	list[Any]	Category labels displayed on the x-axis (one per bar).	required
ys	list[Any]	Numeric values controlling bar heights (one per bar).	required
names	list[str]	Text annotations placed on or near each bar (one per bar).	required
colors	list[str] | str | None	A single CSS color string applied uniformly to all bars, or a list of per-bar CSS color strings. Defaults to the primary theme color THEME['main'].	None

Source code in unaiverse/stats.py

def add_bar(self, xs: list[Any], ys: list[Any], names: list[str],
            colors: list[str] | str | None = None):
    """Add a bar chart trace with automatic text labels to the plot.

    Creates a Plotly ``bar`` trace with text annotations positioned automatically on
    top of each bar. The y-axis title is reset to ``'Value'`` so that it always
    reflects this trace.

    Args:
        xs: Category labels displayed on the x-axis (one per bar).
        ys: Numeric values controlling bar heights (one per bar).
        names: Text annotations placed on or near each bar (one per bar).
        colors: A single CSS color string applied uniformly to all bars, or a list of
            per-bar CSS color strings. Defaults to the primary theme color
            ``THEME['main']``.
    """
    if colors is None:
        colors = THEME['main']
    trace = {
        'type': 'bar',
        'x': xs,
        'y': ys,
        'marker': {'color': colors},
        'showlegend': False,
        'text': names,
        'textposition': 'auto'
    }
    self._data.append(trace)
    self._layout['yaxis'].update({'title': 'Value'})

add_trace

add_trace(trace: dict[str, Any])

Add an arbitrary raw Plotly trace dictionary to the plot.

Use this method when none of the typed helpers (add_line, add_bar, etc.) cover the required trace type. The dictionary is appended as-is to the internal trace list without validation.

Parameters:

Name	Type	Description	Default
trace	dict[str, Any]	A fully-specified Plotly trace dictionary. For example, {'type': 'scatter', 'mode': 'lines', 'x': [...], 'y': [...]}.	required

Source code in unaiverse/stats.py

def add_trace(self, trace: dict[str, Any]):
    """Add an arbitrary raw Plotly trace dictionary to the plot.

    Use this method when none of the typed helpers (``add_line``, ``add_bar``, etc.)
    cover the required trace type. The dictionary is appended as-is to the internal
    trace list without validation.

    Args:
        trace: A fully-specified Plotly trace dictionary. For example,
            ``{'type': 'scatter', 'mode': 'lines', 'x': [...], 'y': [...]}``.
    """
    self._data.append(trace)

set_y_range

set_y_range(min_val: float, max_val: float)

Force the Y-axis to a fixed numeric range.

Overrides Plotly's auto-scaling by writing a 'range' key into the 'yaxis' layout entry. Useful when the data range changes dramatically over time and a stable view is preferred.

Parameters:

Name	Type	Description	Default
min_val	float	Lower bound of the Y-axis.	required
max_val	float	Upper bound of the Y-axis.	required

Source code in unaiverse/stats.py

def set_y_range(self, min_val: float, max_val: float):
    """Force the Y-axis to a fixed numeric range.

    Overrides Plotly's auto-scaling by writing a ``'range'`` key into the
    ``'yaxis'`` layout entry. Useful when the data range changes dramatically over
    time and a stable view is preferred.

    Args:
        min_val: Lower bound of the Y-axis.
        max_val: Upper bound of the Y-axis.
    """
    self._layout.setdefault('yaxis', {})['range'] = [min_val, max_val]

set_layout_opt

set_layout_opt(key: str, value: Any)

Set or merge a top-level Plotly layout option.

When both the existing value at key and the incoming value are dictionaries, the existing entry is updated (shallow-merged) rather than replaced. For all other types the key is simply overwritten. This makes it safe to call multiple times on the same key to progressively configure an axis or other composite layout object.

Parameters:

Name	Type	Description	Default
key	str	Top-level Plotly layout key, such as 'xaxis', 'yaxis', 'legend', or 'font'.	required
value	Any	Value to assign. Dicts are merged into the existing entry; all other types overwrite it.	required

Source code in unaiverse/stats.py

def set_layout_opt(self, key: str, value: Any):
    """Set or merge a top-level Plotly layout option.

    When both the existing value at ``key`` and the incoming ``value`` are
    dictionaries, the existing entry is updated (shallow-merged) rather than
    replaced. For all other types the key is simply overwritten. This makes it safe
    to call multiple times on the same key to progressively configure an axis or
    other composite layout object.

    Args:
        key: Top-level Plotly layout key, such as ``'xaxis'``, ``'yaxis'``,
            ``'legend'``, or ``'font'``.
        value: Value to assign. Dicts are merged into the existing entry; all other
            types overwrite it.
    """
    if isinstance(value, dict) and key in self._layout:
        self._layout[key].update(value)
    else:
        self._layout[key] = value

set_legend

set_legend(orientation: str = 'v', x: float = 1.0, y: float = 1.0, xanchor: str = 'left', yanchor: str = 'top')

Configure the legend visibility, position, and orientation.

Enables the legend (showlegend: True) and writes a fully-specified 'legend' entry into the layout, styled with theme-consistent background and border colors.

Parameters:

Name	Type	Description	Default
orientation	str	Legend box orientation. Use 'v' for a vertical list (default) or 'h' for a horizontal row. Defaults to 'v'.	'v'
x	float	Horizontal anchor position in Plotly paper coordinates (0.0 = left edge, 1.0 = right edge). Defaults to 1.0.	1.0
y	float	Vertical anchor position in Plotly paper coordinates (0.0 = bottom, 1.0 = top). Defaults to 1.0.	1.0
xanchor	str	Horizontal alignment of the legend box relative to x. One of 'left', 'center', or 'right'. Defaults to 'left'.	'left'
yanchor	str	Vertical alignment of the legend box relative to y. One of 'top', 'middle', or 'bottom'. Defaults to 'top'.	'top'

Source code in unaiverse/stats.py

def set_legend(self, orientation: str = 'v', x: float = 1.0, y: float = 1.0,
               xanchor: str = 'left', yanchor: str = 'top'):
    """Configure the legend visibility, position, and orientation.

    Enables the legend (``showlegend: True``) and writes a fully-specified
    ``'legend'`` entry into the layout, styled with theme-consistent background and
    border colors.

    Args:
        orientation: Legend box orientation. Use ``'v'`` for a vertical list (default)
            or ``'h'`` for a horizontal row. Defaults to ``'v'``.
        x: Horizontal anchor position in Plotly paper coordinates (0.0 = left edge,
            1.0 = right edge). Defaults to 1.0.
        y: Vertical anchor position in Plotly paper coordinates (0.0 = bottom,
            1.0 = top). Defaults to 1.0.
        xanchor: Horizontal alignment of the legend box relative to ``x``. One of
            ``'left'``, ``'center'``, or ``'right'``. Defaults to ``'left'``.
        yanchor: Vertical alignment of the legend box relative to ``y``. One of
            ``'top'``, ``'middle'``, or ``'bottom'``. Defaults to ``'top'``.
    """
    self._layout['showlegend'] = True
    self._layout['legend'] = {
        'orientation': orientation,
        'x': x,
        'y': y,
        'xanchor': xanchor,
        'yanchor': yanchor,
        'bgcolor': THEME['bg_paper'],
        'bordercolor': THEME['edge'],
        'borderwidth': 1
    }

to_json

to_json() -> str

Serialize the panel to a JSON string expected by the frontend.

Returns:

Type	Description
str	A JSON string containing a top-level object with 'data' (list of trace
str	dicts) and 'layout' (layout configuration dict) keys, ready to be passed
str	directly to Plotly.

Source code in unaiverse/stats.py

def to_json(self) -> str:
    """Serialize the panel to a JSON string expected by the frontend.

    Returns:
        A JSON string containing a top-level object with ``'data'`` (list of trace
        dicts) and ``'layout'`` (layout configuration dict) keys, ready to be passed
        directly to Plotly.
    """
    return json.dumps({'data': self._data, 'layout': self._layout})

DefaultBaseDash

DefaultBaseDash(title: str = 'Network Overview')

A 2x2 grid dashboard that composes four UIPlot panels into a single Plotly figure.

Each of the four fixed cells ('top_left', 'top_right', 'bot_left', 'bot_right') receives one UIPlot via add_panel. The dashboard uses a shared Plotly layout with independent sub-plot axes so traces from different panels do not interfere with each other. Call to_json to obtain the assembled figure ready for the frontend.

The styling follows the dark theme defined by the module-level THEME constant.

Attributes:

Name	Type	Description
traces		Accumulated list of raw Plotly trace dicts contributed by each panel.
layout		Shared Plotly layout dict controlling the 2x2 grid axes, legend, and global appearance.

Initialize the 2x2 grid dashboard with a dark-themed Plotly layout.

Sets up two rows and two columns of independent sub-plot axes with fixed domain coordinates, a shared horizontal legend positioned between the rows, and theme- consistent margins and colors. The four axis pairs are stored in an internal _map dict for use by add_panel.

Parameters:

Name	Type	Description	Default
title	str	Main title text displayed at the top of the dashboard. Defaults to "Network Overview".	'Network Overview'

Source code in unaiverse/stats.py

def __init__(self, title: str = "Network Overview"):
    """Initialize the 2x2 grid dashboard with a dark-themed Plotly layout.

    Sets up two rows and two columns of independent sub-plot axes with fixed domain
    coordinates, a shared horizontal legend positioned between the rows, and theme-
    consistent margins and colors. The four axis pairs are stored in an internal
    ``_map`` dict for use by ``add_panel``.

    Args:
        title: Main title text displayed at the top of the dashboard. Defaults to
            ``"Network Overview"``.
    """
    self.traces = []
    self.layout = {
        "title": title,
        "height": 800,
        "template": "plotly_dark",
        "paper_bgcolor": THEME['bg_paper'],
        "grid": {"rows": 2, "columns": 2, "pattern": "independent"},

        # --- ROW 1 ---
        # Top Left (Graph)
        "xaxis1": {"domain": [0, 0.48]},
        "yaxis1": {"domain": [0.56, 1]},
        # "xaxis1": {"domain": [0, 0.48], "visible": False}, 
        # "yaxis1": {"domain": [0.58, 1], "visible": False},
        # Top Right (Timeseries)
        "xaxis2": {"domain": [0.52, 1]},
        "yaxis2": {"domain": [0.56, 1]},

        # --- ROW 2 ---
        # Bot Left (Bar)
        "xaxis3": {"domain": [0, 0.48]},
        "yaxis3": {"domain": [0, 0.44]},
        # Bot Right (Bar)
        "xaxis4": {"domain": [0.52, 1]},
        "yaxis4": {"domain": [0, 0.44]},

        "showlegend": True,
        "legend": {
            "orientation": "h",
            "y": 0.55,
            "x": 0.55,
            "xanchor": "left",
            "yanchor": "top",
            "bgcolor": "rgba(0,0,0,0)",
            "font": {"color": THEME['text_main']}
        },
        "margin": {"l": 50, "r": 50, "t": 80, "b": 50}
    }
    self._map = {
        "top_left": ("xaxis1", "yaxis1"),
        "top_right": ("xaxis2", "yaxis2"),
        "bot_left": ("xaxis3", "yaxis3"),
        "bot_right": ("xaxis4", "yaxis4")
    }

traces instance-attribute

traces = []

layout instance-attribute

layout = {'title': title, 'height': 800, 'template': 'plotly_dark', 'paper_bgcolor': THEME['bg_paper'], 'grid': {'rows': 2, 'columns': 2, 'pattern': 'independent'}, 'xaxis1': {'domain': [0, 0.48]}, 'yaxis1': {'domain': [0.56, 1]}, 'xaxis2': {'domain': [0.52, 1]}, 'yaxis2': {'domain': [0.56, 1]}, 'xaxis3': {'domain': [0, 0.48]}, 'yaxis3': {'domain': [0, 0.44]}, 'xaxis4': {'domain': [0.52, 1]}, 'yaxis4': {'domain': [0, 0.44]}, 'showlegend': True, 'legend': {'orientation': 'h', 'y': 0.55, 'x': 0.55, 'xanchor': 'left', 'yanchor': 'top', 'bgcolor': 'rgba(0,0,0,0)', 'font': {'color': THEME['text_main']}}, 'margin': {'l': 50, 'r': 50, 't': 80, 'b': 50}}

add_panel

add_panel(ui_plot: UIPlot, position: str)

Merge a UIPlot into one of the four fixed grid positions.

Each trace from ui_plot is copied and assigned the axis references that correspond to the target cell (e.g., 'x1'/'y1' for 'top_left'). Table traces use a 'domain' key instead of axis references. Axis-level layout settings (grid lines, titles, etc.) from the panel are merged into the shared layout without overwriting the fixed domain boundaries. If the panel has a title, it is added as a centered annotation just above the cell.

If position is not one of the four recognized cell names the call is silently ignored.

Parameters:

Name	Type	Description	Default
ui_plot	UIPlot	The UIPlot instance whose traces and layout settings are merged into the dashboard.	required
position	str	Target grid cell. One of 'top_left', 'top_right', 'bot_left', or 'bot_right'.	required

Source code in unaiverse/stats.py

def add_panel(self, ui_plot: UIPlot, position: str):
    """Merge a UIPlot into one of the four fixed grid positions.

    Each trace from ``ui_plot`` is copied and assigned the axis references that
    correspond to the target cell (e.g., ``'x1'``/``'y1'`` for ``'top_left'``).
    Table traces use a ``'domain'`` key instead of axis references. Axis-level layout
    settings (grid lines, titles, etc.) from the panel are merged into the shared
    layout without overwriting the fixed domain boundaries. If the panel has a title,
    it is added as a centered annotation just above the cell.

    If ``position`` is not one of the four recognized cell names the call is silently
    ignored.

    Args:
        ui_plot: The ``UIPlot`` instance whose traces and layout settings are merged
            into the dashboard.
        position: Target grid cell. One of ``'top_left'``, ``'top_right'``,
            ``'bot_left'``, or ``'bot_right'``.
    """
    if position not in self._map:
        return

    xa, ya = self._map[position]
    self.layout: dict[str, dict[str, list[float]]]
    x_dom: list[float] = self.layout[xa]["domain"]
    y_dom: list[float] = self.layout[ya]["domain"]

    # Merge Traces
    for t in ui_plot._data:
        nt = t.copy()
        if nt.get("type") == "table":
            nt["domain"] = {"x": x_dom, "y": y_dom}
        else:
            # Cartesian plots use axis references
            nt["xaxis"] = xa.replace("xaxis", "x")
            nt["yaxis"] = ya.replace("yaxis", "y")
        self.traces.append(nt)

    # Merge Layout
    src_l = ui_plot._layout
    dest_x = self.layout.setdefault(xa, {})
    dest_y = self.layout.setdefault(ya, {})
    if "xaxis" in src_l:
        dest_x.update({k: v for k, v in src_l["xaxis"].items() if k != "domain"})
    if "yaxis" in src_l:
        dest_y.update({k: v for k, v in src_l["yaxis"].items() if k != "domain"})

    # Add Title via Annotation
    if src_l.get("title"):
        self.layout: dict[str, list]
        self.layout.setdefault("annotations", []).append({
            "text": f"<b>{src_l['title']}</b>",
            "x": (x_dom[0] + x_dom[1]) / 2,
            "y": y_dom[1] + 0.02,
            "xref": "paper", "yref": "paper",
            "showarrow": False, "xanchor": "center", "yanchor": "bottom",
            "font": {"size": 14, "color": THEME['text_main']}
        })

to_json

to_json() -> str

Serialize the assembled dashboard to a JSON string expected by the frontend.

Returns:

Type	Description
str	A JSON string containing a top-level object with 'data' (the combined
str	trace list from all panels) and 'layout' (the shared grid layout), ready
str	to be passed directly to Plotly.

Source code in unaiverse/stats.py

def to_json(self) -> str:
    """Serialize the assembled dashboard to a JSON string expected by the frontend.

    Returns:
        A JSON string containing a top-level object with ``'data'`` (the combined
        trace list from all panels) and ``'layout'`` (the shared grid layout), ready
        to be passed directly to Plotly.
    """
    return json.dumps({"data": self.traces, "layout": self.layout})

Stats

Stats(is_world: bool, db_path: str | None = None, cache_window_hours: float = 2.0)

Statistics engine for UNaIVERSE agents and worlds.

Stats operates in two distinct modes controlled by is_world:

• Agent mode (is_world=False): maintains a lightweight send buffer (_update_batch) and a local world-view snapshot (_world_view). Agents call store_stat to queue measurements for transmission and get_payload_for_world to retrieve the pending batch before flushing it over the network.

• World mode (is_world=True): runs the full persistence stack. Incoming measurements are type-validated, written to an in-memory rolling hot-cache (_stats), and also queued in write buffers that are flushed to an SQLite database on save_to_disk. Historical queries are served from SQLite via query_history.

Schema design

Statistics are declared by setting class-level schema dictionaries (e.g., CORE_WORLD_STATS_DYNAMIC_SCHEMA) before any instance is created. Custom subclasses may extend the built-in schemas via the CUSTOM_* class variables. Each schema maps a stat name to a (type, default) pair. At construction time all schemas are merged and the resulting master key sets (all_static_keys, all_dynamic_keys, etc.) are used throughout for fast O(1) lookups.

Stat categories

• Static stats: single latest value per group_key (e.g., current HSM state). On the Agent side, duplicate entries for the same group_key are removed from the buffer before the new value is appended (de-duplication).
• Dynamic stats: time-series values stored in a SortedDict keyed by millisecond timestamps on the World side. The hot cache retains only the rolling window defined by cache_window_hours; older data lives only on disk.

Grouping

Stats are either ungrouped (world-level, e.g., membership counts) or grouped under a group_key (usually a peer ID). Ungrouped world stats live directly in _stats[stat_name]; grouped stats live in _stats[GROUP_KEY][group_key][stat_name].

Plotting

Call plot to obtain a self-contained HTML dashboard that visualizes the world topology, agent-count history, state distribution, and last-action distribution. Agents that receive a world-view snapshot via update_view can also call plot to render their cached view.

Attributes:

Name	Type	Description
is_world	bool	True if this instance operates in World mode with full persistence.
max_seen_timestamp	int	Largest millisecond timestamp encountered so far; used for rolling-window pruning.
all_static_keys	set[str]	Set of all stat names classified as static across all schemas.
all_dynamic_keys	set[str]	Set of all stat names classified as dynamic across all schemas.
all_keys	set[str]	Union of all_static_keys and all_dynamic_keys.
stat_types		Mapping from stat name to its schema-declared Python type.
STORE_DYNAMIC_IF_CHANGED		Class-level flag; when True, dynamic stats are only stored when the value differs from the previous measurement. Defaults to False.
GROUP_KEY		Class-level constant used as the top-level key in _stats for all grouped (peer) statistics. Value is 'peer_stats'.

Initialize the Stats engine in either World or Agent mode.

When is_world is True, the full World-side stack is set up: all CORE and CUSTOM schemas are merged, master key sets are populated, the SQLite database is opened (or created) at db_path, the in-memory hot-cache structure is initialized, and any existing data within the rolling window is loaded from disk. The parent directory of db_path is created automatically if it does not exist.

When is_world is False, only a lightweight send buffer (_update_batch) and an empty world-view snapshot (_world_view) are allocated. No database connection is opened.

Parameters:

Name	Type	Description	Default
is_world	bool	True to initialize as a World (full persistence stack), False to initialize as an Agent (lightweight buffer only).	required
db_path	str | None	Filesystem path for the SQLite database file. Required when is_world is True; ignored when False. Defaults to None.	None
cache_window_hours	float	Duration in hours of the in-memory rolling window kept by the World hot cache. Dynamic data older than this window is evicted from RAM but remains on disk and is queryable via query_history. Only meaningful when is_world is True. Defaults to 2.0.	2.0

Source code in unaiverse/stats.py

def __init__(self, is_world: bool,
             db_path: str | None = None,  # only needed by the world
             cache_window_hours: float = 2.0):  # only needed by the world
    """Initialize the Stats engine in either World or Agent mode.

    When ``is_world`` is ``True``, the full World-side stack is set up: all CORE and
    CUSTOM schemas are merged, master key sets are populated, the SQLite database is
    opened (or created) at ``db_path``, the in-memory hot-cache structure is
    initialized, and any existing data within the rolling window is loaded from disk.
    The parent directory of ``db_path`` is created automatically if it does not exist.

    When ``is_world`` is ``False``, only a lightweight send buffer (``_update_batch``)
    and an empty world-view snapshot (``_world_view``) are allocated. No database
    connection is opened.

    Args:
        is_world: ``True`` to initialize as a World (full persistence stack), ``False``
            to initialize as an Agent (lightweight buffer only).
        db_path: Filesystem path for the SQLite database file. Required when
            ``is_world`` is ``True``; ignored when ``False``. Defaults to None.
        cache_window_hours: Duration in hours of the in-memory rolling window kept
            by the World hot cache. Dynamic data older than this window is evicted
            from RAM but remains on disk and is queryable via ``query_history``.
            Only meaningful when ``is_world`` is ``True``. Defaults to 2.0.
    """
    self.is_world: bool = is_world
    self.max_seen_timestamp: int = 0

    # --- Integrate custom statistics ---
    self.WORLD_STATS_STATIC_SCHEMA = self.CORE_WORLD_STATS_STATIC_SCHEMA | self.CUSTOM_WORLD_STATS_STATIC_SCHEMA
    self.WORLD_STATS_DYNAMIC_SCHEMA = self.CORE_WORLD_STATS_DYNAMIC_SCHEMA | self.CUSTOM_WORLD_STATS_DYNAMIC_SCHEMA
    self.AGENT_STATS_STATIC_SCHEMA = self.CORE_AGENT_STATS_STATIC_SCHEMA | self.CUSTOM_AGENT_STATS_STATIC_SCHEMA
    self.AGENT_STATS_DYNAMIC_SCHEMA = self.CORE_AGENT_STATS_DYNAMIC_SCHEMA | self.CUSTOM_AGENT_STATS_DYNAMIC_SCHEMA
    self.OUTER_STATS_STATIC_SCHEMA = self.CORE_OUTER_STATS_STATIC_SCHEMA | self.CUSTOM_OUTER_STATS_STATIC_SCHEMA
    self.OUTER_STATS_DYNAMIC_SCHEMA = self.CORE_OUTER_STATS_DYNAMIC_SCHEMA | self.CUSTOM_OUTER_STATS_DYNAMIC_SCHEMA

    # --- Master key sets for easier lookup ---
    self.all_static_keys: set[str] = set()
    self.all_dynamic_keys: set[str] = set()
    self.all_keys: set[str] = set()
    self.world_grouped_keys: set[str] = set()
    self.world_ungrouped_keys: set[str] = set()
    self.agent_grouped_keys: set[str] = set()
    self.agent_ungrouped_keys: set[str] = set()
    self.stat_types = {}
    self._stat_defaults: dict[str, Any] = {}
    self._initialize_key_sets()

    if self.is_world:
        # --- World Configuration ---
        self._stats: dict[str, Any] = {self.GROUP_KEY: {}}
        self.min_window_duration = timedelta(hours=cache_window_hours)
        self.db_path = db_path
        self._db_conn: sqlite3.Connection | None = None
        self._static_db_buffer = []
        self._dynamic_db_buffer = []

        # --- World Initialization ---
        self._init_db()  # Connect and create tables
        self._initialize_cache_structure()  # Ensures all keys exist
        self._load_existing_stats()  # Hydrates _stats from disk
    else:
        # --- Agent Initialization (Simple Buffer) ---
        self._world_view: dict[str, Any] = {}
        self._update_batch: list[dict[str, Any]] = []

CORE_WORLD_STATS_STATIC_SCHEMA class-attribute instance-attribute

CORE_WORLD_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {'graph': (dict, {'nodes': {}, 'edges': {}})}

CORE_WORLD_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CORE_WORLD_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {'world_masters': (int, 0), 'world_agents': (int, 0), 'human_agents': (int, 0), 'artificial_agents': (int, 0)}

CORE_AGENT_STATS_STATIC_SCHEMA class-attribute instance-attribute

CORE_AGENT_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {'connected_peers': (list, []), 'state': (str, None), 'action': (str, None), 'last_action': (str, None)}

CORE_AGENT_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CORE_AGENT_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CORE_OUTER_STATS_STATIC_SCHEMA class-attribute instance-attribute

CORE_OUTER_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CORE_OUTER_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CORE_OUTER_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_WORLD_STATS_STATIC_SCHEMA class-attribute instance-attribute

CUSTOM_WORLD_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_WORLD_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CUSTOM_WORLD_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_AGENT_STATS_STATIC_SCHEMA class-attribute instance-attribute

CUSTOM_AGENT_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_AGENT_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CUSTOM_AGENT_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_OUTER_STATS_STATIC_SCHEMA class-attribute instance-attribute

CUSTOM_OUTER_STATS_STATIC_SCHEMA: dict[str, tuple[type, Any]] = {}

CUSTOM_OUTER_STATS_DYNAMIC_SCHEMA class-attribute instance-attribute

CUSTOM_OUTER_STATS_DYNAMIC_SCHEMA: dict[str, tuple[type, Any]] = {}

STORE_DYNAMIC_IF_CHANGED class-attribute instance-attribute

STORE_DYNAMIC_IF_CHANGED = False

GROUP_KEY class-attribute instance-attribute

GROUP_KEY = 'peer_stats'

DEBUG class-attribute instance-attribute

DEBUG = False

is_world instance-attribute

is_world: bool = is_world

max_seen_timestamp instance-attribute

max_seen_timestamp: int = 0

WORLD_STATS_STATIC_SCHEMA instance-attribute

WORLD_STATS_STATIC_SCHEMA = CORE_WORLD_STATS_STATIC_SCHEMA | CUSTOM_WORLD_STATS_STATIC_SCHEMA

WORLD_STATS_DYNAMIC_SCHEMA instance-attribute

WORLD_STATS_DYNAMIC_SCHEMA = CORE_WORLD_STATS_DYNAMIC_SCHEMA | CUSTOM_WORLD_STATS_DYNAMIC_SCHEMA

AGENT_STATS_STATIC_SCHEMA instance-attribute

AGENT_STATS_STATIC_SCHEMA = CORE_AGENT_STATS_STATIC_SCHEMA | CUSTOM_AGENT_STATS_STATIC_SCHEMA

AGENT_STATS_DYNAMIC_SCHEMA instance-attribute

AGENT_STATS_DYNAMIC_SCHEMA = CORE_AGENT_STATS_DYNAMIC_SCHEMA | CUSTOM_AGENT_STATS_DYNAMIC_SCHEMA

OUTER_STATS_STATIC_SCHEMA instance-attribute

OUTER_STATS_STATIC_SCHEMA = CORE_OUTER_STATS_STATIC_SCHEMA | CUSTOM_OUTER_STATS_STATIC_SCHEMA

OUTER_STATS_DYNAMIC_SCHEMA instance-attribute

OUTER_STATS_DYNAMIC_SCHEMA = CORE_OUTER_STATS_DYNAMIC_SCHEMA | CUSTOM_OUTER_STATS_DYNAMIC_SCHEMA

all_static_keys instance-attribute

all_static_keys: set[str] = set()

all_dynamic_keys instance-attribute

all_dynamic_keys: set[str] = set()

all_keys instance-attribute

all_keys: set[str] = set()

world_grouped_keys instance-attribute

world_grouped_keys: set[str] = set()

world_ungrouped_keys instance-attribute

world_ungrouped_keys: set[str] = set()

agent_grouped_keys instance-attribute

agent_grouped_keys: set[str] = set()

agent_ungrouped_keys instance-attribute

agent_ungrouped_keys: set[str] = set()

stat_types instance-attribute

stat_types = {}

min_window_duration instance-attribute

min_window_duration = timedelta(hours=cache_window_hours)

db_path instance-attribute

db_path = db_path

store_stat

store_stat(stat_name: str, value: Any, group_key: str, timestamp: int)

Store a single measurement, dispatching to static or dynamic storage.

The stat name is looked up in all_static_keys and all_dynamic_keys to determine the storage path. Static stats are routed to _store_static; dynamic stats are routed to _store_dynamic. If stat_name is not recognized, an error is logged and the call returns without storing anything.

Parameters:

Name	Type	Description	Default
stat_name	str	Name of the statistic as declared in one of the schema dictionaries.	required
value	Any	Measurement value. Type-validated and cast to the schema-declared type before storage.	required
group_key	str	Grouping identifier for this measurement, typically a peer ID. Used to bucket related records in the world hot-cache and in the database.	required
timestamp	int	Millisecond Unix timestamp associated with the measurement.	required

Source code in unaiverse/stats.py

def store_stat(self, stat_name: str, value: Any, group_key: str, timestamp: int):
    """Store a single measurement, dispatching to static or dynamic storage.

    The stat name is looked up in ``all_static_keys`` and ``all_dynamic_keys`` to
    determine the storage path. Static stats are routed to ``_store_static``; dynamic
    stats are routed to ``_store_dynamic``. If ``stat_name`` is not recognized, an
    error is logged and the call returns without storing anything.

    Args:
        stat_name: Name of the statistic as declared in one of the schema dictionaries.
        value: Measurement value. Type-validated and cast to the schema-declared type
            before storage.
        group_key: Grouping identifier for this measurement, typically a peer ID. Used
            to bucket related records in the world hot-cache and in the database.
        timestamp: Millisecond Unix timestamp associated with the measurement.
    """
    if stat_name not in self.all_keys:
        log.error(f'Stat "{stat_name}" is not defined')
        return

    # disambiguate between static and dynamic stats
    if stat_name in self.all_static_keys:
        self._store_static(stat_name, value, group_key, timestamp)
    else:
        self._store_dynamic(stat_name, value, group_key, timestamp)

update_view

update_view(view_data: dict[str, Any] | None = None, overwrite: bool = False)

Merge a world-view snapshot received from the World into the local cache (Agent-side).

This method is a no-op on the World side. On the Agent side it incrementally updates _world_view with the incoming data. Dynamic stats (whose names appear in all_dynamic_keys) are stored as [[timestamp, value], ...] lists and are extended (new pairs appended) rather than replaced on repeated calls. Static stats are simply overwritten. The running maximum seen timestamp is updated as dynamic data arrives.

If overwrite is True, the existing _world_view is discarded before merging, giving a clean snapshot of the new data only.

The expected structure of view_data is the same dictionary produced by the World's get_view method::

{
    "world": { "stat_name": value_or_timeseries },
    "peers": { "peer_id": { "stat_name": value_or_timeseries } }
}

Parameters:

Name	Type	Description	Default
view_data	dict[str, Any] | None	The snapshot dictionary received from the world, matching the structure produced by get_view. Pass None to perform no update.	None
overwrite	bool	If True, the current local view is cleared before merging. Defaults to False.	False

Source code in unaiverse/stats.py

def update_view(self, view_data: dict[str, Any] | None = None, overwrite: bool = False):
    """Merge a world-view snapshot received from the World into the local cache (Agent-side).

    This method is a no-op on the World side. On the Agent side it incrementally
    updates ``_world_view`` with the incoming data. Dynamic stats (whose names appear
    in ``all_dynamic_keys``) are stored as ``[[timestamp, value], ...]`` lists and are
    *extended* (new pairs appended) rather than replaced on repeated calls. Static
    stats are simply overwritten. The running maximum seen timestamp is updated as
    dynamic data arrives.

    If ``overwrite`` is ``True``, the existing ``_world_view`` is discarded before
    merging, giving a clean snapshot of the new data only.

    The expected structure of ``view_data`` is the same dictionary produced by the
    World's ``get_view`` method::

        {
            "world": { "stat_name": value_or_timeseries },
            "peers": { "peer_id": { "stat_name": value_or_timeseries } }
        }

    Args:
        view_data: The snapshot dictionary received from the world, matching the
            structure produced by ``get_view``. Pass ``None`` to perform no update.
        overwrite: If ``True``, the current local view is cleared before merging.
            Defaults to False.
    """
    if self.is_world:
        return

    # Initialize empty structure if needed
    if not self._world_view or overwrite:
        self._world_view = {'world': {}, 'peers': {}}

    def _update_max_ts(ts):
        """Helper to update the max seen timestamp from a time-series."""
        # Dynamic stats come as [[ts, val], [ts, val]...]
        if isinstance(ts, list) and len(ts) > 0 and isinstance(ts[0], list):
            # The last item is usually the newest in sorted time-series
            last_ts = ts[-1][0]
            if last_ts > self.max_seen_timestamp:
                self.max_seen_timestamp = int(last_ts)

    def _merge_dict(target: dict, source: dict):
        """
        Helper to merge source into target with special handling for dynamic stats.
        Copies a source dict { "stat_name": value_or_timeseries } into target.
        """
        for stat_name, val_or_ts in source.items():
            if stat_name in self.all_dynamic_keys:
                _update_max_ts(val_or_ts)
                if stat_name not in target:
                    target[stat_name] = []
                target[stat_name].extend(val_or_ts)
            else:
                target[stat_name] = val_or_ts

    # 1. Merge World (Ungrouped) Stats
    if 'world' in view_data:
        _merge_dict(self._world_view.setdefault('world', {}), view_data['world'])

    # 2. Merge Peer (Grouped) Stats
    if 'peers' in view_data:
        target_peers = self._world_view.setdefault('peers', {})
        for peer_id, peer_data in view_data['peers'].items():
            target_peer = target_peers.setdefault(peer_id, {})
            _merge_dict(target_peer, peer_data)

get_stats

get_stats() -> dict[str, Any]

Return the raw internal hot-cache dictionary (World-only).

Provides direct access to the live _stats structure for callers that need to read or manipulate the cache in ways not covered by the higher-level query API. The returned object is the actual internal dict, not a copy; mutations will affect the running state of the Stats engine.

Returns:

Type	Description
dict[str, Any]	The _stats hot-cache dictionary. Top-level keys are the ungrouped world
dict[str, Any]	stat names (e.g., 'world_masters', 'graph') plus GROUP_KEY
dict[str, Any]	('peer_stats'), which maps each peer ID to its per-stat sub-dict.
dict[str, Any]	On the Agent side this attribute does not exist; this method should only be
dict[str, Any]	called from World-mode instances.

Source code in unaiverse/stats.py

def get_stats(self) -> dict[str, Any]:
    """Return the raw internal hot-cache dictionary (World-only).

    Provides direct access to the live ``_stats`` structure for callers that need
    to read or manipulate the cache in ways not covered by the higher-level query
    API. The returned object is the actual internal dict, not a copy; mutations
    will affect the running state of the Stats engine.

    Returns:
        The ``_stats`` hot-cache dictionary. Top-level keys are the ungrouped world
        stat names (e.g., ``'world_masters'``, ``'graph'``) plus ``GROUP_KEY``
        (``'peer_stats'``), which maps each peer ID to its per-stat sub-dict.
        On the Agent side this attribute does not exist; this method should only be
        called from World-mode instances.
    """
    return self._stats

get_payload_for_world

get_payload_for_world(clear_buffer: bool = True) -> list[dict[str, Any]]

Return the pending stats batch to be transmitted to the world (Agent-only).

Retrieves the current _update_batch and, unless clear_buffer is False, replaces it with an empty list so that subsequent calls do not re-send already dispatched updates. On the World side this always returns an empty list.

Parameters:

Name	Type	Description	Default
clear_buffer	bool	If True, the internal send buffer is cleared after the batch is captured. Set to False to peek at the buffer without consuming it. Defaults to True.	True

Returns:

Type	Description
list[dict[str, Any]]	A list of stat-update dicts ready for transmission. Each dict contains the
list[dict[str, Any]]	keys 'group_key', 'stat_name', 'timestamp', and 'value'.
list[dict[str, Any]]	Returns an empty list when called on the World side.

Source code in unaiverse/stats.py

def get_payload_for_world(self, clear_buffer: bool = True) -> list[dict[str, Any]]:
    """Return the pending stats batch to be transmitted to the world (Agent-only).

    Retrieves the current ``_update_batch`` and, unless ``clear_buffer`` is ``False``,
    replaces it with an empty list so that subsequent calls do not re-send already
    dispatched updates. On the World side this always returns an empty list.

    Args:
        clear_buffer: If ``True``, the internal send buffer is cleared after the
            batch is captured. Set to ``False`` to peek at the buffer without
            consuming it. Defaults to True.

    Returns:
        A list of stat-update dicts ready for transmission. Each dict contains the
        keys ``'group_key'``, ``'stat_name'``, ``'timestamp'``, and ``'value'``.
        Returns an empty list when called on the World side.
    """
    if self.is_world:
        return []

    # self._update_agent_static()  # Ensure static stats are fresh in the batch
    payload = self._update_batch
    if clear_buffer:
        self._update_batch = []  # Clear after getting
    return payload

get_view

get_view(since_timestamp: int = 0) -> dict[str, Any]

Return a JSON-serializable snapshot of the current in-memory hot cache (World-side).

Serializes both ungrouped world stats and all peer-grouped stats from the _stats hot cache. Dynamic stats (SortedDict caches) are sliced to include only entries with a timestamp strictly greater than since_timestamp, making this method efficient for incremental polling: pass the last received timestamp to retrieve only new data points.

Used for initial handshakes (since_timestamp=0) and for lightweight incremental streaming. For historical data beyond the rolling window, use query_history instead.

Parameters:

Name	Type	Description	Default
since_timestamp	int	Millisecond Unix timestamp lower bound. Only dynamic data points recorded after this value are included. Pass 0 to include all data currently in the rolling cache. Defaults to 0.	0

Returns:

Type	Description
dict[str, Any]	A dictionary with the structure:: { "world": { "stat_name": value_or_timeseries }, "peers": { "peer_id": { "stat_name": value_or_timeseries } } }
dict[str, Any]	For dynamic stats, value_or_timeseries is a list of [timestamp, value]
dict[str, Any]	pairs. Empty peer entries (no data points after the cutoff) are omitted.
dict[str, Any]	Returns an empty dict when called on the Agent side.

Source code in unaiverse/stats.py

def get_view(self, since_timestamp: int = 0) -> dict[str, Any]:
    """Return a JSON-serializable snapshot of the current in-memory hot cache (World-side).

    Serializes both ungrouped world stats and all peer-grouped stats from the
    ``_stats`` hot cache. Dynamic stats (``SortedDict`` caches) are sliced to include
    only entries with a timestamp strictly greater than ``since_timestamp``, making
    this method efficient for incremental polling: pass the last received timestamp
    to retrieve only new data points.

    Used for initial handshakes (``since_timestamp=0``) and for lightweight
    incremental streaming. For historical data beyond the rolling window, use
    ``query_history`` instead.

    Args:
        since_timestamp: Millisecond Unix timestamp lower bound. Only dynamic data
            points recorded after this value are included. Pass ``0`` to include all
            data currently in the rolling cache. Defaults to 0.

    Returns:
        A dictionary with the structure::

            {
                "world": { "stat_name": value_or_timeseries },
                "peers": { "peer_id": { "stat_name": value_or_timeseries } }
            }

        For dynamic stats, ``value_or_timeseries`` is a list of ``[timestamp, value]``
        pairs. Empty peer entries (no data points after the cutoff) are omitted.
        Returns an empty dict when called on the Agent side.
    """
    if not self.is_world:
        return {}
    snapshot = {'world': {}, 'peers': {}}

    # 1. Process World (Ungrouped) Stats
    for stat_name in self.world_ungrouped_keys:
        val = self._stats.get(stat_name)
        if val is not None:
            snapshot['world'][stat_name] = self._serialize_value(val, since_timestamp)

    # 2. Process Peer (Grouped) Stats
    peer_groups = self._stats.get(self.GROUP_KEY, {})

    for pid in peer_groups.keys():
        peer_data = {}
        for stat_name, val in peer_groups[pid].items():
            serialized = self._serialize_value(val, since_timestamp)
            # Optimize: Don't send empty lists if polling
            if isinstance(serialized, list) and len(serialized) == 0:
                continue
            peer_data[stat_name] = serialized

        if peer_data:
            snapshot['peers'][pid] = peer_data

    return snapshot

get_last_value

get_last_value(stat_name: str, group_key: str | None = None) -> Any | None

Return the most recent value of a stat, regardless of whether it is static or dynamic.

Dispatches to _get_last_static_value or _get_last_dynamic_value based on stat_name's membership in the master key sets. If the stat name is not recognised an error is logged and None is returned.

Parameters:

Name	Type	Description	Default
stat_name	str	Name of the statistic to retrieve. Must be registered in one of the schema dictionaries.	required
group_key	str | None	When None, the lookup targets ungrouped (world-level) stats. When provided, the lookup searches the grouped stats for that key (usually a peer ID). Defaults to None.	None

Returns:

Type	Description
Any | None	The most recently recorded value for the stat, or None if the stat is
Any | None	unknown, has no data, or the cache does not hold a value for the requested key.

Source code in unaiverse/stats.py

def get_last_value(self, stat_name: str, group_key: str | None = None) -> Any | None:
    """Return the most recent value of a stat, regardless of whether it is static or dynamic.

    Dispatches to ``_get_last_static_value`` or ``_get_last_dynamic_value`` based on
    ``stat_name``'s membership in the master key sets. If the stat name is not
    recognised an error is logged and ``None`` is returned.

    Args:
        stat_name: Name of the statistic to retrieve. Must be registered in one of
            the schema dictionaries.
        group_key: When ``None``, the lookup targets ungrouped (world-level) stats.
            When provided, the lookup searches the grouped stats for that key (usually
            a peer ID). Defaults to None.

    Returns:
        The most recently recorded value for the stat, or ``None`` if the stat is
        unknown, has no data, or the cache does not hold a value for the requested key.
    """
    if stat_name in self.all_static_keys:
        return self._get_last_static_value(stat_name, group_key)
    elif stat_name in self.all_dynamic_keys:
        return self._get_last_dynamic_value(stat_name, group_key)
    else:
        log.error(f'get_last_value: Unknown stat_name "{stat_name}"')
        return None

save_to_disk

save_to_disk()

Flush all pending buffers to SQLite and prune stale data (World-only).

Sequentially calls _save_static_to_db, _save_dynamic_to_db, _prune_cache, and _prune_db, then commits the transaction. If the instance is not in World mode or the database connection is unavailable, the call returns immediately without doing anything. Any exception raised during the save is caught, logged, and the transaction is rolled back so that no partial writes are committed.

Note

On the Agent side this method is a no-op.

Source code in unaiverse/stats.py

def save_to_disk(self):
    """Flush all pending buffers to SQLite and prune stale data (World-only).

    Sequentially calls ``_save_static_to_db``, ``_save_dynamic_to_db``,
    ``_prune_cache``, and ``_prune_db``, then commits the transaction. If the
    instance is not in World mode or the database connection is unavailable, the
    call returns immediately without doing anything. Any exception raised during
    the save is caught, logged, and the transaction is rolled back so that no
    partial writes are committed.

    Note:
        On the Agent side this method is a no-op.
    """
    if not self.is_world or not self._db_conn:
        return
    log.debug('Saving world stats to DB...')
    try:
        self._save_static_to_db()
        self._save_dynamic_to_db()
        self._prune_cache()
        self._prune_db()

        self._db_conn.commit()
        log.debug('Save complete.')
    except Exception as e:
        log.error(f'CRITICAL: Save_to_disk failed: {e}')
        if self._db_conn:
            self._db_conn.rollback()

query_history

query_history(stat_names: list[str] | None = None, group_keys: list[str] | None = None, time_range: tuple[int, int] | int | None = None, value_range: tuple[float, float] | None = None, limit: int | None = None) -> dict[str, Any]

Query the SQLite database for historical stats with optional filters (World-only).

Automatically flushes the in-memory static and dynamic buffers to the database before executing the query, ensuring "read-your-writes" consistency. Both static and dynamic tables are queried separately and merged into the same structure produced by get_view, so the result can be fed directly into update_view on the Agent side.

Static stats are loaded without a time filter (each stat has at most one row per peer). Dynamic stats are filtered by time_range, value_range, and limit, then ordered by timestamp ascending.

Parameters:

Name	Type	Description	Default
stat_names	list[str] | None	Restrict results to these stat names. Pass None or an empty list to include all registered stats. Defaults to None.	None
group_keys	list[str] | None	Restrict results to these group keys (peer IDs). Pass None or an empty list to include all groups. Defaults to None.	None
time_range	tuple[int, int] | int | None	Time filter applied only to dynamic stats. An int is treated as a lower-bound ("since X ms"). A two-element tuple or list (start, end) filters to the closed interval [start, end]. Pass None to apply no time filter. Defaults to None.	None
value_range	tuple[float, float] | None	(min, max) numeric range applied to the val_num column. Dynamic rows that have no numeric value are excluded when this filter is active. Pass None for no range filter. Defaults to None.	None
limit	int | None	Maximum number of dynamic-stat rows to return. Defaults to 5000 when None.	None

Returns:

Type	Description
dict[str, Any]	A dictionary with the same structure as get_view:: { "world": { "stat_name": value_or_timeseries }, "peers": { "peer_id": { "stat_name": value_or_timeseries } } }
dict[str, Any]	Returns an empty dict when called on the Agent side or when the database
dict[str, Any]	connection is unavailable. Individual query errors are caught and logged;
dict[str, Any]	partial results may be returned.

Source code in unaiverse/stats.py

def query_history(self,
                  stat_names: list[str] | None = None,
                  group_keys: list[str] | None = None,
                  time_range: tuple[int, int] | int | None = None,
                  value_range: tuple[float, float] | None = None,
                  limit: int | None = None) -> dict[str, Any]:
    """Query the SQLite database for historical stats with optional filters (World-only).

    Automatically flushes the in-memory static and dynamic buffers to the database
    before executing the query, ensuring "read-your-writes" consistency. Both static
    and dynamic tables are queried separately and merged into the same structure
    produced by ``get_view``, so the result can be fed directly into ``update_view``
    on the Agent side.

    Static stats are loaded without a time filter (each stat has at most one row per
    peer). Dynamic stats are filtered by ``time_range``, ``value_range``, and
    ``limit``, then ordered by timestamp ascending.

    Args:
        stat_names: Restrict results to these stat names. Pass ``None`` or an empty
            list to include all registered stats. Defaults to None.
        group_keys: Restrict results to these group keys (peer IDs). Pass ``None`` or
            an empty list to include all groups. Defaults to None.
        time_range: Time filter applied only to dynamic stats. An ``int`` is treated
            as a lower-bound ("since X ms"). A two-element tuple or list
            ``(start, end)`` filters to the closed interval ``[start, end]``. Pass
            ``None`` to apply no time filter. Defaults to None.
        value_range: ``(min, max)`` numeric range applied to the ``val_num`` column.
            Dynamic rows that have no numeric value are excluded when this filter is
            active. Pass ``None`` for no range filter. Defaults to None.
        limit: Maximum number of dynamic-stat rows to return. Defaults to 5000 when
            ``None``.

    Returns:
        A dictionary with the same structure as ``get_view``::

            {
                "world": { "stat_name": value_or_timeseries },
                "peers": { "peer_id": { "stat_name": value_or_timeseries } }
            }

        Returns an empty dict when called on the Agent side or when the database
        connection is unavailable. Individual query errors are caught and logged;
        partial results may be returned.
    """
    if stat_names is None:
        stat_names = []
    if group_keys is None:
        group_keys = []
    if not self.is_world or not self._db_conn:
        return {}

    # Flush the cached upadtes to db before querying
    self._save_static_to_db()
    self._save_dynamic_to_db()
    self._db_conn.commit()

    snapshot = {'world': {}, 'peers': {}}

    # A. Query the static stats
    query_static = ['SELECT peer_id, stat_name, val_json FROM static_stats']
    params_static = []

    where_added = False
    if stat_names:
        query_static.append("WHERE")
        where_added = True
        query_static.append(f"stat_name IN ({','.join(['?'] * len(stat_names))})")
        params_static.extend(stat_names)
    if group_keys:
        if not where_added:
            query_static.append("WHERE")
        else:
            query_static.append(f"AND")
        query_static.append(f"peer_id IN ({','.join(['?'] * len(group_keys))})")
        params_static.extend(group_keys)

    try:
        cursor = self._db_conn.execute(' '.join(query_static), params_static)
        for pid, sname, vjson in cursor:
            val = self._validate_type(sname, json.loads(vjson))
            # Handle special Graph reconstruction if needed (legacy format support)
            if sname == 'graph':
                # Handle both legacy format (just edges) and new format (nodes+edges) safely
                if isinstance(val, dict) and 'edges' in val:
                    # Convert the edge lists back to sets
                    val['edges'] = {k: set(v) for k, v in val['edges'].items()}
                    # Ensure nodes dict exists
                    if 'nodes' not in val:
                        val['nodes'] = {}
                else:
                    val: dict
                    # Convert entire dict to sets (as it was before)
                    edges_set = {k: set(v) for k, v in val.items()}
                    # Migrate to new structure on the fly
                    val = {'nodes': {}, 'edges': edges_set}

            # Static stats format: value (direct)
            if pid in (None, 'None', ''):
                snapshot['world'][sname] = val
            else:
                snapshot['peers'].setdefault(pid, {})[sname] = val
    except Exception as e:
        log.error(f'Query history (static) failed: {e}')

    # B. Query the dynamic stats
    query_dyn = ['SELECT timestamp, peer_id, stat_name, val_num, val_str, val_json FROM dynamic_stats']
    params_dyn = []

    # 1. Stat Names
    where_added = False
    if stat_names:
        query_static.append("WHERE")
        where_added = True
        query_dyn.append(f"stat_name IN ({','.join(['?'] * len(stat_names))})")
        params_dyn.extend(stat_names)

    # 2. Group Keys
    if group_keys:
        if not where_added:
            query_static.append("WHERE")
        else:
            query_static.append("AND")
        query_dyn.append(f"peer_id IN ({','.join(['?'] * len(group_keys))})")
        params_dyn.extend(group_keys)

    if time_range is not None:
        if isinstance(time_range, int):
            # Treated as "Since X"
            if not where_added:
                query_static.append("WHERE")
            else:
                query_static.append("AND")
            query_dyn.append("timestamp >= ?")
            params_dyn.append(time_range)
        elif isinstance(time_range, (tuple, list)) and len(time_range) == 2:
            # Treated as "Between X and Y"
            if not where_added:
                query_static.append("WHERE")
            else:
                query_static.append("AND")
            query_dyn.append("timestamp >= ? AND timestamp <= ?")
            params_dyn.extend([time_range[0], time_range[1]])

    # 4. Value Range (The logic requested by user)
    if value_range:
        if not where_added:
            query_static.append("WHERE")
        else:
            query_static.append("AND")
        query_dyn.append("val_num IS NOT NULL AND val_num >= ? AND val_num <= ?")
        params_dyn.extend([value_range[0], value_range[1]])

    query_dyn.append("ORDER BY timestamp ASC")

    # add the limit
    query_dyn.append("LIMIT 5000" if limit is None else f"LIMIT {limit}")

    try:
        cursor = self._db_conn.execute(' '.join(query_dyn), params_dyn)
        for ts, pid, sname, vnum, vstr, vjson in cursor:
            ts = int(ts)
            val = vnum if vnum is not None else (vstr if vstr is not None else json.loads(vjson))
            val = self._validate_type(sname, val)

            # Structure construction
            if pid in (None, 'None', ''):  # Handling world stats
                target_ts = snapshot['world'].setdefault(sname, [])
            else:  # Handling peer stats
                target_ts = snapshot['peers'].setdefault(pid, {}).setdefault(sname, [])
            target_ts.append([ts, val])

    except Exception as e:
        log.error(f'Query history failed: {e}')

    return snapshot

shutdown

shutdown() -> None

Flush all pending stats to disk and close the SQLite connection gracefully.

Calls save_to_disk to persist any buffered measurements, then closes the database connection and sets _db_conn to None. Any exception raised by the final save is caught and logged rather than propagated, so the connection is always closed even when the flush fails. On the Agent side this method is a no-op.

Note

Call this method explicitly at application shutdown. The destructor (__del__) also attempts a final save, but relying on the destructor for critical I/O is not recommended.

Source code in unaiverse/stats.py

def shutdown(self) -> None:
    """Flush all pending stats to disk and close the SQLite connection gracefully.

    Calls ``save_to_disk`` to persist any buffered measurements, then closes the
    database connection and sets ``_db_conn`` to ``None``. Any exception raised by
    the final save is caught and logged rather than propagated, so the connection is
    always closed even when the flush fails. On the Agent side this method is a no-op.

    Note:
        Call this method explicitly at application shutdown. The destructor
        (``__del__``) also attempts a final save, but relying on the destructor for
        critical I/O is not recommended.
    """
    if self.is_world and self._db_conn:
        log.debug('Shutdown: Saving final stats...')
        try:
            self.save_to_disk()
        except Exception as e:
            log.error(f'Shutdown save failed: {e}')
        self._db_conn.close()
        self._db_conn = None
        log.debug('SQLite connection closed.')

plot

plot(since_timestamp: int = 0) -> str | None

Build and return the default 2x2 stats dashboard as a self-contained HTML document.

Composes four panels into a DefaultBaseDash and delegates HTML rendering to the stats_html_renderer module (imported lazily). Plotly.js is loaded from CDN inside the generated HTML; the Python plotly package is never imported.

On the World side, get_view(since_timestamp) is called to obtain the data. On the Agent side, the locally cached _world_view is used directly. If neither source provides any data, None is returned.

The four panels are: - Top left: Network topology graph (circular layout of connected peers). - Top right: World agent-count history (world masters, agents, humans, artificial). - Bottom left: Network topology graph (duplicate, reserved for future state distribution). - Bottom right: Last-action distribution bar chart across all peers.

Parameters:

Name	Type	Description	Default
since_timestamp	int	Millisecond Unix timestamp lower bound. Only dynamic data points recorded after this value are included in the time-series panels. Pass 0 to include all available cached data. Defaults to 0.	0

Returns:

Type	Description
str | None	A complete <!DOCTYPE html> document string, or None if no view data
str | None	is available to render.

Source code in unaiverse/stats.py

def plot(self, since_timestamp: int = 0) -> str | None:
    """Build and return the default 2x2 stats dashboard as a self-contained HTML document.

    Composes four panels into a ``DefaultBaseDash`` and delegates HTML rendering to
    the ``stats_html_renderer`` module (imported lazily). Plotly.js is loaded from
    CDN inside the generated HTML; the Python ``plotly`` package is never imported.

    On the World side, ``get_view(since_timestamp)`` is called to obtain the data.
    On the Agent side, the locally cached ``_world_view`` is used directly. If
    neither source provides any data, ``None`` is returned.

    The four panels are:
    - **Top left**: Network topology graph (circular layout of connected peers).
    - **Top right**: World agent-count history (world masters, agents, humans, artificial).
    - **Bottom left**: Network topology graph (duplicate, reserved for future state distribution).
    - **Bottom right**: Last-action distribution bar chart across all peers.

    Args:
        since_timestamp: Millisecond Unix timestamp lower bound. Only dynamic data
            points recorded after this value are included in the time-series panels.
            Pass ``0`` to include all available cached data. Defaults to 0.

    Returns:
        A complete ``<!DOCTYPE html>`` document string, or ``None`` if no view data
        is available to render.
    """
    from .stats_html_renderer import render_plotly_html

    # 1. Get Data view
    view = self.get_view(since_timestamp) if self.is_world else self._world_view
    if not view:
        return None

    dash = DefaultBaseDash("World Overview")

    # --- Panel 1: Network Topology (Top Left) ---
    p1 = UIPlot(title="World Topology")
    self._populate_graph(p1, view, "graph")
    clean_axis = {'showgrid': False, 'showticklabels': False, 'zeroline': False}
    p1.set_layout_opt('xaxis', clean_axis)
    p1.set_layout_opt('yaxis', clean_axis)
    dash.add_panel(p1, "top_left")

    # --- Panel 2: System Counters (Table) ---
    p2 = UIPlot(title="World Agents History")
    metrics = [
        ("world_masters", "World Masters", THEME['peers'][0]),
        ("world_agents", "World Agents", THEME['peers'][1]),
        ("human_agents", "Human Agents", THEME['peers'][2]),
        ("artificial_agents", "Artificial Agents", THEME['peers'][3]),
    ]
    for stat_key, label, color in metrics:
        self._populate_time_series(
            panel=p2,
            view=view,
            stat_name=stat_key,
            color_override=color,
            title_override=label
        )
    p2.set_layout_opt('xaxis', {'title': None, 'showticklabels': False})
    p2.set_layout_opt('yaxis', {'title': None})
    dash.add_panel(p2, "top_right")

    # --- Panel 3: State Distribution (Bar) ---
    p3 = UIPlot(title="State Distribution")
    self._populate_graph(p3, view, "graph")
    clean_axis = {'showgrid': False, 'showticklabels': False, 'zeroline': False}
    p3.set_layout_opt('xaxis', clean_axis)
    p3.set_layout_opt('yaxis', clean_axis)
    p3.set_layout_opt("xaxis", {"title": None})
    dash.add_panel(p3, "bot_left")

    # --- Panel 4: Action Distribution (Bar) ---
    p4 = UIPlot(title="Last Action Distribution")
    self._populate_distribution(p4, view, "last_action")
    p4.set_layout_opt("xaxis", {"title": None})
    dash.add_panel(p4, "bot_right")

    return render_plotly_html(dash.to_json())