""" E-Ink Photo Frame Server Serves photos for the ESP32 photo frame and provides a web UI for management. Photos come from pluggable provider backends (local directory, SFTP, cloud, etc). Tracks frame status via heartbeat reports from the ESP32. """ import argparse import io import json import sys from datetime import datetime, timezone from pathlib import Path from flask import Flask, Response, jsonify, request, render_template_string from werkzeug.utils import secure_filename from PIL import Image from dither import dither_floyd_steinberg from providers import ( ProviderManager, available_provider_types, migrate_image_settings, ) # Import providers to trigger registration import providers.local # noqa: F401 app = Flask(__name__) PHYSICAL_WIDTH = 800 PHYSICAL_HEIGHT = 480 DATA_DIR = Path("/data") STATE_FILE = DATA_DIR / "frame_state.json" SETTINGS_FILE = DATA_DIR / "settings.json" IMAGE_SETTINGS_FILE = DATA_DIR / "image_settings.json" # --- Persistence helpers --- def _load_json(path: Path, default): try: if path.exists(): return json.loads(path.read_text()) except Exception: pass return default() if callable(default) else default def _save_json(path: Path, data): try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2)) except Exception as e: print(f"Warning: could not save {path}: {e}", file=sys.stderr) # Frame state (runtime, updated by heartbeats) frame_state = _load_json(STATE_FILE, lambda: { "last_update": None, "current_photo": None, "ip": None, "updates": 0 }) # Frame settings (user-configured) DEFAULT_SETTINGS = { "orientation": "landscape", "default_mode": "zoom", "letterbox_color": [255, 255, 255], } frame_settings = _load_json(SETTINGS_FILE, lambda: dict(DEFAULT_SETTINGS)) # Per-image settings keyed by composite key: "provider_id:photo_id" image_settings: dict = _load_json(IMAGE_SETTINGS_FILE, dict) # Provider manager — initialized in main provider_manager: ProviderManager = None # type: ignore def _parse_photo_key(photo_key: str) -> tuple[str, str]: """Parse a composite key into (provider_id, photo_id).""" if ":" not in photo_key: # Legacy bare filename — assume local-default return "local-default", photo_key return photo_key.split(":", 1) def get_image_settings(composite_key: str) -> dict: """Get effective settings for an image (per-image overrides merged with defaults).""" defaults = {"mode": None, "pan_x": 0.5, "pan_y": 0.5} per_image = image_settings.get(composite_key, {}) merged = {**defaults, **per_image} if merged["mode"] is None: merged["mode"] = frame_settings.get("default_mode", "zoom") return merged # --- Image processing (unchanged — operates on PIL Images) --- def get_logical_dimensions() -> tuple[int, int]: orientation = frame_settings.get("orientation", "landscape") if orientation == "landscape": return PHYSICAL_WIDTH, PHYSICAL_HEIGHT else: return PHYSICAL_HEIGHT, PHYSICAL_WIDTH def process_image(img: Image.Image, settings: dict) -> Image.Image: logical_w, logical_h = get_logical_dimensions() mode = settings.get("mode", "zoom") pan_x = settings.get("pan_x", 0.5) pan_y = settings.get("pan_y", 0.5) if mode == "zoom": img = _zoom_and_pan(img, logical_w, logical_h, pan_x, pan_y) else: img = _letterbox(img, logical_w, logical_h) orientation = frame_settings.get("orientation", "landscape") if orientation == "portrait_cw": img = img.rotate(-90, expand=True) elif orientation == "portrait_ccw": img = img.rotate(90, expand=True) return img def _zoom_and_pan(img: Image.Image, tw: int, th: int, pan_x: float, pan_y: float) -> Image.Image: target_ratio = tw / th img_ratio = img.width / img.height if img_ratio > target_ratio: new_height = th new_width = int(img.width * (th / img.height)) else: new_width = tw new_height = int(img.height * (tw / img.width)) img = img.resize((new_width, new_height), Image.LANCZOS) max_left = new_width - tw max_top = new_height - th left = int(max_left * pan_x) top = int(max_top * pan_y) return img.crop((left, top, left + tw, top + th)) def _letterbox(img: Image.Image, tw: int, th: int) -> Image.Image: lb_color = tuple(frame_settings.get("letterbox_color", [255, 255, 255])) target_ratio = tw / th img_ratio = img.width / img.height if img_ratio > target_ratio: new_width = tw new_height = int(img.height * (tw / img.width)) else: new_height = th new_width = int(img.width * (th / img.height)) img = img.resize((new_width, new_height), Image.LANCZOS) canvas = Image.new("RGB", (tw, th), lb_color) paste_x = (tw - new_width) // 2 paste_y = (th - new_height) // 2 canvas.paste(img, (paste_x, paste_y)) return canvas # --- ESP32 endpoints --- @app.route("/photo") def random_photo(): ref = provider_manager.pick_random_photo() if not ref: return jsonify({"error": "No photos found"}), 404 try: img = provider_manager.get_photo_image(ref) settings = get_image_settings(ref.composite_key) img = process_image(img, settings) buf = io.BytesIO() img.save(buf, format="JPEG", quality=85) buf.seek(0) return Response(buf.getvalue(), mimetype="image/jpeg", headers={"X-Photo-Name": ref.composite_key}) except Exception as e: print(f"Error processing {ref.composite_key}: {e}", file=sys.stderr) return jsonify({"error": str(e)}), 500 @app.route("/heartbeat", methods=["POST"]) def heartbeat(): global frame_state data = request.get_json(silent=True) or {} frame_state["last_update"] = datetime.now(timezone.utc).isoformat() frame_state["current_photo"] = data.get("photo", None) frame_state["ip"] = request.remote_addr frame_state["free_heap"] = data.get("free_heap", None) frame_state["updates"] = frame_state.get("updates", 0) + 1 _save_json(STATE_FILE, frame_state) return jsonify({"ok": True}) # --- Frame settings API --- @app.route("/api/settings", methods=["GET"]) def api_get_settings(): return jsonify(frame_settings) @app.route("/api/settings", methods=["PUT"]) def api_put_settings(): global frame_settings data = request.get_json() allowed_keys = {"orientation", "default_mode", "letterbox_color"} for k, v in data.items(): if k in allowed_keys: frame_settings[k] = v _save_json(SETTINGS_FILE, frame_settings) return jsonify(frame_settings) # --- Per-image settings API (uses composite keys) --- @app.route("/api/photos//settings", methods=["GET"]) def api_get_image_settings(photo_key: str): return jsonify(get_image_settings(photo_key)) @app.route("/api/photos//settings", methods=["PUT"]) def api_put_image_settings(photo_key: str): data = request.get_json() allowed_keys = {"mode", "pan_x", "pan_y"} current = image_settings.get(photo_key, {}) for k, v in data.items(): if k in allowed_keys: current[k] = v image_settings[photo_key] = current _save_json(IMAGE_SETTINGS_FILE, image_settings) return jsonify(get_image_settings(photo_key)) @app.route("/api/photos//settings", methods=["DELETE"]) def api_delete_image_settings(photo_key: str): image_settings.pop(photo_key, None) _save_json(IMAGE_SETTINGS_FILE, image_settings) return jsonify(get_image_settings(photo_key)) # --- Provider management API --- @app.route("/api/providers/types") def api_provider_types(): return jsonify(available_provider_types()) @app.route("/api/providers", methods=["GET"]) def api_list_providers(): result = [] for iid, cfg in provider_manager.all_configs().items(): entry = {"id": iid, **cfg} provider = provider_manager.get_instance(iid) if provider: entry["health"] = provider.health_check().__dict__ else: entry["health"] = {"status": "disabled"} result.append(entry) return jsonify(result) @app.route("/api/providers", methods=["POST"]) def api_add_provider(): data = request.get_json() iid = data.get("id", "").strip() ptype = data.get("type", "") config = data.get("config", {}) if not iid or not ptype: return jsonify({"error": "id and type are required"}), 400 try: provider_manager.add_provider(iid, ptype, config) return jsonify({"id": iid, "status": "created"}) except ValueError as e: return jsonify({"error": str(e)}), 400 @app.route("/api/providers/", methods=["GET"]) def api_get_provider(provider_id: str): configs = provider_manager.all_configs() if provider_id not in configs: return jsonify({"error": "not found"}), 404 entry = {"id": provider_id, **configs[provider_id]} provider = provider_manager.get_instance(provider_id) if provider: entry["health"] = provider.health_check().__dict__ return jsonify(entry) @app.route("/api/providers/", methods=["PUT"]) def api_update_provider(provider_id: str): data = request.get_json() try: if "config" in data: provider_manager.update_provider_config(provider_id, data["config"]) if "enabled" in data: provider_manager.set_enabled(provider_id, data["enabled"]) return jsonify({"id": provider_id, "status": "updated"}) except ValueError as e: return jsonify({"error": str(e)}), 400 @app.route("/api/providers/", methods=["DELETE"]) def api_delete_provider(provider_id: str): provider_manager.remove_provider(provider_id) return jsonify({"id": provider_id, "status": "removed"}) @app.route("/api/providers//cache/clear", methods=["POST"]) def api_clear_provider_cache(provider_id: str): provider_manager.cache.clear(provider_id) return jsonify({"status": "cleared"}) # Auth flow endpoints @app.route("/api/providers//auth/start", methods=["POST"]) def api_auth_start(provider_id: str): provider = provider_manager.get_instance(provider_id) if not provider: return jsonify({"error": "not found"}), 404 callback_url = request.json.get("callback_url", "") if request.json else "" return jsonify(provider.auth_start(callback_url)) @app.route("/api/providers//auth/callback") def api_auth_callback(provider_id: str): provider = provider_manager.get_instance(provider_id) if not provider: return jsonify({"error": "not found"}), 404 result = provider.auth_callback(dict(request.args)) if "config" in result: provider_manager.save_provider_auth_state(provider_id, result.pop("config")) return jsonify(result) @app.route("/api/providers//auth/code", methods=["POST"]) def api_auth_code(provider_id: str): provider = provider_manager.get_instance(provider_id) if not provider: return jsonify({"error": "not found"}), 404 code = request.json.get("code", "") if request.json else "" return jsonify(provider.auth_submit_code(code)) @app.route("/api/providers//auth/status") def api_auth_status(provider_id: str): provider = provider_manager.get_instance(provider_id) if not provider: return jsonify({"error": "not found"}), 404 return jsonify(provider.get_auth_state()) # --- Photo API endpoints --- @app.route("/api/status") def api_status(): all_photos = provider_manager.get_all_photos() return jsonify({ "state": "online" if frame_state.get("last_update") else "waiting", "last_update": frame_state.get("last_update"), "current_photo": frame_state.get("current_photo"), "frame_ip": frame_state.get("ip"), "total_photos": len(all_photos), "total_updates": frame_state.get("updates", 0), "orientation": frame_settings.get("orientation", "landscape"), }) @app.route("/api/photos") def api_photos(): all_photos = provider_manager.get_all_photos() result = [] for ref in all_photos: entry = { "key": ref.composite_key, "name": ref.display_name, "provider": ref.provider_id, } provider = provider_manager.get_instance(ref.provider_id) if provider: size = provider.get_photo_size(ref.photo_id) if size is not None: entry["size"] = size if ref.composite_key in image_settings: entry["settings"] = image_settings[ref.composite_key] result.append(entry) return jsonify(result) @app.route("/api/upload", methods=["POST"]) def api_upload(): target_id = request.args.get("provider", None) # Find a provider that supports upload provider = None if target_id: provider = provider_manager.get_instance(target_id) else: for p in provider_manager.all_instances().values(): if p.supports_upload(): provider = p break if not provider or not provider.supports_upload(): return jsonify({"error": "No upload-capable provider found"}), 400 uploaded = [] for f in request.files.getlist("photos"): if not f.filename: continue fname = secure_filename(f.filename) try: ref = provider.upload_photo(fname, f.read()) uploaded.append(ref.composite_key) except Exception as e: print(f"Upload error for {fname}: {e}", file=sys.stderr) # Invalidate the provider's photo list cache provider_manager.cache.invalidate_list(provider.instance_id) return jsonify({"uploaded": uploaded, "count": len(uploaded)}) @app.route("/api/photos/", methods=["DELETE"]) def api_delete(photo_key: str): provider_id, photo_id = _parse_photo_key(photo_key) provider = provider_manager.get_instance(provider_id) if not provider: return jsonify({"error": "provider not found"}), 404 if not provider.supports_delete(): return jsonify({"error": "provider does not support deletion"}), 400 try: provider.delete_photo(photo_id) except FileNotFoundError: return jsonify({"error": "not found"}), 404 image_settings.pop(photo_key, None) _save_json(IMAGE_SETTINGS_FILE, image_settings) provider_manager.cache.invalidate_list(provider_id) return jsonify({"deleted": photo_key}) @app.route("/thumb/") def thumbnail(photo_key: str): ref = provider_manager.find_photo_ref(photo_key) if not ref: return "", 404 try: size = request.args.get("size", 300, type=int) size = min(size, 1600) data = provider_manager.get_photo_thumbnail(ref, size) return Response(data, mimetype="image/jpeg") except Exception: return "", 500 # --- Simulator / Preview --- @app.route("/preview/") def preview_photo(photo_key: str): ref = provider_manager.find_photo_ref(photo_key) if not ref: return jsonify({"error": "not found"}), 404 settings = get_image_settings(ref.composite_key) if "mode" in request.args: settings["mode"] = request.args["mode"] if "pan_x" in request.args: settings["pan_x"] = float(request.args["pan_x"]) if "pan_y" in request.args: settings["pan_y"] = float(request.args["pan_y"]) try: img = provider_manager.get_photo_image(ref) img = process_image(img, settings) _, preview = dither_floyd_steinberg(img) buf = io.BytesIO() preview.save(buf, format="PNG") buf.seek(0) return Response(buf.getvalue(), mimetype="image/png") except Exception as e: print(f"Preview error: {e}", file=sys.stderr) return jsonify({"error": str(e)}), 500 @app.route("/preview") def preview_random(): ref = provider_manager.pick_random_photo() if not ref: return jsonify({"error": "No photos found"}), 404 return preview_photo(ref.composite_key) @app.route("/api/photos//geometry") def photo_geometry(photo_key: str): ref = provider_manager.find_photo_ref(photo_key) if not ref: return jsonify({"error": "not found"}), 404 img = provider_manager.get_photo_image(ref) lw, lh = get_logical_dimensions() return jsonify({ "source_width": img.width, "source_height": img.height, "logical_width": lw, "logical_height": lh, "logical_aspect": lw / lh, }) @app.route("/simulate/") def simulate_page(photo_key: str): ref = provider_manager.find_photo_ref(photo_key) if not ref: return "", 404 settings = get_image_settings(ref.composite_key) orientation = frame_settings.get("orientation", "landscape") lw, lh = get_logical_dimensions() return render_template_string(SIMULATE_UI, photo_key=ref.composite_key, photo_name=ref.display_name, settings=settings, orientation=orientation, logical_w=lw, logical_h=lh) SIMULATE_UI = """ {{ photo_name }} — Frame Preview
← Gallery {{ photo_name }}
E-Paper Preview
""" # --- Web UI --- @app.route("/") def index(): all_photos = provider_manager.get_all_photos() orientation = frame_settings.get("orientation", "landscape") default_mode = frame_settings.get("default_mode", "zoom") return render_template_string(WEB_UI, photos=all_photos, state=frame_state, photo_count=len(all_photos), orientation=orientation, default_mode=default_mode) @app.route("/health") def health(): all_photos = provider_manager.get_all_photos() return jsonify({"status": "ok", "photo_count": len(all_photos)}) WEB_UI = """ Photo Frame

