""" TeslaCam Video Player This web app displays TeslaCam videos and allows users to change the camera view. It scans the TeslaCam folder for all videos and stores a list sorted by filename and categorized by camera view (format YYYY-MM-DD_HH-MM-SS-${camera_view}.mp4). The app then displays the videos in a web interface, allowing users to select and play videos from different camera views. """ import os import re import time import secrets from functools import wraps from dotenv import load_dotenv from flask import ( Flask, render_template, send_from_directory, abort, jsonify, request, session, redirect, url_for ) # Load variables from a .env file if present (does not override real env vars) load_dotenv() # Password for web login. Default: 'patreon'. Override via LOGIN_PASSWORD env var. LOGIN_PASSWORD = os.environ.get('LOGIN_PASSWORD', 'patreon') # Root TeslaCam directory containing RecentClips, SavedClips, SentryClips, etc. # Override via the TESLACAM_DIRECTORY environment variable (see .env.example). TESLACAM_DIRECTORY = os.environ.get('TESLACAM_DIRECTORY', '/mnt/TeslaCam') app = Flask(__name__) # Session secret key — MUST be set in production. Generate with: python -c "import secrets; print(secrets.token_hex(32))" app.secret_key = os.environ.get('SECRET_KEY', secrets.token_hex(32)) # Regular expression matching TeslaCam filenames: YYYY-MM-DD_HH-MM-SS-${camera_view}.mp4 VIDEO_FILENAME_REGEX = re.compile( r'^(\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2})-(\w+)\.mp4$' ) # Friendly display names for camera views CAMERA_VIEW_MAPPING = { 'front': 'Front', 'left_repeater': 'Left Repeater', 'right_repeater': 'Right Repeater', 'rear': 'Rear', 'left_pillar': 'Left Pillar', 'right_pillar': 'Right Pillar' } # In-memory scan cache. Scanning a large TeslaCam tree on every request is # expensive, so the result is cached and only refreshed after this many seconds # (or when /refresh is hit). Override via the SCAN_CACHE_TTL env var. SCAN_CACHE_TTL = int(os.environ.get('SCAN_CACHE_TTL', '3600')) _scan_cache = {'data': None, 'expires_at': 0.0} def _scan_teslacam_videos_uncached(): """ Recursively scans the TeslaCam directory tree for video files and categorizes them by camera view. Returns: dict: Keys are raw camera view identifiers (e.g. 'front'), values are lists of dictionaries containing file details sorted chronologically. """ videos_by_camera = {} if not os.path.exists(TESLACAM_DIRECTORY): return videos_by_camera # Recursive directory walk to catch nested folders in SavedClips/SentryClips for root, _, files in os.walk(TESLACAM_DIRECTORY): for filename in files: match = VIDEO_FILENAME_REGEX.match(filename) if match: timestamp, camera_view = match.groups() # Get path relative to TESLACAM_DIRECTORY for safe web serving full_path = os.path.join(root, filename) rel_path = os.path.relpath( full_path, TESLACAM_DIRECTORY).replace('\\', '/') video_entry = { 'filename': filename, 'timestamp': timestamp, 'rel_path': rel_path, 'url': f'/video/{rel_path}' } if camera_view not in videos_by_camera: videos_by_camera[camera_view] = [] videos_by_camera[camera_view].append(video_entry) # Sort each camera list chronologically by filename/timestamp for camera_view in videos_by_camera: videos_by_camera[camera_view].sort(key=lambda item: item['filename']) return videos_by_camera def scan_teslacam_videos(force_refresh=False): """ Returns the cached TeslaCam video scan, refreshing it when the TTL expires or when `force_refresh` is True. Safe to call on every request. """ now = time.time() if ( _scan_cache['data'] is None or now >= _scan_cache['expires_at'] or force_refresh ): _scan_cache['data'] = _scan_teslacam_videos_uncached() _scan_cache['expires_at'] = now + SCAN_CACHE_TTL return _scan_cache['data'] def get_camera_display_name(camera_view): """Returns human-readable title for a camera view key.""" return CAMERA_VIEW_MAPPING.get(camera_view, camera_view.replace('_', ' ').title()) def login_required(f): """Decorator: redirect to /login if not authenticated.""" @wraps(f) def decorated_function(*args, **kwargs): if not session.get('logged_in'): return redirect(url_for('login')) return f(*args, **kwargs) return decorated_function @app.route('/login', methods=['GET', 'POST']) def login(): """Simple password-based login page.""" error = None if request.method == 'POST': if request.form.get('password') == LOGIN_PASSWORD: session['logged_in'] = True return redirect(url_for('index')) error = 'Invalid password. Please try again.' return render_template('login.html', error=error) @app.route('/logout') def logout(): """Clear session and return to login page.""" session.pop('logged_in', None) return redirect(url_for('login')) @app.route('/') @login_required def index(): """ Main page listing all videos categorized by human-readable camera view. """ all_videos = scan_teslacam_videos() # Map raw camera keys to display names for rendering categorized_videos = {} for camera_view, video_list in all_videos.items(): display_name = get_camera_display_name(camera_view) categorized_videos[display_name] = { 'raw_key': camera_view, 'videos': video_list } return render_template('index.html', categorized_videos=categorized_videos) @app.route('/camera/') @login_required def camera_view(camera_view): """ Displays all available videos for a single camera view. """ all_videos = scan_teslacam_videos() videos = all_videos.get(camera_view, []) display_name = get_camera_display_name(camera_view) return render_template( 'camera_view.html', camera_view_key=camera_view, camera_view_name=display_name, videos=videos ) @app.route('/video/') @login_required def serve_video(filepath): """ Safely streams video files from the TeslaCam directory structure. """ full_path = os.path.join(TESLACAM_DIRECTORY, filepath) if not os.path.exists(full_path): abort(404, description="Video file not found") return send_from_directory(TESLACAM_DIRECTORY, filepath) @app.route('/refresh') @login_required def refresh(): """ Forces a re-scan of the TeslaCam directory, bypassing the cache TTL. Useful after copying new clips onto the drive. """ scan_teslacam_videos(force_refresh=True) return 'Cache refreshed', 200 @app.route('/health') def health(): """ Health-check endpoint for load balancers / monitoring. Returns 200 with a JSON status when the app is running. """ return jsonify({'status': 'ok'}) @app.route('/static/') def serve_static(filename): """ Serves static assets (CSS, JS, images) from the static/ directory. """ static_dir = app.static_folder or os.path.join(app.root_path, 'static') return send_from_directory(static_dir, filename) if __name__ == '__main__': # Local dev server. For production, run behind a WSGI server instead, e.g.: # gunicorn -w 4 -b 0.0.0.0:5000 'videoplayer:app' # or # waitress-serve --host=0.0.0.0 --port=5000 videoplayer:app host = os.environ.get('FLASK_RUN_HOST', '0.0.0.0') port = int(os.environ.get('FLASK_RUN_PORT', '5000')) debug = os.environ.get('FLASK_DEBUG', '0').lower() in ('1', 'true', 'yes') app.run(debug=debug, host=host, port=port)