remoll/rpi-eink-picture-frame
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added other sources for more variety

Browse Source
This commit is contained in:
Remy Moll 2023-11-17 22:25:55 +01:00
parent b684c293b0
commit 08bdfe970c
10 changed files with 140 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,5 +1,5 @@
keys.py
*.pyc
.image-cache/
.cache/
test.png
wifi_networks.yml

3
deploy/deploy.playbook.yml
View File

@ -4,9 +4,12 @@
# become: true
vars:
# mostly fixed
repo_dest: /home/remy/eink
code_dest: /home/remy/eink/src
service_target_dir: /etc/systemd/system/
# adapt to taste
own_image_percent: 90
tasks:
- name: Pull the latest version of the code

1
deploy/templates/eink-show.service.j2
View File

@ -6,6 +6,7 @@ Type=oneshot
User={{ ansible_user }}
WorkingDirectory={{ code_dest }}
ExecStart=python main.py
Environment="OWN_IMAGE_PERCENT={{ own_image_percent }}"
[Install]

25
src/get/combined.py
View File

@ -0,0 +1,25 @@
from .immich import ImageGetImmich
from .unsplash import ImageGetUnsplash
from .museum import ImageGetMuseum
import random
import os
igetter = ImageGetImmich()
ugettter = ImageGetUnsplash()
mgetter = ImageGetMuseum()
class ImageGetCombined:
def __init__(self) -> None:
own_image_percent = int(os.getenv("OWN_IMAGE_PERCENT", 90))
other_image_percent = int(100 - own_image_percent)
self.getters = [igetter] * own_image_percent \
+ [ugettter] * int(other_image_percent/2) \
+ [mgetter] * int(other_image_percent/2)
# representing weighted probabilities
print(f"Using {len(self.getters)} image sources: ~{own_image_percent}% own, ~{other_image_percent}% foreign images.")
def get_random_image(self) -> bytearray:
getter = random.choice(self.getters)
return getter.get_random_image()

11
src/get/immich.py
View File

@ -17,8 +17,9 @@ class ImageGetImmich(ImageGet):
headers=headers
)
def get_random_image_ids(self, num=1) -> str:
url = keys.immich_api_root_url + "album/" + keys.immich_album_id
url = self.base_url + "album/" + keys.immich_album_id
headers = self.headers | {"Accept": "application/json"}
response = h.request("GET", url, headers=headers, data={})
@ -41,7 +42,7 @@ class ImageGetImmich(ImageGet):
def get_image_file(self, image_id: str) -> bytearray:
url = keys.immich_api_root_url + "asset/download/" + image_id
url = self.base_url + "asset/download/" + image_id
headers = self.headers | {"Accept": "application/octet-stream"}
response = h.request("POST", url, headers=headers, data={})
@ -49,9 +50,3 @@ class ImageGetImmich(ImageGet):
raise ImageGetException("Error in step get_image_file: " + str(response.status_code))
return response.content
# def save_cached_files(self) -> None:
# in super
# return

66
src/get/museum.py Normal file
View File

@ -0,0 +1,66 @@
import httpx as h
import random
import json
from .template import ImageGet, ImageGetException
class ImageGetMuseum(ImageGet):
def __init__(self) -> None:
headers = {
"AIC-User-Agent": "Eink Art Display (remoll@ethz.ch)"
}
super().__init__(
base_url="https://api.artic.edu/api/v1/artworks/",
headers=headers
)
def get_random_image_ids(self, num=1) -> list[str]:
post_data = {
"resources": "artworks",
"fields": ["id","image_id"],
"boost": False,
"limit": num,
"query": {
"function_score": {
"query": {
"bool": {
"filter": [
{"term": {"is_public_domain": True}},
{"exists": {"field": "image_id"}}
]
}
},
"boost_mode": "replace",
"random_score": {
"field": "id",
"seed": f"{random.randint(1, 1000000)}"
}
}
}
}
response = h.post(self.base_url + "search", json=post_data, headers=self.headers)
if not response.status_code == 200:
raise ImageGetException("Error in step get_image_file: " + str(response.status_code))
response = json.loads(response.text)
images = response["data"]
ids = [img["id"] for img in images]
return ids
def get_image_file(self, image_id: str) -> bytearray:
url = self.base_url + f"{image_id}"
print(url)
response = h.get(url, headers=self.headers)
if not response.status_code == 200:
raise ImageGetException("Error in step get_image_file: " + str(response.status_code))
response = json.loads(response.text)
image_url = response["config"]["iiif_url"] + "/" + response["data"]["image_id"] + "/full/843,/0/default.jpg"
image = h.get(image_url).content
return image

