Quellcode für 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__)




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()



    @classmethod
    def get_running_world(cls) -> Optional["World"]:
        return cls._get_state_bridge().get_running_world()




    @classmethod
    def get_running_app(cls) -> Optional["App"]:
        return cls._get_state_bridge().get_running_app()




    @classmethod
    def get_window(cls) -> Optional["Window"]:
        return cls._get_state_bridge().get_window()




    @classmethod
    def get_path(cls) -> str | None:
        return cls._get_state_bridge().get_path()




    @classmethod
    def get_platform(cls) -> platform_mod.PlatformAdapter:
        running_app = cls.get_running_app()
        if running_app and hasattr(running_app, "platform"):
            return running_app.platform
        return cls._fallback_platform




    @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()




    @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"
                )




    @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)



    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()




    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)




    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()




    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)




    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()



    def set_running_world(self, world: Optional["World"]) -> None:
        self._get_state_bridge().set_running_world(world)




    def add_running_world(self, world: "World") -> None:
        self._get_state_bridge().add_running_world(world)




    def remove_running_world(self, world: "World") -> None:
        self._get_state_bridge().remove_running_world(world)




    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




    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()




    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 = []




    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 = []




    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))




    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()