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
|
import requests
import random
from io import BytesIO
from queue import Queue
from .texture import ImageTexture
from .manager import PixMan
class ImmichConnector:
def __init__(self):
config = PixMan().config
self.server_url = config.immich_url.removesuffix("/")
self.api_key = config.immich_api_key
self.texture_load_queue = Queue()
def _request(self, endpoint):
if not self.server_url:
return None
return requests.get(f"{self.server_url}/api/{endpoint}", headers={ "x-api-key": self.api_key })
def load_all_albums(self):
response = self._request("albums")
if not response or response.status_code != 200: return
data = response.json()
return data
def load_album_assets(self, key):
response = self._request(f"albums/{key}")
if not response or response.status_code != 200: return
data = response.json()
return data["assets"]
def load_albums_assets(self, keys):
config = PixMan().config
assets = []
for key in keys:
assets += self.load_album_assets(key)
if config.order == "newest-first" or config.order == "oldest-first":
# NOTE: asset.createdAt is a ISO 8601 date string, which are able to be sorted via python string sorting
assets_sorted = sorted(assets, key=lambda asset: asset["createdAt"], reverse=(config.order == "newest-first"))
elif config.order == "random":
assets_sorted = assets
random.shuffle(assets_sorted)
assert assets_sorted, f"Invalid config value for 'order': {config.order}"
return assets_sorted
def load_image(self, key, size="preview"):
response = self._request(f"assets/{key}/thumbnail?size={size}")
if not response or response.status_code != 200: return None, None
image_data = BytesIO(response.content)
mimetype = response.headers.get("Content-Type")
return image_data, mimetype
def load_image_filename(self, key):
response = self._request(f"assets/{key}")
if not response or response.status_code != 200: return None, None
data = response.json()
return data["originalFileName"]
def load_texture_async(self, texture_list, image_texture):
self.texture_load_queue.put((texture_list, image_texture))
def idle(self):
config = PixMan().config
size = config.display_size
while True:
texture_list, image_texture = self.texture_load_queue.get()
if not texture_list.index_in_cache_range(image_texture.asset_index) or texture_list.void:
continue # Texture was never loaded so it doesn't need to be free'd
image_data, _ = self.load_image(image_texture.asset_key, size)
image_texture.initialize(image_data)
def validate_connection(self):
# TODO
return True
|
