472 lines
17 KiB
Python
472 lines
17 KiB
Python
from flask import Flask, render_template, make_response, json, jsonify, request, send_file, Response
|
|
from modules import app_gps, app_events, app_car, app_account, app_obd, app_sql, app_dash, app_upload, app_video, app_health, app_alerts, app_settings, app_network, app_device_config, app_device_sync, app_remote_config, app_status
|
|
import os, subprocess, time
|
|
from gevent.pywsgi import WSGIServer
|
|
|
|
|
|
app = Flask(__name__)
|
|
app_events.socketio.init_app(app)
|
|
|
|
account = app_account.Account()
|
|
obd = app_obd.start_obd()
|
|
car = app_car.Car()
|
|
gps = app_gps.start_gps()
|
|
app_sql.start_logging()
|
|
app_upload.start_uploading()
|
|
app_alerts.start_alerting()
|
|
|
|
|
|
def _reload_runtime_device_config():
|
|
account.reload_runtime_config()
|
|
car.reload_runtime_config()
|
|
|
|
|
|
app_device_sync.register_reload_callback(_reload_runtime_device_config)
|
|
app_device_sync.start_sync()
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html', account=account, car=car)
|
|
|
|
@app.route('/dtc')
|
|
def dtc():
|
|
return render_template('dtc.html', account = account, car = car)
|
|
|
|
@app.route('/dash')
|
|
def dash():
|
|
return render_template('dash.html', account = account, car = car)
|
|
|
|
@app.route('/history')
|
|
def history():
|
|
return render_template('history.html', account=account, car=car)
|
|
|
|
@app.route('/settings')
|
|
def settings():
|
|
return render_template('settings.html', account=account, car=car)
|
|
|
|
@app.route('/status')
|
|
def status_page():
|
|
return render_template('status.html', account=account, car=car)
|
|
|
|
@app.route('/api/settings', methods=['GET', 'POST'])
|
|
def api_settings():
|
|
if request.method == 'GET':
|
|
return jsonify(app_settings.get_settings())
|
|
errors = app_settings.update_settings(request.get_json(silent=True))
|
|
if errors:
|
|
return jsonify(error="Invalid settings.", fields=errors), 400
|
|
return jsonify(app_settings.get_settings())
|
|
|
|
@app.route('/api/device-config', methods=['GET', 'POST'])
|
|
def api_device_config():
|
|
if request.method == 'GET':
|
|
return jsonify(app_device_config.get_device_config())
|
|
errors = {}
|
|
app_device_config.update_device_config(request.get_json(silent=True), errors)
|
|
if errors:
|
|
return jsonify(error="Invalid device config.", fields=errors), 400
|
|
_reload_runtime_device_config()
|
|
return jsonify(app_device_config.get_device_config())
|
|
|
|
@app.route('/api/remote-config', methods=['GET', 'POST'])
|
|
def api_remote_config():
|
|
if request.method == 'GET':
|
|
return jsonify(app_remote_config.get_remote_config())
|
|
errors = {}
|
|
app_remote_config.update_remote_config(request.get_json(silent=True), errors)
|
|
if errors:
|
|
return jsonify(error="Invalid remote config.", fields=errors), 400
|
|
app_upload.start_uploading()
|
|
app_device_sync.start_sync()
|
|
return jsonify(app_remote_config.get_remote_config())
|
|
|
|
@app.route('/api/remote-config/test', methods=['POST'])
|
|
def api_remote_config_test():
|
|
result = app_remote_config.test_connection()
|
|
return jsonify(result), 200 if result.get("ok") else 400
|
|
|
|
@app.route('/api/remote-config/sync-now', methods=['POST'])
|
|
def api_remote_config_sync_now():
|
|
try:
|
|
result = app_device_sync.sync_now()
|
|
except Exception as error:
|
|
return jsonify(status="error", message=str(error)), 502
|
|
status_codes = {
|
|
"synced": 200,
|
|
"no_change": 200,
|
|
"not_configured": 400,
|
|
"offline": 503,
|
|
}
|
|
return jsonify(result), status_codes.get(result["status"], 500)
|
|
|
|
@app.route('/api/uplink/status')
|
|
def uplink_status():
|
|
try:
|
|
return jsonify(app_network.get_status())
|
|
except app_network.NetworkConfigurationError as error:
|
|
return jsonify(error=str(error)), 503
|
|
|
|
@app.route('/api/uplink/config', methods=['POST', 'DELETE'])
|
|
def uplink_config():
|
|
try:
|
|
if request.method == 'DELETE':
|
|
app_network.remove_phone_hotspot()
|
|
else:
|
|
body = request.get_json(silent=True)
|
|
if not isinstance(body, dict):
|
|
return jsonify(error="Expected a JSON object."), 400
|
|
app_network.configure_phone_hotspot(
|
|
body.get("ssid"),
|
|
body.get("password"),
|
|
)
|
|
return jsonify(app_network.get_status())
|
|
except app_network.NetworkConfigurationError as error:
|
|
return jsonify(error=str(error)), 400
|
|
|
|
@app.route('/api/uplink/connect', methods=['POST'])
|
|
def uplink_connect():
|
|
try:
|
|
app_network.request_phone_connection()
|
|
except app_network.NetworkConfigurationError as error:
|
|
return jsonify(error=str(error)), 503
|
|
return jsonify(status="connection_requested"), 202
|
|
|
|
_HOTSPOT_PROFILE = "Hotspot-local" # wlan0 AP profile name in NetworkManager
|
|
|
|
|
|
@app.route('/api/hotspot')
|
|
def hotspot_config():
|
|
"""Read current Pi hotspot SSID from NetworkManager (password is never returned)."""
|
|
ssid = ""
|
|
try:
|
|
result = subprocess.run(
|
|
["sudo", "-n", "nmcli", "-t", "-f", "802-11-wireless.ssid", "con", "show", _HOTSPOT_PROFILE],
|
|
capture_output=True, text=True, timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
for line in result.stdout.strip().splitlines():
|
|
if line.startswith("802-11-wireless.ssid:"):
|
|
ssid = line.split(":", 1)[1]
|
|
break
|
|
except Exception:
|
|
pass
|
|
return jsonify(ssid=ssid)
|
|
|
|
@app.route('/api/hotspot', methods=['POST'])
|
|
def hotspot_config_update():
|
|
"""Update Pi hotspot SSID and/or password via NetworkManager."""
|
|
body = request.get_json(silent=True) or {}
|
|
ssid = (body.get("ssid") or "").strip()
|
|
password = (body.get("password") or "")
|
|
|
|
if not ssid:
|
|
return jsonify(error="SSID is required."), 400
|
|
if len(ssid) > 32:
|
|
return jsonify(error="SSID must be at most 32 characters."), 400
|
|
if password and len(password) < 8:
|
|
return jsonify(error="Password must be at least 8 characters."), 400
|
|
if password and len(password) > 63:
|
|
return jsonify(error="Password must be at most 63 characters."), 400
|
|
|
|
try:
|
|
subprocess.run(
|
|
["sudo", "-n", "nmcli", "con", "mod", _HOTSPOT_PROFILE, "802-11-wireless.ssid", ssid],
|
|
capture_output=True, text=True, timeout=10, check=True,
|
|
)
|
|
if password:
|
|
subprocess.run(
|
|
["sudo", "-n", "nmcli", "con", "mod", _HOTSPOT_PROFILE, "802-11-wireless-security.psk", password],
|
|
capture_output=True, text=True, timeout=10, check=True,
|
|
)
|
|
# Restart the hotspot so changes take effect immediately.
|
|
subprocess.run(
|
|
["sudo", "-n", "nmcli", "con", "down", _HOTSPOT_PROFILE],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
time.sleep(1)
|
|
subprocess.run(
|
|
["sudo", "-n", "nmcli", "con", "up", _HOTSPOT_PROFILE],
|
|
capture_output=True, text=True, timeout=15, check=True,
|
|
)
|
|
return jsonify(message=f"Hotspot updated to '{ssid}' and restarted."
|
|
+ (" New password saved." if password else ""))
|
|
except subprocess.CalledProcessError as e:
|
|
return jsonify(error=f"Could not update hotspot: {e.stderr.strip() or 'nmcli failed'}"), 500
|
|
|
|
@app.route('/api/trips')
|
|
def trips():
|
|
return jsonify({"trips": app_sql.list_trips()})
|
|
|
|
@app.route('/api/gps/status')
|
|
def gps_status():
|
|
return jsonify(app_gps.get_status())
|
|
|
|
@app.route('/api/obd/status')
|
|
def obd_status():
|
|
port = app_obd._find_obd_port()
|
|
connected = app_obd._is_car_connected()
|
|
return jsonify({
|
|
"adapter_present": bool(port),
|
|
"port": port,
|
|
"car_connected": connected,
|
|
"status_label": "Connected" if connected else "Disconnected",
|
|
"speed": app_obd.get_fresh_speed(),
|
|
"rpm": app_obd.rpm,
|
|
"coolant_temp": app_obd.coolant_temp,
|
|
"fuel_level": app_obd.fuel_level,
|
|
"throttle_position": app_obd.throttle_position,
|
|
"mil_on": app_obd.get_fresh_mil(),
|
|
})
|
|
|
|
@app.route('/api/trip/<int:trip_id>')
|
|
def trip_route(trip_id):
|
|
trip = app_sql.get_trip_route(trip_id)
|
|
if trip is None:
|
|
return jsonify(error=f"Trip not found: {trip_id}"), 404
|
|
return jsonify(trip)
|
|
|
|
@app.route('/api/trip/<int:trip_id>', methods=['DELETE'])
|
|
def api_delete_trip(trip_id):
|
|
deleted = app_sql.delete_trip(trip_id)
|
|
if not deleted:
|
|
return jsonify(error=f"Trip not found: {trip_id}"), 404
|
|
return jsonify(status="deleted"), 200
|
|
|
|
import functools
|
|
import reverse_geocoder as rg
|
|
|
|
@functools.lru_cache(maxsize=512)
|
|
def _geocode_cached(lat4, lon4):
|
|
"""Reverse-geocode a coordinate to city/province using the offline
|
|
reverse_geocoder library. Always returns a non-empty label string."""
|
|
lat = lat4 / 10000.0
|
|
lon = lon4 / 10000.0
|
|
try:
|
|
results = rg.search([(lat, lon)], mode=1, verbose=False)
|
|
if results:
|
|
r = results[0]
|
|
label = ", ".join(p for p in [r.get("name", ""), r.get("admin1", "")] if p)
|
|
if label:
|
|
return label
|
|
except Exception:
|
|
pass
|
|
return f"{lat:.5f}, {lon:.5f}"
|
|
|
|
@app.route('/api/geocode')
|
|
def geocode_route():
|
|
try:
|
|
lat = float(request.args['lat'])
|
|
lon = float(request.args['lon'])
|
|
except (KeyError, ValueError):
|
|
return jsonify(error="lat and lon required"), 400
|
|
# Round to 3 decimals (~111m precision) for cache key
|
|
label = _geocode_cached(int(round(lat, 4) * 10000), int(round(lon, 4) * 10000))
|
|
return jsonify(label=label)
|
|
|
|
@app.route('/api/geocode/batch', methods=['POST'])
|
|
def geocode_batch_route():
|
|
body = request.get_json(silent=True) or {}
|
|
coords = body.get("coords")
|
|
if not isinstance(coords, list) or not coords:
|
|
return jsonify(error="expected {coords: [{lat, lon}, ...]}"), 400
|
|
# Deduplicate — same (lat4,lon4) would hit lru_cache anyway but avoids
|
|
# submitting redundant work.
|
|
seen = set()
|
|
result = {}
|
|
for item in coords:
|
|
try:
|
|
lat = float(item["lat"])
|
|
lon = float(item["lon"])
|
|
except (KeyError, ValueError, TypeError):
|
|
continue
|
|
lat4 = int(round(lat, 4) * 10000)
|
|
lon4 = int(round(lon, 4) * 10000)
|
|
if (lat4, lon4) not in seen:
|
|
seen.add((lat4, lon4))
|
|
result[f"{lat:.3f},{lon:.3f}"] = _geocode_cached(lat4, lon4)
|
|
return jsonify(result)
|
|
|
|
@app.route('/api/dash/devices')
|
|
def dash_devices():
|
|
devices = app_dash.list_usb_storage_partitions()
|
|
return jsonify({"devices": devices})
|
|
|
|
@app.route('/api/dash/import', methods=['POST'])
|
|
def dash_import():
|
|
body = request.get_json(silent=True)
|
|
relative_paths = body.get("paths") if isinstance(body, dict) else None
|
|
started, err = app_dash.start_import_async(relative_paths=relative_paths)
|
|
if not started:
|
|
return jsonify({"status": "busy", "message": err}), 409
|
|
return jsonify({"status": "started"})
|
|
|
|
@app.route('/api/dash/import/status')
|
|
def dash_import_status():
|
|
job = app_dash.get_import_status()
|
|
if job is None:
|
|
return jsonify({"status": "none"})
|
|
return jsonify(job)
|
|
|
|
@app.route('/api/dash/files')
|
|
def dash_files():
|
|
summary = app_dash.list_dashcam_files()
|
|
return jsonify(summary)
|
|
|
|
@app.route('/api/dash/files', methods=['DELETE'])
|
|
def dash_delete_all():
|
|
summary = app_dash.delete_dashcam_files(delete_all=True)
|
|
status_code = 200 if summary["status"] in {"ok", "partial"} else 409
|
|
return jsonify(summary), status_code
|
|
|
|
@app.route('/api/dash/file', methods=['DELETE'])
|
|
def dash_delete_file():
|
|
body = request.get_json(silent=True) or {}
|
|
summary = app_dash.delete_dashcam_files(relative_path=body.get("path"))
|
|
status_codes = {
|
|
"ok": 200,
|
|
"partial": 409,
|
|
"invalid_path": 400,
|
|
"not_found": 404,
|
|
}
|
|
return jsonify(summary), status_codes.get(summary["status"], 409)
|
|
|
|
@app.route('/api/local-videos')
|
|
def local_videos():
|
|
return jsonify(app_video.list_local_videos())
|
|
|
|
@app.route('/api/local-video/file')
|
|
def local_video_file():
|
|
relative_path = request.args.get("path", "")
|
|
try:
|
|
path = app_video.get_local_video_path(relative_path)
|
|
except ValueError as error:
|
|
return jsonify(error=str(error)), 400
|
|
except FileNotFoundError:
|
|
return jsonify(error=f"Local video not found: {relative_path}"), 404
|
|
as_attachment = request.args.get("download") == "1"
|
|
return send_file(path, conditional=True, as_attachment=as_attachment)
|
|
|
|
@app.route('/api/dash/file/download')
|
|
def dash_file_download():
|
|
relative_path = request.args.get("path", "")
|
|
try:
|
|
path = app_dash.safe_video_path(app_dash.DEFAULT_MOUNT_POINT, relative_path)
|
|
except (ValueError, FileNotFoundError) as error:
|
|
return jsonify(error=str(error)), 400
|
|
return send_file(path, as_attachment=True)
|
|
|
|
@app.route('/api/local-video/thumbnail')
|
|
def local_video_thumbnail():
|
|
relative_path = request.args.get("path", "")
|
|
try:
|
|
path = app_video.get_thumbnail_path(relative_path)
|
|
except ValueError as error:
|
|
return jsonify(error=str(error)), 400
|
|
except FileNotFoundError:
|
|
return jsonify(error=f"Local video not found: {relative_path}"), 404
|
|
except RuntimeError as error:
|
|
if str(error) == "generating":
|
|
return jsonify(status="generating"), 202
|
|
return jsonify(error=str(error)), 422
|
|
return send_file(path, conditional=True, mimetype="image/jpeg")
|
|
|
|
@app.route('/api/local-video', methods=['DELETE'])
|
|
def local_video_delete():
|
|
body = request.get_json(silent=True) or {}
|
|
summary = app_video.delete_local_video(body.get("path"))
|
|
status_codes = {
|
|
"ok": 200,
|
|
"invalid_path": 400,
|
|
"not_found": 404,
|
|
"busy": 409,
|
|
"delete_failed": 409,
|
|
}
|
|
return jsonify(summary), status_codes.get(summary["status"], 409)
|
|
|
|
@app.route('/api/local-video/convert', methods=['POST'])
|
|
def local_video_convert():
|
|
body = request.get_json(silent=True) or {}
|
|
path = body.get("path", "")
|
|
if not path:
|
|
return jsonify({"status": "invalid_path", "message": "No path provided."}), 400
|
|
queued = app_video.queue_convert(path, force=True)
|
|
if queued is False:
|
|
return jsonify({"status": "busy", "message": "Already converting."}), 409
|
|
return jsonify({"status": "started"})
|
|
|
|
@app.route('/api/upload/status')
|
|
def upload_status():
|
|
return jsonify(app_upload.get_status())
|
|
|
|
@app.route('/api/health')
|
|
def health():
|
|
return jsonify(app_health.collect_health())
|
|
|
|
@app.route('/api/status')
|
|
def api_status():
|
|
return jsonify(app_status.collect_status())
|
|
|
|
@app.route('/api/status/restart-app', methods=['POST'])
|
|
def api_status_restart_app():
|
|
try:
|
|
return jsonify(app_status.request_app_restart()), 202
|
|
except RuntimeError as error:
|
|
return jsonify(error=str(error)), 503
|
|
|
|
@app.route('/api/alerts/status')
|
|
def alert_status():
|
|
return jsonify(app_alerts.get_status())
|
|
|
|
@app.route('/api/upload/dashcam_toggle', methods=['POST'])
|
|
def upload_dashcam_toggle():
|
|
body = request.get_json(silent=True) or {}
|
|
if not isinstance(body.get("enabled"), bool):
|
|
return jsonify(error="expected JSON body {'enabled': true|false}"), 400
|
|
app_upload.set_dashcam_upload_enabled(body["enabled"])
|
|
return jsonify(app_upload.get_status())
|
|
|
|
# not sure if im gonna use this
|
|
@app.route('/api/obd_speed/')
|
|
def data():
|
|
data = app_obd.speed
|
|
response = make_response(json.dumps(data))
|
|
response.content_type = 'application/json'
|
|
return response
|
|
|
|
@app.route('/api/dtc')
|
|
def api_dtc():
|
|
fresh_mil = app_obd.get_fresh_mil()
|
|
fresh_codes = app_obd.get_fresh_dtc_codes()
|
|
if fresh_mil is None and not fresh_codes:
|
|
# OBD offline/stale (disconnected or not yet polled): fall back to the
|
|
# last-persisted DTC snapshot so codes survive a disconnect/reboot.
|
|
persisted = app_sql.get_persisted_dtc_state()
|
|
fresh_mil = persisted["mil"]
|
|
fresh_codes = persisted["codes"]
|
|
return jsonify({
|
|
"mil": fresh_mil,
|
|
"codes": fresh_codes,
|
|
})
|
|
|
|
@app.route('/api/dtc/clear', methods=['POST'])
|
|
def api_dtc_clear():
|
|
result = app_obd.clear_dtc_codes()
|
|
status_codes = {
|
|
"cleared": 200,
|
|
"not_connected": 409,
|
|
"error": 502,
|
|
}
|
|
return jsonify(result), status_codes.get(result["status"], 500)
|
|
|
|
@app.route('/api/stats')
|
|
def api_stats():
|
|
return jsonify(app_sql.get_overview_stats())
|
|
|
|
if __name__ == '__main__':
|
|
# Debug/Development
|
|
# app.run(debug=True, host="0.0.0.0", port="5000")
|
|
# Production
|
|
http_server = WSGIServer(('', 5000), app)
|
|
print("server on http://localhost:5000")
|
|
http_server.serve_forever()
|
|
|