from __future__ import annotations import os import secrets from copy import deepcopy from datetime import datetime, timedelta from functools import wraps from io import BytesIO from uuid import uuid4 import requests from flask import Flask, Response, jsonify, redirect, render_template, request, session, url_for from config import CONFIG from amundsen.core.analysis_service import AnalysisService from amundsen.core.analysis_cache import AnalysisCache from amundsen.core.gpx_exporter import GPXExporter from amundsen.core.gpx_simplifier import GPXSimplifier from amundsen.core.gpx_trimmer import GPXTrimmer from amundsen.core.google_maps_importer import GoogleMapsImportError, GoogleMapsWaypointImporter from amundsen.core.models import GPXDocument, Waypoint from amundsen.core.project_store import MemoryProjectStore, ProjectStore from amundsen.core.raspberry_amundsen_client import RaspberryAmundsenClient, RaspberryAmundsenError from amundsen.core.report_normalizer import ReportNormalizer from amundsen.core.smart_gpx_combiner import SmartCombineSource, SmartGPXCombiner from amundsen.core.upload_store import ServerUploadStore app = Flask( __name__, static_url_path="/wandelen/static", static_folder="static", ) DEFAULT_RPI_DATA_ROOT = os.path.join(app.root_path, "data", "raspberryai", "amundsen") DEFAULT_RPI_JOB_TIMEOUT_SECONDS = 1800.0 DEFAULT_RPI_JOB_POLL_SECONDS = 1.0 app.secret_key = os.getenv("AMUNDSEN_SECRET_KEY", secrets.token_hex(32)) app.permanent_session_lifetime = timedelta(days=180) service = AnalysisService() report_normalizer = ReportNormalizer(CONFIG) exporter = GPXExporter(service.distance) smart_combiner = SmartGPXCombiner(service.distance) gpx_trimmer = GPXTrimmer(service.distance) gpx_simplifier = GPXSimplifier(service.distance) google_maps_importer = GoogleMapsWaypointImporter(lambda: os.getenv("AMUNDSEN_GOOGLE_MAPS_API_KEY")) raspberry_amundsen = RaspberryAmundsenClient( os.getenv("AMUNDSEN_RPI_GATEWAY_URL", "http://192.168.68.230:8000"), os.getenv("AMUNDSEN_RPI_SERVICE_PREFIX", "/services/amundsen"), os.getenv("AMUNDSEN_RPI_DATA_ROOT", os.getenv("AMUNDSEN_RPI_NAS_ROOT", DEFAULT_RPI_DATA_ROOT)), os.getenv( "AMUNDSEN_RPI_SHARED_ROOT", os.getenv("AMUNDSEN_RPI_SERVICE_ROOT", os.getenv("AMUNDSEN_RPI_DATA_ROOT", DEFAULT_RPI_DATA_ROOT)), ), ) project_store = MemoryProjectStore() if os.getenv("AMUNDSEN_TEST_STORE") == "memory" else ProjectStore() upload_store = ServerUploadStore(os.getenv("AMUNDSEN_UPLOAD_DIR")) analysis_cache = AnalysisCache(os.getenv("AMUNDSEN_ANALYSIS_CACHE_DIR")) STORE: dict[str, GPXDocument] = {} COMBINE_DRAFTS: dict[str, dict] = {} TRIM_DRAFTS: dict[str, dict] = {} SIMPLIFY_DRAFTS: dict[str, dict] = {} TOOL_TREE = [ { "id": "gpx_combine", "name": "GPX combineren", "endpoint": "gpx_combine_tool", "description": "Plak losse GPX-bestanden slim aan elkaar.", }, { "id": "gpx_trim", "name": "GPX knippen", "endpoint": "gpx_trim_tool", "description": "Kort het begin of einde van een GPX-route in.", }, { "id": "gpx_simplify", "name": "GPX verkleinen", "endpoint": "gpx_simplify_tool", "description": "Maak grote GPX-bestanden kleiner met minder routepunten.", }, { "id": "raspberry_analysis", "name": "Raspberry analyse", "endpoint": "raspberry_analysis_tool", "description": "Start een losse analyse via de Raspberry-service.", }, ] PUSHOVER_API_TOKEN = os.getenv("PUSHOVER_API_TOKEN") PUSHOVER_USER_KEY = os.getenv("PUSHOVER_USER_KEY") def login_required(view): @wraps(view) def wrapped(*args, **kwargs): if not session.get("user_id"): return redirect(url_for("login_page")) return view(*args, **kwargs) return wrapped @app.context_processor def navigation_context() -> dict: return {"tool_tree": TOOL_TREE} @app.get("/") def root_redirect() -> str: return redirect(url_for("index")) @app.get("/wandelen") @login_required def index() -> str: return render_template( "index.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="home", ) @app.get("/wandelen/projecten") @login_required def projects_manager_page() -> str: return render_template( "projects_manage.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="projects", ) @app.get("/wandelen/instellingen/wandeltijd") @login_required def hiking_settings_page() -> str: return render_template( "hiking_settings.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="hiking", ) @app.get("/wandelen/tools") @login_required def gpx_tools_page() -> str: return render_template( "gpx_tools.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="home", ) @app.get("/wandelen/tools/gpx-combine") @login_required def gpx_combine_tool() -> str: return render_template( "gpx_combine.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_tool="gpx_combine", current_user=session.get("username"), current_menu="tools", ) @app.post("/wandelen/tools/gpx-combine") @login_required def preview_combined_gpx() -> str: files = [file for file in request.files.getlist("gpx_files") if file.filename] output_name = (request.form.get("name") or "Gecombineerde route").strip() or "Gecombineerde route" if len(files) < 2: return _render_gpx_combine(error="Kies minimaal twee GPX-bestanden om samen te voegen.", output_name=output_name) sources = [] try: for uploaded in files: raw_bytes = uploaded.read() xml_text = raw_bytes.decode("utf-8-sig") document = service.parser.parse_text(xml_text, uploaded.filename.rsplit(".", 1)[0]) if not document.all_points(): raise ValueError("GPX zonder trackpunten") sources.append(SmartCombineSource(document, uploaded.filename, raw_bytes, xml_text)) except Exception: return _render_gpx_combine(error="Een van de GPX-bestanden kon niet worden gelezen.", output_name=output_name) plan = smart_combiner.build_plan(sources, output_name) draft_token = secrets.token_urlsafe(24) COMBINE_DRAFTS[draft_token] = { "output_name": output_name, "sources": sources, "plan": plan, } return _render_gpx_combine( output_name=output_name, preview=plan, draft_token=draft_token, ) @app.post("/wandelen/tools/gpx-combine/execute") @login_required def execute_combined_gpx() -> Response | str: draft_token = request.form.get("draft_token", "") draft = COMBINE_DRAFTS.pop(draft_token, None) if draft is None: return _render_gpx_combine(error="De voorvertoning is verlopen. Upload de GPX-bestanden opnieuw.") selected_routes = _selected_combine_routes( request.form.getlist("source_index"), request.form.getlist("reversed"), len(draft["sources"]), ) plan = draft["plan"] if selected_routes: plan = smart_combiner.build_plan_from_selection(draft["sources"], draft["output_name"], selected_routes) combined = plan["document"] xml_text = exporter.to_xml(combined) combined_filename = _gpx_filename(draft["output_name"]) return Response( xml_text, mimetype="application/gpx+xml", headers={"Content-Disposition": f'attachment; filename="{combined_filename}"'}, ) @app.get("/wandelen/tools/gpx-trim") @login_required def gpx_trim_tool() -> str: return _render_gpx_trim() @app.post("/wandelen/tools/gpx-trim") @login_required def preview_trimmed_gpx() -> str: uploaded = request.files.get("gpx_file") output_name = (request.form.get("name") or "Geknipte route").strip() or "Geknipte route" if uploaded is None or not uploaded.filename: return _render_gpx_trim(error="Kies een GPX-bestand om te knippen.", output_name=output_name) try: raw_bytes = uploaded.read() xml_text = raw_bytes.decode("utf-8-sig") document = service.parser.parse_text(xml_text, uploaded.filename.rsplit(".", 1)[0]) if len(document.all_points()) < 2: raise ValueError("GPX zonder route") except Exception: return _render_gpx_trim(error="Het GPX-bestand kon niet worden gelezen.", output_name=output_name) draft_token = secrets.token_urlsafe(24) preview = gpx_trimmer.build_preview(document) TRIM_DRAFTS[draft_token] = { "output_name": output_name, "document": document, } return _render_gpx_trim(output_name=output_name, preview=preview, draft_token=draft_token) @app.post("/wandelen/tools/gpx-trim/execute") @login_required def execute_trimmed_gpx() -> Response | str: draft_token = request.form.get("draft_token", "") draft = TRIM_DRAFTS.pop(draft_token, None) if draft is None: return _render_gpx_trim(error="De voorvertoning is verlopen. Upload het GPX-bestand opnieuw.") start_ratio = _ratio_from_form("start_ratio", 0.0) end_ratio = _ratio_from_form("end_ratio", 1.0) trimmed = gpx_trimmer.trim(draft["document"], draft["output_name"], start_ratio, end_ratio) xml_text = exporter.to_xml(trimmed) return Response( xml_text, mimetype="application/gpx+xml", headers={"Content-Disposition": f'attachment; filename="{_gpx_filename(draft["output_name"])}"'}, ) @app.get("/wandelen/tools/gpx-simplify") @login_required def gpx_simplify_tool() -> str: return _render_gpx_simplify() @app.post("/wandelen/tools/gpx-simplify") @login_required def preview_simplified_gpx() -> str: uploaded = request.files.get("gpx_file") output_name = (request.form.get("name") or "Verkleinde route").strip() or "Verkleinde route" tolerance_m = _float_from_form("tolerance_m", 10.0, minimum=0.0, maximum=250.0) if uploaded is None or not uploaded.filename: return _render_gpx_simplify(error="Kies een GPX-bestand om te verkleinen.", output_name=output_name, tolerance_m=tolerance_m) try: raw_bytes = uploaded.read() xml_text = raw_bytes.decode("utf-8-sig") document = service.parser.parse_text(xml_text, uploaded.filename.rsplit(".", 1)[0]) if len(document.all_points()) < 2: raise ValueError("GPX zonder route") except Exception: return _render_gpx_simplify(error="Het GPX-bestand kon niet worden gelezen.", output_name=output_name, tolerance_m=tolerance_m) draft_token = secrets.token_urlsafe(24) preview = gpx_simplifier.build_preview(document, tolerance_m) SIMPLIFY_DRAFTS[draft_token] = { "output_name": output_name, "document": document, "tolerance_m": tolerance_m, } return _render_gpx_simplify(output_name=output_name, tolerance_m=tolerance_m, preview=preview, draft_token=draft_token) @app.post("/wandelen/tools/gpx-simplify/execute") @login_required def execute_simplified_gpx() -> Response | str: draft_token = request.form.get("draft_token", "") draft = SIMPLIFY_DRAFTS.pop(draft_token, None) if draft is None: return _render_gpx_simplify(error="De voorvertoning is verlopen. Upload het GPX-bestand opnieuw.") simplified = gpx_simplifier.simplify(draft["document"], draft["output_name"], draft["tolerance_m"]) xml_text = exporter.to_xml(simplified) return Response( xml_text, mimetype="application/gpx+xml", headers={"Content-Disposition": f'attachment; filename="{_gpx_filename(draft["output_name"])}"'}, ) @app.get("/wandelen/tools/raspberry-analysis") @login_required def raspberry_analysis_tool() -> str: return render_template( "raspberry_analysis.html", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_tool="raspberry_analysis", current_user=session.get("username"), current_menu="tools", gateway_url=raspberry_amundsen.gateway_url, service_prefix=raspberry_amundsen.service_prefix, data_root=raspberry_amundsen.service_root, local_data_root=str(raspberry_amundsen.local_root), ) @app.get("/wandelen/login") def login_page() -> str: return render_template("login.html") @app.post("/wandelen/request-login") def request_login() -> str: username = request.form.get("username", "").strip() if not username: return render_template("login.html", error="Vul een gebruikersnaam in.") login_key = f"{secrets.randbelow(1_000_000):06}" project_store.create_login_request(username, login_key, request.remote_addr) sent = send_pushover( "Amundsen login", f"Login-aanvraag voor {username}. Bevestigingscode: {login_key}", ) message = "Aanvraag verstuurd. Vraag de beheerder om je bevestigingscode." if not sent: message = "Aanvraag opgeslagen, maar Pushover is niet ingesteld of kon niet versturen." return render_template("login.html", message=message, requested_username=username) @app.post("/wandelen/login") def login() -> Response | str: username = request.form.get("username", "").strip() login_key = request.form.get("login_key", "").strip() remember = request.form.get("remember") == "on" user = project_store.verify_login_key(username, login_key) if user is None: return render_template("login.html", error="Gebruikersnaam of bevestigingscode klopt niet.", requested_username=username) session.clear() session.permanent = remember session["user_id"] = user["id"] session["username"] = user["username"] return redirect(url_for("index")) @app.post("/wandelen/logout") def logout() -> Response: session.clear() return redirect(url_for("login_page")) @app.post("/wandelen/projects") @login_required def create_project() -> Response: project_store.create_project(request.form.get("name", "")) return redirect(request.form.get("next") or url_for("index")) @app.get("/wandelen/projects/") @login_required def project_detail(project_id: str) -> str: project = _project_from_tree(project_id) if project is None: return render_template( "index.html", error="Project niet gevonden.", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="home", ) return render_template( "project.html", project=project, routes=_routes_for_project(project), project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_project_id=project_id, current_user=session.get("username"), current_menu="projects", ) @app.post("/wandelen/projects//rename") @login_required def rename_project(project_id: str) -> Response: project_store.rename_project(project_id, request.form.get("name", "")) return redirect(request.form.get("next") or url_for("index")) @app.post("/wandelen/projects//delete") @login_required def delete_project(project_id: str) -> Response: project_store.delete_project(project_id) return redirect(request.form.get("next") or url_for("index")) @app.post("/wandelen/routes//metadata") @login_required def update_route_metadata(route_id: str) -> Response: route = project_store.get_route(route_id) if route is None: return redirect(request.form.get("next") or url_for("index")) route_date = _normalize_route_date(request.form.get("route_date", "")) route_name = request.form.get("name", "") changed = project_store.update_route_metadata(route_id, route_name, route_date) cached_report = analysis_cache.read(route_id) if changed and cached_report is not None: cached_report["name"] = route_name.strip() analysis_cache.write(route_id, cached_report) return redirect(request.form.get("next") or url_for("project_detail", project_id=route["project_id"])) @app.post("/wandelen/routes//delete") @login_required def delete_route(route_id: str) -> Response: route = project_store.get_route(route_id) next_url = request.form.get("next") if route is not None: project_store.delete_route(route_id) analysis_cache.clear(route_id) next_url = next_url or url_for("project_detail", project_id=route["project_id"]) return redirect(next_url or url_for("index")) @app.post("/wandelen/settings/hiking") @login_required def save_hiking_settings() -> Response: project_store.save_hiking_settings(request.form) return redirect(request.form.get("next") or url_for("index")) @app.post("/wandelen/analyze") @login_required def analyze_upload() -> str: files = [file for file in request.files.getlist("gpx_files") if file.filename] if not files: return render_template( "index.html", error="Kies minimaal een GPX-bestand.", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="home", ) project = _project_from_upload() first_route_id: str | None = None for uploaded in files: raw_bytes = uploaded.read() xml_text = raw_bytes.decode("utf-8-sig") route_name = _filename_route_name(uploaded.filename) route = project_store.add_route(project["id"], route_name, uploaded.filename, xml_text, store_xml=False) saved_path = upload_store.save_gpx(project["id"], route["id"], uploaded.filename, raw_bytes) project_store.update_route_file_path(route["id"], str(saved_path)) try: report = _analyze_bytes_and_cache(route["id"], uploaded.filename, raw_bytes) except RaspberryAmundsenError as error: return _render_index_error(f"Amundsen job aanmaken mislukt: {error.message}") _rename_new_route_from_report(route["id"], report) if first_route_id is None: first_route_id = route["id"] return render_report(first_route_id or uuid4().hex) @app.post("/wandelen/api/import/start") @login_required def start_import_jobs(): files = [file for file in request.files.getlist("gpx_files") if file.filename] if not files: return jsonify({"error": "Kies minimaal een GPX-bestand."}), 400 project = _project_from_upload() jobs = [] for uploaded in files: raw_bytes = uploaded.read() xml_text = raw_bytes.decode("utf-8-sig") route_name = _filename_route_name(uploaded.filename) route = project_store.add_route(project["id"], route_name, uploaded.filename, xml_text, store_xml=False) saved_path = upload_store.save_gpx(project["id"], route["id"], uploaded.filename, raw_bytes) project_store.update_route_file_path(route["id"], str(saved_path)) def start_job() -> dict: gpx_path, original_filename = raspberry_amundsen.save_bytes(uploaded.filename, raw_bytes) job = raspberry_amundsen.create_job(gpx_path, original_filename, _raspberry_analysis_settings()) job_id = _job_id_from_payload(job) return { **job, "job_id": job_id, "route_id": route["id"], "filename": uploaded.filename, "status_url": url_for("raspberry_amundsen_job_status", job_id=job_id), "complete_url": url_for("complete_import_job", route_id=route["id"], job_id=job_id), "report_url": url_for("render_report", report_id=route["id"]), } try: jobs.append(start_job()) except RaspberryAmundsenError as error: return jsonify(error.to_dict()), error.status_code return jsonify({"jobs": jobs, "first_report_url": jobs[0]["report_url"]}) @app.post("/wandelen/api/import//jobs//result") @login_required def complete_import_job(route_id: str, job_id: str): if project_store.get_route(route_id) is None: return jsonify({"error": "Rapport niet gevonden."}), 404 def complete_job() -> dict: report = _complete_analysis_job(route_id, job_id) _rename_new_route_from_report(route_id, report) return {"message": "Route geimporteerd.", "report": report, "report_url": url_for("render_report", report_id=route_id)} return _raspberry_response(complete_job) @app.get("/wandelen/report/") @login_required def render_report(report_id: str) -> str: document = _document_for_report(report_id) if document is None: return render_template( "index.html", error="Rapport niet gevonden. Upload het GPX-bestand opnieuw.", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="home", ) route_meta = project_store.get_route(report_id) try: report = _report_for_route(report_id, document) except RaspberryAmundsenError as error: return _render_index_error(f"Analyse via Raspberry-service mislukt: {error.message}") if route_meta: report["name"] = route_meta["name"] return render_template( "report.html", report_id=report_id, report=report, project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_route_id=report_id, current_project=route_meta["project_name"] if route_meta else "", current_project_id=route_meta["project_id"] if route_meta else "", route_date_label=_format_date(route_meta.get("route_date")) if route_meta else "", route_date_value=_date_input_value(route_meta.get("route_date")) if route_meta else "", current_user=session.get("username"), current_menu="projects", ) @app.post("/wandelen/report//refresh") @login_required def refresh_report_analysis(report_id: str) -> Response | str: document = _document_for_report(report_id) if document is None: return render_template( "index.html", error="Rapport niet gevonden. Upload het GPX-bestand opnieuw.", project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), ) try: _analyze_document_and_cache(report_id, document) except RaspberryAmundsenError as error: return _render_index_error(f"Analyse vernieuwen mislukt: {error.message}") return redirect(url_for("render_report", report_id=report_id)) @app.post("/wandelen/api/report//refresh/start") @login_required def start_report_refresh_job(report_id: str): xml_text = project_store.read_route_xml(report_id) route = project_store.get_route(report_id) if xml_text is None or route is None: return jsonify({"error": "Rapport niet gevonden."}), 404 def start_job() -> dict: filename = str(route.get("filename") or f"{report_id}.gpx") gpx_path, original_filename = raspberry_amundsen.save_bytes(filename, xml_text.encode("utf-8")) job = raspberry_amundsen.create_job(gpx_path, original_filename, _raspberry_analysis_settings()) job_id = _job_id_from_payload(job) return { **job, "job_id": job_id, "route_id": report_id, "status_url": url_for("raspberry_amundsen_job_status", job_id=job_id), "complete_url": url_for("complete_report_refresh_job", report_id=report_id, job_id=job_id), "report_url": url_for("render_report", report_id=report_id), } return _raspberry_response(start_job) @app.get("/wandelen/api/report//cache") @login_required def report_cache_status(report_id: str): if project_store.get_route(report_id) is None: return jsonify({"error": "Rapport niet gevonden."}), 404 return jsonify({"cached": analysis_cache.read(report_id) is not None}) @app.post("/wandelen/api/report//refresh/jobs//result") @login_required def complete_report_refresh_job(report_id: str, job_id: str): if project_store.get_route(report_id) is None: return jsonify({"error": "Rapport niet gevonden."}), 404 def complete_job() -> dict: report = _complete_analysis_job(report_id, job_id) return {"message": "Analyse vernieuwd.", "report": report} return _raspberry_response(complete_job) @app.get("/wandelen/api/report/") @login_required def report_json(report_id: str): document = _document_for_report(report_id) if document is None: return jsonify({"error": "Rapport niet gevonden."}), 404 try: return jsonify(_report_for_route(report_id, document)) except RaspberryAmundsenError as error: return jsonify(error.to_dict()), error.status_code @app.post("/wandelen/api/plan/") @login_required def plan_days(report_id: str): document = _document_for_report(report_id) if document is None: return jsonify({"error": "Rapport niet gevonden."}), 404 payload = request.get_json(silent=True) or {} return jsonify(_day_plan_payload(document, payload)) @app.post("/wandelen/api/plan//save") @login_required def save_day_plan(report_id: str): document = _document_for_report(report_id) if document is None: return jsonify({"error": "Rapport niet gevonden."}), 404 payload = request.get_json(silent=True) or {} plan_payload = _day_plan_payload(document, payload) cached_report = analysis_cache.read(report_id) or _report_for_route(report_id, document) cached_report["route_views"] = plan_payload["route_views"] cached_report["route_directions"] = plan_payload["route_directions"] cached_report["day_plan"] = plan_payload["days"] cached_report["saved_day_planning"] = { "direction": plan_payload["direction"], "start_date": _clean_waypoint_text(payload.get("planning_start_date")), "manual_days_by_direction": plan_payload["manual_days_by_direction"], "day_notes_by_direction": plan_payload["day_notes_by_direction"], "days_by_direction": plan_payload["days_by_direction"], "legs_by_direction": plan_payload["legs_by_direction"], } analysis_cache.write(report_id, cached_report) return jsonify({"message": "Dagplanning opgeslagen.", "report": cached_report, **plan_payload}) def _day_plan_payload(document: GPXDocument, payload: dict) -> dict: min_minutes = float(payload.get("min_hours", 5)) * 60 max_minutes = float(payload.get("max_hours", 7)) * 60 plan_document = deepcopy(document) payload_waypoints = _waypoints_from_payload(document) if payload_waypoints is not None: plan_document.waypoints = payload_waypoints hiking_settings = project_store.get_hiking_settings() time_predictor = service._time_predictor(hiking_settings) direction = str(payload.get("direction") or "forward") manual_day_ids = _manual_day_ids_from_payload(payload) if manual_day_ids is None: route_directions, day_plans, legs_by_direction = service.route_views_for_day_plan( plan_document, min_minutes=min_minutes, max_minutes=max_minutes, hiking_settings=hiking_settings, ) else: route_directions, day_plans, legs_by_direction = service.manual_route_views_for_day_plan( plan_document, direction, manual_day_ids, min_minutes=min_minutes, max_minutes=max_minutes, hiking_settings=hiking_settings, ) days_by_direction = { direction_id: _planned_days_payload(direction_days, time_predictor, legs_by_direction.get(direction_id, [])) for direction_id, direction_days in day_plans.items() } legs_payload = { direction_id: _planning_legs_payload(direction_legs, time_predictor) for direction_id, direction_legs in legs_by_direction.items() } active_days = days_by_direction.get(direction) or days_by_direction.get("forward", []) manual_days_by_direction = _manual_days_by_direction_from_payload(payload, direction) if manual_day_ids is not None else {} day_notes_by_direction = _day_notes_by_direction_from_payload(payload, direction) return { "direction": direction if direction in days_by_direction else "forward", "days": active_days, "days_by_direction": days_by_direction, "legs_by_direction": legs_payload, "manual_days_by_direction": manual_days_by_direction, "day_notes_by_direction": day_notes_by_direction, "route_views": route_directions["forward"]["route_views"], "route_directions": route_directions, } @app.post("/wandelen/api/export/") @login_required def export_gpx(report_id: str) -> Response: document = _document_for_report(report_id) if document is None: return Response("Rapport niet gevonden.", status=404) waypoints = _waypoints_from_payload(document) xml_text = exporter.to_xml(document, waypoints=waypoints) return Response( xml_text, mimetype="application/gpx+xml", headers={"Content-Disposition": f'attachment; filename="{document.name}.gpx"'}, ) @app.post("/wandelen/api/waypoints/") @login_required def save_waypoints(report_id: str): document = _document_for_report(report_id) if document is None: return jsonify({"error": "Rapport niet gevonden."}), 404 waypoints = _waypoints_from_payload(document) or [] xml_text = exporter.to_xml(document, waypoints=waypoints) if not project_store.update_route_xml(report_id, xml_text): return jsonify({"error": "Route kon niet worden opgeslagen."}), 404 STORE.pop(report_id, None) updated_document = service.parser.parse_text(xml_text, document.name) try: report = _analyze_document_xml_and_cache(report_id, updated_document.name, xml_text, updated_document) except RaspberryAmundsenError as error: return jsonify(error.to_dict()), error.status_code return jsonify( { "message": "Waypoints opgeslagen in GPX.", "waypoints": [waypoint.__dict__ for waypoint in updated_document.waypoints], "report": report, } ) @app.post("/wandelen/api/google-waypoint") @login_required def import_google_waypoint(): payload = request.get_json(silent=True) or {} try: waypoint = google_maps_importer.import_waypoint(str(payload.get("query") or "")) except GoogleMapsImportError as error: return jsonify({"error": str(error)}), 400 return jsonify({"waypoint": waypoint}) @app.get("/wandelen/api/raspberry-amundsen/health") @login_required def raspberry_amundsen_health(): return _raspberry_response(lambda: raspberry_amundsen.health()) @app.get("/wandelen/api/raspberry-amundsen/info") @login_required def raspberry_amundsen_info(): return _raspberry_response(lambda: raspberry_amundsen.info()) @app.post("/wandelen/api/raspberry-amundsen/jobs") @login_required def raspberry_amundsen_create_job(): def create_job() -> dict: uploaded = request.files.get("gpx_file") data_path = (request.form.get("data_path") or request.form.get("nas_path") or "").strip() if uploaded is not None and uploaded.filename: gpx_path, original_filename = raspberry_amundsen.save_upload(uploaded) elif data_path: gpx_path = raspberry_amundsen.validate_service_path(data_path) original_filename = request.form.get("original_filename", "") or gpx_path.rsplit("/", 1)[-1] else: raise RaspberryAmundsenError("file_not_found", "Kies een GPX-bestand of vul een data-pad in.", 400) return raspberry_amundsen.create_job(gpx_path, original_filename, _raspberry_analysis_settings()) return _raspberry_response(create_job) @app.get("/wandelen/api/raspberry-amundsen/jobs") @login_required def raspberry_amundsen_jobs(): return _raspberry_response(lambda: raspberry_amundsen.list_jobs()) @app.get("/wandelen/api/raspberry-amundsen/jobs/open") @login_required def raspberry_amundsen_open_jobs(): return _raspberry_response(lambda: raspberry_amundsen.list_open_jobs()) @app.get("/wandelen/api/raspberry-amundsen/jobs/") @login_required def raspberry_amundsen_job_status(job_id: str): return _raspberry_response(lambda: raspberry_amundsen.job_status(job_id)) @app.get("/wandelen/api/raspberry-amundsen/jobs//result") @login_required def raspberry_amundsen_job_result(job_id: str): return _raspberry_response(lambda: raspberry_amundsen.job_result(job_id)) @app.post("/wandelen/api/raspberry-amundsen/jobs//cancel") @login_required def raspberry_amundsen_cancel_job(job_id: str): return _raspberry_response(lambda: raspberry_amundsen.cancel_job(job_id)) @app.get("/wandelen/api/split/") @login_required def split_gpx(report_id: str) -> Response: document = _document_for_report(report_id) if document is None: return Response("Rapport niet gevonden.", status=404) archive = exporter.split_zip(document) return Response( BytesIO(archive).getvalue(), mimetype="application/zip", headers={"Content-Disposition": f'attachment; filename="{document.name}-split.zip"'}, ) def _waypoints_from_payload(document: GPXDocument | None = None) -> list[Waypoint] | None: payload = request.get_json(silent=True) or {} if "waypoints" not in payload: return None return [ Waypoint( lat=float(item["lat"]), lon=float(item["lon"]), name=str(item.get("name") or "Waypoint"), elevation=_waypoint_elevation_from_item(item, document), contact_person=_clean_waypoint_text(item.get("contact_person")), plus_code=_clean_waypoint_text(item.get("plus_code") or item.get("pluscode") or item.get("google_plus_code")), email=_clean_waypoint_text(item.get("email")), website=_clean_waypoint_text(item.get("website")), phone=_clean_waypoint_text(item.get("phone")), ) for item in payload["waypoints"] ] def _planned_days_payload(days: list[list], time_predictor, all_legs: list | None = None) -> list[dict]: planned_days = [] leg_ids = {id(leg): index for index, leg in enumerate(all_legs or [])} for index, day in enumerate(days, start=1): if not day: continue names = [day[0].start_name, *[leg.end_name for leg in day]] planned_days.append( { "day": index, "from": day[0].start_name, "to": day[-1].end_name, "walking_time": time_predictor.humanize(sum(leg.walking_minutes for leg in day)), "distance_km": round(sum(leg.distance_m for leg in day) / 1000.0, 2), "ascent_m": round(sum(leg.ascent_m for leg in day)), "descent_m": round(sum(leg.descent_m for leg in day)), "waypoints": [{"name": name} for name in names], "legs": [ _planning_leg_payload(leg, leg_ids.get(id(leg), leg_index), time_predictor) for leg_index, leg in enumerate(day) ], } ) return planned_days def _planning_legs_payload(legs: list, time_predictor) -> list[dict]: return [_planning_leg_payload(leg, index, time_predictor) for index, leg in enumerate(legs)] def _planning_leg_payload(leg, leg_id: int, time_predictor) -> dict: return { "id": str(leg_id), "from": leg.start_name, "to": leg.end_name, "start_name": leg.start_name, "end_name": leg.end_name, "distance_km": round(leg.distance_m / 1000.0, 2), "ascent_m": round(leg.ascent_m), "descent_m": round(leg.descent_m), "walking_time": time_predictor.humanize(leg.walking_minutes), "walking_minutes": round(leg.walking_minutes, 1), } def _manual_day_ids_from_payload(payload: dict) -> list[list[int]] | None: raw_days = payload.get("manual_days") if not isinstance(raw_days, list): return None return _clean_manual_day_ids(raw_days) def _manual_days_by_direction_from_payload(payload: dict, fallback_direction: str) -> dict: raw_by_direction = payload.get("manual_days_by_direction") if isinstance(raw_by_direction, dict): return { str(direction_id): _clean_manual_day_ids(raw_days) for direction_id, raw_days in raw_by_direction.items() if isinstance(raw_days, list) } raw_days = payload.get("manual_days") if isinstance(raw_days, list): return {fallback_direction: _clean_manual_day_ids(raw_days)} return {} def _day_notes_by_direction_from_payload(payload: dict, fallback_direction: str) -> dict: raw_by_direction = payload.get("day_notes_by_direction") if isinstance(raw_by_direction, dict): return { str(direction_id): _clean_day_notes(raw_notes) for direction_id, raw_notes in raw_by_direction.items() if isinstance(raw_notes, list) } raw_notes = payload.get("day_notes") if isinstance(raw_notes, list): return {fallback_direction: _clean_day_notes(raw_notes)} return {} def _clean_day_notes(raw_notes: list) -> list[str]: notes = [] for raw_note in raw_notes: notes.append(str(raw_note or "").strip()) return notes def _clean_manual_day_ids(raw_days: list) -> list[list[int]]: days: list[list[int]] = [] for raw_day in raw_days: if not isinstance(raw_day, list): days.append([]) continue day_ids = [] for raw_id in raw_day: try: day_ids.append(int(raw_id)) except (TypeError, ValueError): continue days.append(day_ids) return days def _clean_waypoint_text(value) -> str | None: text = str(value or "").strip() return text or None def _waypoint_elevation_from_item(item: dict, document: GPXDocument | None) -> float | None: if item.get("elevation") not in (None, ""): return float(item["elevation"]) if document is None: return None return _nearest_route_elevation(float(item["lat"]), float(item["lon"]), document) def _nearest_route_elevation(lat: float, lon: float, document: GPXDocument) -> float | None: points = document.all_points() if not points: return None nearest = min(points, key=lambda point: (point.lat - lat) ** 2 + (point.lon - lon) ** 2) elevation = nearest.analysis_elevation if elevation is None: elevation = nearest.elevation return round(float(elevation), 1) if elevation is not None else None def _raspberry_response(action): try: return jsonify(action()) except RaspberryAmundsenError as error: return jsonify(error.to_dict()), error.status_code def _raspberry_analysis_settings() -> dict: hiking_settings = project_store.get_hiking_settings() return { "smoothing_window_points": CONFIG.smoothing_window_points, "elevation_hysteresis_m": CONFIG.elevation_hysteresis_m, "segment_length_m": CONFIG.segment_length_m, "hiking_model": { "minutes_per_km": float(hiking_settings.get("minutes_per_km", CONFIG.hiking_model.minutes_per_km)), "minutes_per_100m_ascent": float( hiking_settings.get("minutes_per_100m_ascent", CONFIG.hiking_model.minutes_per_100m_ascent) ), "minutes_per_100m_descent": float( hiking_settings.get("minutes_per_100m_descent", CONFIG.hiking_model.minutes_per_100m_descent) ), }, } def _raspberry_job_timeout_seconds() -> float: return _float_from_env("AMUNDSEN_RPI_JOB_TIMEOUT_SECONDS", DEFAULT_RPI_JOB_TIMEOUT_SECONDS, minimum=5.0) def _raspberry_job_poll_seconds() -> float: return _float_from_env("AMUNDSEN_RPI_JOB_POLL_SECONDS", DEFAULT_RPI_JOB_POLL_SECONDS, minimum=0.25) def _document_for_report(report_id: str) -> GPXDocument | None: if report_id in STORE: return STORE[report_id] xml_text = project_store.read_route_xml(report_id) if xml_text is None: return None route = project_store.get_route(report_id) return service.parser.parse_text(xml_text, route["name"] if route else "Wandeling") def _project_from_upload() -> dict: project_name = request.form.get("project_name", "").strip() if project_name: return project_store.create_project(project_name) project_id = request.form.get("project_id", "").strip() for project in project_store.list_projects(): if project["id"] == project_id: return project return project_store.create_project("Mijn wandelingen") def _project_from_tree(project_id: str) -> dict | None: for project in project_store.list_projects(): if project["id"] == project_id: return project return None def _routes_for_project(project: dict) -> list[dict]: routes = [] for route in project.get("routes", []): route_item = dict(route) route_item["created_at_label"] = _format_date(route_item.get("created_at")) route_item["route_date_label"] = _format_date(route_item.get("route_date")) route_item["route_date_value"] = _date_input_value(route_item.get("route_date")) cached_report = analysis_cache.read(route_item["id"]) if cached_report: cached_report = _normalize_report(cached_report) route_item["summary"] = cached_report.get("summary") if cached_report else None routes.append(route_item) return routes def _report_for_route(route_id: str, document: GPXDocument) -> dict: cached_report = analysis_cache.read(route_id) if cached_report is not None: return _with_document_waypoints(_normalize_report(cached_report), document) return _with_document_waypoints(_analyze_document_and_cache(route_id, document), document) def _analyze_document_and_cache(route_id: str, document: GPXDocument) -> dict: xml_text = project_store.read_route_xml(route_id) or exporter.to_xml(document) return _analyze_document_xml_and_cache(route_id, _route_filename(route_id, document.name), xml_text, document) def _analyze_document_xml_and_cache(route_id: str, filename: str, xml_text: str, document: GPXDocument | None = None) -> dict: report = _analyze_bytes_and_cache(route_id, filename, xml_text.encode("utf-8")) if document is not None: report = _with_document_waypoints(report, document) analysis_cache.write(route_id, report) return report def _analyze_bytes_and_cache(route_id: str, filename: str, raw_bytes: bytes) -> dict: report = _normalize_report( raspberry_amundsen.analyze_gpx_bytes( filename, raw_bytes, _raspberry_analysis_settings(), wait_timeout_seconds=_raspberry_job_timeout_seconds(), poll_seconds=_raspberry_job_poll_seconds(), ) ) analysis_cache.write(route_id, report) return report def _complete_analysis_job(route_id: str, job_id: str) -> dict: status_payload = raspberry_amundsen.job_status(job_id) status = str(status_payload.get("status") or "").lower() if status != "done": if status in {"error", "cancelled"}: error_payload = status_payload.get("error") if isinstance(status_payload, dict) else None if isinstance(error_payload, dict): raise RaspberryAmundsenError( str(error_payload.get("code") or status), str(error_payload.get("message") or "Raspberry-analyse mislukt."), 499 if status == "cancelled" else 502, str(error_payload.get("detail") or ""), ) raise RaspberryAmundsenError(status or "processing_failed", "Raspberry-analyse is gestopt.", 502) raise RaspberryAmundsenError("processing_failed", "Analyse is nog bezig.", 409, status or "queued") report = _normalize_report(raspberry_amundsen.analysis_result(job_id)) document = _document_for_report(route_id) if document is not None: report = _with_document_waypoints(report, document) analysis_cache.write(route_id, report) return report def _job_id_from_payload(payload: dict) -> str: job_id = str(payload.get("job_id") or payload.get("id") or "").strip() if not job_id: raise RaspberryAmundsenError("processing_failed", "Raspberry-service gaf geen jobnummer terug.", 502) return job_id def _normalize_report(report: dict) -> dict: return report_normalizer.normalize(report, project_store.get_hiking_settings()) def _with_document_waypoints(report: dict, document: GPXDocument) -> dict: report["waypoints"] = [waypoint.__dict__ for waypoint in document.waypoints] return report def _rename_new_route_from_report(route_id: str, report: dict) -> None: route_name = str(report.get("name") or "").strip() if route_name: project_store.update_route_metadata(route_id, route_name, None) def _route_filename(route_id: str, fallback_name: str) -> str: route = project_store.get_route(route_id) filename = str(route.get("filename") or "") if route else "" if filename: return filename return f"{_filename_route_name(fallback_name)}.gpx" def _filename_route_name(filename: str) -> str: base_name = os.path.splitext(os.path.basename(filename or ""))[0].strip() return base_name or "Wandeling" def _float_from_env(name: str, default: float, minimum: float | None = None) -> float: try: value = float(os.getenv(name, str(default))) except ValueError: return default if minimum is not None: return max(minimum, value) return value def _render_index_error(error: str) -> str: return render_template( "index.html", error=error, project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_user=session.get("username"), current_menu="tools", ) def _render_gpx_combine( error: str | None = None, output_name: str = "Gecombineerde route", preview: dict | None = None, draft_token: str | None = None, ) -> str: return render_template( "gpx_combine.html", error=error, output_name=output_name, preview=preview, draft_token=draft_token, project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_tool="gpx_combine", current_user=session.get("username"), current_menu="tools", ) def _render_gpx_trim( error: str | None = None, output_name: str = "Geknipte route", preview: dict | None = None, draft_token: str | None = None, ) -> str: return render_template( "gpx_trim.html", error=error, output_name=output_name, preview=preview, draft_token=draft_token, project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_tool="gpx_trim", current_user=session.get("username"), current_menu="tools", ) def _render_gpx_simplify( error: str | None = None, output_name: str = "Verkleinde route", tolerance_m: float = 10.0, preview: dict | None = None, draft_token: str | None = None, ) -> str: return render_template( "gpx_simplify.html", error=error, output_name=output_name, tolerance_m=tolerance_m, preview=preview, draft_token=draft_token, project_tree=project_store.list_projects(), hiking_settings=project_store.get_hiking_settings(), current_tool="gpx_simplify", current_user=session.get("username"), current_menu="tools", ) def _selected_combine_routes( source_indexes: list[str], reversed_values: list[str], expected_count: int, ) -> list[tuple[int, bool]]: routes = [] if len(source_indexes) != expected_count or len(reversed_values) != expected_count: return routes seen_indexes = set() for source_index, reversed_value in zip(source_indexes, reversed_values, strict=False): try: parsed_index = int(source_index) except ValueError: return [] if parsed_index < 0 or parsed_index >= expected_count or parsed_index in seen_indexes: return [] seen_indexes.add(parsed_index) routes.append((parsed_index, reversed_value == "1")) return routes def _ratio_from_form(name: str, fallback: float) -> float: try: return float(request.form.get(name, fallback)) except (TypeError, ValueError): return fallback def _float_from_form(name: str, fallback: float, minimum: float, maximum: float) -> float: value = _ratio_from_form(name, fallback) return min(max(value, minimum), maximum) def _gpx_filename(name: str) -> str: clean = "".join( character if character.isascii() and (character.isalnum() or character in ("-", "_")) else "-" for character in name.lower() ) clean = "-".join(part for part in clean.split("-") if part) return f"{clean or 'gecombineerde-route'}.gpx" def _normalize_route_date(value: str) -> str | None: clean_value = str(value or "").strip() if not clean_value: return None try: return datetime.strptime(clean_value, "%Y-%m-%d").date().isoformat() except ValueError: return None def _date_input_value(value: object) -> str: if hasattr(value, "strftime"): return value.strftime("%Y-%m-%d") return str(value or "") def _format_date(value: object) -> str: if hasattr(value, "strftime"): return value.strftime("%d-%m-%Y") if isinstance(value, str): try: return datetime.strptime(value, "%Y-%m-%d").strftime("%d-%m-%Y") except ValueError: pass return str(value or "-") def send_pushover(title: str, message: str) -> bool: if not PUSHOVER_API_TOKEN or not PUSHOVER_USER_KEY: return False try: response = requests.post( "https://api.pushover.net/1/messages.json", data={ "token": PUSHOVER_API_TOKEN, "user": PUSHOVER_USER_KEY, "title": title, "message": message, "priority": 1, }, timeout=10, ) response.raise_for_status() except requests.RequestException: return False return True if __name__ == "__main__": app.run(host="0.0.0.0", port=6006, debug=True)