Mathieu MoranandCopilot
ea03ef7f6c
- yaris.css: add dash-col-size/date/upload hide rules on mobile (<576px), dash-col-file truncation on tablet, btn-icon-mobile label hide - app.py: clamp byte2 to file_size-1 to prevent Content-Length mismatch; detect MIME from file extension (.mov -> video/quicktime) Co-authored-by: Copilot <[email protected]>
271 lines
9.0 KiB
Python
271 lines
9.0 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_gps,app_obd, app_sql, app_dash, app_upload, app_video, app_health, app_alerts, app_settings, app_network
|
|
import os, 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()
|
|
|
|
@app.route('/')
|
|
def index():
|
|
print("hello "+ account.name)
|
|
print("you're car model is:" + car.model)
|
|
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('/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/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
|
|
|
|
@app.route('/api/trips')
|
|
def trips():
|
|
return jsonify({"trips": app_sql.list_trips()})
|
|
|
|
@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/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)
|
|
if body is None:
|
|
relative_paths = None
|
|
elif isinstance(body, dict):
|
|
relative_paths = body.get("paths", [])
|
|
else:
|
|
relative_paths = []
|
|
summary = app_dash.import_dashcam(relative_paths=relative_paths)
|
|
status_code = (
|
|
200
|
|
if summary["status"] == "ok"
|
|
else 400
|
|
if summary["status"] == "invalid_selection"
|
|
else 409
|
|
)
|
|
return jsonify(summary), status_code
|
|
|
|
@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
|
|
|
|
ext = os.path.splitext(path)[1].lower()
|
|
mime_map = {".mp4": "video/mp4", ".mov": "video/quicktime", ".avi": "video/x-msvideo", ".mkv": "video/x-matroska"}
|
|
mime = mime_map.get(ext, "video/mp4")
|
|
|
|
file_size = os.path.getsize(path)
|
|
range_header = request.headers.get("Range")
|
|
|
|
if range_header:
|
|
m = range_header.strip().replace("bytes=", "").split("-")
|
|
byte1 = int(m[0]) if m[0] else 0
|
|
byte2 = int(m[1]) if len(m) > 1 and m[1] else file_size - 1
|
|
byte2 = min(byte2, file_size - 1)
|
|
length = byte2 - byte1 + 1
|
|
|
|
def generate_range():
|
|
with open(path, "rb") as f:
|
|
f.seek(byte1)
|
|
remaining = length
|
|
while remaining > 0:
|
|
chunk = f.read(min(65536, remaining))
|
|
if not chunk:
|
|
break
|
|
remaining -= len(chunk)
|
|
yield chunk
|
|
|
|
resp = Response(generate_range(), 206, mimetype=mime, direct_passthrough=True)
|
|
resp.headers["Content-Range"] = f"bytes {byte1}-{byte2}/{file_size}"
|
|
resp.headers["Accept-Ranges"] = "bytes"
|
|
resp.headers["Content-Length"] = str(length)
|
|
return resp
|
|
|
|
def generate_full():
|
|
with open(path, "rb") as f:
|
|
while True:
|
|
chunk = f.read(65536)
|
|
if not chunk:
|
|
break
|
|
yield chunk
|
|
|
|
resp = Response(generate_full(), 200, mimetype=mime, direct_passthrough=True)
|
|
resp.headers["Accept-Ranges"] = "bytes"
|
|
resp.headers["Content-Length"] = str(file_size)
|
|
return resp
|
|
|
|
@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/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/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():
|
|
return jsonify({
|
|
"mil": app_obd.get_fresh_mil(),
|
|
"codes": app_obd.get_fresh_dtc_codes(),
|
|
})
|
|
|
|
@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()
|
|
|