Source code for miniworlds.base.app
import __main__
import atexit
import os
import sys
import warnings
import asyncio
import logging
import pygame
from typing import List, Optional, TYPE_CHECKING, cast
import miniworlds.appearances.managers.image_manager as image_manager
import miniworlds.base.manager.app_event_manager as event_manager
import miniworlds.base.manager.app_worlds_manager as worlds_manager
import miniworlds.base.manager.app_music_manager as music_manager
import miniworlds.base.manager.app_sound_manager as sound_manager
import miniworlds.base.app_state as app_state
import miniworlds.base.app_state_bridge as app_state_bridge
import miniworlds.base.platform as platform_mod
import miniworlds.base.window as window_mod
from miniworlds.base.window import Window
if TYPE_CHECKING:
from miniworlds.base.manager.app_event_manager import AppEventManager
from miniworlds.base.manager.app_worlds_manager import WorldsManager
from miniworlds.base.manager.app_music_manager import MusicManager
from miniworlds.base.manager.app_sound_manager import SoundManager
from miniworlds.worlds.world import World
logger = logging.getLogger(__name__)
[docs]
class App:
"""
Main application class for Miniworlds.
Created automatically when `world.run()` is called for the first time.
Raises:
NoRunError: If `run()` is not called from the main module.
"""
_state = app_state.AppState()
_state_bridge: "app_state_bridge.AppStateBridge | None" = None
_fallback_platform = platform_mod.PlatformAdapter()
running_world: Optional["World"] = None
running_worlds: List["World"] = []
path: str | None = ""
running_app: Optional["App"] = None
init: bool = False
window: Optional["Window"] = None
_run_started: bool = False
_run_check_registered: bool = False
@classmethod
def _get_state_bridge(cls) -> app_state_bridge.AppStateBridge:
bridge = cls._state_bridge
if bridge is None or bridge.app_class is not cls or bridge.state is not cls._state:
bridge = app_state_bridge.AppStateBridge(cls, cls._state)
cls._state_bridge = bridge
return bridge
@classmethod
def _sync_class_state(cls) -> None:
cls._get_state_bridge().sync_class_state()
[docs]
@classmethod
def get_running_world(cls) -> Optional["World"]:
return cls._get_state_bridge().get_running_world()
[docs]
@classmethod
def get_running_app(cls) -> Optional["App"]:
return cls._get_state_bridge().get_running_app()
[docs]
@classmethod
def get_window(cls) -> Optional["Window"]:
return cls._get_state_bridge().get_window()
[docs]
@classmethod
def get_path(cls) -> str | None:
return cls._get_state_bridge().get_path()
[docs]
@staticmethod
def reset(unittest=False, file=None):
"""
Resets all app globals.
Args:
unittest: Whether the reset is being called in a unit test context.
file: Optional file path to use for setting the base path.
"""
App._get_state_bridge().reset(unittest=unittest, file=file)
App.init = False
import miniworlds.base.manager.app_file_manager as app_file_manager
app_file_manager.FileManager.clear_cache()
[docs]
@staticmethod
def check_for_run_method():
"""
Verifies that `.run()` is called in the user's main module.
Prints a warning if it's not found (except in emscripten or notebooks).
"""
try:
content = App.get_platform().read_main_module()
if content is not None and ".run(" not in content:
warnings.warn(
"""[world_name].run() was not found in your code.
This must be the last line in your code
\ne.g.:\nworld.run()\n if your world-object is named world.""")
except AttributeError:
if not App.get_platform().is_web():
logger.info(
"Skipping run() presence check because the main module could not be read"
)
[docs]
@classmethod
def register_run_check(cls) -> None:
"""
Registers an exit hook that warns when a world was created but
`run()` was never called.
The source-based check in `check_for_run_method()` cannot detect
`run()` calls that are commented out or never reached. This hook
runs after the program ends and reports if no mainloop ever
started, so students learn why no window appeared.
"""
if cls._run_check_registered or cls.get_platform().is_web():
return
if "pytest" in sys.modules:
return
cls._run_check_registered = True
atexit.register(cls._warn_if_never_run)
@classmethod
def _warn_if_never_run(cls) -> None:
if cls._run_started:
return
warnings.warn(
"A world was created, but [world_name].run() was never called - "
"no window was shown.\n"
"Add world.run() as the last line of your program "
"(if your world-object is named world)."
)
def _output_start(self):
"""
Outputs version info at app start (desktop only).
"""
if not self.platform.is_web():
version_str = self.platform.get_package_version("miniworlds")
print(f"miniworlds version: {version_str}")
logger.info("Starting miniworlds window for version %s", version_str)
[docs]
def __init__(self, title, world):
"""
Initializes the App and all its managers.
Args:
title: Title for the window.
world: The initial world object to be run.
"""
self.platform = platform_mod.PlatformAdapter()
self._output_start()
self.check_for_run_method()
self.register_run_check()
self.worlds_manager: "WorldsManager" = worlds_manager.WorldsManager(self)
self.event_manager: "AppEventManager" = event_manager.AppEventManager(self)
self.sound_manager: "SoundManager" = sound_manager.SoundManager(self)
self.music_manager: "MusicManager" = music_manager.MusicManager(self)
self.window: "Window" = window_mod.Window(title, self, self.worlds_manager, self.event_manager)
self._quit = False
self._unittest = False
self._skip_frame_delay = os.getenv("MINIWORLDS_TEST_FAST") == "1"
self._mainloop_started: bool = False
self._exit_code: int = 0
self.image = None
self.repaint_areas: List = []
self._get_state_bridge().bind_app(self, world, self.window)
if self.get_path():
self.path = self.get_path()
[docs]
async def run(self, image, fullscreen: bool = False, fit_desktop: bool = False, replit: bool = False):
"""
Starts the app and enters the mainloop.
Args:
image: The background image to display.
fullscreen: Whether to start in fullscreen mode.
fit_desktop: Whether to adapt the window to desktop size.
replit: Whether running in replit environment.
"""
self.image = image
self.window = cast(Window, self.window)
self.window.fullscreen = fullscreen
self.window.fit_desktop = fit_desktop
self.window.replit = replit
App._run_started = True
self.init_app()
App.init = True
self.prepare_mainloop()
if not self._mainloop_started:
await self.start_mainloop()
else:
for world in self.running_worlds:
world.dirty = 1
world.background.set_dirty("all", 2)
[docs]
def init_app(self):
"""
Initializes global resources (e.g., image cache).
"""
# Avoid preloading all images when running on the web (Pyodide):
# preloading triggers many HTTP requests and slows startup.
if not self.platform.is_web():
image_manager.ImageManager.cache_images_in_image_folder()
[docs]
def prepare_mainloop(self):
"""
Prepares all world objects for drawing.
"""
self.resize()
for world in self.running_worlds:
world.dirty = 1
world.background.set_dirty("all", 2)
[docs]
async def start_mainloop(self):
"""
Starts the main event loop.
"""
self._mainloop_started = True
finished_normally = False
try:
while not self._quit:
await self._update()
finished_normally = True
finally:
self._finalize_mainloop(finished_normally)
def _finalize_mainloop(self, finished_normally: bool) -> None:
self._mainloop_started = False
if self._unittest:
return
if self.platform.is_web():
# SDL's display teardown can segfault under Emscripten/Pyodide
# ("memory access out of bounds"), which kills the interpreter and
# breaks stop -> restart in web hosts. The host owns the canvas
# lifecycle there, and set_mode() re-initializes the display on the
# next run, so quitting the display is neither safe nor needed.
return
self.platform.quit_display()
if finished_normally:
sys.exit(self._exit_code)
async def _update(self):
"""
A single iteration of the mainloop.
Handles events, updates worlds, redraws screen.
"""
self.event_manager.pygame_events_to_event_queue()
if self.window.dirty:
self.resize()
if not self._quit:
self.event_manager.handle_event_queue()
frame_wait = await self.worlds_manager.reload_all_worlds()
self.display_repaint()
if frame_wait is not None:
# One frame wait per app frame for all worlds together
# (a per-world wait would multiply the frame delay).
await self.platform.wait_for_frame(frame_wait, self._skip_frame_delay)
await self.platform.yield_mainloop()
[docs]
def set_running_world(self, world: Optional["World"]) -> None:
self._get_state_bridge().set_running_world(world)
[docs]
def add_running_world(self, world: "World") -> None:
self._get_state_bridge().add_running_world(world)
[docs]
def remove_running_world(self, world: "World") -> None:
self._get_state_bridge().remove_running_world(world)
[docs]
def quit(self, exit_code=0):
"""
Signals the mainloop to exit.
Args:
exit_code: Exit code to use when quitting.
"""
self._exit_code = exit_code
self._quit = True
[docs]
def register_path(self, path):
"""
Registers the app path for relative resource access.
Args:
path: Path to the project directory.
"""
self.path = path
self._get_state_bridge().set_path(path)
import miniworlds.base.manager.app_file_manager as app_file_manager
app_file_manager.FileManager.clear_cache()
[docs]
def display_repaint(self):
"""
Repaints the regions marked as dirty (called every frame).
"""
if not self.repaint_areas:
return
self.platform.update_display(self.repaint_areas)
self.repaint_areas = []
[docs]
def display_update(self):
"""
Repaints the full display if it was marked dirty.
Note:
This could be merged with display_repaint and update_surface.
"""
if self.window.dirty:
self.window.dirty = 0
self.add_display_to_repaint_areas()
self.platform.update_display(self.repaint_areas)
self.repaint_areas = []
[docs]
def add_display_to_repaint_areas(self):
"""
Adds the full screen area to the repaint queue.
"""
self.repaint_areas.append(pygame.Rect(0, 0, self.window.width, self.window.height))
[docs]
def resize(self):
"""
Resizes the window surface and updates all layout-related components.
"""
self.worlds_manager.recalculate_dimensions()
self.window._update_surface()
self.display_update()