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
|
import os
import sys
import signal
import copy
from threading import Thread
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)
# 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()
self.display.update_config()
if oldconfig.album_list != self.config.album_list:
self.update_textures()
else:
self.texture_list.max_cache_items = self.config.max_cache_assets
# If all goes well
self.config.save(self.configfile)
return True
from lazycachelist import LazyCachingTextureList
from window import PixDisplay
from immich import ImmichConnector
from settings import Config
|