24
src/get/public.py
View File

@ -1,24 +0,0 @@
import keys
import httpx as h
import random
import json
from image_get import ImageGet, ImageGetException
class ImageGetImmich(ImageGet):
def __init__(self) -> None:
headers = {
"AIC-User-Agent": "Eink Art Display (remoll@ethz.ch)"
}
super().__init__(
base_url=keys.immich_api_root_url,
headers=self.headers
)
def get_random_image_ids(self) -> list[str]:
raise NotImplementedError
def get_image_file(self, image_id: str) -> bytearray:
raise NotImplementedError

20
src/get/template.py
View File

@ -13,16 +13,18 @@ class ImageGet:
self.base_url = base_url
self.headers = headers
self.cache_num = cache_num
self.cache_dir = Path("../.cache/{self.__class__.__name__}/")
self.cache_dir = Path(f"../.cache/{self.__class__.__name__}/")
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.save_cached_files()
# Main entrypoint, called externally
def get_random_image(self) -> bytearray:
try:
id = self.get_random_image_ids()[0]
bytes = self.get_image_file(id)
# self.save_cached_files()
self.save_cached_files()
except (h.ConnectError, h.HTTPStatusError, h.NetworkError, h.RequestError, h.DecodingError, h.TransportError, ImageGetException):
print("Loading image from cache")
bytes = self.load_cached_file()
@ -33,21 +35,21 @@ class ImageGet:
def save_cached_files(self) -> None:
"""Ensures self.cache_num files are at self.cache_dir at any time"""
present_count = len(list(self.cached_path.glob("*")))
missing = self.cached_num - present_count
present_count = len(list(self.cache_dir.glob("*")))
missing = self.cache_num - present_count
if missing == 0:
return
ids = self.get_random_image_ids(missing)
ids = self.get_random_image_ids(num = missing)
for i, id in enumerate(ids):
print(f"Caching image {i + 1}")
new_cache = self.cached_path / f"{uuid.uuid4()}"
new_cache = self.cache_dir / f"{uuid.uuid4()}"
new_cache.write_bytes(self.get_image_file(id))
def load_cached_file(self) -> bytearray:
"""Returns a random file from self.cached_path"""
files = list(self.cached_path.glob("*"))
"""Returns a random file from self.cache_dir"""
files = list(self.cache_dir.glob("*"))
if len(files) == 0:
raise ImageGetException("Could not load cached file: directory empty")
@ -59,7 +61,7 @@ class ImageGet:
def get_random_image_ids(self) -> list[str]:
def get_random_image_ids(self, num=1) -> list[str]:
raise NotImplementedError

27
src/get/unsplash.py Normal file
View File

@ -0,0 +1,27 @@
import httpx as h
from .template import ImageGet, ImageGetException
class ImageGetUnsplash(ImageGet):
def __init__(self) -> None:
super().__init__(
base_url="https://source.unsplash.com/",
headers={}
)
def get_random_image_ids(self, num=1) -> list[str]:
# there are no ids since the api has a random endpoint
return [0 for i in range(num)]
def get_image_file(self, image_id: str = "doesntmatter") -> bytearray:
url = self.base_url + "random"
response = h.get(url, follow_redirects=True)
if not response.status_code == 200:
raise ImageGetException("Error in step get_image_file: " + str(response.status_code))
return response.content

4
src/main.py
View File

@ -1,6 +1,6 @@
import sys
from image_convert import ImageShrink
from get import immich
from get.combined import ImageGetCombined
from image_show import ImageShow
if len(sys.argv) == 2 and sys.argv[1] == "test":
@ -18,7 +18,7 @@ if "noreduce" in sys.argv:
print("Disabling color reduction")
shrink_kwargs["colors"] = -1
getter = immich.ImageGetImmich()
getter = ImageGetCombined()
converter = ImageShrink(**shrink_kwargs)
shower = ImageShow()