1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
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, configfile, app, socketio, host, port):
assert not cls._initialized, "Already initialized"
if cls._initialized:
return
cls._initialized = True
self = cls()
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 = configfile
self.config = Config.load(self.configfile) if os.path.exists(self.configfile) else Config()
self.init_web(host, 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()
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):
if self.frozen:
return Path(sys._MEIPASS) / "static"
return Path("./static/dist")
@property
def static_public(self):
if self.frozen:
return Path(sys._MEIPASS) / "public"
return Path("./static/public")
from lazycachelist import LazyCachingTextureList
from window import PixDisplay
from immich import ImmichConnector
from settings import Config
|
