Honey
684d706afc
Not needed any more -- the Journey screen already covers the same regional-transit need via EFA backends, and Departures depended on the free db-vendo-client API that DB has been actively blocking third parties from anyway. Removed screens/departures.py, its settings template, its route, its registration in screens/__init__.py, and its nav link. Reworded the docstring/hint references in journey.py and settings_journey.html that pointed at the now-gone screen. Existing config.json entries referencing "departures" (rotation list, pinned screen) are harmless leftovers -- both scheduler lookups already filter to screens that still exist, so nothing needs a migration step. Verified: app boots clean, /settings/departures 404s, every other settings page still loads fine, no leftover references outside of journey.py's own unrelated "departure time" wording.
301 lines
14 KiB
Python
301 lines
14 KiB
Python
"""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
|
|
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/<screen_id>", 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)
|
|
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(),
|
|
)
|
|
|
|
return app
|