445 lines
14 KiB
Python
445 lines
14 KiB
Python
"""
|
|
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 json
|
|
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': 'L Rep.',
|
|
'right_repeater': 'R Rep.',
|
|
'back': 'Back',
|
|
'left_pillar': 'L Pillar',
|
|
'right_pillar': 'R Pillar'
|
|
}
|
|
|
|
# Camera button layout order: three columns around the car graphic
|
|
CAMERA_LAYOUT = [
|
|
['left_pillar', 'left_repeater'],
|
|
['front', 'back'],
|
|
['right_pillar', 'right_repeater'],
|
|
]
|
|
|
|
# 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 _normalize_path(path):
|
|
"""Normalize a filesystem path to use forward slashes and no trailing slash."""
|
|
return os.path.normpath(path).replace('\\', '/').rstrip('/')
|
|
|
|
|
|
def _parse_event_json(event_path):
|
|
"""
|
|
Parse an event.json file from a SavedClips subfolder.
|
|
Returns a dict mapping normalized timestamps (YYYY-MM-DD_HH-MM-SS) to
|
|
geodata dicts with lat, lon, and location fields.
|
|
"""
|
|
result = {}
|
|
try:
|
|
with open(event_path, 'r') as f:
|
|
data = json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
return result
|
|
|
|
# Handle both a single event object and a list of events
|
|
events = data if isinstance(data, list) else [data]
|
|
for evt in events:
|
|
ts_raw = evt.get('timestamp', '')
|
|
# Normalize ISO timestamp
|
|
ts_normalized = _normalize_timestamp(ts_raw)
|
|
lat = evt.get('est_lat', '')
|
|
lon = evt.get('est_lon', '')
|
|
city = evt.get('city', '')
|
|
street = evt.get('street', '')
|
|
parts = [p for p in [street, city] if p]
|
|
location = ', '.join(parts) if parts else ''
|
|
result[ts_normalized] = {
|
|
'lat': lat,
|
|
'lon': lon,
|
|
'location': location
|
|
}
|
|
return result
|
|
|
|
|
|
def _normalize_timestamp(ts_string):
|
|
"""
|
|
Normalizes a timestamp string to the format YYYYMMDD_HHMM.
|
|
"""
|
|
try:
|
|
ts = ts_string.replace('T', '_')
|
|
ts = ts.replace('-', '')
|
|
ts = ts.replace(':', '')
|
|
ts = ts.split('.')[0]
|
|
# remove last 2 digits
|
|
ts = ts[:-2]
|
|
# set last 2 digits to '00' to represent the minute
|
|
ts = ts[:-2] + '00'
|
|
return ts
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _get_video_timestamp_string(ts_string):
|
|
"""
|
|
normalizes a timestamp string to the format YYYYMMDD_HHMM, or returns None if invalid.
|
|
"""
|
|
try:
|
|
ts = ts_string.replace('T', '_')
|
|
ts = ts.replace('-', '')
|
|
ts = ts.replace(':', '')
|
|
ts = ts.split('.')[0]
|
|
# remove last 2 digits
|
|
ts = ts[:-2]
|
|
# set last 2 digits to '00' to represent the minute
|
|
ts = ts[:-2] + '00'
|
|
return ts
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _scan_teslacam_videos_uncached():
|
|
"""
|
|
Recursively scans the TeslaCam directory tree for video files and categorizes
|
|
them by camera view. Also parses event.json files in SavedClips subfolders
|
|
to attach geodata (lat, lon, location) to matching video entries.
|
|
|
|
Two-pass approach: first collect ALL event.json geodata across the entire
|
|
tree, then process videos and attach matching geodata.
|
|
|
|
Returns:
|
|
dict: Keys are raw camera view identifiers (e.g. 'front'), values are lists
|
|
of dictionaries containing file details sorted chronologically.
|
|
"""
|
|
videos_by_camera = {}
|
|
cam_dir = _normalize_path(TESLACAM_DIRECTORY)
|
|
|
|
if not os.path.exists(cam_dir):
|
|
return videos_by_camera
|
|
|
|
# ---- Pass 1: collect ALL event.json files first ----
|
|
geodata_by_timestamp = {} # normalized_ts -> {lat, lon, location}
|
|
for root, _, files in os.walk(cam_dir):
|
|
if 'event.json' in files:
|
|
event_path = _normalize_path(os.path.join(root, 'event.json'))
|
|
geodata_by_timestamp.update(_parse_event_json(event_path))
|
|
|
|
# ---- Pass 2: process video files, attach geodata from fully populated dict ----
|
|
for root, _, files in os.walk(cam_dir):
|
|
for filename in files:
|
|
match = VIDEO_FILENAME_REGEX.match(filename)
|
|
if not match:
|
|
continue
|
|
timestamp, camera_view = match.groups()
|
|
normalized_ts = _get_video_timestamp_string(timestamp)
|
|
|
|
# Get path relative to TESLACAM_DIRECTORY for safe web serving
|
|
full_path = os.path.join(root, filename)
|
|
rel_path = os.path.relpath(
|
|
full_path, cam_dir).replace('\\', '/')
|
|
|
|
video_entry = {
|
|
'filename': filename,
|
|
'timestamp': normalized_ts,
|
|
'rel_path': rel_path,
|
|
'url': f'/video/{rel_path}'
|
|
}
|
|
|
|
# Attach geodata if available for this timestamp
|
|
if normalized_ts in geodata_by_timestamp:
|
|
video_entry.update(geodata_by_timestamp[normalized_ts])
|
|
|
|
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: video player with camera-angle switching via a 3-column
|
|
button grid around a top-down Model Y graphic.
|
|
"""
|
|
all_videos = scan_teslacam_videos()
|
|
|
|
# Build camera info list for the button grid
|
|
camera_list = []
|
|
for col in CAMERA_LAYOUT:
|
|
col_cams = []
|
|
for key in col:
|
|
videos = all_videos.get(key, [])
|
|
col_cams.append({
|
|
'key': key,
|
|
'display': get_camera_display_name(key),
|
|
'count': len(videos),
|
|
'first_url': videos[0]['url'] if videos else None,
|
|
})
|
|
camera_list.append(col_cams)
|
|
|
|
# Default to the oldest front video
|
|
front_videos = all_videos.get('front', [])
|
|
default_url = front_videos[0]['url'] if front_videos else None
|
|
|
|
return render_template(
|
|
'index.html',
|
|
camera_list=camera_list,
|
|
default_url=default_url,
|
|
)
|
|
|
|
|
|
@app.route('/camera/<camera_view>')
|
|
@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/<path:filepath>')
|
|
@login_required
|
|
def serve_video(filepath):
|
|
"""
|
|
Safely streams video files from the TeslaCam directory structure.
|
|
"""
|
|
full_path = os.path.join(_normalize_path(TESLACAM_DIRECTORY), filepath)
|
|
if not os.path.exists(full_path):
|
|
abort(404, description="Video file not found")
|
|
|
|
return send_from_directory(_normalize_path(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('/api/videos')
|
|
@login_required
|
|
def api_videos():
|
|
"""
|
|
Returns all scanned videos as JSON, grouped by camera view.
|
|
Used by the frontend for camera-angle switching.
|
|
"""
|
|
all_videos = scan_teslacam_videos()
|
|
result = {}
|
|
for cam_key, video_list in all_videos.items():
|
|
result[cam_key] = {
|
|
'display': get_camera_display_name(cam_key),
|
|
'videos': video_list
|
|
}
|
|
return jsonify(result)
|
|
|
|
|
|
# US region boundaries for geodata classification
|
|
# West: lon <= -104 (Pacific + Mountain)
|
|
# Midwest: -104 < lon <= -80 AND lat >= 37
|
|
# South: lat < 37 AND lon > -104
|
|
# East: lon > -80
|
|
US_REGION_BOUNDS = [
|
|
('4-Back West', lambda lat, lon: lon <= -104),
|
|
('3-Midwest', lambda lat, lon: -104 < lon <= -80 and lat >= 37),
|
|
('1-Texas', lambda lat, lon: lat < 37 and lon > -104),
|
|
('2-East Coast', lambda lat, lon: lon > -80),
|
|
]
|
|
|
|
|
|
def _classify_region(lat, lon):
|
|
"""Classify a lat/lon pair into a US region name, or None if unclassified."""
|
|
try:
|
|
lat_f = float(lat)
|
|
lon_f = float(lon)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
for name, check in US_REGION_BOUNDS:
|
|
if check(lat_f, lon_f):
|
|
return name
|
|
return None
|
|
|
|
|
|
@app.route('/api/regions')
|
|
@login_required
|
|
def api_regions():
|
|
"""
|
|
Returns available US regions with clip counts and a representative
|
|
front-camera timestamp for jumping to region.
|
|
"""
|
|
all_videos = scan_teslacam_videos()
|
|
regions = {}
|
|
|
|
for cam_key, video_list in all_videos.items():
|
|
for v in video_list:
|
|
lat = v.get('lat', '')
|
|
lon = v.get('lon', '')
|
|
if not lat or not lon:
|
|
continue
|
|
region = _classify_region(lat, lon)
|
|
if not region:
|
|
continue
|
|
ts = v['timestamp']
|
|
if region not in regions:
|
|
regions[region] = {'count': 0, 'timestamps': set()}
|
|
regions[region]['count'] += 1
|
|
regions[region]['timestamps'].add(ts)
|
|
|
|
# Convert sets to sorted lists, find earliest front-cam timestamp per region
|
|
front_videos = all_videos.get('front', [])
|
|
front_ts_set = {v['timestamp'] for v in front_videos}
|
|
|
|
result = {}
|
|
for name, data in sorted(regions.items()):
|
|
ts_list = sorted(data['timestamps'])
|
|
# First timestamp that has a front camera clip
|
|
first_ts = next((ts for ts in ts_list if ts in front_ts_set),
|
|
ts_list[0] if ts_list else None)
|
|
result[name] = {
|
|
'count': data['count'],
|
|
'first_timestamp': first_ts,
|
|
}
|
|
|
|
return jsonify({'regions': result})
|
|
|
|
|
|
@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/<path:filename>')
|
|
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)
|