515 lines
20 KiB
Python
515 lines
20 KiB
Python
#!/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.") |