commit 052f10c083394ae36df27bbabeb07133ac04eb28 Author: William Söderberg Date: Thu Sep 24 21:21:46 2026 +0200 Initial Commit diff --git a/.gitea/workflows/docker-publish.yml b/.gitea/workflows/docker-publish.yml new file mode 100644 index 0000000..fffd194 --- /dev/null +++ b/.gitea/workflows/docker-publish.yml @@ -0,0 +1,44 @@ +name: Publish Docker image + +on: + release: + types: [published] + push: + tags: + - "v*" + +jobs: + push_to_registry: + name: Build and Push Docker image + runs-on: ubuntu-latest + permissions: + packages: write + contents: read + + steps: + - name: Check out the repo + uses: actions/checkout@v4 + + - name: Log in to the Container registry + uses: docker/login-action@v3 + with: + registry: git.stws.cc + username: ${{ github.actor }} + password: ${{ secrets.REGISTRY_TOKEN }} + + - name: Extract metadata (tags, labels) + id: meta + uses: docker/metadata-action@v5 + with: + images: git.stws.cc/${{ github.repository }} + tags: | + type=ref,event=tag + type=raw,value=latest + + - name: Build and push Docker images + uses: docker/build-push-action@v6 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..2648caa --- /dev/null +++ b/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-alpine + +ENV PYTHONUNBUFFERED=1 + +# Install system dependencies +RUN apk add --no-cache \ + usbutils \ + iproute2 \ + dhcpcd \ + curl \ + ffmpeg \ + gcc \ + musl-dev \ + linux-headers \ + libffi-dev + +WORKDIR /app + +# Install Python dependencies and strip compiler tools +RUN pip install --no-cache-dir \ + "open-gopro>=0.22.0" \ + aiohttp \ + && apk del gcc musl-dev linux-headers libffi-dev + +COPY ingest.py /app/ingest.py +RUN chmod +x /app/ingest.py + +CMD ["python3", "/app/ingest.py"] \ No newline at end of file diff --git a/ingest.py b/ingest.py new file mode 100644 index 0000000..b5f61f1 --- /dev/null +++ b/ingest.py @@ -0,0 +1,515 @@ +#!/usr/bin/env python3 +import asyncio +import logging +import os +import re +import subprocess +import sys +import tempfile +import time +import urllib.request +import warnings +import shutil +from collections import defaultdict +from pathlib import Path +import aiohttp +from open_gopro import WiredGoPro + +warnings.filterwarnings("ignore", category=UserWarning) +warnings.filterwarnings("ignore", message=".*unregister_all_services.*") + +DEBUG_MODE = os.getenv("DEBUG", "false").lower() == "true" +LOG_LEVEL = logging.DEBUG if DEBUG_MODE else logging.INFO + +logging.basicConfig( + level=LOG_LEVEL, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[logging.StreamHandler(sys.stdout)] +) +logger = logging.getLogger("gopro-ingest") + +if not DEBUG_MODE: + for lib in ["open_gopro", "zeroconf", "asyncio", "aiohttp", "urllib3"]: + logging.getLogger(lib).setLevel(logging.ERROR) + +DEST_DIR = Path(os.getenv("INGEST_DIR", "/media")) +TEMP_DIR = DEST_DIR / "temp" +HISTORY_FILE = DEST_DIR / ".ingested_history.txt" + +POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "5")) +LOG_PROGRESS_INTERVAL = int(os.getenv("LOG_PROGRESS_INTERVAL", "5")) +DELETE_AFTER_INGEST = os.getenv("DELETE_AFTER_INGEST", "false").lower() == "true" +KEEP_RAW_CHAPTERS = os.getenv("KEEP_RAW_CHAPTERS", "false").lower() == "true" + +GOPRO_USB_VENDOR_ID = "2672" +last_synced_fingerprint = "" + +def format_bytes(size_bytes: int | float) -> str: + if size_bytes <= 0: + return "0 B" + units = ["B", "KB", "MB", "GB", "TB"] + unit_idx = 0 + val = float(size_bytes) + while val >= 1024.0 and unit_idx < len(units) - 1: + val /= 1024.0 + unit_idx += 1 + return f"{val:.2f} {units[unit_idx]}" + +def load_history() -> set[str]: + if not HISTORY_FILE.exists(): + return set() + try: + with open(HISTORY_FILE, "r") as f: + return {line.strip() for line in f if line.strip()} + except Exception as e: + logger.warning(f"Could not read history file: {e}") + return set() + +def record_history(filenames: list[str] | set[str]): + try: + with open(HISTORY_FILE, "a") as f: + for fn in filenames: + f.write(f"{fn}\n") + except Exception as e: + logger.error(f"Failed to update history file: {e}") + +def get_source_creation_time(filepath: Path) -> str | None: + """Extract creation_time string from the original raw GoPro file.""" + cmd = [ + "ffprobe", "-v", "quiet", + "-print_format", "json", + "-show_entries", "format_tags=creation_time:stream_tags=creation_time", + str(filepath) + ] + try: + res = subprocess.run(cmd, capture_output=True, text=True, check=True) + import json + data = json.loads(res.stdout) + c_time = data.get("format", {}).get("tags", {}).get("creation_time") + if not c_time: + for s in data.get("streams", []): + c_time = s.get("tags", {}).get("creation_time") + if c_time: + break + return c_time + except Exception: + return None + +def is_carrier_ready(iface: str) -> bool: + carrier_file = Path(f"/sys/class/net/{iface}/carrier") + try: + return carrier_file.read_text().strip() == "1" + except (FileNotFoundError, OSError): + return False + +def auto_configure_gopro_interfaces(): + net_path = Path("/sys/class/net") + if not net_path.exists(): + return + + for iface_dir in net_path.iterdir(): + iface = iface_dir.name + if iface.startswith(("lo", "docker", "br", "veth", "tun", "wg")): + continue + + dev_path = iface_dir / "device" + if not dev_path.exists(): + continue + + curr = dev_path.resolve() + is_gopro = False + for _ in range(4): + vendor_file = curr / "idVendor" + if vendor_file.exists(): + try: + if vendor_file.read_text().strip().lower() == GOPRO_USB_VENDOR_ID: + is_gopro = True + break + except Exception: + pass + curr = curr.parent + + if is_gopro: + try: + addr_output = subprocess.check_output(["ip", "-4", "addr", "show", "dev", iface], text=True) + if "inet " in addr_output: + continue # Already has an IP address + + # Bring link up administratively + subprocess.run(["ip", "link", "set", iface, "up"], check=False) + + # Wait up to 5 seconds for the hardware carrier to settle + carrier_detected = False + for _ in range(10): + time.sleep(0.5) + if is_carrier_ready(iface): + carrier_detected = True + break + + if not carrier_detected: + logger.debug(f"GoPro interface '{iface}' present, but has NO-CARRIER. Is camera turned on and set to GoPro Connect?") + continue + + logger.info(f"Carrier active on '{iface}'. Requesting DHCP lease...") + # udhcpc in Alpine exits cleanly and does not lock sockets + if shutil.which("udhcpc"): + subprocess.run(["udhcpc", "-i", iface, "-n", "-q", "-t", "4"], check=False) + elif shutil.which("dhcpcd"): + subprocess.run(["dhcpcd", "-1", "-4", "-q", "-t", "10", iface], check=False) + + except Exception as e: + logger.debug(f"Interface configuration error on {iface}: {e}") + +def detect_gopro_ip() -> str | None: + candidates = set() + try: + out = subprocess.check_output(["ip", "-4", "addr", "show"], text=True) + subnets = re.findall(r'inet (172\.(?:2[0-9]|3[0-1])\.\d+)\.\d+/\d+', out) + for subnet in subnets: + candidates.add(f"{subnet}.51") + except Exception: + pass + + try: + out = subprocess.check_output(["ip", "-4", "route"], text=True) + subnets = re.findall(r'(172\.(?:2[0-9]|3[0-1])\.\d+)\.0/\d+', out) + for subnet in subnets: + candidates.add(f"{subnet}.51") + except Exception: + pass + + for ip in candidates: + try: + req = urllib.request.Request(f"http://{ip}:8080/gopro/camera/state", method="GET") + with urllib.request.urlopen(req, timeout=1.0) as resp: + if resp.status == 200: + return ip + except Exception: + continue + return None + +def extract_media_files(media_data) -> list[str]: + files = [] + if isinstance(media_data, dict): + for group in media_data.get("media", []): + dir_name = group.get("directory", group.get("d", "")) + entries = group.get("file_system") or group.get("fs") or group.get("files") or [] + for entry in entries: + if isinstance(entry, dict): + fn = entry.get("filename") or entry.get("n") + if fn: + files.append(f"{dir_name}/{fn}" if (dir_name and "/" not in fn) else fn) + elif isinstance(entry, str): + files.append(f"{dir_name}/{entry}" if dir_name else entry) + elif hasattr(media_data, "files"): + for f in media_data.files: + files.append(f.filename if hasattr(f, "filename") else str(f)) + return files + +def parse_gopro_filename(filename: str) -> tuple[str, int] | None: + m = re.match(r'^(GX|GH)(\d{2})(\d{4})\.MP4$', filename, re.IGNORECASE) + if m: + prefix, chapter, take_num = m.group(1).upper(), int(m.group(2)), m.group(3) + return (f"{prefix}_{take_num}", chapter) + + m = re.match(r'^GOPR(\d{4})\.MP4$', filename, re.IGNORECASE) + if m: + return (f"GOPR_{m.group(1)}", 1) + + m = re.match(r'^GP(\d{2})(\d{4})\.MP4$', filename, re.IGNORECASE) + if m: + return (f"GOPR_{m.group(2)}", int(m.group(1)) + 1) + + return None + +def is_file_already_processed(cam_path: str, history: set[str]) -> bool: + file_name = Path(cam_path).name + if file_name in history or cam_path in history: + return True + + parsed = parse_gopro_filename(file_name) + if parsed: + take_id, _ = parsed + if (DEST_DIR / f"{take_id}.mp4").exists(): + return True + + if (TEMP_DIR / file_name).exists() or (DEST_DIR / file_name).exists(): + return True + + return False + +async def download_file_async(gopro_ip: str, cam_path: str, temp_file: Path, max_retries: int = 3) -> bool: + file_name = Path(cam_path).name + url = f"http://{gopro_ip}:8080/videos/DCIM/{cam_path}" + # Increase socket read timeout to prevent disconnects during SD card lag + timeout = aiohttp.ClientTimeout(total=None, sock_connect=15, sock_read=120) + + for attempt in range(1, max_retries + 1): + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.get(url) as resp: + if resp.status != 200: + logger.error(f"Failed to fetch {cam_path}: HTTP {resp.status} (attempt {attempt}/{max_retries})") + if attempt < max_retries: + await asyncio.sleep(2 * attempt) + continue + return False + + total_bytes = int(resp.headers.get("Content-Length", 0)) + size_label = f" ({format_bytes(total_bytes)})" if total_bytes > 0 else "" + logger.info(f"Downloading {file_name}{size_label} [Attempt {attempt}/{max_retries}]...") + + start_time = time.time() + last_log_time = start_time + last_log_bytes = 0 + downloaded = 0 + + with open(temp_file, "wb") as f: + async for chunk in resp.content.iter_chunked(1024 * 1024): + f.write(chunk) + downloaded += len(chunk) + + now = time.time() + if now - last_log_time >= LOG_PROGRESS_INTERVAL: + delta_t = now - last_log_time + delta_b = downloaded - last_log_bytes + speed = (delta_b / delta_t) if delta_t > 0 else 0 + speed_str = f"{format_bytes(speed)}/s" + + if total_bytes > 0: + pct = (downloaded / total_bytes) * 100 + logger.info( + f"Progress [{file_name}]: {pct:.1f}% " + f"({format_bytes(downloaded)} / {format_bytes(total_bytes)}) @ {speed_str}" + ) + else: + logger.info(f"Progress [{file_name}]: {format_bytes(downloaded)} @ {speed_str}") + + last_log_time = now + last_log_bytes = downloaded + + # Verify received bytes match Content-Length header if provided + if total_bytes > 0 and downloaded < total_bytes: + raise aiohttp.ClientPayloadError(f"Incomplete download: got {downloaded}/{total_bytes} bytes") + + elapsed = max(time.time() - start_time, 0.1) + avg_speed = downloaded / elapsed + logger.info(f"Completed {file_name}: {format_bytes(downloaded)} in {elapsed:.1f}s ({format_bytes(avg_speed)}/s)") + return True + + except Exception as err: + logger.warning(f"Error transferring {cam_path} (attempt {attempt}/{max_retries}): {err}") + if temp_file.exists(): + temp_file.unlink(missing_ok=True) + if attempt < max_retries: + await asyncio.sleep(3 * attempt) + else: + logger.error(f"Exhausted all retries for {cam_path}") + return False + + return False + +async def process_and_finalize_videos(all_camera_files: list[str]): + TEMP_DIR.mkdir(parents=True, exist_ok=True) + + # 1. Map what chapters exist in the camera inventory + camera_takes = defaultdict(set) + for p in all_camera_files: + fn = Path(p).name + parsed = parse_gopro_filename(fn) + if parsed: + take_id, chapter = parsed + camera_takes[take_id].add(chapter) + + # 2. Map what staged chapters we have on disk in TEMP_DIR + staged_takes = defaultdict(dict) + for p in TEMP_DIR.glob("*.[mM][pP]4"): + parsed = parse_gopro_filename(p.name) + if parsed: + take_id, chapter = parsed + staged_takes[take_id][chapter] = p + + for take_id, chapters in staged_takes.items(): + expected_chapters = camera_takes.get(take_id, set()) + staged_chapters = set(chapters.keys()) + + # If any chapter for this take is still missing from TEMP_DIR, wait for future run + if expected_chapters and not expected_chapters.issubset(staged_chapters): + missing = sorted(list(expected_chapters - staged_chapters)) + logger.info(f"Skipping finalize for '{take_id}': waiting for missing chapter(s) {missing}") + continue + + sorted_chapters = [chapters[k] for k in sorted(chapters.keys())] + output_file = DEST_DIR / f"{take_id}.mp4" + temp_output_file = DEST_DIR / f".{take_id}.tmp.mp4" + + if output_file.exists(): + continue + + is_hevc = take_id.startswith("GX") + tag_args = ["-tag:v", "hvc1"] if is_hevc else [] + primary_source_file = sorted_chapters[0] + + creation_time = get_source_creation_time(primary_source_file) + meta_args = ["-metadata", f"creation_time={creation_time}"] if creation_time else [] + + if len(sorted_chapters) == 1: + logger.info(f"Finalizing single-take '{take_id}' -> {output_file.name}...") + cmd = [ + "ffmpeg", "-y", + "-i", str(primary_source_file), + "-map", "0:v:0", + "-map", "0:a?", + *meta_args, + "-c", "copy", + "-sn", "-dn", + *tag_args, + "-movflags", "+faststart", + str(temp_output_file) + ] + else: + logger.info(f"Stitching {len(sorted_chapters)} chapters for '{take_id}' -> {output_file.name} (Timestamp: {creation_time})...") + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as concat_list: + for ch in sorted_chapters: + concat_list.write(f"file '{ch.resolve()}'\n") + concat_path = concat_list.name + + cmd = [ + "ffmpeg", "-y", + "-f", "concat", + "-safe", "0", + "-i", concat_path, + "-map", "0:v:0", + "-map", "0:a?", + *meta_args, + "-c", "copy", + "-sn", "-dn", + *tag_args, + "-movflags", "+faststart", + str(temp_output_file) + ] + + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE + ) + _, stderr = await proc.communicate() + + if proc.returncode == 0 and temp_output_file.exists(): + temp_output_file.rename(output_file) + src_stat = primary_source_file.stat() + os.utime(output_file, (src_stat.st_atime, src_stat.st_mtime)) + + logger.info(f"Ready for upload: {output_file.name} ({format_bytes(output_file.stat().st_size)})") + + raw_names = [ch.name for ch in sorted_chapters] + record_history(raw_names) + + if not KEEP_RAW_CHAPTERS: + logger.info(f"Cleaning raw staging files for '{take_id}' from temp...") + for ch in sorted_chapters: + ch.unlink(missing_ok=True) + else: + logger.error(f"FFmpeg processing failed for {take_id}: {stderr.decode()}") + if temp_output_file.exists(): + temp_output_file.unlink(missing_ok=True) + finally: + if 'concat_path' in locals() and Path(concat_path).exists(): + Path(concat_path).unlink(missing_ok=True) + +async def sync_gopro(gopro_ip: str): + global last_synced_fingerprint + DEST_DIR.mkdir(parents=True, exist_ok=True) + TEMP_DIR.mkdir(parents=True, exist_ok=True) + + try: + async with WiredGoPro(ip=gopro_ip) as gopro: + resp = await gopro.http_command.get_media_list() + if not resp.ok: + return + + all_files = extract_media_files(resp.data) + if not all_files: + return + + history = load_history() + pending_files = [f for f in all_files if not is_file_already_processed(f, history)] + current_fingerprint = f"{gopro_ip}:{len(all_files)}:{len(pending_files)}" + + if not pending_files: + if last_synced_fingerprint != current_fingerprint: + logger.info(f"GoPro connected ({len(all_files)} files found). All media up to date. Idle...") + last_synced_fingerprint = current_fingerprint + return + + logger.info(f"GoPro connected: {len(pending_files)} new file(s) to ingest (Total on SD: {len(all_files)}).") + + ingested_files_this_run = [] + for cam_path in pending_files: + file_name = Path(cam_path).name + + if file_name.upper().endswith(".MP4"): + target_file = TEMP_DIR / file_name + temp_file = TEMP_DIR / f".{file_name}.tmp" + else: + target_file = DEST_DIR / file_name + temp_file = DEST_DIR / f".{file_name}.tmp" + + try: + success = await download_file_async( + gopro_ip=gopro_ip, + cam_path=cam_path, + temp_file=temp_file + ) + + if success and temp_file.exists(): + temp_file.rename(target_file) + logger.info(f"Ingested to staging: {file_name}") + ingested_files_this_run.append(file_name) + + if not file_name.upper().endswith(".MP4"): + record_history([file_name]) + + if DELETE_AFTER_INGEST: + logger.info(f"Deleting {cam_path} from camera SD card...") + await gopro.http_command.delete_file(path=cam_path) + else: + logger.error(f"Failed to ingest: {cam_path}") + if temp_file.exists(): + temp_file.unlink() + + except Exception as err: + logger.error(f"Error transferring {cam_path}: {err}") + if temp_file.exists(): + temp_file.unlink() + + logger.info("Checking staged media for completed takes...") + await process_and_finalize_videos(all_files) + + logger.info("All media finalized in root directory. Camera can be safely unplugged.") + last_synced_fingerprint = f"{gopro_ip}:{len(all_files)}:0" + + except (asyncio.TimeoutError, ConnectionRefusedError, OSError): + last_synced_fingerprint = "" + except Exception as e: + logger.debug(f"Sync event: {e}") + +async def main(): + global last_synced_fingerprint + logger.info(f"GoPro Ingest Daemon active. Target folder: '{DEST_DIR}'") + while True: + auto_configure_gopro_interfaces() + gopro_ip = detect_gopro_ip() + if gopro_ip: + await sync_gopro(gopro_ip) + else: + last_synced_fingerprint = "" + await asyncio.sleep(POLL_INTERVAL) + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + logger.info("Ingest daemon stopped.") \ No newline at end of file