quoije/yaris-obd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add dashcam USB file browser

Browse Source 
Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
2026-08-09 18:16:59 -04:00
co-authored by Copilot
parent 796b34a0b2
commit 5a520bf022
4 changed files with 396 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/app_dash.py
+220 -31
View File
@@ -20,6 +20,7 @@ import json
import os
import shutil
import subprocess
import threading
VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".ts", ".mkv"}
@@ -31,6 +32,7 @@ DEFAULT_MOUNT_POINT = "/mnt/dashcam"
DEFAULT_DESTINATION = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "dash", "videos"
)
_device_lock = threading.Lock()
def list_usb_storage_partitions():
@@ -107,8 +109,8 @@ def find_dashcam_partition():
return (recognized or partitions)[0]
def mount_device(device_path, mount_point=DEFAULT_MOUNT_POINT):
"""Mount device_path read-only at mount_point. Returns True on success.
def mount_device(device_path, mount_point=DEFAULT_MOUNT_POINT, read_only=True):
"""Mount device_path at mount_point. Returns True on success.
The app runs as an unprivileged user while /mnt is root-owned, so the
mount point directory itself has to be created via sudo too (not just
@@ -128,7 +130,14 @@ def mount_device(device_path, mount_point=DEFAULT_MOUNT_POINT):
try:
subprocess.run(
["sudo", "mount", "-o", "ro", device_path, mount_point],
[
"sudo",
"mount",
"-o",
"ro" if read_only else "rw",
device_path,
mount_point,
],
capture_output=True,
text=True,
check=True,
@@ -139,6 +148,28 @@ def mount_device(device_path, mount_point=DEFAULT_MOUNT_POINT):
return False
def remount_device(mount_point=DEFAULT_MOUNT_POINT, read_only=True):
"""Remount an existing dashcam mount read-only or read-write."""
try:
subprocess.run(
[
"sudo",
"mount",
"-o",
f"remount,{'ro' if read_only else 'rw'}",
mount_point,
],
capture_output=True,
text=True,
check=True,
)
return True
except subprocess.CalledProcessError as e:
mode = "read-only" if read_only else "read-write"
print(f"Failed to remount {mount_point} {mode}: {e.stderr.strip()}")
return False
def unmount_device(mount_point=DEFAULT_MOUNT_POINT):
"""Unmount mount_point. Returns True on success (or if already unmounted)."""
try:
@@ -170,6 +201,170 @@ def find_video_files(source_folder):
return found
def _video_metadata(source_folder):
files = []
errors = []
total_size = 0
for abs_path, rel_path in find_video_files(source_folder):
try:
stat = os.stat(abs_path)
except OSError as e:
errors.append(f"{rel_path}: {e}")
continue
files.append(
{
"path": rel_path,
"size": stat.st_size,
"modified_at": stat.st_mtime,
}
)
total_size += stat.st_size
files.sort(key=lambda item: item["path"].lower())
return files, total_size, errors
def _mounted_source(partition, mount_point, read_only):
"""Return (source, mounted_here, remounted_existing) for one operation."""
existing_mount = partition.get("mountpoint")
if existing_mount:
if not read_only and not remount_device(existing_mount, read_only=False):
return None, False, False
return existing_mount, False, not read_only
if not mount_device(partition["path"], mount_point, read_only=read_only):
return None, False, False
return mount_point, True, False
def list_dashcam_files(mount_point=DEFAULT_MOUNT_POINT):
"""List video files on the detected dashcam without copying them."""
with _device_lock:
partition = find_dashcam_partition()
if partition is None:
return {
"status": "no_device",
"message": "No USB dashcam/storage device detected.",
"files": [],
"total_size": 0,
}
source, mounted_here, _remounted = _mounted_source(
partition, mount_point, read_only=True
)
if source is None:
return {
"status": "mount_failed",
"device": partition["path"],
"message": f"Failed to mount {partition['path']} read-only.",
"files": [],
"total_size": 0,
}
try:
files, total_size, errors = _video_metadata(source)
return {
"status": "ok",
"device": partition,
"files": files,
"count": len(files),
"total_size": total_size,
"errors": errors,
}
finally:
if mounted_here:
unmount_device(mount_point)
def _safe_video_path(source_folder, relative_path):
if not isinstance(relative_path, str) or not relative_path.strip():
raise ValueError("A non-empty relative video path is required.")
if os.path.isabs(relative_path):
raise ValueError("Video path must be relative to the dashcam.")
if os.path.splitext(relative_path)[1].lower() not in VIDEO_EXTENSIONS:
raise ValueError("Only recognized video files can be deleted.")
source_real = os.path.realpath(source_folder)
target_real = os.path.realpath(os.path.join(source_real, relative_path))
if os.path.commonpath([source_real, target_real]) != source_real:
raise ValueError("Video path escapes the dashcam mount.")
if not os.path.isfile(target_real):
raise FileNotFoundError(relative_path)
return target_real
def delete_dashcam_files(relative_path=None, delete_all=False, mount_point=DEFAULT_MOUNT_POINT):
"""Delete one or all dashcam videos, restoring read-only state afterward."""
if not delete_all and relative_path is None:
return {"status": "invalid_path", "message": "A video path is required."}
with _device_lock:
partition = find_dashcam_partition()
if partition is None:
return {"status": "no_device", "message": "No USB dashcam/storage device detected."}
source, mounted_here, remounted_existing = _mounted_source(
partition, mount_point, read_only=False
)
if source is None:
return {
"status": "mount_failed",
"device": partition["path"],
"message": f"Failed to mount {partition['path']} read-write.",
}
result = None
cleanup_ok = True
try:
if delete_all:
targets = find_video_files(source)
else:
try:
target = _safe_video_path(source, relative_path)
except ValueError as e:
result = {"status": "invalid_path", "message": str(e)}
except FileNotFoundError:
result = {
"status": "not_found",
"message": f"Video not found: {relative_path}",
}
else:
targets = [(target, relative_path)]
if result is None:
deleted = []
errors = []
for _abs_path, rel_path in targets:
try:
safe_path = _safe_video_path(source, rel_path)
os.remove(safe_path)
deleted.append(rel_path)
except (OSError, ValueError) as e:
errors.append(f"{rel_path}: {e}")
result = {
"status": "ok" if not errors else "partial",
"device": partition["path"],
"deleted": deleted,
"deleted_count": len(deleted),
"errors": errors,
}
finally:
if mounted_here:
cleanup_ok = unmount_device(mount_point)
if not cleanup_ok:
cleanup_ok = remount_device(mount_point, read_only=True)
elif remounted_existing:
cleanup_ok = remount_device(source, read_only=True)
if not cleanup_ok:
return {
"status": "cleanup_failed",
"message": "Files were processed, but the dashcam could not be returned to read-only state.",
"operation": result,
}
return result
def copy_files(source_folder, destination_folder):
"""Copy every video file found under source_folder into destination_folder,
preserving its relative sub-folder path (dashcams often split files into
@@ -210,36 +405,30 @@ def import_dashcam(destination_folder=DEFAULT_DESTINATION, mount_point=DEFAULT_M
Returns a summary dict describing what happened, always unmounting the
device again afterwards (even if copying fails partway through).
"""
partition = find_dashcam_partition()
if partition is None:
return {"status": "no_device", "message": "No USB dashcam/storage device detected."}
with _device_lock:
partition = find_dashcam_partition()
if partition is None:
return {"status": "no_device", "message": "No USB dashcam/storage device detected."}
device_path = partition["path"]
device_path = partition["path"]
source, mounted_here, _remounted = _mounted_source(
partition, mount_point, read_only=True
)
if source is None:
return {
"status": "mount_failed",
"device": device_path,
"message": f"Failed to mount {device_path}.",
}
# Already mounted elsewhere (e.g. by the desktop environment) - read
# straight from there instead of trying to mount it again ourselves.
already_mounted_at = partition.get("mountpoint")
mounted_here = False
try:
source = already_mounted_at
if not source:
if not mount_device(device_path, mount_point):
return {
"status": "mount_failed",
"device": device_path,
"message": f"Failed to mount {device_path}.",
}
mounted_here = True
source = mount_point
summary = copy_files(source, destination_folder)
summary["status"] = "ok"
summary["device"] = device_path
return summary
finally:
if mounted_here:
unmount_device(mount_point)
try:
summary = copy_files(source, destination_folder)
summary["status"] = "ok"
summary["device"] = device_path
return summary
finally:
if mounted_here:
unmount_device(mount_point)
if __name__ == "__main__":