mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-08-21 04:53:47 +08:00
simple clips (#38509)
This commit is contained in:
Adeeb Shihadeh
@@ -4,20 +4,24 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
import itertools
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import queue
|
||||
import random
|
||||
import re
|
||||
import select
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from contextlib import suppress
|
||||
from dataclasses import asdict, dataclass, replace
|
||||
from datetime import datetime
|
||||
from functools import partial, total_ordering
|
||||
from queue import Queue
|
||||
from typing import cast
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Iterable
|
||||
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter, DEFAULT_POOLBLOCK
|
||||
@@ -33,7 +37,9 @@ from openpilot.common.utils import CallbackReader, get_upload_stream
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.common.realtime import set_core_affinity
|
||||
from openpilot.common.hardware import HARDWARE, PC
|
||||
from openpilot.system.loggerd.config import CAMERA_FPS, SEGMENT_LENGTH
|
||||
from openpilot.system.loggerd.xattr_cache import getxattr, setxattr
|
||||
from openpilot.tools.lib.helpers import RE
|
||||
from openpilot.common.swaglog import cloudlog
|
||||
from openpilot.common.version import get_build_metadata
|
||||
from openpilot.common.hardware.hw import Paths
|
||||
@@ -395,6 +401,197 @@ def listDataDirectory(prefix='') -> list[str]:
|
||||
return scan_dir(Paths.log_root(), prefix)
|
||||
|
||||
|
||||
class VideoClips:
|
||||
@dataclass
|
||||
class Clip:
|
||||
route: str
|
||||
camera: str
|
||||
source_start_time: float
|
||||
source_end_time: float
|
||||
bitrate: int
|
||||
speedup: int
|
||||
filename: str
|
||||
requested_at: float
|
||||
|
||||
def __init__(self):
|
||||
self.clip_path = os.path.join(Paths.log_root(), "clips")
|
||||
self.lock = threading.Condition()
|
||||
self.clips: dict[str, VideoClips.Clip] = {}
|
||||
self.transcode_proc: tuple[str, subprocess.Popen] | None = None
|
||||
threading.Thread(target=self._worker, name="video_clip", daemon=True).start()
|
||||
|
||||
def _encode(self, clip: Clip, inputs: Iterable[str], output_path: str, start_time: float, duration: float) -> None:
|
||||
# TODO: use hardware accelerated decoding and encoding
|
||||
command = [
|
||||
"ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin", "-y",
|
||||
"-r", str(CAMERA_FPS * clip.speedup), "-f", "concat", "-safe", "0", "-protocol_whitelist", "file,pipe", "-c:v", "hevc",
|
||||
"-i", "pipe:0", "-ss", str(start_time / clip.speedup), "-t", str(duration / clip.speedup),
|
||||
"-map", "0:v:0", "-an", "-r", str(CAMERA_FPS), "-c:v", "libx264", "-preset", "veryfast",
|
||||
"-b:v", f"{clip.bitrate}M", "-pix_fmt", "yuv420p", "-movflags", "+faststart+use_metadata_tags",
|
||||
"-metadata", f"ai.comma.clip.settings={json.dumps(asdict(clip), separators=(',', ':'))}", output_path,
|
||||
]
|
||||
|
||||
with self.lock:
|
||||
if self.clips.get(clip.filename) is not clip:
|
||||
return
|
||||
process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True)
|
||||
self.transcode_proc = (clip.filename, process)
|
||||
try:
|
||||
if process.stdin is None:
|
||||
raise RuntimeError("ffmpeg stdin is unavailable")
|
||||
process.stdin.write("ffconcat version 1.0\n")
|
||||
for path in inputs:
|
||||
escaped_path = path.replace("'", "'\\''")
|
||||
process.stdin.write(f"file 'file:{escaped_path}'\noption framerate {CAMERA_FPS}\nduration {SEGMENT_LENGTH}\n")
|
||||
process.stdin.close()
|
||||
process.wait()
|
||||
if process.returncode != 0:
|
||||
raise RuntimeError(f"ffmpeg exited with code {process.returncode}")
|
||||
finally:
|
||||
with suppress(OSError):
|
||||
if process.stdin is not None:
|
||||
process.stdin.close()
|
||||
if process.poll() is None:
|
||||
process.terminate()
|
||||
process.wait()
|
||||
with self.lock:
|
||||
if self.transcode_proc is not None and self.transcode_proc[0] == clip.filename:
|
||||
self.transcode_proc = None
|
||||
|
||||
def _worker(self) -> None:
|
||||
while True:
|
||||
with self.lock:
|
||||
while not self.clips:
|
||||
self.lock.wait()
|
||||
clip = next(iter(self.clips.values()))
|
||||
temporary_path = ""
|
||||
try:
|
||||
with self.lock:
|
||||
if self.clips.get(clip.filename) is not clip:
|
||||
continue
|
||||
first_segment = math.floor(clip.source_start_time / SEGMENT_LENGTH)
|
||||
inputs = (
|
||||
os.path.join(Paths.log_root(), f"{clip.route}--{segment}", clip.camera)
|
||||
for segment in range(first_segment, math.ceil(clip.source_end_time / SEGMENT_LENGTH))
|
||||
)
|
||||
os.makedirs(self.clip_path, exist_ok=True)
|
||||
temporary_path = os.path.join(self.clip_path, f".{clip.filename}")
|
||||
output_path = os.path.join(self.clip_path, clip.filename)
|
||||
self._encode(clip, inputs, temporary_path, clip.source_start_time - first_segment * SEGMENT_LENGTH,
|
||||
clip.source_end_time - clip.source_start_time)
|
||||
with self.lock:
|
||||
if self.clips.get(clip.filename) is clip:
|
||||
os.replace(temporary_path, output_path)
|
||||
del self.clips[clip.filename]
|
||||
except Exception:
|
||||
with self.lock:
|
||||
failed = self.clips.get(clip.filename) is clip
|
||||
if failed:
|
||||
del self.clips[clip.filename]
|
||||
if failed:
|
||||
cloudlog.exception("athena.video_clip.failed")
|
||||
finally:
|
||||
with suppress(OSError):
|
||||
if temporary_path:
|
||||
os.unlink(temporary_path)
|
||||
|
||||
def _on_disk(self) -> dict[str, dict]:
|
||||
clips = {}
|
||||
try:
|
||||
entries = os.scandir(self.clip_path)
|
||||
except FileNotFoundError:
|
||||
return clips
|
||||
with entries:
|
||||
for entry in entries:
|
||||
if entry.name.startswith(".") or not entry.is_file():
|
||||
continue
|
||||
probe = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format_tags=ai.comma.clip.settings",
|
||||
"-of", "json", entry.path], capture_output=True, text=True)
|
||||
if probe.returncode != 0:
|
||||
continue
|
||||
try:
|
||||
metadata = json.loads(json.loads(probe.stdout)["format"]["tags"]["ai.comma.clip.settings"])
|
||||
size = entry.stat().st_size
|
||||
except (FileNotFoundError, KeyError, TypeError, json.JSONDecodeError):
|
||||
continue
|
||||
if not isinstance(metadata, dict) or not isinstance(metadata.get("requested_at"), (int, float)):
|
||||
continue
|
||||
clips[entry.name] = {**metadata, "filename": entry.name, "status": "ready",
|
||||
"fn": os.path.relpath(entry.path, Paths.log_root()), "size": size}
|
||||
return clips
|
||||
|
||||
def _available_ranges(self, route: str) -> dict:
|
||||
cameras: dict[str, list[int]] = {}
|
||||
try:
|
||||
with os.scandir(Paths.log_root()) as entries:
|
||||
for entry in entries:
|
||||
entry_route, _, segment = entry.name.rpartition("--")
|
||||
if entry_route != route or not segment.isdigit() or not entry.is_dir():
|
||||
continue
|
||||
with os.scandir(entry.path) as files:
|
||||
for camera in files:
|
||||
if camera.is_file() and camera.name.endswith("camera.hevc"):
|
||||
cameras.setdefault(camera.name, []).append(int(segment))
|
||||
except OSError:
|
||||
return {}
|
||||
|
||||
available = {}
|
||||
for camera, camera_segments in cameras.items():
|
||||
ranges: list[list[int]] = []
|
||||
for segment in sorted(camera_segments):
|
||||
if ranges and ranges[-1][1] == segment * SEGMENT_LENGTH:
|
||||
ranges[-1][1] += SEGMENT_LENGTH
|
||||
else:
|
||||
ranges.append([segment * SEGMENT_LENGTH, (segment + 1) * SEGMENT_LENGTH])
|
||||
available[camera] = {"available_ranges": ranges}
|
||||
return available
|
||||
|
||||
def createClip(self, route: str, source_start_time: float, source_end_time: float, clip: dict):
|
||||
if not PC and not Params().get_bool("IsOffroad"):
|
||||
raise RuntimeError("video clips can only be created while offroad")
|
||||
route_match = re.fullmatch(RE.ROUTE_NAME, route)
|
||||
assert route_match is not None, "invalid route"
|
||||
route_name = route_match.group("log_id")
|
||||
camera = clip["camera"]
|
||||
filename = clip["filename"]
|
||||
assert camera == os.path.basename(camera) and camera.endswith("camera.hevc"), "invalid camera filename"
|
||||
assert filename == os.path.basename(filename), "invalid filename"
|
||||
with self.lock:
|
||||
self.clips[filename] = self.Clip(route_name, camera, source_start_time, source_end_time, clip["bitrate"], clip["speedup"],
|
||||
filename, datetime.now().timestamp())
|
||||
self.lock.notify()
|
||||
|
||||
def getClipState(self, route: str | None = None) -> dict:
|
||||
route_match = re.search(RE.ROUTE_NAME, route or "")
|
||||
with self.lock:
|
||||
transcode_filename = self.transcode_proc[0] if self.transcode_proc is not None else None
|
||||
active_clips = {clip.filename: {**asdict(clip), "status": "encoding" if clip.filename == transcode_filename else "queued"}
|
||||
for clip in self.clips.values()}
|
||||
clips = self._on_disk()
|
||||
clips.update(active_clips)
|
||||
state = {"clips": sorted(clips.values(), key=lambda clip: clip["requested_at"], reverse=True)}
|
||||
if route_match is not None:
|
||||
route_name = route_match.group("log_id")
|
||||
state.update({"route": route_name, "cameras": self._available_ranges(route_name)})
|
||||
return state
|
||||
|
||||
def deleteClip(self, filename: str) -> None:
|
||||
assert filename == os.path.basename(filename), "invalid filename"
|
||||
with self.lock:
|
||||
self.clips.pop(filename, None)
|
||||
output_path = os.path.join(self.clip_path, filename)
|
||||
if self.transcode_proc is not None and self.transcode_proc[0] == filename:
|
||||
self.transcode_proc[1].terminate()
|
||||
if os.path.exists(output_path):
|
||||
os.unlink(output_path)
|
||||
|
||||
|
||||
video_clips = VideoClips()
|
||||
dispatcher.add_method(video_clips.createClip)
|
||||
dispatcher.add_method(video_clips.getClipState)
|
||||
dispatcher.add_method(video_clips.deleteClip)
|
||||
|
||||
|
||||
@dispatcher.add_method
|
||||
def uploadFileToUrl(fn: str, url: str, headers: dict[str, str]) -> UploadFilesToUrlResponse:
|
||||
# this is because mypy doesn't understand that the decorator doesn't change the return type
|
||||
|
||||
Reference in New Issue
Block a user