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
130
131
132
133
134
135
136
137
|
import os
import sys
import signal
import copy
from threading import Thread
from pathlib import Path
from OpenGL.GLUT import glutLeaveMainLoop
class PixMan:
_instance = None
_initialized = False
@staticmethod
def handle_sigint(sig, frame):
try:
# Threads are started as daemons so will automatically shut down
glutLeaveMainLoop()
sys.exit(0)
except:
pass
finally:
print("Exiting...")
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
signal.signal(signal.SIGINT, PixMan.handle_sigint)
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({}) # TODO glut args
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
@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
|
