Spaces:
Runtime error
Runtime error
| """ | |
| Generic 2D world renderer for WorldSmithAI. | |
| This module renders runtime world state into Matplotlib figures, NumPy image | |
| arrays, or PNG files. It is intentionally compatible with a root-level | |
| Hugging Face Spaces ``app.py`` and does not assume an ``app/`` package. | |
| The renderer is domain-agnostic. It does not know about sheep, wolves, | |
| scientists, merchants, cities, vehicles, farms, civilizations, power grids, or | |
| fantasy entities. It reads generic object fields such as ``id``, ``type``, | |
| ``position``, ``state``, ``memory``, ``metadata``, ``amount``, and ``alive``. | |
| Gradio usage: | |
| from visualization.renderer import render_world, render_world_to_array | |
| def run_simulation(prompt: str): | |
| world = build_world_somehow(prompt) | |
| fig = render_world(world) | |
| image = render_world_to_array(world) | |
| return fig, image | |
| Future extensibility: | |
| - Add graph/network rendering for transport, social, and power-grid worlds. | |
| - Add terrain and region layers. | |
| - Add animation hooks in ``visualization.animation``. | |
| - Add renderer plugins for different world projections. | |
| - Add interactive Plotly or Altair adapters while preserving this Matplotlib backend. | |
| - Add event overlays and behavior traces. | |
| """ | |
| from __future__ import annotations | |
| import copy | |
| import logging | |
| import math | |
| import tempfile | |
| from collections.abc import Iterable, Mapping, MutableSequence, Sequence | |
| from dataclasses import dataclass, field | |
| from enum import Enum | |
| from numbers import Real | |
| from pathlib import Path | |
| from types import MappingProxyType | |
| from typing import TYPE_CHECKING, Any, ClassVar | |
| import matplotlib | |
| matplotlib.use("Agg", force=False) | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| from matplotlib.axes import Axes | |
| from matplotlib.figure import Figure | |
| if TYPE_CHECKING: | |
| from core.world import World | |
| logger = logging.getLogger(__name__) | |
| _MISSING = object() | |
| _EPSILON = 1.0e-12 | |
| class RenderCollection(str, Enum): | |
| """World object collections that can be rendered.""" | |
| AGENTS = "agents" | |
| RESOURCES = "resources" | |
| BOTH = "both" | |
| class MissingPositionStrategy(str, Enum): | |
| """Strategies for objects without valid positions.""" | |
| SKIP = "skip" | |
| ORIGIN = "origin" | |
| GRID = "grid" | |
| class LabelMode(str, Enum): | |
| """Label modes for object annotations.""" | |
| NONE = "none" | |
| ID = "id" | |
| TYPE = "type" | |
| ID_TYPE = "id_type" | |
| class ObjectView: | |
| """Serializable view of a world object prepared for rendering.""" | |
| object_id: str | |
| object_type: str | |
| collection: str | |
| x: float | |
| y: float | |
| label: str | |
| group: str | |
| size: float | |
| alive: bool = True | |
| source_has_position: bool = True | |
| metadata: Mapping[str, Any] = field(default_factory=dict) | |
| def to_dict(self) -> dict[str, Any]: | |
| """Return a JSON-friendly representation of this object view.""" | |
| return { | |
| "object_id": self.object_id, | |
| "object_type": self.object_type, | |
| "collection": self.collection, | |
| "x": self.x, | |
| "y": self.y, | |
| "label": self.label, | |
| "group": self.group, | |
| "size": self.size, | |
| "alive": self.alive, | |
| "source_has_position": self.source_has_position, | |
| "metadata": copy.deepcopy(dict(self.metadata)), | |
| } | |
| class RenderSnapshot: | |
| """Serializable snapshot of renderable world state.""" | |
| objects: tuple[ObjectView, ...] | |
| step: int | None = None | |
| bounds: tuple[tuple[float, float], tuple[float, float]] | None = None | |
| metadata: Mapping[str, Any] = field(default_factory=dict) | |
| def object_count(self) -> int: | |
| """Return number of renderable objects.""" | |
| return len(self.objects) | |
| def agent_count(self) -> int: | |
| """Return number of rendered agents.""" | |
| return sum(1 for item in self.objects if item.collection == RenderCollection.AGENTS.value) | |
| def resource_count(self) -> int: | |
| """Return number of rendered resources.""" | |
| return sum(1 for item in self.objects if item.collection == RenderCollection.RESOURCES.value) | |
| def groups(self) -> tuple[str, ...]: | |
| """Return distinct rendered groups.""" | |
| return tuple(sorted({item.group for item in self.objects})) | |
| def to_dict(self) -> dict[str, Any]: | |
| """Return a JSON-friendly snapshot representation.""" | |
| return { | |
| "step": self.step, | |
| "bounds": None | |
| if self.bounds is None | |
| else [list(self.bounds[0]), list(self.bounds[1])], | |
| "object_count": self.object_count, | |
| "agent_count": self.agent_count, | |
| "resource_count": self.resource_count, | |
| "groups": list(self.groups), | |
| "objects": [item.to_dict() for item in self.objects], | |
| "metadata": copy.deepcopy(dict(self.metadata)), | |
| } | |
| class RenderResult: | |
| """Render result containing a figure and the snapshot used to draw it.""" | |
| figure: Figure | |
| snapshot: RenderSnapshot | |
| path: str | None = None | |
| def to_dict(self) -> dict[str, Any]: | |
| """Return a JSON-friendly result representation. | |
| The Matplotlib figure itself is intentionally not serialized. | |
| """ | |
| return { | |
| "path": self.path, | |
| "snapshot": self.snapshot.to_dict(), | |
| } | |
| class RendererConfig: | |
| """Configuration for ``WorldRenderer``. | |
| The defaults are chosen to work well inside a root-level Gradio ``app.py``. | |
| ``render_world`` returns a Matplotlib figure for ``gr.Plot`` and | |
| ``render_world_to_array`` returns a NumPy array for ``gr.Image``. | |
| """ | |
| collection: RenderCollection | str = RenderCollection.BOTH | |
| position_path: str = "position" | |
| group_by_path: str = "type" | |
| label_mode: LabelMode | str = LabelMode.ID | |
| x_index: int = 0 | |
| y_index: int = 1 | |
| missing_position_strategy: MissingPositionStrategy | str = MissingPositionStrategy.GRID | |
| grid_spacing: float = 1.0 | |
| alive_only: bool = False | |
| include_missing_group: bool = True | |
| missing_group_label: str = "unknown" | |
| figure_size: tuple[float, float] = (7.0, 5.0) | |
| dpi: int = 120 | |
| title: str | None = None | |
| show_title: bool = True | |
| show_step: bool = True | |
| show_grid: bool = True | |
| show_legend: bool = True | |
| show_axis_labels: bool = True | |
| equal_aspect: bool = True | |
| agent_marker: str = "o" | |
| resource_marker: str = "^" | |
| missing_position_marker: str = "x" | |
| default_agent_size: float = 64.0 | |
| default_resource_size: float = 90.0 | |
| size_path: str | None = None | |
| resource_size_path: str | None = "amount" | |
| size_scale: float = 1.0 | |
| min_marker_size: float = 20.0 | |
| max_marker_size: float = 400.0 | |
| label_font_size: int = 8 | |
| max_labels: int = 80 | |
| max_label_chars: int = 28 | |
| label_offset: tuple[float, float] = (0.03, 0.03) | |
| use_world_bounds: bool = True | |
| xlim: tuple[float, float] | None = None | |
| ylim: tuple[float, float] | None = None | |
| padding_fraction: float = 0.08 | |
| minimum_span: float = 1.0 | |
| close_after_array: bool = True | |
| close_after_save: bool = True | |
| metadata: Mapping[str, Any] = field(default_factory=dict) | |
| class WorldRenderer: | |
| """Render generic world state into Matplotlib figures or images. | |
| This class is deliberately app-friendly: | |
| - It does not import Gradio. | |
| - It does not import or depend on root-level ``app.py``. | |
| - It returns standard Matplotlib figures, NumPy arrays, or file paths. | |
| - It is safe to call from Hugging Face Spaces callbacks. | |
| """ | |
| config: RendererConfig = field(default_factory=RendererConfig) | |
| name: ClassVar[str] = "world_renderer" | |
| def snapshot(self, world: World) -> RenderSnapshot: | |
| """Build a serializable render snapshot from a runtime world.""" | |
| collection = _normalize_collection(self.config.collection) | |
| raw_items = self._raw_items(world, collection) | |
| positioned: list[tuple[Any, str, tuple[float, float] | None]] = [] | |
| missing_position_items: list[tuple[Any, str]] = [] | |
| for item, item_collection in raw_items: | |
| if self.config.alive_only and _is_agent_like(item) and not _is_alive(item): | |
| continue | |
| position = self._position_for(item) | |
| if position is None: | |
| strategy = _normalize_missing_strategy(self.config.missing_position_strategy) | |
| if strategy is MissingPositionStrategy.SKIP: | |
| continue | |
| missing_position_items.append((item, item_collection)) | |
| else: | |
| positioned.append((item, item_collection, position)) | |
| fallback_positions = self._fallback_positions( | |
| missing_position_items, | |
| existing_count=len(positioned), | |
| ) | |
| objects: list[ObjectView] = [] | |
| for item, item_collection, position in positioned: | |
| if position is None: | |
| continue | |
| objects.append( | |
| self._object_view( | |
| item=item, | |
| collection=item_collection, | |
| x=position[0], | |
| y=position[1], | |
| source_has_position=True, | |
| ) | |
| ) | |
| for (item, item_collection), position in zip(missing_position_items, fallback_positions): | |
| objects.append( | |
| self._object_view( | |
| item=item, | |
| collection=item_collection, | |
| x=position[0], | |
| y=position[1], | |
| source_has_position=False, | |
| ) | |
| ) | |
| objects.sort(key=lambda item: (item.collection, item.group, item.object_id)) | |
| bounds = self._bounds(world, objects) | |
| return RenderSnapshot( | |
| objects=tuple(objects), | |
| step=_world_step(world), | |
| bounds=bounds, | |
| metadata={ | |
| **copy.deepcopy(dict(self.config.metadata)), | |
| "renderer": self.name, | |
| "collection": collection.value, | |
| "missing_position_strategy": _normalize_missing_strategy( | |
| self.config.missing_position_strategy | |
| ).value, | |
| }, | |
| ) | |
| def render(self, world: World) -> Figure: | |
| """Render a world into a Matplotlib ``Figure``. | |
| This return value can be used directly with ``gr.Plot``. | |
| """ | |
| return self.render_result(world).figure | |
| def render_result(self, world: World) -> RenderResult: | |
| """Render a world and return both figure and snapshot.""" | |
| snapshot = self.snapshot(world) | |
| figure, axes = plt.subplots( | |
| figsize=self.config.figure_size, | |
| dpi=int(self.config.dpi), | |
| constrained_layout=True, | |
| ) | |
| self._draw_snapshot(snapshot, figure, axes) | |
| return RenderResult(figure=figure, snapshot=snapshot) | |
| def render_to_array(self, world: World) -> np.ndarray: | |
| """Render a world to an RGB NumPy array. | |
| This return value can be used directly with ``gr.Image``. | |
| """ | |
| result = self.render_result(world) | |
| array = figure_to_rgb_array(result.figure) | |
| if self.config.close_after_array: | |
| plt.close(result.figure) | |
| return array | |
| def save(self, world: World, path: str | Path) -> str: | |
| """Render a world and save it as an image file. | |
| Args: | |
| world: Runtime world object. | |
| path: Output file path, usually ending in ``.png``. | |
| Returns: | |
| String path to the saved image. | |
| """ | |
| output_path = Path(path) | |
| output_path.parent.mkdir(parents=True, exist_ok=True) | |
| result = self.render_result(world) | |
| result.figure.savefig(output_path, dpi=int(self.config.dpi), bbox_inches="tight") | |
| if self.config.close_after_save: | |
| plt.close(result.figure) | |
| return str(output_path) | |
| def save_temp_png( | |
| self, | |
| world: World, | |
| *, | |
| output_dir: str | Path | None = None, | |
| prefix: str = "worldsmithai_", | |
| ) -> str: | |
| """Render a world to a temporary PNG path. | |
| This is convenient for Gradio image/file outputs when a path is easier | |
| than a NumPy array. | |
| """ | |
| directory = None if output_dir is None else str(output_dir) | |
| with tempfile.NamedTemporaryFile( | |
| suffix=".png", | |
| prefix=prefix, | |
| dir=directory, | |
| delete=False, | |
| ) as handle: | |
| path = handle.name | |
| return self.save(world, path) | |
| def _draw_snapshot(self, snapshot: RenderSnapshot, figure: Figure, axes: Axes) -> None: | |
| """Draw a prepared snapshot onto Matplotlib axes.""" | |
| del figure | |
| if not snapshot.objects: | |
| axes.text( | |
| 0.5, | |
| 0.5, | |
| "No renderable world objects", | |
| transform=axes.transAxes, | |
| ha="center", | |
| va="center", | |
| ) | |
| self._style_axes(axes, snapshot) | |
| return | |
| grouped_objects = self._group_objects(snapshot.objects) | |
| for group_name, objects in grouped_objects.items(): | |
| x_values = [item.x for item in objects] | |
| y_values = [item.y for item in objects] | |
| sizes = [item.size for item in objects] | |
| marker = self._marker_for_group(objects) | |
| axes.scatter( | |
| x_values, | |
| y_values, | |
| s=sizes, | |
| marker=marker, | |
| label=group_name, | |
| ) | |
| self._draw_labels(axes, snapshot.objects) | |
| self._style_axes(axes, snapshot) | |
| def _draw_labels(self, axes: Axes, objects: Sequence[ObjectView]) -> None: | |
| """Draw optional labels for objects.""" | |
| label_mode = _normalize_label_mode(self.config.label_mode) | |
| if label_mode is LabelMode.NONE: | |
| return | |
| if self.config.max_labels <= 0: | |
| return | |
| for item in objects[: int(self.config.max_labels)]: | |
| if not item.label: | |
| continue | |
| axes.text( | |
| item.x + self.config.label_offset[0], | |
| item.y + self.config.label_offset[1], | |
| _truncate(item.label, self.config.max_label_chars), | |
| fontsize=int(self.config.label_font_size), | |
| ) | |
| def _style_axes(self, axes: Axes, snapshot: RenderSnapshot) -> None: | |
| """Apply generic axes styling.""" | |
| if self.config.show_axis_labels: | |
| axes.set_xlabel("x") | |
| axes.set_ylabel("y") | |
| if self.config.show_grid: | |
| axes.grid(True, linewidth=0.5, alpha=0.35) | |
| if self.config.equal_aspect: | |
| axes.set_aspect("equal", adjustable="box") | |
| if snapshot.bounds is not None: | |
| (x_min, x_max), (y_min, y_max) = snapshot.bounds | |
| axes.set_xlim(x_min, x_max) | |
| axes.set_ylim(y_min, y_max) | |
| if self.config.show_legend and snapshot.groups: | |
| axes.legend(loc="best", fontsize="small") | |
| if self.config.show_title: | |
| title = self._title(snapshot) | |
| if title: | |
| axes.set_title(title) | |
| def _title(self, snapshot: RenderSnapshot) -> str: | |
| """Return the figure title.""" | |
| if self.config.title is not None: | |
| base_title = self.config.title | |
| else: | |
| base_title = "WorldSmithAI World State" | |
| if self.config.show_step and snapshot.step is not None: | |
| return f"{base_title} — step {snapshot.step}" | |
| return base_title | |
| def _raw_items(self, world: World, collection: RenderCollection) -> tuple[tuple[Any, str], ...]: | |
| """Return raw world objects and their collection labels.""" | |
| items: list[tuple[Any, str]] = [] | |
| if collection in {RenderCollection.AGENTS, RenderCollection.BOTH}: | |
| items.extend( | |
| (item, RenderCollection.AGENTS.value) | |
| for item in _iter_collection(getattr(world, "agents", ())) | |
| ) | |
| if collection in {RenderCollection.RESOURCES, RenderCollection.BOTH}: | |
| items.extend( | |
| (item, RenderCollection.RESOURCES.value) | |
| for item in _iter_collection(getattr(world, "resources", ())) | |
| ) | |
| return tuple(items) | |
| def _position_for(self, item: Any) -> tuple[float, float] | None: | |
| """Return a projected 2D position for an item.""" | |
| raw_position = _read_path(item, self.config.position_path, _MISSING) | |
| if raw_position is _MISSING: | |
| raw_position = getattr(item, "position", None) | |
| return _project_position( | |
| raw_position, | |
| x_index=self.config.x_index, | |
| y_index=self.config.y_index, | |
| ) | |
| def _fallback_positions( | |
| self, | |
| missing_items: Sequence[tuple[Any, str]], | |
| *, | |
| existing_count: int, | |
| ) -> tuple[tuple[float, float], ...]: | |
| """Return deterministic fallback positions for missing-position objects.""" | |
| if not missing_items: | |
| return () | |
| strategy = _normalize_missing_strategy(self.config.missing_position_strategy) | |
| if strategy is MissingPositionStrategy.ORIGIN: | |
| return tuple((0.0, 0.0) for _ in missing_items) | |
| if strategy is MissingPositionStrategy.SKIP: | |
| return () | |
| return _grid_positions( | |
| count=len(missing_items), | |
| spacing=float(self.config.grid_spacing), | |
| offset=existing_count, | |
| ) | |
| def _object_view( | |
| self, | |
| *, | |
| item: Any, | |
| collection: str, | |
| x: float, | |
| y: float, | |
| source_has_position: bool, | |
| ) -> ObjectView: | |
| """Build a renderable object view.""" | |
| object_id = _object_id(item) | |
| object_type = _object_type(item) | |
| group = self._group_for(item, collection) | |
| size = self._size_for(item, collection) | |
| label = self._label_for(object_id, object_type) | |
| return ObjectView( | |
| object_id=object_id, | |
| object_type=object_type, | |
| collection=collection, | |
| x=float(x), | |
| y=float(y), | |
| label=label, | |
| group=group, | |
| size=size, | |
| alive=_is_alive(item), | |
| source_has_position=source_has_position, | |
| metadata=self._metadata_for(item), | |
| ) | |
| def _group_for(self, item: Any, collection: str) -> str: | |
| """Return the legend group for an object.""" | |
| raw_group = _read_path(item, self.config.group_by_path, _MISSING) | |
| if raw_group is _MISSING or raw_group is None or str(raw_group) == "": | |
| if self.config.include_missing_group: | |
| raw_group = self.config.missing_group_label | |
| else: | |
| raw_group = collection | |
| return f"{collection}:{_stable_label(raw_group)}" | |
| def _label_for(self, object_id: str, object_type: str) -> str: | |
| """Return the text label for an object.""" | |
| label_mode = _normalize_label_mode(self.config.label_mode) | |
| if label_mode is LabelMode.NONE: | |
| return "" | |
| if label_mode is LabelMode.TYPE: | |
| return object_type | |
| if label_mode is LabelMode.ID_TYPE: | |
| return f"{object_id} ({object_type})" | |
| return object_id | |
| def _size_for(self, item: Any, collection: str) -> float: | |
| """Return marker size for an object.""" | |
| if collection == RenderCollection.RESOURCES.value: | |
| default_size = float(self.config.default_resource_size) | |
| size_path = self.config.resource_size_path | |
| else: | |
| default_size = float(self.config.default_agent_size) | |
| size_path = self.config.size_path | |
| if size_path is None: | |
| raw_size = default_size | |
| else: | |
| value = _read_path(item, size_path, _MISSING) | |
| raw_size = default_size if not _is_number(value) else default_size + float(value) | |
| scaled_size = float(raw_size) * float(self.config.size_scale) | |
| return _clamp( | |
| scaled_size, | |
| minimum=float(self.config.min_marker_size), | |
| maximum=float(self.config.max_marker_size), | |
| ) | |
| def _metadata_for(item: Any) -> Mapping[str, Any]: | |
| """Return object metadata as a safe mapping.""" | |
| metadata = getattr(item, "metadata", {}) | |
| if isinstance(metadata, Mapping): | |
| return copy.deepcopy(dict(metadata)) | |
| return {} | |
| def _group_objects(objects: Sequence[ObjectView]) -> dict[str, tuple[ObjectView, ...]]: | |
| """Group object views by legend group.""" | |
| groups: dict[str, list[ObjectView]] = {} | |
| for item in objects: | |
| groups.setdefault(item.group, []).append(item) | |
| return { | |
| group: tuple(items) | |
| for group, items in sorted(groups.items(), key=lambda pair: pair[0]) | |
| } | |
| def _marker_for_group(self, objects: Sequence[ObjectView]) -> str: | |
| """Return a marker style for a plotted group.""" | |
| if not objects: | |
| return self.config.agent_marker | |
| if any(not item.source_has_position for item in objects): | |
| return self.config.missing_position_marker | |
| collection = objects[0].collection | |
| if collection == RenderCollection.RESOURCES.value: | |
| return self.config.resource_marker | |
| return self.config.agent_marker | |
| def _bounds( | |
| self, | |
| world: World, | |
| objects: Sequence[ObjectView], | |
| ) -> tuple[tuple[float, float], tuple[float, float]] | None: | |
| """Return render bounds.""" | |
| if self.config.xlim is not None and self.config.ylim is not None: | |
| return self.config.xlim, self.config.ylim | |
| if self.config.use_world_bounds: | |
| world_bounds = _bounds_from_world(world) | |
| if world_bounds is not None: | |
| x_bounds, y_bounds = world_bounds | |
| return ( | |
| self.config.xlim if self.config.xlim is not None else x_bounds, | |
| self.config.ylim if self.config.ylim is not None else y_bounds, | |
| ) | |
| if not objects: | |
| return None | |
| x_values = np.asarray([item.x for item in objects], dtype=float) | |
| y_values = np.asarray([item.y for item in objects], dtype=float) | |
| x_bounds = self.config.xlim or _padded_bounds( | |
| x_values, | |
| padding_fraction=float(self.config.padding_fraction), | |
| minimum_span=float(self.config.minimum_span), | |
| ) | |
| y_bounds = self.config.ylim or _padded_bounds( | |
| y_values, | |
| padding_fraction=float(self.config.padding_fraction), | |
| minimum_span=float(self.config.minimum_span), | |
| ) | |
| return x_bounds, y_bounds | |
| def render_world(world: World, config: RendererConfig | None = None) -> Figure: | |
| """Render a world to a Matplotlib figure. | |
| This is the simplest function to use with ``gr.Plot``. | |
| """ | |
| return WorldRenderer(config=config or RendererConfig()).render(world) | |
| def render_world_result(world: World, config: RendererConfig | None = None) -> RenderResult: | |
| """Render a world and return both figure and snapshot.""" | |
| return WorldRenderer(config=config or RendererConfig()).render_result(world) | |
| def render_world_snapshot(world: World, config: RendererConfig | None = None) -> RenderSnapshot: | |
| """Return a serializable snapshot of renderable world state.""" | |
| return WorldRenderer(config=config or RendererConfig()).snapshot(world) | |
| def render_world_to_array(world: World, config: RendererConfig | None = None) -> np.ndarray: | |
| """Render a world to an RGB NumPy array. | |
| This is the simplest function to use with ``gr.Image``. | |
| """ | |
| return WorldRenderer(config=config or RendererConfig()).render_to_array(world) | |
| def render_world_to_png_path( | |
| world: World, | |
| *, | |
| path: str | Path | None = None, | |
| output_dir: str | Path | None = None, | |
| config: RendererConfig | None = None, | |
| ) -> str: | |
| """Render a world to a PNG file path. | |
| If ``path`` is omitted, a temporary PNG path is created. This is convenient | |
| for root-level Hugging Face Spaces ``app.py`` callbacks. | |
| """ | |
| renderer = WorldRenderer(config=config or RendererConfig()) | |
| if path is not None: | |
| return renderer.save(world, path) | |
| return renderer.save_temp_png(world, output_dir=output_dir) | |
| def figure_to_rgb_array(figure: Figure) -> np.ndarray: | |
| """Convert a Matplotlib figure into an RGB NumPy array.""" | |
| figure.canvas.draw() | |
| width, height = figure.canvas.get_width_height() | |
| try: | |
| buffer = figure.canvas.buffer_rgba() | |
| image = np.frombuffer(buffer, dtype=np.uint8).reshape(height, width, 4) | |
| return image[:, :, :3].copy() | |
| except AttributeError: | |
| raw = figure.canvas.tostring_rgb() | |
| image = np.frombuffer(raw, dtype=np.uint8).reshape(height, width, 3) | |
| return image.copy() | |
| def close_figure(figure: Figure) -> None: | |
| """Close a Matplotlib figure to release memory in long Gradio sessions.""" | |
| plt.close(figure) | |
| def _normalize_collection(value: RenderCollection | str) -> RenderCollection: | |
| """Normalize a render collection value.""" | |
| if isinstance(value, RenderCollection): | |
| return value | |
| return RenderCollection(str(value)) | |
| def _normalize_missing_strategy(value: MissingPositionStrategy | str) -> MissingPositionStrategy: | |
| """Normalize a missing-position strategy value.""" | |
| if isinstance(value, MissingPositionStrategy): | |
| return value | |
| return MissingPositionStrategy(str(value)) | |
| def _normalize_label_mode(value: LabelMode | str) -> LabelMode: | |
| """Normalize a label mode value.""" | |
| if isinstance(value, LabelMode): | |
| return value | |
| return LabelMode(str(value)) | |
| def _world_step(world: World) -> int | None: | |
| """Return current world step if available.""" | |
| value = getattr(world, "step_count", None) | |
| if isinstance(value, Real) and not isinstance(value, bool): | |
| return int(value) | |
| return None | |
| def _is_number(value: Any) -> bool: | |
| """Return whether a value is a real numeric scalar, excluding booleans.""" | |
| return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool) | |
| def _is_alive(item: Any) -> bool: | |
| """Return whether an item is alive if it exposes an alive field.""" | |
| return bool(getattr(item, "alive", True)) | |
| def _is_agent_like(item: Any) -> bool: | |
| """Return whether an item looks like an agent.""" | |
| return hasattr(item, "behaviors") or hasattr(item, "policy") or hasattr(item, "alive") | |
| def _object_id(item: Any) -> str: | |
| """Return stable object id.""" | |
| value = _read_path(item, "id", _MISSING) | |
| if value is not _MISSING and value is not None: | |
| return str(value) | |
| return str(id(item)) | |
| def _object_type(item: Any) -> str: | |
| """Return stable object type.""" | |
| value = _read_path(item, "type", _MISSING) | |
| if value is not _MISSING and value is not None: | |
| return str(value) | |
| return item.__class__.__name__ | |
| def _iter_collection(raw_collection: Any) -> tuple[Any, ...]: | |
| """Return items from a mapping-backed or sequence-backed collection.""" | |
| if raw_collection is None: | |
| return () | |
| if isinstance(raw_collection, Mapping): | |
| values = raw_collection.values() | |
| elif isinstance(raw_collection, Iterable) and not isinstance(raw_collection, (str, bytes)): | |
| values = raw_collection | |
| else: | |
| values = (raw_collection,) | |
| return tuple(item for item in values if item is not None) | |
| def _split_path(path: str) -> tuple[str, ...]: | |
| """Split a dot-separated path into components.""" | |
| return tuple(part for part in str(path).split(".") if part) | |
| def _get_mapping_path(container: Mapping[str, Any], path: str, default: Any = _MISSING) -> Any: | |
| """Read a nested mapping value using dot notation.""" | |
| parts = _split_path(path) | |
| if not parts: | |
| return default | |
| current: Any = container | |
| for part in parts: | |
| if not isinstance(current, Mapping) or part not in current: | |
| return default | |
| current = current[part] | |
| return current | |
| def _get_object_path(root: Any, path: str, default: Any = _MISSING) -> Any: | |
| """Read nested values from mappings, sequences, or object attributes.""" | |
| parts = _split_path(path) | |
| if not parts: | |
| return root | |
| current: Any = root | |
| for part in parts: | |
| if isinstance(current, Mapping): | |
| if part not in current: | |
| return default | |
| current = current[part] | |
| continue | |
| if isinstance(current, Sequence) and not isinstance(current, (str, bytes)) and part.isdigit(): | |
| index = int(part) | |
| if index >= len(current): | |
| return default | |
| current = current[index] | |
| continue | |
| if not hasattr(current, part): | |
| return default | |
| current = getattr(current, part) | |
| return current | |
| def _read_path(item: Any, path: str, default: Any = _MISSING) -> Any: | |
| """Read a generic path from an object or mapping. | |
| Supported prefixes: | |
| - ``state.foo`` | |
| - ``state:foo`` | |
| - ``memory.foo`` | |
| - ``memory:foo`` | |
| - ``metadata.foo`` | |
| - ``metadata:foo`` | |
| Without a prefix, object attributes and mapping keys are read directly. | |
| """ | |
| normalized_path = str(path) | |
| if normalized_path.startswith("state."): | |
| state = getattr(item, "state", {}) | |
| return _get_mapping_path( | |
| state if isinstance(state, Mapping) else {}, | |
| normalized_path.removeprefix("state."), | |
| default, | |
| ) | |
| if normalized_path.startswith("state:"): | |
| state = getattr(item, "state", {}) | |
| return _get_mapping_path( | |
| state if isinstance(state, Mapping) else {}, | |
| normalized_path.removeprefix("state:"), | |
| default, | |
| ) | |
| if normalized_path.startswith("memory."): | |
| memory = getattr(item, "memory", {}) | |
| return _get_mapping_path( | |
| memory if isinstance(memory, Mapping) else {}, | |
| normalized_path.removeprefix("memory."), | |
| default, | |
| ) | |
| if normalized_path.startswith("memory:"): | |
| memory = getattr(item, "memory", {}) | |
| return _get_mapping_path( | |
| memory if isinstance(memory, Mapping) else {}, | |
| normalized_path.removeprefix("memory:"), | |
| default, | |
| ) | |
| if normalized_path.startswith("metadata."): | |
| metadata = getattr(item, "metadata", {}) | |
| return _get_mapping_path( | |
| metadata if isinstance(metadata, Mapping) else {}, | |
| normalized_path.removeprefix("metadata."), | |
| default, | |
| ) | |
| if normalized_path.startswith("metadata:"): | |
| metadata = getattr(item, "metadata", {}) | |
| return _get_mapping_path( | |
| metadata if isinstance(metadata, Mapping) else {}, | |
| normalized_path.removeprefix("metadata:"), | |
| default, | |
| ) | |
| return _get_object_path(item, normalized_path, default) | |
| def _project_position( | |
| value: Any, | |
| *, | |
| x_index: int, | |
| y_index: int, | |
| ) -> tuple[float, float] | None: | |
| """Project a position-like value into 2D coordinates.""" | |
| if value is None or value is _MISSING: | |
| return None | |
| try: | |
| array = np.asarray(value, dtype=float).reshape(-1) | |
| except (TypeError, ValueError): | |
| return None | |
| if array.size == 0: | |
| return None | |
| if not np.all(np.isfinite(array)): | |
| return None | |
| x = float(array[x_index]) if 0 <= x_index < array.size else float(array[0]) | |
| if 0 <= y_index < array.size: | |
| y = float(array[y_index]) | |
| elif array.size >= 2: | |
| y = float(array[1]) | |
| else: | |
| y = 0.0 | |
| return x, y | |
| def _grid_positions( | |
| *, | |
| count: int, | |
| spacing: float, | |
| offset: int = 0, | |
| ) -> tuple[tuple[float, float], ...]: | |
| """Return deterministic grid positions.""" | |
| if count <= 0: | |
| return () | |
| safe_spacing = spacing if abs(spacing) > _EPSILON else 1.0 | |
| total = count + max(0, offset) | |
| columns = max(1, int(math.ceil(math.sqrt(total)))) | |
| positions: list[tuple[float, float]] = [] | |
| for index in range(offset, offset + count): | |
| row = index // columns | |
| column = index % columns | |
| positions.append((column * safe_spacing, row * safe_spacing)) | |
| return tuple(positions) | |
| def _bounds_from_world(world: World) -> tuple[tuple[float, float], tuple[float, float]] | None: | |
| """Return first two spatial bounds from world configuration if present.""" | |
| raw_space = getattr(world, "space_config", None) | |
| if raw_space is None: | |
| raw_space = getattr(world, "space", None) | |
| bounds: Any = None | |
| if isinstance(raw_space, Mapping): | |
| bounds = raw_space.get("bounds") | |
| elif raw_space is not None: | |
| bounds = getattr(raw_space, "bounds", None) | |
| if bounds is None: | |
| return None | |
| if not isinstance(bounds, Sequence) or isinstance(bounds, (str, bytes)) or len(bounds) < 2: | |
| return None | |
| try: | |
| x_bounds = (float(bounds[0][0]), float(bounds[0][1])) | |
| y_bounds = (float(bounds[1][0]), float(bounds[1][1])) | |
| except (TypeError, ValueError, IndexError): | |
| return None | |
| if not all(math.isfinite(value) for value in (*x_bounds, *y_bounds)): | |
| return None | |
| if x_bounds[0] >= x_bounds[1] or y_bounds[0] >= y_bounds[1]: | |
| return None | |
| return x_bounds, y_bounds | |
| def _padded_bounds( | |
| values: np.ndarray, | |
| *, | |
| padding_fraction: float, | |
| minimum_span: float, | |
| ) -> tuple[float, float]: | |
| """Return padded numeric bounds for plotted values.""" | |
| finite_values = values[np.isfinite(values)] | |
| if finite_values.size == 0: | |
| half_span = max(1.0, minimum_span) / 2.0 | |
| return -half_span, half_span | |
| lower = float(np.min(finite_values)) | |
| upper = float(np.max(finite_values)) | |
| span = max(upper - lower, float(minimum_span)) | |
| center = (lower + upper) / 2.0 | |
| half_span = span / 2.0 | |
| padding = span * max(0.0, float(padding_fraction)) | |
| return center - half_span - padding, center + half_span + padding | |
| def _stable_label(value: Any) -> str: | |
| """Return a stable string label for arbitrary values.""" | |
| if value is None: | |
| return "none" | |
| if isinstance(value, bool): | |
| return "true" if value else "false" | |
| if _is_number(value): | |
| numeric_value = float(value) | |
| if numeric_value.is_integer(): | |
| return str(int(numeric_value)) | |
| return str(numeric_value) | |
| if isinstance(value, Mapping): | |
| parts = [f"{key}={_stable_label(value[key])}" for key in sorted(value.keys(), key=str)] | |
| return "{" + ",".join(parts) + "}" | |
| if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): | |
| return "[" + ",".join(_stable_label(item) for item in value) + "]" | |
| return str(value) | |
| def _truncate(value: str, max_chars: int) -> str: | |
| """Truncate a label to a configured length.""" | |
| if max_chars <= 0: | |
| return "" | |
| if len(value) <= max_chars: | |
| return value | |
| if max_chars <= 3: | |
| return value[:max_chars] | |
| return value[: max_chars - 3] + "..." | |
| def _clamp(value: float, *, minimum: float, maximum: float) -> float: | |
| """Clamp a value to inclusive numeric bounds.""" | |
| return min(max(float(value), float(minimum)), float(maximum)) | |
| RENDERER_REGISTRY: Mapping[str, type[WorldRenderer]] = MappingProxyType( | |
| { | |
| WorldRenderer.name: WorldRenderer, | |
| } | |
| ) | |
| __all__ = [ | |
| "LabelMode", | |
| "MissingPositionStrategy", | |
| "ObjectView", | |
| "RENDERER_REGISTRY", | |
| "RenderCollection", | |
| "RenderResult", | |
| "RenderSnapshot", | |
| "RendererConfig", | |
| "WorldRenderer", | |
| "close_figure", | |
| "figure_to_rgb_array", | |
| "render_world", | |
| "render_world_result", | |
| "render_world_snapshot", | |
| "render_world_to_array", | |
| "render_world_to_png_path", | |
| ] |