""" Root-level Gradio callback bridge for WorldSmithAI. This module is designed to sit beside ``app.py`` in a Hugging Face Space. It does not import Gradio and does not assume an ``app/`` package or folder. Responsibilities: - Convert prompts into WorldSpec DSL. - Parse and validate DSL JSON. - Construct runtime World objects. - Run deterministic simulations. - Generate animation, charts, final render, metrics, and narrative. - Return simple values that root-level ``app.py`` can bind to Gradio outputs. Default callback output order: 1. world_spec_json 2. animation_path 3. population_chart_path 4. resource_chart_path 5. final_image_path 6. narrative 7. metrics_json 8. status_message Example app.py: import gradio as gr from callbacks import run_worldsmith_callback with gr.Blocks() as demo: prompt = gr.Textbox(label="World prompt") steps = gr.Slider(1, 200, value=60, step=1) run = gr.Button("Run") dsl = gr.Code(language="json") animation = gr.Image() population = gr.Image() resources = gr.Image() final_state = gr.Image() narrative = gr.Markdown() metrics = gr.Code(language="json") status = gr.Textbox() run.click( run_worldsmith_callback, inputs=[prompt, steps], outputs=[dsl, animation, population, resources, final_state, narrative, metrics, status], ) demo.launch() Future extensibility: - Add progress-yielding generator callbacks for Gradio progress bars. - Add user-selectable SLM backends. - Add example-world dropdown support. - Add multi-run comparison callbacks. - Add downloadable artifact bundles. - Add persistent run logs for demo evaluation. """ from __future__ import annotations import copy import json import logging import math import tempfile from collections.abc import Callable, Mapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from pathlib import Path from typing import Any from dsl.parser import DSLParseError, parse_world_file, parse_world_spec from dsl.schema import WorldSpec from dsl.validator import ValidationConfig, ValidationReport, ValidationSeverity, validate_world_spec from factory.world_factory import ( WorldBuildResult, WorldFactory, WorldFactoryConfig, WorldFactoryValidationError, ) from llm.narrator import NarrationAudience, NarrationMode, NarrationStyle, NarratorConfig, WorldNarrator from llm.world_generator import ( GenerationMode, WorldGenerationConfig, WorldGenerationResult, WorldGenerator, ) from metrics.diversity import compute_agent_type_diversity, compute_resource_type_diversity from metrics.entropy import compute_agent_type_entropy, compute_resource_type_entropy from metrics.interestingness import InterestingnessTracker from metrics.stability import StabilityMetric, StabilityTracker from visualization.animation import ( AnimationConfig, AnimationFormat, AnimationFrame, AnimationWriterError, WorldAnimator, normalize_animation_format, ) from visualization.charts import ( ChartConfig, ChartTracker, WorldChartRenderer, population_chart_config, resource_chart_config, ) from visualization.renderer import ( RendererConfig, WorldRenderer, close_figure, figure_to_rgb_array, ) logger = logging.getLogger(__name__) DEFAULT_GRADIO_OUTPUT_FIELDS: tuple[str, ...] = ( "world_spec_json", "animation_path", "population_chart_path", "resource_chart_path", "final_image_path", "narrative", "metrics_json", "status_message", ) DEFAULT_OUTPUT_SUBDIR_PREFIX = "worldsmithai_run_" class CallbackMode(str, Enum): """Supported callback entry modes.""" PROMPT = "prompt" DSL = "dsl" @dataclass(frozen=True) class CallbackArtifacts: """File artifacts produced by a callback run.""" output_dir: str animation_path: str | None = None population_chart_path: str | None = None resource_chart_path: str | None = None final_image_path: str | None = None def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly artifact dictionary.""" return { "output_dir": self.output_dir, "animation_path": self.animation_path, "population_chart_path": self.population_chart_path, "resource_chart_path": self.resource_chart_path, "final_image_path": self.final_image_path, } @dataclass(frozen=True) class CallbackResult: """Result returned by high-level callback pipelines.""" success: bool status_message: str world_spec_json: str = "" validation_json: str = "{}" metrics_json: str = "{}" narrative: str = "" animation_path: str | None = None population_chart_path: str | None = None resource_chart_path: str | None = None final_image_path: str | None = None artifacts: CallbackArtifacts | None = None metadata: Mapping[str, Any] = field(default_factory=dict) def as_gradio_tuple( self, fields: Sequence[str] = DEFAULT_GRADIO_OUTPUT_FIELDS, ) -> tuple[Any, ...]: """Return a tuple matching a Gradio output component order.""" values: list[Any] = [] for field_name in fields: if not hasattr(self, field_name): raise AttributeError(f"CallbackResult has no field {field_name!r}") values.append(getattr(self, field_name)) return tuple(values) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly result dictionary.""" return { "success": self.success, "status_message": self.status_message, "world_spec_json": self.world_spec_json, "validation_json": self.validation_json, "metrics_json": self.metrics_json, "narrative": self.narrative, "animation_path": self.animation_path, "population_chart_path": self.population_chart_path, "resource_chart_path": self.resource_chart_path, "final_image_path": self.final_image_path, "artifacts": None if self.artifacts is None else self.artifacts.to_dict(), "metadata": _json_safe(copy.deepcopy(dict(self.metadata))), } @dataclass class CallbackConfig: """Configuration for root-level Gradio callbacks. Defaults favor a robust demo: - deterministic fallback world generation is enabled, - unknown behavior/policy names become validation warnings instead of UI crashes, - GIF animation is used by default, - charts and narrative are always attempted. """ output_dir: str | Path | None = None steps: int | None = None max_animation_frames: int = 80 animation_format: AnimationFormat | str = AnimationFormat.GIF fps: float = 8.0 generate_animation: bool = True generate_charts: bool = True generate_final_image: bool = True generate_narrative: bool = True collect_metric_history: bool = True fallback_to_gif_on_animation_error: bool = True renderer_config: RendererConfig = field(default_factory=RendererConfig) population_chart_config: ChartConfig = field(default_factory=population_chart_config) resource_chart_config: ChartConfig = field(default_factory=resource_chart_config) world_generation_config: WorldGenerationConfig | None = None world_factory_config: WorldFactoryConfig | None = None narrator_config: NarratorConfig | None = None validation_config: ValidationConfig | None = None strict_validation_for_app: bool = False include_validation_json: bool = True include_generation_diagnostics: bool = True metadata: Mapping[str, Any] = field(default_factory=dict) def resolved_animation_format(self) -> AnimationFormat: """Return normalized animation format.""" return normalize_animation_format(self.animation_format) def resolved_output_dir(self) -> Path: """Return an existing directory for callback artifacts.""" if self.output_dir is not None: path = Path(self.output_dir) path.mkdir(parents=True, exist_ok=True) return path path = Path(tempfile.mkdtemp(prefix=DEFAULT_OUTPUT_SUBDIR_PREFIX)) path.mkdir(parents=True, exist_ok=True) return path def resolved_generation_config(self) -> WorldGenerationConfig: """Return generation config.""" if self.world_generation_config is not None: return self.world_generation_config return WorldGenerationConfig( mode=GenerationMode.AUTO, default_steps=max(1, int(self.steps or 60)), semantic_validation=True, strict_semantic_validation=False, fallback_on_model_error=True, fallback_on_parse_error=True, fallback_on_semantic_error=False, ) def resolved_factory_config(self) -> WorldFactoryConfig: """Return world factory config.""" if self.world_factory_config is not None: return self.world_factory_config return WorldFactoryConfig( validate_before_build=True, strict_unknown_behaviors=False, strict_unknown_policies=False, strict_constructor_params=False, include_disabled_behaviors=False, include_disabled_events=False, attach_default_policy=True, default_policy_type="rule_policy", attach_scheduler=True, attach_dsl_spec_to_world=True, ) def resolved_narrator_config(self) -> NarratorConfig: """Return narrator config.""" if self.narrator_config is not None: return self.narrator_config return NarratorConfig( mode=NarrationMode.DETERMINISTIC, style=NarrationStyle.ANALYTICAL, audience=NarrationAudience.HACKATHON_JUDGE, compute_default_metrics=True, ) def resolved_validation_config(self) -> ValidationConfig: """Return semantic validation config.""" if self.validation_config is not None: return self.validation_config severity = ( ValidationSeverity.ERROR if self.strict_validation_for_app else ValidationSeverity.WARNING ) return ValidationConfig( require_known_behaviors=True, require_known_policies=True, validate_constructor_params=True, validate_references=True, unknown_registry_item_severity=severity, constructor_param_severity=severity, unresolved_reference_severity=severity, ) @dataclass(frozen=True) class SimulationRunData: """Internal structured result of a runtime simulation.""" final_world: Any artifacts: CallbackArtifacts metrics: Mapping[str, Any] timeline: tuple[Mapping[str, Any], ...] population_snapshots: tuple[Any, ...] resource_snapshots: tuple[Any, ...] def run_worldsmith_callback( prompt: str, steps: int | float | None = None, constraints: str | Mapping[str, Any] | None = None, animation_format: str = "gif", client: Any | None = None, ) -> tuple[Any, ...]: """Gradio-friendly prompt-to-simulation callback. Default return order: 1. world_spec_json 2. animation_path 3. population_chart_path 4. resource_chart_path 5. final_image_path 6. narrative 7. metrics_json 8. status_message This function catches exceptions and returns a failure tuple instead of crashing the UI. """ config = CallbackConfig( steps=_coerce_optional_int(steps), animation_format=animation_format, ) try: result = run_worldsmith_pipeline( prompt=prompt, constraints=constraints, client=client, config=config, ) return result.as_gradio_tuple() except Exception as exc: logger.exception("WorldSmithAI callback failed") return callback_failure_result(exc).as_gradio_tuple() def run_worldsmith_pipeline( *, prompt: str, constraints: str | Mapping[str, Any] | None = None, client: Any | None = None, config: CallbackConfig | None = None, ) -> CallbackResult: """Run the full prompt-to-world-to-artifacts pipeline. This function raises exceptions. Use ``run_worldsmith_callback`` for a Gradio-safe wrapper that catches errors. """ callback_config = config or CallbackConfig() parsed_constraints = parse_constraints(constraints) generation_result = generate_world_from_prompt( prompt=prompt, constraints=parsed_constraints, client=client, config=callback_config, ) world_spec = apply_step_override(generation_result.spec, callback_config.steps) validation_report = validate_world_spec( world_spec, config=callback_config.resolved_validation_config(), ) if callback_config.strict_validation_for_app and not validation_report.is_valid: raise WorldFactoryValidationError(validation_report) build_result = build_world_from_spec(world_spec, config=callback_config) simulation = simulate_world_for_app( build_result.world, steps=world_spec.simulation.steps, config=callback_config, ) narrative = "" if callback_config.generate_narrative: narrative = narrate_world_for_app( simulation.final_world, timeline=simulation.timeline, metrics=simulation.metrics, client=client, config=callback_config, ) metrics_payload = { "success": True, "world": { "id": world_spec.id, "name": world_spec.name, "description": world_spec.description, "steps": world_spec.simulation.steps, }, "generation": generation_result.to_dict() if callback_config.include_generation_diagnostics else {"mode": generation_result.mode.value}, "validation": validation_report.to_dict(), "factory": build_result.report.to_dict(), "artifacts": simulation.artifacts.to_dict(), "metrics": _json_safe(simulation.metrics), "timeline": _json_safe(simulation.timeline), } status = ( f"Built and simulated world {world_spec.id!r} for " f"{world_spec.simulation.steps} step(s). " f"Validation: {len(validation_report.errors)} error(s), " f"{len(validation_report.warnings)} warning(s)." ) return CallbackResult( success=True, status_message=status, world_spec_json=world_spec.to_json_string(indent=2, exclude_none=True), validation_json=_safe_json_dumps(validation_report.to_dict()), metrics_json=_safe_json_dumps(metrics_payload), narrative=narrative, animation_path=simulation.artifacts.animation_path, population_chart_path=simulation.artifacts.population_chart_path, resource_chart_path=simulation.artifacts.resource_chart_path, final_image_path=simulation.artifacts.final_image_path, artifacts=simulation.artifacts, metadata={ "mode": CallbackMode.PROMPT.value, "output_dir": simulation.artifacts.output_dir, }, ) def generate_dsl_callback( prompt: str, constraints: str | Mapping[str, Any] | None = None, client: Any | None = None, ) -> tuple[str, str, str]: """Gradio-friendly callback that only generates and validates DSL JSON.""" try: config = CallbackConfig() parsed_constraints = parse_constraints(constraints) generation_result = generate_world_from_prompt( prompt=prompt, constraints=parsed_constraints, client=client, config=config, ) validation_report = validate_world_spec( generation_result.spec, config=config.resolved_validation_config(), ) status = ( f"Generated WorldSpec {generation_result.spec.id!r}. " f"Validation: {len(validation_report.errors)} error(s), " f"{len(validation_report.warnings)} warning(s)." ) return ( generation_result.spec.to_json_string(indent=2, exclude_none=True), _safe_json_dumps(validation_report.to_dict()), status, ) except Exception as exc: logger.exception("DSL generation callback failed") return "", "{}", f"Generation failed: {exc}" def validate_dsl_callback(dsl_json: str) -> tuple[str, str]: """Gradio-friendly callback that validates DSL JSON.""" try: spec = parse_world_spec(dsl_json) report = validate_world_spec( spec, config=CallbackConfig().resolved_validation_config(), ) status = ( f"Validation complete: {len(report.errors)} error(s), " f"{len(report.warnings)} warning(s)." ) return _safe_json_dumps(report.to_dict()), status except Exception as exc: logger.exception("DSL validation callback failed") return "{}", f"Validation failed: {exc}" def simulate_dsl_callback( dsl_json: str, steps: int | float | None = None, animation_format: str = "gif", client: Any | None = None, ) -> tuple[Any, ...]: """Gradio-friendly callback that simulates provided DSL JSON.""" config = CallbackConfig( steps=_coerce_optional_int(steps), animation_format=animation_format, ) try: result = simulate_dsl_pipeline( dsl_json=dsl_json, client=client, config=config, ) return result.as_gradio_tuple() except Exception as exc: logger.exception("DSL simulation callback failed") return callback_failure_result(exc, world_spec_json=dsl_json).as_gradio_tuple() def simulate_dsl_pipeline( *, dsl_json: str, client: Any | None = None, config: CallbackConfig | None = None, ) -> CallbackResult: """Parse, validate, build, and simulate a supplied DSL JSON string.""" callback_config = config or CallbackConfig() world_spec = apply_step_override(parse_world_spec(dsl_json), callback_config.steps) validation_report = validate_world_spec( world_spec, config=callback_config.resolved_validation_config(), ) if callback_config.strict_validation_for_app and not validation_report.is_valid: raise WorldFactoryValidationError(validation_report) build_result = build_world_from_spec(world_spec, config=callback_config) simulation = simulate_world_for_app( build_result.world, steps=world_spec.simulation.steps, config=callback_config, ) narrative = "" if callback_config.generate_narrative: narrative = narrate_world_for_app( simulation.final_world, timeline=simulation.timeline, metrics=simulation.metrics, client=client, config=callback_config, ) metrics_payload = { "success": True, "world": { "id": world_spec.id, "name": world_spec.name, "description": world_spec.description, "steps": world_spec.simulation.steps, }, "validation": validation_report.to_dict(), "factory": build_result.report.to_dict(), "artifacts": simulation.artifacts.to_dict(), "metrics": _json_safe(simulation.metrics), "timeline": _json_safe(simulation.timeline), } status = ( f"Simulated supplied DSL world {world_spec.id!r} for " f"{world_spec.simulation.steps} step(s)." ) return CallbackResult( success=True, status_message=status, world_spec_json=world_spec.to_json_string(indent=2, exclude_none=True), validation_json=_safe_json_dumps(validation_report.to_dict()), metrics_json=_safe_json_dumps(metrics_payload), narrative=narrative, animation_path=simulation.artifacts.animation_path, population_chart_path=simulation.artifacts.population_chart_path, resource_chart_path=simulation.artifacts.resource_chart_path, final_image_path=simulation.artifacts.final_image_path, artifacts=simulation.artifacts, metadata={ "mode": CallbackMode.DSL.value, "output_dir": simulation.artifacts.output_dir, }, ) def load_example_callback(example_path: str | Path) -> tuple[str, str]: """Load an example JSON file for a Gradio dropdown or button.""" try: spec = parse_world_file(example_path) return spec.to_json_string(indent=2, exclude_none=True), f"Loaded {example_path}" except Exception as exc: logger.exception("Could not load example world") return "", f"Could not load example: {exc}" def generate_world_from_prompt( *, prompt: str, constraints: Mapping[str, Any] | str | None, client: Any | None, config: CallbackConfig, ) -> WorldGenerationResult: """Generate a ``WorldSpec`` from natural language.""" generation_config = config.resolved_generation_config() if config.steps is not None: generation_config.default_steps = max(1, int(config.steps)) generator = WorldGenerator( client=client, config=generation_config, ) return generator.generate( prompt, constraints=constraints, ) def build_world_from_spec( spec: WorldSpec, *, config: CallbackConfig, ) -> WorldBuildResult: """Build a runtime world from a ``WorldSpec``.""" factory = WorldFactory(config=config.resolved_factory_config()) return factory.create_world_result(spec) def simulate_world_for_app( world: Any, *, steps: int, config: CallbackConfig, ) -> SimulationRunData: """Run a deterministic simulation and create app artifacts. The same simulation run is used for: - animation frames, - population chart snapshots, - resource chart snapshots, - metric history, - final render, - narrative context. """ output_dir = config.resolved_output_dir() safe_steps = max(0, int(steps)) renderer = WorldRenderer(config=config.renderer_config) animator_config = AnimationConfig( frame_count=max(1, min(config.max_animation_frames, safe_steps + 1)), steps_per_frame=1, format=config.resolved_animation_format(), fps=float(config.fps), output_dir=output_dir, renderer_config=config.renderer_config, ) animator = WorldAnimator(config=animator_config) population_tracker = ChartTracker( renderer=WorldChartRenderer(config=config.population_chart_config) ) resource_tracker = ChartTracker( renderer=WorldChartRenderer(config=config.resource_chart_config) ) stability_tracker = StabilityTracker( metric=StabilityMetric(collection="agents", group_by_path="type") ) interestingness_tracker = InterestingnessTracker() capture_interval = _capture_interval( steps=safe_steps, max_frames=max(1, int(config.max_animation_frames)), ) frames: list[AnimationFrame] = [] metric_history: list[dict[str, Any]] = [] timeline: list[Mapping[str, Any]] = [] for step_index in range(safe_steps + 1): if config.generate_charts: population_tracker.update(world) resource_tracker.update(world) if config.collect_metric_history: metrics_at_step = compute_metric_bundle( world, stability_tracker=stability_tracker, interestingness_tracker=interestingness_tracker, ) metric_history.append( { "step": _world_step(world), "metrics": _compact_metric_history_entry(metrics_at_step), } ) timeline.append(_timeline_snapshot(world, index=step_index)) should_capture = ( config.generate_animation and ( step_index == 0 or step_index == safe_steps or step_index % capture_interval == 0 ) ) if should_capture: frames.append(_capture_animation_frame(world, renderer, index=len(frames))) if step_index < safe_steps: animator.advance_world(world, steps=1) final_metrics = compute_metric_bundle( world, stability_tracker=stability_tracker, interestingness_tracker=interestingness_tracker, ) animation_path: str | None = None if config.generate_animation and frames: animation_path = _write_animation_with_fallback( animator=animator, frames=frames, output_dir=output_dir, requested_format=config.resolved_animation_format(), fallback_to_gif=config.fallback_to_gif_on_animation_error, ) population_chart_path: str | None = None resource_chart_path: str | None = None if config.generate_charts: population_chart_path = population_tracker.save_temp_png(output_dir=output_dir) resource_chart_path = resource_tracker.save_temp_png(output_dir=output_dir) final_image_path: str | None = None if config.generate_final_image: final_image_path = renderer.save_temp_png(world, output_dir=output_dir) artifacts = CallbackArtifacts( output_dir=str(output_dir), animation_path=animation_path, population_chart_path=population_chart_path, resource_chart_path=resource_chart_path, final_image_path=final_image_path, ) metrics_payload = { "final": _json_safe(final_metrics), "history": _json_safe(metric_history), } return SimulationRunData( final_world=world, artifacts=artifacts, metrics=metrics_payload, timeline=tuple(timeline), population_snapshots=tuple(population_tracker.snapshots), resource_snapshots=tuple(resource_tracker.snapshots), ) def compute_metric_bundle( world: Any, *, stability_tracker: StabilityTracker | None = None, interestingness_tracker: InterestingnessTracker | None = None, ) -> dict[str, Any]: """Compute default metrics for app display.""" bundle: dict[str, Any] = {} try: bundle["agent_diversity"] = compute_agent_type_diversity(world).to_dict() except Exception as exc: bundle["agent_diversity_error"] = str(exc) try: bundle["resource_diversity"] = compute_resource_type_diversity( world, weight_by_amount=True, ).to_dict() except Exception as exc: bundle["resource_diversity_error"] = str(exc) try: bundle["agent_entropy"] = compute_agent_type_entropy(world).to_dict() except Exception as exc: bundle["agent_entropy_error"] = str(exc) try: bundle["resource_entropy"] = compute_resource_type_entropy( world, weight_by_amount=True, ).to_dict() except Exception as exc: bundle["resource_entropy_error"] = str(exc) try: if stability_tracker is not None: bundle["stability"] = stability_tracker.update(world).to_dict() else: bundle["stability"] = StabilityTracker().update(world).to_dict() except Exception as exc: bundle["stability_error"] = str(exc) try: if interestingness_tracker is not None: bundle["interestingness"] = interestingness_tracker.update(world).to_dict() else: bundle["interestingness"] = InterestingnessTracker().update(world).to_dict() except Exception as exc: bundle["interestingness_error"] = str(exc) return bundle def narrate_world_for_app( world: Any, *, timeline: Sequence[Mapping[str, Any]], metrics: Mapping[str, Any], client: Any | None, config: CallbackConfig, ) -> str: """Create a narrative summary for app display.""" narrator = WorldNarrator( client=client, config=config.resolved_narrator_config(), ) result = narrator.narrate( world, history=timeline, metric_results=_metric_results_for_narrator(metrics), extra_context={ "source": "callbacks.py", "artifact_mode": "gradio_root_app", }, ) return result.text def parse_constraints(value: str | Mapping[str, Any] | None) -> Mapping[str, Any] | str | None: """Parse optional UI constraints. If the input is valid JSON, a mapping/list/scalar is returned from JSON. If it is plain text, the text is returned so the LLM prompt can still use it. """ if value is None: return None if isinstance(value, Mapping): return copy.deepcopy(dict(value)) text = str(value).strip() if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: return text if isinstance(parsed, Mapping): return copy.deepcopy(dict(parsed)) return parsed def apply_step_override(spec: WorldSpec, steps: int | None) -> WorldSpec: """Return a copy of ``spec`` with simulation steps overridden when provided.""" if steps is None: return spec safe_steps = max(0, int(steps)) simulation = spec.simulation.model_copy(update={"steps": safe_steps}) return spec.model_copy(update={"simulation": simulation}) def callback_failure_result( error: BaseException, *, world_spec_json: str = "", ) -> CallbackResult: """Return a Gradio-safe failure result.""" message = f"WorldSmithAI callback failed: {error.__class__.__name__}: {error}" payload = { "success": False, "error_type": error.__class__.__name__, "error": str(error), } if isinstance(error, WorldFactoryValidationError): payload["validation_report"] = error.report.to_dict() if isinstance(error, DSLParseError): payload["diagnostics"] = copy.deepcopy(dict(error.diagnostics)) return CallbackResult( success=False, status_message=message, world_spec_json=world_spec_json, metrics_json=_safe_json_dumps(payload), narrative=( "The simulation could not be completed. Check the status message " "and validation output for details." ), metadata=payload, ) def _capture_animation_frame( world: Any, renderer: WorldRenderer, *, index: int, ) -> AnimationFrame: """Capture one animation frame from the current world state.""" render_result = renderer.render_result(world) image = figure_to_rgb_array(render_result.figure) close_figure(render_result.figure) return AnimationFrame( index=index, step=_world_step(world), image=image, snapshot=render_result.snapshot, metadata={ "world_step": _world_step(world), "object_count": render_result.snapshot.object_count, }, ) def _write_animation_with_fallback( *, animator: WorldAnimator, frames: Sequence[AnimationFrame], output_dir: Path, requested_format: AnimationFormat, fallback_to_gif: bool, ) -> str | None: """Write animation frames and optionally fall back from MP4 to GIF.""" try: result = animator.write_frames( frames, output_path=_artifact_path(output_dir, "simulation", requested_format.value), format=requested_format, ) return result.path except AnimationWriterError: if not fallback_to_gif or requested_format is AnimationFormat.GIF: raise logger.warning("Animation writer failed for %s; falling back to GIF", requested_format.value) result = animator.write_frames( frames, output_path=_artifact_path(output_dir, "simulation", "gif"), format=AnimationFormat.GIF, ) return result.path def _artifact_path(output_dir: Path, stem: str, suffix: str) -> str: """Return a deterministic artifact path inside an output directory.""" suffix_text = suffix.lstrip(".") return str(output_dir / f"{stem}.{suffix_text}") def _capture_interval(*, steps: int, max_frames: int) -> int: """Return how often animation frames should be captured.""" if steps <= 0: return 1 if max_frames <= 1: return steps return max(1, int(math.ceil((steps + 1) / float(max_frames)))) def _timeline_snapshot(world: Any, *, index: int) -> Mapping[str, Any]: """Return compact timeline snapshot for narration.""" agents = _iter_collection(getattr(world, "agents", ())) resources = _iter_collection(getattr(world, "resources", ())) alive_count = sum(1 for agent in agents if _is_alive(agent)) total_resource_amount = sum(_as_float(getattr(resource, "amount", 0.0)) for resource in resources) return { "index": index, "step": _world_step(world), "agent_count": len(agents), "alive_agent_count": alive_count, "resource_count": len(resources), "total_resource_amount": total_resource_amount, } def _compact_metric_history_entry(metrics: Mapping[str, Any]) -> dict[str, Any]: """Return compact metric history values for charts and JSON display.""" output: dict[str, Any] = {} interestingness = _nested_get(metrics, ("interestingness", "score")) if _is_number(interestingness): output["interestingness_score"] = float(interestingness) stability = _nested_get(metrics, ("stability", "stability_score")) if _is_number(stability): output["stability_score"] = float(stability) entropy = _nested_get(metrics, ("agent_entropy", "normalized_entropy")) if _is_number(entropy): output["agent_normalized_entropy"] = float(entropy) diversity = _nested_get(metrics, ("agent_diversity", "gini_simpson_index")) if _is_number(diversity): output["agent_diversity"] = float(diversity) return output def _metric_results_for_narrator(metrics: Mapping[str, Any]) -> Mapping[str, Any]: """Extract final metric result mapping for the narrator.""" final_metrics = metrics.get("final") if isinstance(metrics.get("final"), Mapping) else metrics if not isinstance(final_metrics, Mapping): return {} return { "diversity": final_metrics.get("agent_diversity", {}), "entropy": final_metrics.get("agent_entropy", {}), "stability": final_metrics.get("stability", {}), "interestingness": final_metrics.get("interestingness", {}), "resource_diversity": final_metrics.get("resource_diversity", {}), "resource_entropy": final_metrics.get("resource_entropy", {}), } def _nested_get(mapping: Mapping[str, Any], path: Sequence[str]) -> Any: """Read a nested mapping path.""" current: Any = mapping for key in path: if not isinstance(current, Mapping) or key not in current: return None current = current[key] return current def _world_step(world: Any) -> 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 _iter_collection(raw_collection: Any) -> tuple[Any, ...]: """Return items from mapping-backed or sequence-backed collections.""" if raw_collection is None: return () if isinstance(raw_collection, Mapping): values = raw_collection.values() elif isinstance(raw_collection, Sequence) 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 _is_alive(item: Any) -> bool: """Return whether an item is alive when it exposes an alive field.""" return bool(getattr(item, "alive", True)) def _is_number(value: Any) -> bool: """Return whether a value is a real numeric scalar, excluding booleans.""" return isinstance(value, Real) and not isinstance(value, bool) def _as_float(value: Any, default: float = 0.0) -> float: """Convert numeric-like value to float.""" if _is_number(value): return float(value) return default def _coerce_optional_int(value: int | float | str | None) -> int | None: """Convert optional UI numeric value to int.""" if value is None or value == "": return None if isinstance(value, Real) and not isinstance(value, bool): return int(value) return int(float(str(value))) def _safe_json_dumps(value: Any) -> str: """Serialize a value as pretty JSON for Gradio code components.""" return json.dumps( _json_safe(value), indent=2, sort_keys=True, ensure_ascii=False, ) def _json_safe(value: Any) -> Any: """Return a JSON-friendly representation of arbitrary callback data.""" if value is None or isinstance(value, (str, bool)): return value if isinstance(value, int) and not isinstance(value, bool): return value if isinstance(value, float): if not math.isfinite(value): return None return value if isinstance(value, Mapping): return {str(key): _json_safe(nested) for key, nested in value.items()} if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return [_json_safe(item) for item in value] if hasattr(value, "to_dict") and callable(value.to_dict): return _json_safe(value.to_dict()) if hasattr(value, "model_dump") and callable(value.model_dump): return _json_safe(value.model_dump(mode="json")) return str(value) def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append a value while enforcing a max length.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] __all__ = [ "CallbackArtifacts", "CallbackConfig", "CallbackMode", "CallbackResult", "DEFAULT_GRADIO_OUTPUT_FIELDS", "SimulationRunData", "apply_step_override", "build_world_from_spec", "callback_failure_result", "compute_metric_bundle", "generate_dsl_callback", "generate_world_from_prompt", "load_example_callback", "narrate_world_for_app", "parse_constraints", "run_worldsmith_callback", "run_worldsmith_pipeline", "simulate_dsl_callback", "simulate_dsl_pipeline", "simulate_world_for_app", "validate_dsl_callback", ]