Spaces:
Configuration error
Configuration error
File size: 7,357 Bytes
3624d0b 2206380 3624d0b 2206380 3624d0b 2206380 3624d0b 2206380 3624d0b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 | """Render native C++ decoded lanes in a separate, untimed video pass."""
import argparse
from collections import deque
import json
import os
import subprocess
import sys
from pathlib import Path
import cv2
import imageio_ffmpeg
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from decode import Lane
from bev import BevRange, CameraCalibration
from test_video_bev_onnx import draw_bev, make_composite
from test_video_onnx import draw_predictions
def parse_args():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--video", required=True)
parser.add_argument("--results", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--benchmark-report", default=None)
return parser.parse_args()
def load_lane(payload):
lane = Lane(800, 320)
lane.points = np.asarray(payload["points"], dtype=np.float32)
lane._score_sum = float(payload["score"]) * len(lane.points)
lane.lane_id = int(payload["lane_id"])
lane.lane_role = payload["role"]
lane.is_ego_boundary = lane.lane_role in ("ego_left", "ego_right")
lane.lateral_rank = None
return lane
def load_bev_result(payload):
fit_payload = payload["fit"]
fit = None
if payload["fit_accepted"] and fit_payload is not None:
fit = {
"coefficients": np.asarray(
fit_payload["coefficients"], dtype=np.float64
),
"x_min": float(fit_payload["x_min"]),
"x_max": float(fit_payload["x_max"]),
"rmse": float(fit_payload["rmse"]),
"point_count": int(fit_payload["point_count"]),
"inlier_count": int(fit_payload["inlier_count"]),
}
lane_id = int(payload["lane_id"])
record = {
"lane_id": f"P{lane_id}",
"lane_index": lane_id,
"role": payload["role"],
"score": float(payload["score"]),
"valid_fit": fit is not None,
"parallel_repair_applied": bool(
payload.get("parallel_repaired", False)
),
"synthetic_bev_lane": bool(payload.get("synthetic", False)),
}
if fit_payload is not None:
record["rmse_m"] = float(fit_payload["rmse"])
points = np.asarray(payload["points"], dtype=np.float64)
return {
"record": record,
"points": points[:, :2] if points.size else np.empty((0, 2)),
"fit": fit,
}
def funnel_report(frame_payload, bev_payloads):
topology = frame_payload.get("bev_topology", {})
mode = frame_payload.get("bev_mode", topology.get("mode", "raw"))
return {
"mode": mode,
"parallel_assumption": bool(
topology.get("parallel_assumption", mode != "raw")
),
"synthetic_lanes": any(
item.get("synthetic", False) for item in bev_payloads
),
"parallel_repair_applied": bool(topology.get("applied", False)),
"parallel_repair_forced": bool(topology.get("forced", False)),
"parallel_reference_lane": topology.get("reference_lane"),
"trigger_pairs": topology.get("trigger_pairs", []),
"clipped_lanes": [
f"P{item['lane_id']}" for item in bev_payloads
if item["funnel_clipped"]
],
"rejected_lanes": [
f"P{item['lane_id']}" for item in bev_payloads
if item["fit"] is not None and not item["fit_accepted"]
],
}
def main():
args = parse_args()
video_path = Path(args.video).expanduser().resolve()
results_path = Path(args.results).expanduser().resolve()
output_path = Path(args.output).expanduser().resolve()
output_path.parent.mkdir(parents=True, exist_ok=True)
calibration = CameraCalibration()
bev_range = BevRange()
capture = cv2.VideoCapture(str(video_path))
if not capture.isOpened():
raise RuntimeError(f"cannot open video: {video_path}")
fps = float(capture.get(cv2.CAP_PROP_FPS))
width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
temporary = output_path.with_name(output_path.stem + ".mp4v.mp4")
writer = cv2.VideoWriter(
str(temporary), cv2.VideoWriter_fourcc(*"mp4v"), fps,
(width, height),
)
if not writer.isOpened():
capture.release()
raise RuntimeError(f"cannot create temporary video: {temporary}")
rendered = 0
rolling_core_ms = deque(maxlen=20)
try:
with results_path.open() as results:
for line in results:
payload = json.loads(line)
ok, frame = capture.read()
if not ok:
raise RuntimeError(
f"video ended before result frame {payload['frame_index']}"
)
lanes = [load_lane(item) for item in payload["lanes"]]
bev_payloads = payload["bev_lanes"]
lane_results = [
load_bev_result(item) for item in bev_payloads
]
guard = funnel_report(payload, bev_payloads)
timing = payload["timing"]
rolling_core_ms.append(float(timing["core_ms"]))
rolling_fps = 1000.0 / max(
float(np.mean(rolling_core_ms)), 1e-9
)
draw_predictions(frame, lanes)
bev_canvas = draw_bev(
lane_results, bev_range, calibration, guard
)
composite = make_composite(
frame, bev_canvas, int(payload["frame_index"]),
sum(item["fit"] is not None for item in lane_results),
float(timing["core_ms"]), guard,
result_generation_fps=rolling_fps,
)
detail = (
f"C++ current={timing['core_ms']:.1f}ms | "
f"infer={timing['inference_ms']:.1f} "
f"decode={timing['decode_ms']:.1f} "
f"BEV-result={timing['bev_result_ms']:.2f}ms | "
"draw/encode excluded"
)
cv2.putText(
composite, detail, (664, 101),
cv2.FONT_HERSHEY_SIMPLEX, 0.50, (0, 0, 0), 4,
cv2.LINE_AA,
)
cv2.putText(
composite, detail, (664, 101),
cv2.FONT_HERSHEY_SIMPLEX, 0.50, (255, 255, 255), 1,
cv2.LINE_AA,
)
writer.write(composite)
rendered += 1
if rendered % 200 == 0:
print(f"rendered {rendered} frames", flush=True)
finally:
capture.release()
writer.release()
encoded = output_path.with_name(output_path.stem + ".h264.tmp.mp4")
subprocess.run(
[
imageio_ffmpeg.get_ffmpeg_exe(), "-y", "-loglevel", "error",
"-i", str(temporary), "-c:v", "libx264", "-preset", "fast",
"-crf", "18", "-pix_fmt", "yuv420p", "-movflags", "+faststart",
str(encoded),
],
check=True,
)
os.replace(encoded, output_path)
temporary.unlink(missing_ok=True)
print(f"rendered video: {output_path} ({rendered} frames)")
if __name__ == "__main__":
main()
|