6.6 KiB
Dashcam Transfer Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Automatic transfer of dashcam MP4 files from SD card to USB-SSD.
Architecture: Single Python script (dashcam-transfer.py) using stdlib only, triggered by udev rule via systemd oneshot service.
Tech Stack: Python 3 (stdlib), systemd, udev
Task 1: Write dashcam-transfer.py
Files:
- Create:
dashcam-transfer.py(local repo) - Deploy to:
/usr/local/bin/dashcam-transferon nextcloud
The script must:
- Parse all
.MP4files from/mnt/sdcard/Novatek/Movie/ - Extract date (first 8 chars of filename) →
YYYYMMDD/dir - Extract camera suffix (last char before
.MP4):F→front/B→back/
- Create target dirs:
/mnt/ultratouch/dashcam/YYYYMMDD/{front,back}/ - Calculate total space needed, check available space
- If insufficient: sort date dirs on ultratouch by name, delete oldest dirs until enough space
mveach file (not copy)- Log every action to
/var/log/dashcam-transfer.log - Use
flocklockfile to prevent concurrent runs - Error handling: skip unreadable files, abort if SD not mounted
- Write dashcam-transfer.py
#!/usr/bin/env python3
"""Transfer dashcam MP4 files from SD card to USB-SSD with date/camera structure."""
import fcntl
import logging
import os
import re
import shutil
import sys
# Config
SD_CARD = "/mnt/sdcard"
SOURCE_DIR = os.path.join(SD_CARD, "Novatek", "Movie")
SSD = "/mnt/ultratouch"
TARGET_BASE = os.path.join(SSD, "dashcam")
LOG_FILE = "/var/log/dashcam-transfer.log"
LOCK_FILE = "/var/run/dashcam-transfer.lock"
CAMERA_MAP = {"F": "front", "B": "back"}
FILE_PATTERN = re.compile(r"^(\d{8})\d{6}_\d+([FB])\.MP4$")
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="[%(asctime)s] %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
def acquire_lock():
lock_fd = open(LOCK_FILE, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError:
logging.warning("Lockfile actief — script al bezig")
sys.exit(1)
return lock_fd
def check_mount(path, name):
if not os.path.ismount(path):
logging.error("%s niet gemount: %s", name, path)
sys.exit(1)
def get_mp4_files(src_dir):
if not os.path.isdir(src_dir):
logging.info("Geen bronmap: %s", src_dir)
return []
files = []
for fname in os.listdir(src_dir):
match = FILE_PATTERN.match(fname)
if match:
files.append((fname, match.group(1), match.group(2).upper()))
return files
def ensure_free_space(needed_bytes, target_base):
stat = os.statvfs(target_base)
free = stat.f_frsize * stat.f_bavail
if free >= needed_bytes:
return
logging.info("Onvoldoende ruimte: %d nodig, %d vrij", needed_bytes, free)
date_dirs = sorted(
d for d in os.listdir(target_base)
if re.match(r"^\d{8}$", d) and os.path.isdir(os.path.join(target_base, d))
)
freed = 0
removed = []
for d in date_dirs:
if free + freed >= needed_bytes:
break
dir_path = os.path.join(target_base, d)
dir_size = sum(
os.path.getsize(os.path.join(dirpath, f))
for dirpath, _, filenames in os.walk(dir_path)
for f in filenames
)
shutil.rmtree(dir_path)
freed += dir_size
removed.append(d)
logging.info("Verwijderd (ruimte gebrek): %s (%d bytes)", d, dir_size)
if free + freed < needed_bytes:
logging.error(
"Nog steeds onvoldoende ruimte na verwijderen %d mappen", len(removed)
)
sys.exit(1)
logging.info("Ruimte vrijgemaakt: %d bytes door %d mappen", freed, len(removed))
def main():
lock_fd = acquire_lock()
check_mount(SD_CARD, "SD-kaart")
check_mount(SSD, "Ultratouch SSD")
files = get_mp4_files(SOURCE_DIR)
if not files:
logging.info("Geen MP4 bestanden gevonden in %s", SOURCE_DIR)
return
total_size = sum(
os.path.getsize(os.path.join(SOURCE_DIR, fname)) for fname, _, _ in files
)
logging.info(
"Start transfer: %d bestanden, %d bytes", len(files), total_size
)
ensure_free_space(total_size, TARGET_BASE)
ok = 0
errors = 0
for fname, date_str, cam in files:
target_dir = os.path.join(TARGET_BASE, date_str, CAMERA_MAP[cam])
os.makedirs(target_dir, exist_ok=True)
src = os.path.join(SOURCE_DIR, fname)
dst = os.path.join(target_dir, fname)
try:
shutil.move(src, dst)
logging.info("Verplaatst: %s → %s/%s/", fname, date_str, CAMERA_MAP[cam])
ok += 1
except OSError as e:
logging.error("Fout bij %s: %s", fname, e)
errors += 1
logging.info(
"Transfer voltooid: %d ok, %d fouten, %d bytes", ok, errors, total_size
)
if __name__ == "__main__":
main()
- Commit to repo
git add dashcam-transfer.py
git commit -m "feat: dashcam-transfer.py script"
git push
Task 2: Deploy and test on nextcloud
Files:
-
Modify:
/usr/local/bin/dashcam-transferon nextcloud -
Copy script to nextcloud
scp dashcam-transfer.py aiaiaiaiagent@nextcloud:/tmp/
ssh aiaiaiaiagent@nextcloud 'sudo -A install -m 755 /tmp/dashcam-transfer.py /usr/local/bin/dashcam-transfer'
- Dry-run test (check detection, no transfer yet)
ssh aiaiaiaiagent@nextcloud '/usr/local/bin/dashcam-transfer'
Expected: Script should find 338 MP4 files, log the start, and proceed with transfer.
Task 3: Add udev rule + systemd service (if needed)
Files:
-
Create:
/etc/udev/rules.d/99-dashcam-sdcard.ruleson nextcloud -
Create:
/etc/systemd/system/dashcam-transfer.serviceon nextcloud -
Create:
dashcam-transfer-triggerscript on nextcloud -
Create trigger script
ssh aiaiaiaiagent@nextcloud 'sudo tee /usr/local/bin/dashcam-transfer-trigger << "TRIGGER"
#!/bin/bash
sleep 5
/usr/local/bin/dashcam-transfer
TRIGGER'
ssh aiaiaiaiagent@nextcloud 'sudo -A chmod +x /usr/local/bin/dashcam-transfer-trigger'
- Create systemd service
[Unit]
Description=Dashcam SD card transfer
After=mnt-sdcard.mount
[Service]
Type=oneshot
ExecStart=/usr/local/bin/dashcam-transfer
- Create udev rule
ACTION=="add", KERNEL=="sdd1", RUN+="/usr/local/bin/dashcam-transfer-trigger"