"""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 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 date/event sub-folders - flattening them risks filename collisions). 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) videos = find_video_files(source_folder) copied = 0 skipped = 0 errors = [] 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): """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) summary["status"] = "ok" summary["device"] = device_path return summary finally: if mounted_here: unmount_device(mount_point) if __name__ == "__main__": print(import_dashcam())