492 lines
17 KiB
Python
492 lines
17 KiB
Python
"""Dashcam USB retrieval.
|
|
|
|
Detects a dashcam plugged in as a USB mass-storage device (no hardcoded
|
|
/dev/sda, no brand-specific assumptions), mounts it read-only, copies any
|
|
video files it finds (any folder layout - DCIM/, MOVIE/, NORM/EVENT/PARKING/,
|
|
etc. - all common dashcam brands are covered since we just walk the whole
|
|
filesystem for video extensions instead of assuming a specific folder name),
|
|
then unmounts it again.
|
|
|
|
Detection is done via `lsblk`, filtering for volumes whose parent disk is
|
|
attached over USB (TRAN == "usb"). This is what actually distinguishes a
|
|
plugged-in dashcam/USB stick from the Pi's own storage (the boot/root
|
|
partitions are on the onboard SD card, transport "mmc", never "usb") - so
|
|
it's safe to run against real hardware without risk of touching the Pi's own
|
|
filesystem, regardless of which /dev/sdX letter the kernel happens to assign.
|
|
Both real partitions and whole-disk "superfloppy" (no partition table)
|
|
layouts are matched - some dashcam SD cards ship factory-formatted that way.
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
|
|
VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".ts", ".mkv"}
|
|
|
|
# Filesystem types dashcams commonly ship formatted as (SD cards are almost
|
|
# always FAT32/exFAT; a few older units use NTFS).
|
|
DASHCAM_FS_TYPES = {"vfat", "exfat", "ntfs", "ntfs3"}
|
|
|
|
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():
|
|
"""Return a list of dicts describing USB-attached storage volumes.
|
|
|
|
Each dict has: path, fstype, label, size, mountpoint. Only volumes whose
|
|
parent disk reports TRAN "usb" are included - this is what makes the
|
|
detection brand/device-agnostic (no hardcoded /dev/sda) while still
|
|
being safe (never matches the Pi's own onboard mmc storage).
|
|
|
|
Includes both real partitions (type "part", the common case) *and*
|
|
whole-disk USB devices with an `fstype` set directly on them (type
|
|
"disk" but no partition table - a "superfloppy" layout with no MBR/GPT).
|
|
Several real dashcam SD cards ship formatted this way straight from the
|
|
factory - lsblk reports them as a bare "disk" with an fstype, never as a
|
|
child "part", so excluding them would silently miss those cameras.
|
|
"""
|
|
try:
|
|
result = subprocess.run(
|
|
["lsblk", "-J", "-o", "NAME,PATH,TRAN,FSTYPE,LABEL,SIZE,MOUNTPOINT,TYPE"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
except (subprocess.CalledProcessError, FileNotFoundError) as e:
|
|
print("Failed to run lsblk:", e)
|
|
return []
|
|
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError as e:
|
|
print("Failed to parse lsblk output:", e)
|
|
return []
|
|
|
|
partitions = []
|
|
|
|
def walk(devices, inherited_tran=None):
|
|
for device in devices:
|
|
tran = device.get("tran") or inherited_tran
|
|
children = device.get("children") or []
|
|
is_partition = device.get("type") == "part"
|
|
is_superfloppy_disk = (
|
|
device.get("type") == "disk" and not children and device.get("fstype")
|
|
)
|
|
if tran == "usb" and (is_partition or is_superfloppy_disk):
|
|
partitions.append(
|
|
{
|
|
"path": device.get("path"),
|
|
"fstype": device.get("fstype"),
|
|
"label": device.get("label"),
|
|
"size": device.get("size"),
|
|
"mountpoint": device.get("mountpoint"),
|
|
}
|
|
)
|
|
if children:
|
|
walk(children, tran)
|
|
|
|
walk(data.get("blockdevices", []))
|
|
return partitions
|
|
|
|
|
|
def find_dashcam_partition():
|
|
"""Pick the best candidate USB partition to import from, or None.
|
|
|
|
Prefers partitions with a recognized dashcam filesystem type; if several
|
|
are found, the first one is returned (a single dashcam SD card should
|
|
only expose one data partition).
|
|
"""
|
|
partitions = list_usb_storage_partitions()
|
|
if not partitions:
|
|
return None
|
|
|
|
recognized = [p for p in partitions if p["fstype"] in DASHCAM_FS_TYPES]
|
|
return (recognized or partitions)[0]
|
|
|
|
|
|
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
|
|
the mount call) - first real-hardware test surfaced this as a plain
|
|
PermissionError from a bare os.makedirs().
|
|
"""
|
|
try:
|
|
subprocess.run(
|
|
["sudo", "mkdir", "-p", mount_point],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Failed to create mount point {mount_point}: {e.stderr.strip()}")
|
|
return False
|
|
|
|
try:
|
|
subprocess.run(
|
|
[
|
|
"sudo",
|
|
"mount",
|
|
"-o",
|
|
"ro" if read_only else "rw",
|
|
device_path,
|
|
mount_point,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Failed to mount {device_path}: {e.stderr.strip()}")
|
|
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:
|
|
subprocess.run(
|
|
["sudo", "umount", mount_point],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
# Not mounted is fine - nothing to clean up.
|
|
if "not mounted" in e.stderr.lower():
|
|
return True
|
|
print(f"Failed to unmount {mount_point}: {e.stderr.strip()}")
|
|
return False
|
|
|
|
|
|
def find_video_files(source_folder):
|
|
"""Walk source_folder, returning (absolute_path, relative_path) for every
|
|
file whose extension looks like a video, regardless of folder layout."""
|
|
found = []
|
|
for root, _dirs, files in os.walk(source_folder):
|
|
for file in files:
|
|
if os.path.splitext(file)[1].lower() in VIDEO_EXTENSIONS:
|
|
abs_path = os.path.join(root, file)
|
|
rel_path = os.path.relpath(abs_path, source_folder)
|
|
found.append((abs_path, rel_path))
|
|
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 _remove_video_file(path):
|
|
try:
|
|
os.remove(path)
|
|
return
|
|
except PermissionError:
|
|
pass
|
|
|
|
try:
|
|
subprocess.run(
|
|
["sudo", "rm", "--", path],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
raise OSError(e.stderr.strip() or f"Failed to delete {path}") from e
|
|
|
|
|
|
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)
|
|
_remove_video_file(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, relative_paths=None):
|
|
"""Copy selected videos, or every video when relative_paths is omitted.
|
|
|
|
Files already present at the destination (same relative path, same size)
|
|
are skipped so re-running an import doesn't re-copy everything.
|
|
|
|
Returns a summary dict: {found, copied, skipped, errors}.
|
|
"""
|
|
os.makedirs(destination_folder, exist_ok=True)
|
|
errors = []
|
|
|
|
if relative_paths is None:
|
|
videos = find_video_files(source_folder)
|
|
else:
|
|
if not isinstance(relative_paths, list) or not relative_paths:
|
|
return {
|
|
"found": 0,
|
|
"copied": 0,
|
|
"skipped": 0,
|
|
"errors": ["Select at least one video to import."],
|
|
"invalid_selection": True,
|
|
}
|
|
|
|
videos = []
|
|
seen = set()
|
|
for relative_path in relative_paths:
|
|
if relative_path in seen:
|
|
continue
|
|
seen.add(relative_path)
|
|
try:
|
|
abs_path = _safe_video_path(source_folder, relative_path)
|
|
except ValueError as e:
|
|
return {
|
|
"found": 0,
|
|
"copied": 0,
|
|
"skipped": 0,
|
|
"errors": [str(e)],
|
|
"invalid_selection": True,
|
|
}
|
|
except FileNotFoundError:
|
|
errors.append(f"Video not found: {relative_path}")
|
|
continue
|
|
videos.append((abs_path, relative_path))
|
|
|
|
copied = 0
|
|
skipped = 0
|
|
|
|
for abs_path, rel_path in videos:
|
|
dest_path = os.path.join(destination_folder, rel_path)
|
|
|
|
if os.path.exists(dest_path) and os.path.getsize(dest_path) == os.path.getsize(abs_path):
|
|
skipped += 1
|
|
continue
|
|
|
|
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
|
try:
|
|
shutil.copy2(abs_path, dest_path)
|
|
copied += 1
|
|
except OSError as e:
|
|
errors.append(f"{rel_path}: {e}")
|
|
|
|
return {"found": len(videos), "copied": copied, "skipped": skipped, "errors": errors}
|
|
|
|
|
|
def import_dashcam(
|
|
destination_folder=DEFAULT_DESTINATION,
|
|
mount_point=DEFAULT_MOUNT_POINT,
|
|
relative_paths=None,
|
|
):
|
|
"""Full detect -> mount -> copy -> unmount pipeline.
|
|
|
|
Returns a summary dict describing what happened, always unmounting the
|
|
device again afterwards (even if copying fails partway through).
|
|
"""
|
|
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"]
|
|
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}.",
|
|
}
|
|
|
|
try:
|
|
summary = copy_files(source, destination_folder, relative_paths)
|
|
summary["status"] = (
|
|
"invalid_selection" if summary.pop("invalid_selection", False) else "ok"
|
|
)
|
|
if summary["status"] == "invalid_selection":
|
|
summary["message"] = summary["errors"][0]
|
|
summary["device"] = device_path
|
|
return summary
|
|
finally:
|
|
if mounted_here:
|
|
unmount_device(mount_point)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(import_dashcam())
|