E-Paper Admin

Edit Code: reddit-photo

Select File:

Font Size: 14px

import time
import threading
import requests
import urllib.request
import re
import random
from io import BytesIO
from PIL import Image, ImageDraw, ImageFont

try:
    from gpiozero import Button
    HAS_GPIO = True
except Exception:
    HAS_GPIO = False

try:
    from waveshare_epd import epd2in7
    HAS_DISPLAY = True
    epd = epd2in7.EPD()
    epd.init()
    epd.Clear(0xFF)
except Exception:
    HAS_DISPLAY = False
    epd = None

# Global state
import json
import os

config_path = os.path.join(os.path.dirname(__file__), 'config.json')
try:
    with open(config_path, 'r') as f:
        config = json.load(f)
except Exception:
    config = {}

subreddit = config.get("subreddit", "EarthPorn")
display_orientation = config.get("display_orientation", "landscape")
refresh_interval = config.get("refresh_interval", 180)
current_orientation = "landscape"
cached_image = None
photo_queue = []
fetch_trigger = threading.Event()
fetch_trigger.set() # trigger initially

def fetch_rss_queue():
    url = f"https://www.reddit.com/r/{subreddit}/hot.rss"
    req = urllib.request.Request(
        url,
        headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
    )
    with urllib.request.urlopen(req, timeout=10) as response:
        content = response.read().decode('utf-8')
        
    entries = content.split('<entry>')
    photos = []
    for entry in entries[1:]:
        title_match = re.search(r'<title>([^<]+)</title>', entry)
        link_match = re.search(r'<media:thumbnail url="([^"]+)"', entry)
        if title_match and link_match:
            title = title_match.group(1).replace('&amp;', '&')
            link = link_match.group(1).replace('&amp;', '&')
            photos.append((title, link))
            
    if photos:
        random.shuffle(photos)
        return photos
    else:
        raise Exception("No photos found in RSS feed")

def fetch_photo():
    global photo_queue
    
    # Refill queue if empty
    if not photo_queue:
        print("Fetching new photos from RSS...")
        photo_queue = fetch_rss_queue()
        
    title, img_url = photo_queue.pop(0)
    
    img_response = requests.get(img_url, timeout=10)
    img = Image.open(BytesIO(img_response.content))
    return img, title

import sys
import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'lib')))
import epaper_utils

def process_image(img, title):
    global current_orientation
    
    show_details = config.get("show_details", "no") == "yes"
    
    # Process it via lib
    processed_img = epaper_utils.process_image(
        img, 
        display_orientation, 
        post_details=title, 
        show_details=show_details
    )
    
    # We still need to track current_orientation for updating the display hardware later
    current_orientation = epaper_utils.resolve_orientation(img.width, img.height, display_orientation)
    
    return processed_img

def background_fetch_loop():
    global cached_image
    while True:
        fetch_trigger.wait(timeout=refresh_interval)
        fetch_trigger.clear()
        
        try:
            raw_img, title = fetch_photo()
            processed_img = process_image(raw_img, title)
            cached_image = processed_img
            update_display()
        except Exception as e:
            print("Fetch failed:", e)
            # If we got rate limited (429), photo_queue will be empty
            # Next loop will try again. We could sleep longer here.
            time.sleep(10)

def update_display():
    if cached_image is None:
        image = epaper_utils.create_loading_screen(f"Loading r/{subreddit}...", display_orientation)
    else:
        image = cached_image
        
    if HAS_DISPLAY:
        epd.display(epd.getbuffer(image))
    else:
        image.save("dashboard_preview.png")

def main():
    fetch_thread = threading.Thread(target=background_fetch_loop, daemon=True)
    fetch_thread.start()
    
    update_display()
    
    if HAS_GPIO:
        btn1 = Button(5, bounce_time=0.1)
        btn2 = Button(6, bounce_time=0.1)
        btn3 = Button(13, bounce_time=0.1)
        btn4 = Button(19, bounce_time=0.1)
        
        try:
            while True:
                if btn1.is_pressed or btn2.is_pressed or btn3.is_pressed or btn4.is_pressed:
                    fetch_trigger.set()
                    time.sleep(0.5)
                time.sleep(0.1)
        except KeyboardInterrupt:
            pass
    else:
        print("Simulation mode. Sleeping forever...")
        try:
            while True:
                time.sleep(10)
        except KeyboardInterrupt:
            pass

if __name__ == '__main__':
    main()

Simulation Output

Click "Save & Start Live Simulation" to run the app locally and view the generated dashboard_preview.png in real-time.

✨ Antigravity Assistant

Ask the AI agent to write or modify code for this app. The agent can read and write files directly in the app directory.