# -*- coding: utf-8 -*- """ AssetService für die ServiceContainer-Integration. Wrapper um den AssetManager, der als IService im ServiceContainer registriert werden kann. """ from pathlib import Path from typing import TYPE_CHECKING, ClassVar, Iterator from trixy_core.service import IService, ServicePriority, ServiceGroup, ServiceState from trixy_core.assets.manager import AssetManager, AssetType, AssetInfo, AssetNotFoundError from trixy_core.utils.debug import pinfo, pdebug if TYPE_CHECKING: from trixy_core.application import IApplication class AssetService(IService): """ Service-Wrapper für den AssetManager. Integriert den AssetManager in den ServiceContainer und übernimmt die Konfiguration aus der Anwendungskonfiguration. Der Service lädt die Assets-Konfiguration (base_dir, profile) aus der Anwendungskonfiguration und stellt den AssetManager für andere Services und Plugins bereit. Beispiel: # Im ServiceContainer asset_service = container.get(AssetService) # Asset-Zugriff path = asset_service.get("audio/success.wav") # Manager direkt manager = asset_service.manager for info in manager.list_audio(): print(info.name) """ NAME: ClassVar[str] = "AssetService" PRIORITY: ClassVar[ServicePriority] = ServicePriority.CRITICAL GROUP: ClassVar[ServiceGroup] = ServiceGroup.UTILITY DEPENDENCIES: ClassVar[list[str]] = [] def __init__(self, application: "IApplication") -> None: """ Initialisiert den AssetService. Args: application: Referenz zur Hauptanwendung """ super().__init__(application) self._manager: AssetManager | None = None @property def manager(self) -> AssetManager: """ Gibt den AssetManager zurück. Returns: AssetManager-Instanz Raises: RuntimeError: Wenn der Service nicht gestartet ist """ if self._manager is None: raise RuntimeError("AssetService ist nicht gestartet") return self._manager async def start(self) -> None: """Startet den AssetService.""" self._set_state(ServiceState.STARTING) # Konfiguration aus Application lesen config = self._application.config base_dir = getattr(config, "assets_directory", "assets") profile = getattr(config, "profile", "default") # Absoluten Pfad auflösen (relativ zum Arbeitsverzeichnis) base_path = Path(base_dir) if not base_path.is_absolute(): base_path = Path.cwd() / base_dir # Manager initialisieren self._manager = AssetManager( base_dir=base_path, profile=profile, cache_enabled=True, cache_ttl=60.0, ) self._set_state(ServiceState.RUNNING) pinfo(f"AssetService gestartet: profile={profile}, base={base_path}") async def stop(self) -> None: """Stoppt den AssetService.""" self._set_state(ServiceState.STOPPING) if self._manager is not None: self._manager.clear_cache() self._manager = None self._set_state(ServiceState.STOPPED) pdebug("AssetService gestoppt") async def health_check(self) -> bool: """Prüft die Gesundheit des Services.""" if self._manager is None: return False # Prüfe ob Basis-Verzeichnis existiert return self._manager.base_dir.exists() # ========================================================================= # Delegierte Methoden für einfachen Zugriff # ========================================================================= def get(self, asset_path: str, raise_on_missing: bool = False) -> Path | None: """ Gibt den Pfad zu einem Asset zurück. Args: asset_path: Relativer Pfad zum Asset raise_on_missing: Bei True wird AssetNotFoundError geworfen Returns: Pfad oder None """ return self.manager.get(asset_path, raise_on_missing) def get_info(self, asset_path: str) -> AssetInfo | None: """ Gibt detaillierte Informationen über ein Asset zurück. Args: asset_path: Relativer Pfad zum Asset Returns: AssetInfo oder None """ return self.manager.get_info(asset_path) def exists(self, asset_path: str) -> bool: """ Prüft, ob ein Asset existiert. Args: asset_path: Relativer Pfad zum Asset Returns: True wenn das Asset existiert """ return self.manager.exists(asset_path) def get_audio(self, filename: str) -> Path | None: """Gibt den Pfad zu einer Audio-Datei zurück.""" return self.manager.get_audio(filename) def get_music(self, filename: str) -> Path | None: """Gibt den Pfad zu einer Musik-Datei zurück.""" return self.manager.get_music(filename) def get_image(self, filename: str) -> Path | None: """Gibt den Pfad zu einer Bild-Datei zurück.""" return self.manager.get_image(filename) def list_assets( self, asset_type: AssetType | None = None, directory: str | None = None, include_default: bool = True, ) -> Iterator[AssetInfo]: """Listet alle Assets auf.""" return self.manager.list_assets(asset_type, directory, include_default) def list_audio(self) -> Iterator[AssetInfo]: """Listet alle Audio-Assets.""" return self.manager.list_audio() def list_music(self) -> Iterator[AssetInfo]: """Listet alle Musik-Assets.""" return self.manager.list_music() def list_profiles(self) -> list[str]: """Listet alle verfügbaren Profile.""" return self.manager.list_profiles() @property def profile(self) -> str: """Gibt das aktive Profil zurück.""" return self.manager.profile @profile.setter def profile(self, value: str) -> None: """Setzt das aktive Profil.""" self.manager.profile = value @property def base_dir(self) -> Path: """Gibt das Basis-Verzeichnis zurück.""" return self.manager.base_dir def get_stats(self) -> dict: """Gibt Statistiken zurück.""" return self.manager.get_stats() def clear_cache(self) -> None: """Leert den Asset-Cache.""" self.manager.clear_cache()