Mathieu MoranandCopilot
acaf1747f3
First real-hardware test revealed the connected dashcam's SD card ships factory-formatted with no partition table - lsblk reports it as a bare "disk" with an fstype set directly, never as a child "part". The previous detection logic only matched type=="part", so it silently missed this real device entirely. list_usb_storage_partitions() now also matches whole-disk USB devices that have no children and an fstype set directly on them, in addition to the existing partition-table case. Verified against a fixture built from this dashcam's actual `lsblk -J` output. Co-authored-by: Copilot <[email protected]>
231 lines
8.4 KiB
Python
231 lines
8.4 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
|
|
|
|
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"
|
|
)
|
|
|
|
|
|
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):
|
|
"""Mount device_path read-only at mount_point. Returns True on success."""
|
|
os.makedirs(mount_point, exist_ok=True)
|
|
try:
|
|
subprocess.run(
|
|
["sudo", "mount", "-o", "ro", 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 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 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).
|
|
"""
|
|
partition = find_dashcam_partition()
|
|
if partition is None:
|
|
return {"status": "no_device", "message": "No USB dashcam/storage device detected."}
|
|
|
|
device_path = partition["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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(import_dashcam())
|