"""Flask app: status/preview page + one settings form per screen + rotation settings.""" from io import BytesIO import requests from flask import Flask, redirect, render_template, request, url_for from PIL import Image from werkzeug.utils import secure_filename from screens.picture import UPLOAD_DIR ALLOWED_IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "bmp", "webp"} PIL_FORMAT_TO_EXT = {"JPEG": "jpg", "PNG": "png", "GIF": "gif", "BMP": "bmp", "WEBP": "webp"} PICTURE_URL_TIMEOUT = 15 MAX_SCHEDULE_RULES = 4 PICTURE_URL_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", } def create_app(config, screens: dict, scheduler) -> Flask: app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 20 * 1024 * 1024 # 20 MB, generous for a photo @app.route("/") def index(): status = scheduler.get_status() return render_template( "index.html", status=status, screens=screens, rotation=config.get_rotation(), navidrome_priority=config.get_navidrome_priority(), ) @app.route("/actions/refresh", methods=["POST"]) def refresh_now(): scheduler.request_manual_refresh() return redirect(url_for("index")) @app.route("/actions/switch/", methods=["POST"]) def switch_now(screen_id): if screen_id in screens: scheduler.request_switch(screen_id) return redirect(url_for("index")) @app.route("/actions/resume", methods=["POST"]) def resume_rotation(): scheduler.request_resume_rotation() return redirect(url_for("index")) @app.route("/settings/navidrome", methods=["GET", "POST"]) def settings_navidrome(): if request.method == "POST": config.update_screen_settings("navidrome", { "url": request.form.get("url", "").strip(), "username": request.form.get("username", "").strip(), "password": request.form.get("password", ""), "filter_user": request.form.get("filter_user", "").strip(), "show_progress": "show_progress" in request.form, "show_genre": "show_genre" in request.form, "show_source": "show_source" in request.form, "boost_cover": "boost_cover" in request.form, "grayscale_cover": "grayscale_cover" in request.form, }) scheduler.invalidate_screen("navidrome") return redirect(url_for("settings_navidrome", saved=1)) return render_template("settings_navidrome.html", settings=config.get_screen_settings("navidrome")) @app.route("/settings/weather", methods=["GET", "POST"]) def settings_weather(): error = None if request.method == "POST": city = request.form.get("city", "").strip() lat_raw = request.form.get("latitude", "").strip() lon_raw = request.form.get("longitude", "").strip() forecast_days = int(request.form.get("forecast_days", 5) or 5) if forecast_days not in (3, 5, 7): forecast_days = 5 update_minutes = max(1, int(request.form.get("update_minutes", 30) or 30)) display_toggles = { "show_city": "show_city" in request.form, "show_date": "show_date" in request.form, "show_hourly_graph": "show_hourly_graph" in request.form, "show_feels_like": "show_feels_like" in request.form, "show_high_low": "show_high_low" in request.form, "show_rain_chance": "show_rain_chance" in request.form, } if city: try: result = screens["weather"].geocode(city) except requests.exceptions.RequestException: result = None if result is None: error = f'Could not find "{city}" -- try a different spelling, or fill in coordinates manually below instead.' else: config.update_screen_settings("weather", { "city": result["label"], "latitude": result["latitude"], "longitude": result["longitude"], "forecast_days": forecast_days, "update_minutes": update_minutes, **display_toggles, }) scheduler.invalidate_screen("weather") return redirect(url_for("settings_weather", saved=1)) else: # The city text field isn't pre-filled (unlike the manual # lat/lon fields below it), so it's blank on every # settings-only save -- toggles, forecast_days, etc. Only # clear the resolved city label when the coordinates # actually changed; otherwise this would silently wipe the # city on every save even though lat/lon round-trip fine. current = config.get_screen_settings("weather") new_lat = float(lat_raw) if lat_raw else None new_lon = float(lon_raw) if lon_raw else None coords_unchanged = new_lat == current.get("latitude") and new_lon == current.get("longitude") config.update_screen_settings("weather", { "city": current.get("city", "") if coords_unchanged else "", "latitude": new_lat, "longitude": new_lon, "forecast_days": forecast_days, "update_minutes": update_minutes, **display_toggles, }) scheduler.invalidate_screen("weather") return redirect(url_for("settings_weather", saved=1)) return render_template("settings_weather.html", settings=config.get_screen_settings("weather"), error=error) @app.route("/settings/journey", methods=["GET", "POST"]) def settings_journey(): error = None if request.method == "POST": origin_query = request.form.get("origin", "").strip() destination_query = request.form.get("destination", "").strip() try: results = max(1, min(5, int(request.form.get("results", 3) or 3))) except (TypeError, ValueError): results = 3 update_minutes = max(1, int(request.form.get("update_minutes", 2) or 2)) connection_type = request.form.get("connection_type", "all") if connection_type not in ("all", "regional", "long_distance"): connection_type = "all" updates = {"results": results, "update_minutes": update_minutes, "connection_type": connection_type} journey_screen = screens["journey"] if origin_query: try: found = journey_screen.geocode(origin_query) except requests.exceptions.RequestException: found = None if found is None: error = f'Could not find "{origin_query}" as a starting point -- try a different spelling.' else: updates.update({ "origin_query": origin_query, "origin_label": found["label"], "origin_lat": found["latitude"], "origin_lon": found["longitude"], }) if not error and destination_query: try: found = journey_screen.geocode(destination_query) except requests.exceptions.RequestException: found = None if found is None: error = f'Could not find "{destination_query}" as a destination -- try a different spelling.' else: updates.update({ "destination_query": destination_query, "destination_label": found["label"], "destination_lat": found["latitude"], "destination_lon": found["longitude"], }) if not error: config.update_screen_settings("journey", updates) scheduler.invalidate_screen("journey") found_parts = [v for k, v in (("origin", origin_query), ("destination", destination_query)) if v] if found_parts: current = config.get_screen_settings("journey") return redirect(url_for( "settings_journey", found=f'{current["origin_label"]} → {current["destination_label"]}', )) return redirect(url_for("settings_journey", saved=1)) return render_template("settings_journey.html", settings=config.get_screen_settings("journey"), error=error) @app.route("/settings/picture", methods=["GET", "POST"]) def settings_picture(): error = None if request.method == "POST": url = request.form.get("image_url", "").strip() common_updates = { "fit": request.form.get("fit", "cover"), "boost_contrast": "boost_contrast" in request.form, "grayscale": "grayscale" in request.form, } uploaded = request.files.get("image") if uploaded and uploaded.filename: filename = secure_filename(uploaded.filename) ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext not in ALLOWED_IMAGE_EXTENSIONS: error = "Unsupported file type -- use PNG, JPG, GIF, BMP, or WebP." else: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) # Only one picture is ever active -- clear any previous # upload first so a different extension doesn't leave an # orphaned file behind. for old in UPLOAD_DIR.glob("current.*"): old.unlink() dest = UPLOAD_DIR / f"current.{ext}" uploaded.save(dest) try: Image.open(dest).verify() except Exception: dest.unlink(missing_ok=True) error = "That file doesn't look like a valid image." else: config.update_screen_settings("picture", { "mode": "upload", "uploaded_filename": dest.name, **common_updates, }) scheduler.invalidate_screen("picture") return redirect(url_for("settings_picture", saved=1)) elif url: # Downloaded once, here, rather than re-fetched by the # screen's own check() on every poll -- otherwise a remote # image that comes back even slightly byte-different between # polls (common with dynamically-generated images) would # force a real e-ink flash every poll_interval for no # visible reason. try: # Some image hosts (Wikimedia confirmed live -- 403s with # "set a user-agent") reject the default requests/python # User-Agent as bot traffic. response = requests.get(url, timeout=PICTURE_URL_TIMEOUT, headers=PICTURE_URL_HEADERS) response.raise_for_status() content = response.content fmt = Image.open(BytesIO(content)).format except (requests.exceptions.RequestException, OSError): fmt = None ext = PIL_FORMAT_TO_EXT.get(fmt or "") if not ext: error = "Could not download a valid image from that URL -- try a different link, or upload a file instead." else: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) # Only one picture is ever active -- clear any previous # upload/download first so a different extension doesn't # leave an orphaned file behind. for old in UPLOAD_DIR.glob("current.*"): old.unlink() dest = UPLOAD_DIR / f"current.{ext}" dest.write_bytes(content) try: Image.open(dest).verify() except Exception: dest.unlink(missing_ok=True) error = "That URL doesn't look like a valid image." else: config.update_screen_settings("picture", { "mode": "url", "image_url": url, "uploaded_filename": dest.name, **common_updates, }) scheduler.invalidate_screen("picture") return redirect(url_for("settings_picture", saved=1)) else: # Neither a new upload nor a URL change -- just update fit/boost/grayscale. config.update_screen_settings("picture", common_updates) scheduler.invalidate_screen("picture") return redirect(url_for("settings_picture", saved=1)) return render_template("settings_picture.html", settings=config.get_screen_settings("picture"), error=error) @app.route("/settings/rotation", methods=["GET", "POST"]) def settings_rotation(): if request.method == "POST": raw = request.form.get("screens", "") selected = [s.strip() for s in raw.split(",") if s.strip() in screens] interval_minutes = int(request.form.get("interval_minutes", 5) or 5) enabled = "rotation_enabled" in request.form config.set_rotation(selected, interval_minutes, enabled) config.set_navidrome_priority("navidrome_priority" in request.form) schedule_rules = [] for i in range(1, MAX_SCHEDULE_RULES + 1): rule_screen = request.form.get(f"schedule_screen_{i}", "").strip() start = request.form.get(f"schedule_start_{i}", "").strip() end = request.form.get(f"schedule_end_{i}", "").strip() if rule_screen in screens and start and end: schedule_rules.append({"screen_id": rule_screen, "start": start, "end": end}) config.set_schedules(schedule_rules) return redirect(url_for("settings_rotation", saved=1)) return render_template( "settings_rotation.html", screens=screens, rotation=config.get_rotation(), navidrome_priority=config.get_navidrome_priority(), schedules=config.get_schedules(), max_schedule_rules=MAX_SCHEDULE_RULES, ) return app