diff --git a/Pi_Bluesky_TFT_Scroller/.data/.gitkeep b/Pi_Bluesky_TFT_Scroller/.data/.gitkeep new file mode 100644 index 000000000..e69de29bb diff --git a/Pi_Bluesky_TFT_Scroller/code.py b/Pi_Bluesky_TFT_Scroller/code.py new file mode 100644 index 000000000..69be8635d --- /dev/null +++ b/Pi_Bluesky_TFT_Scroller/code.py @@ -0,0 +1,196 @@ +# SPDX-FileCopyrightText: 2024 Tim Cocks +# +# SPDX-License-Identifier: MIT +""" +Bluesky_RPi_TFT_Scroller code.py +Infinitely scroll Bluesky posts on a 320x240 pixel TFT +""" +import json +import os +import sys + +import requests +import webview + +FEEDLINK_RETROCOMPUTING = ( + "https://bsky.app/profile/did:plc:tbo4hkau3p2itkar2vsnb3gp/feed/aaabo5oe7bzok" +) + +# Un-comment a single key inside of FEED_ARGS and set it's value to the feed, list or search +# that you want to scroll. +FETCH_ARGS = { + # "feed_share_link": FEEDLINK_RETROCOMPUTING, + # "feed_share_link": "https://bsky.app/profile/did:plc:463touruejpokvutnn5ikxb5/lists/3lbfdtahfzt2a", # pylint: disable=line-too-long + # "search_args": {"q": "Adafruit", "sort": "latest"} + "search_args": {"q": "#circuitpython", "sort": "latest"} +} + + +def at_feed_uri_from_share_link(share_link): + """ + Converts a share link into an AT URI for that resource. + + :param share_link: The share link to convert. + :return str: The AT URI pointing at the resource. + """ + at_feed_uri = share_link.replace("https://bsky.app/profile/", "at://") + if "/feed/" in share_link: + at_feed_uri = at_feed_uri.replace("/feed/", "/app.bsky.feed.generator/") + if "/lists/" in share_link: + at_feed_uri = at_feed_uri.replace("/lists/", "/app.bsky.graph.list/") + return at_feed_uri + + +def fetch_data(feed_share_link=None, search_args=None): + """ + Fetch posts from Bluesky API and write them into the local cached + data files. After posts are written locally iterates over them + and downloads the relevant photos from them. + + Must pass either feed_share_link or search_args. + + :param feed_share_link: The link copied from Bluesky front end to share the feed or list. + :param search_args: A dictionary containing at minimum a ``q`` key with string value of + the hashtag or term to search for. See bsky API docs for other supported keys. + :return: None + """ + # pylint: disable=too-many-statements,too-many-branches + if feed_share_link is None and search_args is None: + # If both inputs are None, just use retrocomputing feed. + feed_share_link = FEEDLINK_RETROCOMPUTING + + # if a share link input was provided + if feed_share_link is not None: + FEED_AT = at_feed_uri_from_share_link(feed_share_link) + # print(FEED_AT) + + # if it's a feed + if "/app.bsky.feed.generator/" in FEED_AT: + URL = (f"https://public.api.bsky.app/xrpc/app.bsky.feed.getFeed?" + f"feed={FEED_AT}&limit=30") + headers = {"Accept-Language": "en"} + resp = requests.get(URL, headers=headers) + + # if it's a list + elif "/app.bsky.graph.list/" in FEED_AT: + URL = (f"https://public.api.bsky.app/xrpc/app.bsky.feed.getListFeed?" + f"list={FEED_AT}&limit=30") + headers = {"Accept-Language": "en"} + resp = requests.get(URL, headers=headers) + + # raise error if it's an unknown type + else: + raise ValueError( + "Only 'app.bsky.feed.generator' and 'app.bsky.graph.list' URIs are supported." + ) + + # if a search input was provided + if search_args is not None: + URL = "https://public.api.bsky.app/xrpc/app.bsky.feed.searchPosts" + headers = {"Accept-Language": "en"} + resp = requests.get(URL, headers=headers, params=search_args) + + with open(".data/raw_data.json", "wb") as f: + # write raw response to cache + f.write(resp.content) + + # Process the post data into a smaller subset + # containing just the bits we need for showing + # on the TFT. + resp_json = json.loads(resp.text) + processed_posts = {"posts": []} + fetched_posts = None + if "feed" in resp_json.keys(): + fetched_posts = resp_json["feed"] + elif "posts" in resp_json.keys(): + fetched_posts = resp_json["posts"] + + for post in fetched_posts: + cur_post = {} + if "post" in post.keys(): + post = post["post"] + cur_post["author"] = post["author"]["handle"] + cur_post["text"] = post["record"]["text"] + + # image handling + if "embed" in post.keys(): + cid = post["cid"] + if "images" in post["embed"].keys(): + cur_post["image_url"] = post["embed"]["images"][0]["thumb"] + elif "thumbnail" in post["embed"].keys(): + cur_post["image_url"] = post["embed"]["thumbnail"] + elif ( + "external" in post["embed"].keys() + and "thumb" in post["embed"]["external"].keys() + ): + cur_post["image_url"] = post["embed"]["external"]["thumb"] + + # if we actually have an image to show + if "image_url" in cur_post.keys(): + # check if we already downloaded this image + if f"{cid}.jpg" not in os.listdir("static/imgs/"): + print(f"downloading: {cur_post['image_url']}") + + # download image and write to file + img_resp = requests.get(cur_post["image_url"]) + with open(f"static/imgs/{cid}.jpg", "wb") as f: + f.write(img_resp.content) + + cur_post["image_file"] = f"{cid}.jpg" + processed_posts["posts"].append(cur_post) + + # save the processed data to a file + with open(".data/processed_data.json", "w", encoding="utf-8") as f: + f.write(json.dumps(processed_posts)) + + +def read_cached_data(): + """ + Load the cached processed data file and return + the data from within it. + + :return: The posts data loaded from JSON + """ + with open(".data/processed_data.json", "r") as f: + return json.load(f) + + +class Api: + """ + API object for interaction between python code here + and JS code running inside the page. + """ + + # pylint: disable=no-self-use + def get_posts(self): + """ + Fetch new posts data from Bluesky API, cache and return it. + :return: Processed data containing everything necessary to show + posts on the TFT. + """ + fetch_data(**FETCH_ARGS) + return read_cached_data() + + def check_quit(self): + """ + Allows the python code to correctly handle KeyboardInterrupt + more quickly. + + :return: None + """ + # pylint: disable=unnecessary-pass + pass + + def quit(self): + window.destroy() + sys.exit(0) + + +# create a webview and load the index.html page +window = webview.create_window( + "bsky posts", "static/index.html", js_api=Api(), width=320, height=240, + x=0, y=0, frameless=True, fullscreen=True + +) +webview.start() +# webview.start(debug=True) # use this one to enable chromium dev tools to see console.log() output from the page. diff --git a/Pi_Bluesky_TFT_Scroller/install_apt_requirements.sh b/Pi_Bluesky_TFT_Scroller/install_apt_requirements.sh new file mode 100644 index 000000000..22a27f28e --- /dev/null +++ b/Pi_Bluesky_TFT_Scroller/install_apt_requirements.sh @@ -0,0 +1,3 @@ +sudo apt install fonts-noto-color-emoji +sudo apt install python3-webview +sudo apt install python3-requests diff --git a/Pi_Bluesky_TFT_Scroller/static/index.html b/Pi_Bluesky_TFT_Scroller/static/index.html new file mode 100644 index 000000000..effa6792a --- /dev/null +++ b/Pi_Bluesky_TFT_Scroller/static/index.html @@ -0,0 +1,46 @@ + + +
+ +