Photo Frame

Frame Status {{ 'Online' if state.get('last_update') else 'Waiting for first update' }}
{% if state.get('last_update') %}
Last Update {{ state.last_update[:19] }}
Showing {{ state.get('current_photo', 'Unknown') }}
{% endif %}
Photos {{ photo_count }}
Orientation
Default Mode

Drop photos here or click to upload

{% if photos %} {% else %}
No photos yet. Upload some!
{% endif %}
""" if __name__ == "__main__": parser = argparse.ArgumentParser(description="E-Ink Photo Frame Server") parser.add_argument("--port", type=int, default=8473) parser.add_argument("--photos-dir", type=str, default="./photos") args = parser.parse_args() photos_dir = Path(args.photos_dir) DATA_DIR = photos_dir / ".data" STATE_FILE = DATA_DIR / "frame_state.json" SETTINGS_FILE = DATA_DIR / "settings.json" IMAGE_SETTINGS_FILE = DATA_DIR / "image_settings.json" if not photos_dir.exists(): photos_dir.mkdir(parents=True, exist_ok=True) # Reload state with correct paths frame_state = _load_json(STATE_FILE, lambda: { "last_update": None, "current_photo": None, "ip": None, "updates": 0 }) frame_settings = _load_json(SETTINGS_FILE, lambda: dict(DEFAULT_SETTINGS)) image_settings = _load_json(IMAGE_SETTINGS_FILE, dict) # Initialize provider manager provider_manager = ProviderManager(DATA_DIR, default_photos_dir=str(photos_dir.resolve())) provider_manager.load() # Migrate legacy image settings to composite keys migrate_image_settings(DATA_DIR) # Reload image settings after migration image_settings = _load_json(IMAGE_SETTINGS_FILE, dict) all_photos = provider_manager.get_all_photos() print(f"Photos: {len(all_photos)} across {len(provider_manager.all_instances())} provider(s)") print(f"Orientation: {frame_settings['orientation']}, Default mode: {frame_settings['default_mode']}") print(f"Listening on port {args.port}") app.run(host="0.0.0.0", port=args.port)