import os import sys import copy from threading import Thread from pathlib import Path class PixMan: _instance = None _initialized = False def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance = super().__new__(cls) else: assert not args and not kwargs, f"Singleton {cls.__name__} cannot be called with arguments more than once" return cls._instance def __init__(self): pass @classmethod def initialize(cls, args, app, socketio): assert not cls._initialized, "Already initialized" if cls._initialized: return cls._initialized = True self = cls() self.args = args self.app = app self.socketio = socketio self.immich_connector = None self.texture_list = None self.display = None self.t_flask = None self.t_idle_download = None self.configfile = self.args.config self.config = Config.load(self.configfile) if os.path.exists(self.configfile) else Config() self.init_web(self.args.host, self.args.port) self.update_config(self.config, replace=True) def init_web(self, host, port): self.t_flask = Thread(target=self.app.run, kwargs={ "host": host, "port": port }) self.t_flask.start() def init_window(self): # Initialize immich connector self.immich_connector = ImmichConnector() if not self.immich_connector.validate_connection(): self.immich_connector = None return # Initialize texture list self.update_textures() # Begin downloading images self.t_idle_download = Thread(target=self.immich_connector.idle, daemon=True) self.t_idle_download.start() # Create display self.display = PixDisplay(self.texture_list) self.display.main(fullscreen=self.args.fullscreen) self.die() def update_textures(self): if self.texture_list: self.texture_list.free() change_callback = lambda d: self.socketio.emit("seek", d) self.texture_list = LazyCachingTextureList(self.config.album_list, max_cache_items=self.config.max_cache_assets, change_callback=change_callback) if self.display: self.display.update_textures(self.texture_list) def update_config(self, config, replace=False): oldconfig = copy.deepcopy(self.config) if replace: self.config = config else: self.config.update(**config) if self.display: self.display.update_config() if oldconfig.album_list != self.config.album_list: self.update_textures() elif self.texture_list: self.texture_list.max_cache_items = self.config.max_cache_assets # If all goes well self.config.save(self.configfile) # Initialize window if immich parameters are valid if self.config.immich_url and self.config.immich_api_key and not self.display: self.init_window() return True def die(self): # Join threads and exit self.t_flask.join() self.t_idle_download.join() sys.exit(0) @property def frozen(self): # For pyinstaller # https://api.arcade.academy/en/latest/tutorials/bundling_with_pyinstaller/index.html#handling-data-files if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"): return True return False @property def static_dir(self): user_assets = os.environ.get("IMMICH_FRAME_STATIC_WEB_ASSETS") if user_assets: return Path(user_assets) if self.frozen: return Path(sys._MEIPASS) / "dist" return Path("../../dist") from .lazycachelist import LazyCachingTextureList from .window import PixDisplay from .immich import ImmichConnector from .settings import Config