2026-06-02 17:32:27 +08:00
|
|
|
#!/usr/bin/env python3
|
|
|
|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
"""
|
|
|
|
|
auto_remove_redlight.py
|
|
|
|
|
================================================================
|
|
|
|
|
安全帽第一人稱機車影片 — 自動去紅燈剪輯工具(支援整個資料夾批次處理)
|
|
|
|
|
|
|
|
|
|
功能流程(對「每一支」影片):
|
|
|
|
|
1. 用 OpenCV 每隔 N 秒(預設 1 秒)抽一幀
|
|
|
|
|
2. 裁切出畫面上「GPS 時速表」的固定像素區域 (ROI)
|
|
|
|
|
3. 用 EasyOCR 只辨識數字,讀出當下時速
|
|
|
|
|
4. 當時速 = 0 且「連續超過 4 秒」→ 判定為等紅燈,標記為移除區段
|
|
|
|
|
5. 其餘「時速 > 0 的行駛片段」全部保留
|
|
|
|
|
6. 用 FFmpeg 進行【無損】切割 (-c copy) 與拼接,輸出去紅燈影片
|
|
|
|
|
|
|
|
|
|
批次模式:
|
|
|
|
|
- 輸入可以直接打「資料夾名稱」(例如 20260502),程式會到 BASE_DIR(預設 E:\\videos)
|
|
|
|
|
底下找該資料夾,並把裡面所有影片(01.MOV、02.MOV…)「各自獨立」去紅燈、各輸出一支。
|
|
|
|
|
- 也可以直接給單一影片檔或完整資料夾路徑。
|
|
|
|
|
|
|
|
|
|
設計重點:
|
|
|
|
|
- 偵測座標 (ROI) 與所有門檻參數都集中在最上方的「使用者設定區」,方便修改
|
|
|
|
|
- 提供 --preview 模式: 先輸出一張畫了紅框的預覽圖,確認 ROI 框對位置再正式跑
|
|
|
|
|
- 每支影片都會輸出一份 *_speeds.csv: 紀錄每秒辨識結果,方便檢查 OCR、微調參數
|
|
|
|
|
|
|
|
|
|
需要安裝(測試前請先裝好):
|
|
|
|
|
1. Python 3.x 本體(目前這台電腦的 python 只是 Microsoft Store 捷徑,不是真的 Python)
|
|
|
|
|
2. pip install easyocr opencv-python numpy
|
|
|
|
|
3. FFmpeg / ffprobe,並加入系統 PATH(目前尚未安裝)
|
|
|
|
|
|
|
|
|
|
基本用法:
|
|
|
|
|
# 1) 先確認框選位置(會抽資料夾第一支影片的一幀,畫紅框)
|
|
|
|
|
python auto_remove_redlight.py 20260502 --preview
|
|
|
|
|
|
|
|
|
|
# 2) 確認 ROI 沒問題後,批次處理整個資料夾(每支各自輸出)
|
|
|
|
|
python auto_remove_redlight.py 20260502
|
|
|
|
|
|
|
|
|
|
# 也可以只處理單一檔案
|
|
|
|
|
python auto_remove_redlight.py E:\\videos\\20260502\\01.MOV -o 01_clean.MOV
|
|
|
|
|
|
|
|
|
|
※ 一般使用建議直接執行同資料夾的「啟動.bat」,用問答方式輸入即可,免記指令。
|
|
|
|
|
================================================================
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import argparse
|
|
|
|
|
import csv
|
|
|
|
|
import json
|
|
|
|
|
import os
|
|
|
|
|
import re
|
|
|
|
|
import shutil
|
|
|
|
|
import subprocess
|
|
|
|
|
import sys
|
|
|
|
|
import tempfile
|
|
|
|
|
from dataclasses import dataclass
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
from typing import List, Optional, Tuple
|
|
|
|
|
|
|
|
|
|
import cv2
|
|
|
|
|
import numpy as np
|
|
|
|
|
|
|
|
|
|
# 讓 print 在任何主控台編碼(cp950 / utf-8)下都不會因無法顯示的字元而崩潰
|
|
|
|
|
try:
|
|
|
|
|
sys.stdout.reconfigure(errors="replace")
|
|
|
|
|
sys.stderr.reconfigure(errors="replace")
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
# ============================================================
|
|
|
|
|
# 【使用者設定區】 — 最常修改的參數都集中在這裡
|
|
|
|
|
# ============================================================
|
|
|
|
|
|
|
|
|
|
# --- 影片根目錄 ---------------------------------------------
|
|
|
|
|
# 當你只輸入資料夾名稱(例如 20260502)時,程式會到這個目錄底下找。
|
|
|
|
|
BASE_DIR = r"E:\videos"
|
|
|
|
|
|
|
|
|
|
# --- GPS 時速表在畫面上的像素區域 (ROI) ----------------------
|
|
|
|
|
# 格式: (x, y, w, h)
|
|
|
|
|
# x, y = 區域「左上角」座標 (像素)
|
|
|
|
|
# w, h = 區域的寬、高 (像素)
|
|
|
|
|
# 4K 影片解析度通常為 3840 x 2160。
|
|
|
|
|
# ⚠️ 下面是「範例佔位值」,請務必改成你影片中時速表的實際位置!
|
|
|
|
|
# 小技巧: 先跑 preview(啟動.bat 選 1,或加 --preview),
|
|
|
|
|
# 它會輸出 preview_roi.png(整張畫面+紅框)和 preview_crop.png(裁切結果),
|
|
|
|
|
# 開圖確認紅框剛好框住時速數字,再回來微調這四個數字即可。
|
|
|
|
|
ROI = (3300, 1770, 300, 140) # <--改這裡!(x, y, w, h)
|
|
|
|
|
# 目前值對應 20260516 那批:右下角圓形時速錶內的白色數字
|
|
|
|
|
# (置中數字,避開左側會移動的指針三角形)
|
|
|
|
|
|
|
|
|
|
# --- 偵測邏輯參數 -------------------------------------------
|
|
|
|
|
SAMPLE_INTERVAL = 1.0 # 每隔幾秒辨識一次 (秒)。1.0 = 每秒一次
|
|
|
|
|
STOP_SECONDS = 4.0 # 時速為 0 連續超過幾秒,才算等紅燈 (秒)
|
|
|
|
|
SPEED_THRESHOLD = 0 # 時速 <= 此值視為「停止」。0 = 只有讀到 0 才算停
|
|
|
|
|
# 若想把「龜速 1~2 km/h」也當停止,可改成 1 或 2
|
|
|
|
|
MIN_CONFIDENCE = 0.50 # OCR 信心低於此值的結果視為「未知」(交給平滑補值)
|
|
|
|
|
# 提高到 0.5 可濾掉停車時偶發的低信心雜訊(如 60@0.39)
|
|
|
|
|
MAX_SPEED = 300 # 合理時速上限,超過視為誤判(雜訊),當成未知
|
|
|
|
|
|
|
|
|
|
# --- 平滑濾波(關鍵!解決停車時 OCR 在 0 與雜訊間跳動的問題)----
|
|
|
|
|
# 停車時 OCR 多數讀到 0,但偶爾蹦出單格雜訊(6、60…)或空值,會把「連續停車」
|
|
|
|
|
# 打斷,導致紅燈段沒被完整移除。對每秒速度做「補空值 + 中位數濾波」後再判定:
|
|
|
|
|
# - 孤立的雜訊讀數會被周圍的 0 吃掉(中位數)
|
|
|
|
|
# - 辨識失敗的空值會用前後最近的有效讀數補上
|
|
|
|
|
SMOOTH_WINDOW = 5 # 中位數濾波視窗(取樣點數,奇數)。5 = 看前後各 2 秒
|
2026-06-03 21:07:03 +08:00
|
|
|
DEPART_SECONDS = 8.0 # 起步門檻(秒): 兩段停車中間「持續行駛」需超過此秒數,
|
|
|
|
|
# 才算真正起步而保留;否則(起步 N 秒內又停)視為仍在等待,
|
|
|
|
|
# 連同那段短暫蠕動一起剪掉(處理車陣走走停停)。
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
# --- 剪輯 / 輸出參數 ----------------------------------------
|
2026-06-02 18:46:33 +08:00
|
|
|
# 紅燈移除段的頭尾「不對稱」微調:
|
|
|
|
|
CUT_BEFORE_STOP = 2.0 # 進入紅燈端(速度到 0 那邊): 移除起點再往前幾秒,
|
|
|
|
|
# 連減速進站那段也一起砍掉 (秒)
|
|
|
|
|
KEEP_AFTER_STOP = 2.0 # 綠燈起步端(速度從 0 開始跑): 移除終點提早幾秒,
|
|
|
|
|
# 多留幾秒「卡達起步」的畫面 (秒)
|
2026-06-02 17:32:27 +08:00
|
|
|
MIN_KEEP = 0.8 # 保留片段若短於此秒數就丟棄,避免產生超碎片段 (秒)
|
|
|
|
|
USE_GPU = True # EasyOCR 是否用 GPU。需 CUDA + GPU 版 PyTorch。
|
|
|
|
|
# 設 True 時若沒有 CUDA,EasyOCR 會自動退回 CPU(不會壞)
|
|
|
|
|
REENCODE = False # False = 無損 -c copy(切點對齊關鍵幀,可能 ±1~2 秒)
|
|
|
|
|
# True = 重新編碼(切點精準到幀,但較慢、非無損)
|
2026-06-02 21:36:26 +08:00
|
|
|
VIDEO_QUALITY = 30 # 重新編碼(轉場/REENCODE)畫質。NVENC 用 -rc vbr -cq,
|
|
|
|
|
# 數字越大檔案越小、畫質越低。參考(4K 來源約 54Mbps):
|
|
|
|
|
# 28≈高畫質(~69Mbps,檔大) / 30≈接近原檔 / 33≈省空間(~37Mbps)
|
|
|
|
|
# / 36≈更小。同一數字也用作 x264 -crf(CPU 退回時)。
|
2026-06-02 17:32:27 +08:00
|
|
|
PROGRESS_EVERY = 60 # 進度回報: 每處理幾秒影片印一行目前時速 (秒)。
|
|
|
|
|
# 純顯示用,改小只是印多幾行,不影響速度與偵測精度
|
|
|
|
|
|
2026-06-02 18:59:35 +08:00
|
|
|
# --- 剪接點轉場(淡出淡入 / 恆定功率交叉淡化)------------------
|
|
|
|
|
# 注意: 一旦開啟轉場,就「必須重新編碼」(濾鏡無法用 -c copy),非無損且較慢。
|
|
|
|
|
# 有 NVIDIA 顯卡時會自動用 NVENC 硬體編碼加速 4K。
|
|
|
|
|
TRANSITION = True # 是否在每個剪接點加轉場。False = 維持無損快剪
|
2026-06-03 21:07:03 +08:00
|
|
|
TRANSITION_DURATION = 1.0 # 轉場長度 (秒)。畫面與聲音共用同一長度,確保同步
|
2026-06-02 18:59:35 +08:00
|
|
|
VIDEO_TRANSITION = "fadeblack" # 畫面轉場 (FFmpeg xfade 類型):
|
|
|
|
|
# fadeblack = 淡出到黑再淡入(較接近「淡出淡入」)
|
|
|
|
|
# fade = 交叉溶接(兩段直接互溶)
|
|
|
|
|
# 其它如 dissolve, smoothleft... 見 ffmpeg xfade 文件
|
|
|
|
|
# 聲音固定用 acrossfade 等功率曲線(c1=qsin:c2=qsin)=恆定功率
|
|
|
|
|
|
2026-06-02 19:29:44 +08:00
|
|
|
# --- 音訊處理(命名參考 Adobe Premiere Pro 效果;需重新編碼,即 TRANSITION=true 時生效)---
|
|
|
|
|
VOLUME_BOOST = True # 音量增益: 是否放大音量(對應 PR「增益 / 音量」)
|
|
|
|
|
VOLUME_BOOST_PERCENT = 30 # 音量增加幅度 (%)。30 = 放大成 130% (x1.3 ≈ +2.3dB)
|
|
|
|
|
HARD_LIMITER = True # 強制壓限 (PR「強制壓限 / Hard Limiter」):
|
|
|
|
|
# 拉大音量後把峰值壓在 0dB 以下,避免破音(擺在最後當煞車)
|
|
|
|
|
LOWPASS = True # 低通 (PR「低通 / Lowpass」): 濾掉高頻刺耳聲(風切/嘶聲)
|
|
|
|
|
LOWPASS_HZ = 15000 # 低通截止頻率 (Hz)。低於此保留,高於此衰減
|
|
|
|
|
|
2026-06-02 17:32:27 +08:00
|
|
|
# --- 批次模式 -----------------------------------------------
|
|
|
|
|
# 會被當成影片來處理的副檔名(小寫比較)
|
|
|
|
|
VIDEO_EXTS = {".mov", ".mp4", ".m4v", ".avi", ".mkv", ".mts", ".m2ts", ".insv"}
|
|
|
|
|
# 輸出檔名後綴與預設輸出子資料夾名稱
|
|
|
|
|
OUTPUT_SUFFIX = "_no_redlight" # 輸出檔名 = <原檔名><此後綴>.<原副檔名>
|
|
|
|
|
OUTPUT_SUBDIR = "no_redlight" # 批次輸出會放到 <資料夾>\no_redlight\ 底下
|
|
|
|
|
|
|
|
|
|
# ============================================================
|
|
|
|
|
# 以下為程式邏輯,一般情況不需更動
|
|
|
|
|
# ============================================================
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
|
|
class Sample:
|
|
|
|
|
"""單一取樣點的辨識結果。"""
|
|
|
|
|
t: float # 時間戳 (秒)
|
|
|
|
|
speed: Optional[int] # 辨識到的時速;None 代表辨識失敗/信心不足
|
|
|
|
|
conf: float # OCR 信心值 (0~1)
|
|
|
|
|
raw: str # OCR 原始文字 (除錯用)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- EasyOCR 讀取器 (延遲載入,避免 --preview 也要等模型) ----------
|
|
|
|
|
|
|
|
|
|
def build_reader(use_gpu: bool):
|
|
|
|
|
"""建立 EasyOCR 讀取器。第一次執行會自動下載模型 (約 100MB)。"""
|
|
|
|
|
import easyocr # 延遲匯入: 載入較慢,且 preview 模式用不到
|
|
|
|
|
print("[INFO] 正在載入 EasyOCR 模型 (第一次會下載,請稍候)...")
|
|
|
|
|
reader = easyocr.Reader(['en'], gpu=use_gpu)
|
|
|
|
|
print("[INFO] EasyOCR 載入完成。")
|
|
|
|
|
return reader
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 影像前處理 ----------
|
|
|
|
|
|
|
|
|
|
def preprocess_roi(roi_bgr: np.ndarray, upscale: int = 3) -> np.ndarray:
|
|
|
|
|
"""
|
|
|
|
|
對 ROI 做前處理以利數字辨識。
|
|
|
|
|
依你的時速表樣式不同,可能需要調整這裡(這是最影響辨識率的地方):
|
|
|
|
|
- 放大: 小數字放大後辨識率較高
|
|
|
|
|
- 二值化 / 反相: 視「亮底暗字」或「暗底亮字」決定是否啟用、是否反相
|
|
|
|
|
"""
|
|
|
|
|
gray = cv2.cvtColor(roi_bgr, cv2.COLOR_BGR2GRAY)
|
|
|
|
|
|
|
|
|
|
# 放大(對小字很有效)
|
|
|
|
|
gray = cv2.resize(gray, None, fx=upscale, fy=upscale,
|
|
|
|
|
interpolation=cv2.INTER_CUBIC)
|
|
|
|
|
|
|
|
|
|
# ↓↓↓ 如果辨識率不佳,試著「取消註解」下面幾行做二值化 ↓↓↓
|
|
|
|
|
# # Otsu 自動二值化:
|
|
|
|
|
# _, gray = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
|
|
|
|
|
# # 若你的時速表是「暗底亮字」,二值化後可能要反相(讓字變黑、底變白):
|
|
|
|
|
# gray = cv2.bitwise_not(gray)
|
|
|
|
|
|
|
|
|
|
return gray
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 單張 ROI → 時速 ----------
|
|
|
|
|
|
|
|
|
|
def detect_speed(reader, roi_bgr: np.ndarray) -> Tuple[Optional[int], float, str]:
|
|
|
|
|
"""
|
|
|
|
|
對一張 ROI 影像辨識時速。
|
|
|
|
|
回傳 (speed, confidence, raw_text);辨識不到數字時 speed = None。
|
|
|
|
|
"""
|
|
|
|
|
proc = preprocess_roi(roi_bgr)
|
|
|
|
|
|
|
|
|
|
# allowlist 限制只辨識 0~9,可大幅降低把背景雜訊誤認成文字的機率
|
|
|
|
|
results = reader.readtext(proc, allowlist='0123456789',
|
|
|
|
|
detail=1, paragraph=False)
|
|
|
|
|
|
|
|
|
|
# results 為 [(bbox, text, conf), ...];挑出「信心最高且能解析成數字」的那個
|
|
|
|
|
best: Optional[Tuple[int, float, str]] = None
|
|
|
|
|
for (_bbox, text, conf) in results:
|
|
|
|
|
digits = ''.join(ch for ch in text if ch.isdigit())
|
|
|
|
|
if not digits:
|
|
|
|
|
continue
|
|
|
|
|
value = int(digits)
|
|
|
|
|
if value > MAX_SPEED: # 超過合理上限,視為誤判
|
|
|
|
|
continue
|
|
|
|
|
if best is None or conf > best[1]:
|
|
|
|
|
best = (value, float(conf), text)
|
|
|
|
|
|
|
|
|
|
if best is None:
|
|
|
|
|
return None, 0.0, ''
|
|
|
|
|
return best
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 逐秒取樣整支影片 ----------
|
|
|
|
|
|
|
|
|
|
def sample_speeds(video_path: str, roi: Tuple[int, int, int, int],
|
|
|
|
|
interval: float, reader,
|
|
|
|
|
debug_dir: Optional[str] = None
|
|
|
|
|
) -> Tuple[List[Sample], float, float]:
|
|
|
|
|
"""
|
|
|
|
|
用 OpenCV 每隔 interval 秒抽一幀並辨識時速。
|
|
|
|
|
回傳 (samples, duration, fps)。
|
|
|
|
|
"""
|
|
|
|
|
cap = cv2.VideoCapture(video_path)
|
|
|
|
|
if not cap.isOpened():
|
|
|
|
|
raise RuntimeError(f"無法開啟影片(OpenCV 可能不支援此編碼): {video_path}")
|
|
|
|
|
|
|
|
|
|
fps = cap.get(cv2.CAP_PROP_FPS) or 60.0
|
|
|
|
|
frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0
|
|
|
|
|
duration = frame_count / fps if frame_count else 0.0
|
|
|
|
|
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
|
|
|
|
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
|
|
|
|
|
|
|
|
|
x, y, w, h = roi
|
|
|
|
|
# 邊界保護: 避免 ROI 超出畫面導致裁切錯誤
|
|
|
|
|
if x < 0 or y < 0 or x + w > width or y + h > height:
|
|
|
|
|
print(f"[WARN] ROI {roi} 超出畫面範圍 {width}x{height},請檢查座標!")
|
|
|
|
|
|
|
|
|
|
print(f"[INFO] 影片資訊: {width}x{height}, {fps:.2f} fps, "
|
|
|
|
|
f"長度 {fmt_ts(duration)} ({duration:.1f}s)")
|
|
|
|
|
print(f"[INFO] 開始逐秒辨識 (每 {interval}s 一次,共約 "
|
|
|
|
|
f"{int(duration / interval) if duration else '?'} 個取樣點)...")
|
|
|
|
|
|
|
|
|
|
samples: List[Sample] = []
|
|
|
|
|
t = 0.0
|
|
|
|
|
while duration == 0.0 or t < duration:
|
|
|
|
|
frame_idx = int(round(t * fps))
|
|
|
|
|
cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
|
|
|
|
|
ok, frame = cap.read()
|
|
|
|
|
if not ok:
|
|
|
|
|
break # 讀到結尾
|
|
|
|
|
|
|
|
|
|
roi_bgr = frame[y:y + h, x:x + w]
|
|
|
|
|
speed, conf, raw = detect_speed(reader, roi_bgr)
|
|
|
|
|
|
|
|
|
|
# 信心不足 → 視為未知 (None),避免把不確定的結果當成「停止」而誤刪畫面
|
|
|
|
|
if speed is not None and conf < MIN_CONFIDENCE:
|
|
|
|
|
speed = None
|
|
|
|
|
|
|
|
|
|
samples.append(Sample(t=t, speed=speed, conf=conf, raw=raw))
|
|
|
|
|
|
|
|
|
|
# 除錯: 把每個取樣點的 ROI 存檔,方便回頭檢查 OCR 為何讀錯
|
|
|
|
|
if debug_dir:
|
|
|
|
|
cv2.imwrite(os.path.join(debug_dir, f"t{int(t):05d}_{raw or 'NA'}.png"),
|
|
|
|
|
roi_bgr)
|
|
|
|
|
|
|
|
|
|
# 進度列 (每 PROGRESS_EVERY 秒印一次;<=0 則不印)
|
|
|
|
|
if PROGRESS_EVERY > 0 and int(t) % PROGRESS_EVERY == 0:
|
|
|
|
|
shown = speed if speed is not None else '??'
|
|
|
|
|
print(f" [{fmt_ts(t)}] 時速={shown}", flush=True)
|
|
|
|
|
|
|
|
|
|
t += interval
|
|
|
|
|
|
|
|
|
|
cap.release()
|
|
|
|
|
print(f"[INFO] 辨識完成,共 {len(samples)} 個取樣點。")
|
|
|
|
|
return samples, (duration or t), fps
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 由取樣結果計算「要保留的片段」 ----------
|
|
|
|
|
|
|
|
|
|
def is_stopped(s: Sample) -> bool:
|
|
|
|
|
"""判斷一個取樣點是否為『停止』(僅供 CSV 顯示用,未平滑)。未知(None)不算停止。"""
|
|
|
|
|
if s.speed is None:
|
|
|
|
|
return False
|
|
|
|
|
return s.speed <= SPEED_THRESHOLD
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def smooth_speeds(samples: List[Sample], window: int) -> List[int]:
|
|
|
|
|
"""
|
2026-06-03 21:07:03 +08:00
|
|
|
對逐秒速度做「讀不到當 0 + 中位數濾波」,壓掉停車時的跳動雜訊。
|
|
|
|
|
1. 讀不到(None/??)一律當作 GPS=0(停止)。停車時 OCR 常讀不到或誤讀成小數字,
|
|
|
|
|
當 0 處理後,孤立的誤讀(如把 0 看成 70)會被中位數吃掉,避免假性「起步」。
|
|
|
|
|
2. 中位數濾波: 每個點取前後 window//2 範圍的中位數,孤立雜訊(單格 6/60/70)會被吃掉。
|
2026-06-02 17:32:27 +08:00
|
|
|
回傳與 samples 等長的整數速度陣列。
|
|
|
|
|
"""
|
|
|
|
|
n = len(samples)
|
|
|
|
|
if n == 0:
|
|
|
|
|
return []
|
|
|
|
|
|
2026-06-03 21:07:03 +08:00
|
|
|
# --- 1) 讀不到(None)一律當 0 ---
|
|
|
|
|
filled: List[int] = [s.speed if s.speed is not None else 0 for s in samples]
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
# --- 2) 中位數濾波 ---
|
|
|
|
|
half = max(0, window // 2)
|
|
|
|
|
smoothed: List[int] = [0] * n
|
|
|
|
|
for i in range(n):
|
|
|
|
|
lo = max(0, i - half)
|
|
|
|
|
hi = min(n, i + half + 1)
|
|
|
|
|
win = sorted(filled[lo:hi])
|
|
|
|
|
smoothed[i] = win[len(win) // 2]
|
|
|
|
|
return smoothed
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def find_keep_intervals(samples: List[Sample], duration: float, interval: float
|
|
|
|
|
) -> Tuple[List[Tuple[float, float]], List[Tuple[float, float]]]:
|
|
|
|
|
"""
|
|
|
|
|
根據取樣結果,找出:
|
|
|
|
|
- removals: 需要移除的紅燈區段 [(start, end), ...]
|
|
|
|
|
- keeps : 需要保留的行駛區段 [(start, end), ...] (= 全片扣掉 removals)
|
|
|
|
|
"""
|
|
|
|
|
n = len(samples)
|
|
|
|
|
# 用平滑後的速度判定停止,避免停車時 OCR 跳動把連續紅燈打斷
|
|
|
|
|
smoothed = smooth_speeds(samples, SMOOTH_WINDOW)
|
|
|
|
|
stopped = [v <= SPEED_THRESHOLD for v in smoothed]
|
|
|
|
|
|
2026-06-02 21:01:01 +08:00
|
|
|
# 1) 先抓出所有「原始連續停止」區段(不論長短)
|
|
|
|
|
raw_runs: List[List[float]] = []
|
2026-06-02 17:32:27 +08:00
|
|
|
i = 0
|
|
|
|
|
while i < n:
|
|
|
|
|
if stopped[i]:
|
|
|
|
|
j = i
|
|
|
|
|
while j + 1 < n and stopped[j + 1]:
|
|
|
|
|
j += 1
|
|
|
|
|
run_start = samples[i].t
|
2026-06-02 21:01:01 +08:00
|
|
|
run_end = min(samples[j].t + interval, duration) # 持續到下個取樣點
|
|
|
|
|
raw_runs.append([run_start, run_end])
|
2026-06-02 17:32:27 +08:00
|
|
|
i = j + 1
|
|
|
|
|
else:
|
|
|
|
|
i += 1
|
|
|
|
|
|
2026-06-03 21:07:03 +08:00
|
|
|
# 2) 合併「中間行駛不超過 DEPART_SECONDS 秒」的相鄰停止段:
|
2026-06-02 21:01:01 +08:00
|
|
|
# 起步後 N 秒內又停 → 視為仍在等待(車陣走走停停),連那段短暫蠕動一起算停止。
|
|
|
|
|
# 只有行駛「超過」DEPART_SECONDS 才算真正起步,該行駛段才會被保留。
|
|
|
|
|
merged_runs: List[List[float]] = []
|
|
|
|
|
for run in raw_runs:
|
|
|
|
|
if merged_runs and (run[0] - merged_runs[-1][1]) <= DEPART_SECONDS:
|
|
|
|
|
merged_runs[-1][1] = run[1] # 行駛 <= 門檻 → 併入前一段停止
|
|
|
|
|
else:
|
|
|
|
|
merged_runs.append(list(run))
|
|
|
|
|
|
2026-06-03 21:07:03 +08:00
|
|
|
# 3) 套用門檻與頭尾偏移 → 產生移除區段(每段附帶移除原因,供 log 使用)
|
|
|
|
|
removals: List[dict] = []
|
2026-06-02 21:01:01 +08:00
|
|
|
for (run_start, run_end) in merged_runs:
|
|
|
|
|
run_dur = run_end - run_start
|
|
|
|
|
at_end = run_end >= duration - 1e-3 # 是否一路停到影片結尾
|
|
|
|
|
|
|
|
|
|
# 列為移除: 達到紅燈門檻,或停到影片結尾(結尾停車一律砍掉)
|
|
|
|
|
if run_dur >= STOP_SECONDS or at_end:
|
|
|
|
|
# 進入紅燈端往前多砍 CUT_BEFORE_STOP 秒(連減速進站一起砍)。
|
|
|
|
|
rstart = max(0.0, run_start - CUT_BEFORE_STOP)
|
|
|
|
|
if at_end:
|
|
|
|
|
# 停到影片結尾: 沒有起步要留 → 砍到底,不保留 KEEP_AFTER_STOP
|
|
|
|
|
rend = run_end
|
2026-06-03 21:07:03 +08:00
|
|
|
reason = f"停車至影片結尾(停止約 {run_dur:.0f} 秒)"
|
2026-06-02 21:01:01 +08:00
|
|
|
else:
|
|
|
|
|
# 一般紅燈: 綠燈起步端提早 KEEP_AFTER_STOP 秒結束(多留卡達起步)
|
|
|
|
|
rend = run_end - KEEP_AFTER_STOP
|
2026-06-03 21:07:03 +08:00
|
|
|
reason = f"等紅燈/停車約 {run_dur:.0f} 秒"
|
2026-06-02 21:01:01 +08:00
|
|
|
if rend > rstart:
|
2026-06-03 21:07:03 +08:00
|
|
|
removals.append({"start": rstart, "end": rend,
|
|
|
|
|
"stop_dur": run_dur, "at_end": at_end,
|
|
|
|
|
"reason": reason})
|
2026-06-02 21:01:01 +08:00
|
|
|
|
|
|
|
|
# 4) 頭尾偏移後若仍有相鄰移除段重疊/相接,合併之
|
2026-06-02 17:32:27 +08:00
|
|
|
if removals:
|
2026-06-03 21:07:03 +08:00
|
|
|
merged: List[dict] = [dict(removals[0])]
|
|
|
|
|
for r in removals[1:]:
|
|
|
|
|
if r["start"] <= merged[-1]["end"]:
|
|
|
|
|
merged[-1]["end"] = max(merged[-1]["end"], r["end"])
|
|
|
|
|
merged[-1]["stop_dur"] += r["stop_dur"]
|
|
|
|
|
merged[-1]["at_end"] = merged[-1]["at_end"] or r["at_end"]
|
|
|
|
|
merged[-1]["reason"] = "等紅燈/停車(多段相連合併)"
|
2026-06-02 17:32:27 +08:00
|
|
|
else:
|
2026-06-03 21:07:03 +08:00
|
|
|
merged.append(dict(r))
|
|
|
|
|
removals = merged
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
# keeps = 全片 [0, duration] 扣掉所有 removals
|
|
|
|
|
keeps: List[Tuple[float, float]] = []
|
|
|
|
|
cursor = 0.0
|
2026-06-03 21:07:03 +08:00
|
|
|
for r in removals:
|
|
|
|
|
rs, re = r["start"], r["end"]
|
2026-06-02 17:32:27 +08:00
|
|
|
if rs > cursor:
|
|
|
|
|
keeps.append((cursor, rs))
|
|
|
|
|
cursor = max(cursor, re)
|
|
|
|
|
if cursor < duration:
|
|
|
|
|
keeps.append((cursor, duration))
|
|
|
|
|
|
|
|
|
|
# 丟棄過短的保留片段
|
|
|
|
|
keeps = [(s, e) for (s, e) in keeps if (e - s) >= MIN_KEEP]
|
|
|
|
|
return keeps, removals
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- FFmpeg: 無損切割 + 拼接 ----------
|
|
|
|
|
|
|
|
|
|
def run_cmd(cmd: List[str]) -> None:
|
|
|
|
|
"""執行外部指令,失敗時印出 stderr 並丟出例外。"""
|
|
|
|
|
proc = subprocess.run(cmd, capture_output=True, text=True)
|
|
|
|
|
if proc.returncode != 0:
|
|
|
|
|
sys.stderr.write("[ERROR] 指令執行失敗:\n " + " ".join(cmd) + "\n")
|
|
|
|
|
sys.stderr.write((proc.stderr or "")[-2000:] + "\n") # 只印最後一段錯誤訊息
|
|
|
|
|
raise RuntimeError("FFmpeg 指令失敗")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def ffmpeg_cut_concat(input_path: str, keeps: List[Tuple[float, float]],
|
|
|
|
|
output_path: str, reencode: bool,
|
|
|
|
|
tmp_parent: Optional[str] = None) -> None:
|
|
|
|
|
"""
|
|
|
|
|
將 keeps 中的每個片段切出來,再拼接成 output_path。
|
|
|
|
|
|
|
|
|
|
無損模式 (reencode=False):
|
|
|
|
|
使用 -c copy 串流複製,完全不重新編碼 → 畫質無損、速度快。
|
|
|
|
|
代價: 切點會對齊到最近的「關鍵幀(keyframe)」,實際起點可能比指定時間稍早,
|
|
|
|
|
因此切點精度約 ±1~2 秒(取決於影片 GOP 長度)。對「去紅燈」用途通常足夠。
|
|
|
|
|
|
|
|
|
|
精準模式 (reencode=True):
|
|
|
|
|
重新編碼 → 切點精準到幀,但會耗時且非無損(此處用 CRF 18 接近視覺無損)。
|
|
|
|
|
|
|
|
|
|
tmp_parent:
|
|
|
|
|
暫存切片要放的「父目錄」。建議放在與輸出同一顆硬碟(例如 E:),
|
|
|
|
|
避免大量 4K 暫存檔塞爆 C: 系統碟。
|
|
|
|
|
"""
|
|
|
|
|
# 暫存檔放在輸出同顆硬碟,避免塞爆系統碟(4K 切片可能很大)
|
|
|
|
|
tmpdir = tempfile.mkdtemp(prefix="redlight_", dir=tmp_parent)
|
|
|
|
|
seg_files: List[str] = []
|
|
|
|
|
try:
|
|
|
|
|
# 1) 逐段切出
|
|
|
|
|
for idx, (start, end) in enumerate(keeps):
|
|
|
|
|
dur = end - start
|
|
|
|
|
ext = os.path.splitext(output_path)[1] or ".mp4"
|
|
|
|
|
seg = os.path.join(tmpdir, f"seg_{idx:04d}{ext}")
|
|
|
|
|
|
|
|
|
|
if reencode:
|
|
|
|
|
cmd = [
|
|
|
|
|
"ffmpeg", "-y",
|
|
|
|
|
"-ss", f"{start:.3f}", "-i", input_path,
|
|
|
|
|
"-t", f"{dur:.3f}",
|
2026-06-02 21:36:26 +08:00
|
|
|
"-c:v", "libx264", "-preset", "medium", "-crf", str(VIDEO_QUALITY),
|
2026-06-02 17:32:27 +08:00
|
|
|
"-c:a", "aac", "-b:a", "192k",
|
|
|
|
|
"-avoid_negative_ts", "make_zero",
|
|
|
|
|
seg,
|
|
|
|
|
]
|
|
|
|
|
else:
|
|
|
|
|
cmd = [
|
|
|
|
|
"ffmpeg", "-y",
|
|
|
|
|
"-ss", f"{start:.3f}", "-i", input_path,
|
|
|
|
|
"-t", f"{dur:.3f}",
|
|
|
|
|
"-c", "copy",
|
|
|
|
|
"-avoid_negative_ts", "make_zero",
|
|
|
|
|
seg,
|
|
|
|
|
]
|
|
|
|
|
print(f" 切割片段 {idx + 1}/{len(keeps)}: "
|
|
|
|
|
f"{fmt_ts(start)} ~ {fmt_ts(end)} ({dur:.1f}s)")
|
|
|
|
|
run_cmd(cmd)
|
|
|
|
|
seg_files.append(seg)
|
|
|
|
|
|
|
|
|
|
# 2) 寫 concat 清單 (concat demuxer 要求每行: file '路徑')
|
|
|
|
|
# Windows 路徑的反斜線在清單檔中容易出問題,統一改成正斜線
|
|
|
|
|
listfile = os.path.join(tmpdir, "concat_list.txt")
|
|
|
|
|
with open(listfile, "w", encoding="utf-8") as f:
|
|
|
|
|
for seg in seg_files:
|
|
|
|
|
f.write(f"file '{seg.replace(os.sep, '/')}'\n")
|
|
|
|
|
|
|
|
|
|
# 3) 無損拼接 (-c copy)
|
|
|
|
|
print(" 正在拼接所有片段...")
|
|
|
|
|
cmd = [
|
|
|
|
|
"ffmpeg", "-y",
|
|
|
|
|
"-f", "concat", "-safe", "0",
|
|
|
|
|
"-i", listfile,
|
|
|
|
|
"-c", "copy",
|
|
|
|
|
output_path,
|
|
|
|
|
]
|
|
|
|
|
run_cmd(cmd)
|
|
|
|
|
print(f"[INFO] 完成! 輸出檔案: {output_path}")
|
|
|
|
|
|
|
|
|
|
finally:
|
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True) # 清掉暫存檔
|
|
|
|
|
|
|
|
|
|
|
2026-06-02 19:29:44 +08:00
|
|
|
def _audio_filter_chain() -> List[str]:
|
|
|
|
|
"""
|
|
|
|
|
依設定組音訊濾鏡鏈(對應 Adobe PR 的 音量 / 強制壓限 / 低通)。
|
|
|
|
|
回傳 ffmpeg 濾鏡片段清單,順序為: 音量增益 → 低通 → 強制壓限(壓限擺最後當煞車)。
|
|
|
|
|
"""
|
|
|
|
|
chain: List[str] = []
|
|
|
|
|
if VOLUME_BOOST and abs(VOLUME_BOOST_PERCENT) > 0:
|
|
|
|
|
mult = 1.0 + VOLUME_BOOST_PERCENT / 100.0 # +30% → x1.3
|
|
|
|
|
chain.append(f"volume={mult:.4f}") # 音量增益
|
|
|
|
|
if LOWPASS:
|
|
|
|
|
chain.append(f"lowpass=f={int(LOWPASS_HZ)}") # 低通 (Lowpass)
|
|
|
|
|
if HARD_LIMITER:
|
2026-06-02 19:49:30 +08:00
|
|
|
# 強制壓限 (Hard Limiter): limit=0.95 → 峰值天花板約 -0.45dBFS;
|
|
|
|
|
# level=disabled 關掉「壓限後自動回補到 0dB」,才能真正留下 headroom 防破音
|
|
|
|
|
chain.append("alimiter=limit=0.95:level=disabled")
|
2026-06-02 19:29:44 +08:00
|
|
|
return chain
|
|
|
|
|
|
|
|
|
|
|
2026-06-02 18:59:35 +08:00
|
|
|
def has_audio(path: str) -> bool:
|
|
|
|
|
"""用 ffprobe 判斷影片是否有音訊軌。出錯時保守假設『有』。"""
|
|
|
|
|
if shutil.which("ffprobe") is None:
|
|
|
|
|
return True
|
|
|
|
|
try:
|
|
|
|
|
out = subprocess.run(
|
|
|
|
|
["ffprobe", "-v", "error", "-select_streams", "a",
|
|
|
|
|
"-show_entries", "stream=index", "-of", "csv=p=0", path],
|
|
|
|
|
capture_output=True, text=True)
|
|
|
|
|
return bool(out.stdout.strip())
|
|
|
|
|
except Exception:
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def ffmpeg_cut_concat_xfade(input_path: str, keeps: List[Tuple[float, float]],
|
|
|
|
|
output_path: str, use_gpu: bool,
|
|
|
|
|
tmp_parent: Optional[str] = None) -> None:
|
|
|
|
|
"""
|
|
|
|
|
把 keeps 各片段接起來,並在「每個剪接點」加上轉場:
|
|
|
|
|
- 畫面: xfade(VIDEO_TRANSITION,例如 fadeblack 淡出淡入)
|
|
|
|
|
- 聲音: acrossfade 等功率曲線(c1=qsin:c2=qsin)= 恆定功率交叉淡化
|
|
|
|
|
兩者重疊長度相同(TRANSITION_DURATION),因此畫面與聲音同步。
|
|
|
|
|
|
|
|
|
|
※ 轉場需要重新編碼(濾鏡無法 -c copy)。有 NVIDIA 顯卡時用 h264_nvenc 硬體加速。
|
|
|
|
|
"""
|
|
|
|
|
n = len(keeps)
|
|
|
|
|
durs = [e - s for (s, e) in keeps]
|
|
|
|
|
|
2026-06-02 19:29:44 +08:00
|
|
|
# 只有一段 → 沒有剪接點,不需轉場
|
2026-06-02 18:59:35 +08:00
|
|
|
if n == 1:
|
|
|
|
|
s, e = keeps[0]
|
2026-06-02 19:29:44 +08:00
|
|
|
af = _audio_filter_chain()
|
|
|
|
|
if af and has_audio(input_path):
|
|
|
|
|
# 仍要做音量處理: 畫面照樣無損複製,只重編音訊
|
|
|
|
|
run_cmd(["ffmpeg", "-y", "-ss", f"{s:.3f}", "-t", f"{e - s:.3f}",
|
|
|
|
|
"-i", input_path, "-af", ",".join(af),
|
|
|
|
|
"-c:v", "copy", "-c:a", "aac", "-b:a", "256k",
|
|
|
|
|
"-avoid_negative_ts", "make_zero", output_path])
|
|
|
|
|
else:
|
|
|
|
|
run_cmd(["ffmpeg", "-y", "-ss", f"{s:.3f}", "-t", f"{e - s:.3f}",
|
|
|
|
|
"-i", input_path, "-c", "copy",
|
|
|
|
|
"-avoid_negative_ts", "make_zero", output_path])
|
2026-06-02 18:59:35 +08:00
|
|
|
print(f"[INFO] 完成(單一片段,免轉場)! 輸出檔案: {output_path}")
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
# 轉場長度不能 >= 任何片段,否則 xfade/acrossfade 會出錯 → 夾到安全值
|
|
|
|
|
d = min(TRANSITION_DURATION, min(durs) - 0.1)
|
|
|
|
|
if d <= 0.05:
|
|
|
|
|
print("[WARN] 片段太短,改用無轉場拼接。")
|
|
|
|
|
ffmpeg_cut_concat(input_path, keeps, output_path, reencode=True,
|
|
|
|
|
tmp_parent=tmp_parent)
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
audio = has_audio(input_path)
|
|
|
|
|
|
|
|
|
|
# 每個保留片段當成一個輸入(同一來源檔多次 -ss/-t 取段)
|
|
|
|
|
cmd: List[str] = ["ffmpeg", "-y"]
|
|
|
|
|
for (s, e) in keeps:
|
|
|
|
|
cmd += ["-ss", f"{s:.3f}", "-t", f"{e - s:.3f}", "-i", input_path]
|
|
|
|
|
|
|
|
|
|
# --- 組 filter_complex ---
|
|
|
|
|
parts: List[str] = []
|
|
|
|
|
|
|
|
|
|
# 畫面: 先把每段正規化(統一時間基準/格式),再逐段 xfade
|
|
|
|
|
for i in range(n):
|
|
|
|
|
parts.append(f"[{i}:v]settb=AVTB,format=yuv420p[v{i}]")
|
|
|
|
|
prev = "v0"
|
|
|
|
|
run_len = durs[0]
|
|
|
|
|
for j in range(1, n):
|
|
|
|
|
offset = run_len - d # 轉場在「目前累積長度 - d」處開始
|
|
|
|
|
out = f"vx{j}"
|
|
|
|
|
parts.append(
|
|
|
|
|
f"[{prev}][v{j}]xfade=transition={VIDEO_TRANSITION}:"
|
|
|
|
|
f"duration={d:.3f}:offset={offset:.3f}[{out}]")
|
|
|
|
|
prev = out
|
|
|
|
|
run_len = run_len + durs[j] - d # xfade 後總長
|
|
|
|
|
vlabel = prev
|
|
|
|
|
|
|
|
|
|
# 聲音: 逐段 acrossfade(等功率 = 恆定功率)
|
|
|
|
|
alabel = None
|
|
|
|
|
if audio:
|
|
|
|
|
prev_a = "0:a"
|
|
|
|
|
for j in range(1, n):
|
|
|
|
|
out = f"ax{j}"
|
|
|
|
|
parts.append(
|
|
|
|
|
f"[{prev_a}][{j}:a]acrossfade=d={d:.3f}:c1=qsin:c2=qsin[{out}]")
|
|
|
|
|
prev_a = out
|
|
|
|
|
alabel = prev_a
|
2026-06-02 19:29:44 +08:00
|
|
|
# 串接完再做音量處理(音量增益 / 低通 / 強制壓限)
|
|
|
|
|
af = _audio_filter_chain()
|
|
|
|
|
if af:
|
|
|
|
|
parts.append(f"[{alabel}]" + ",".join(af) + "[aout]")
|
|
|
|
|
alabel = "aout"
|
2026-06-02 18:59:35 +08:00
|
|
|
|
|
|
|
|
filtergraph = ";".join(parts)
|
|
|
|
|
|
2026-06-02 21:36:26 +08:00
|
|
|
# --- 編碼器: 有 GPU 用 NVENC,否則 libx264;畫質由 VIDEO_QUALITY 控制 ---
|
|
|
|
|
# NVENC 必須搭配 -rc vbr,-cq 才會真正控制位元率/檔案大小(否則位元率會爆高)
|
2026-06-02 18:59:35 +08:00
|
|
|
if use_gpu:
|
2026-06-02 21:36:26 +08:00
|
|
|
venc = ["-c:v", "h264_nvenc", "-preset", "p5",
|
|
|
|
|
"-rc", "vbr", "-cq", str(VIDEO_QUALITY)]
|
2026-06-02 18:59:35 +08:00
|
|
|
else:
|
2026-06-02 21:36:26 +08:00
|
|
|
venc = ["-c:v", "libx264", "-preset", "medium", "-crf", str(VIDEO_QUALITY)]
|
2026-06-02 18:59:35 +08:00
|
|
|
|
|
|
|
|
cmd += ["-filter_complex", filtergraph, "-map", f"[{vlabel}]"]
|
|
|
|
|
if audio:
|
|
|
|
|
cmd += ["-map", f"[{alabel}]", "-c:a", "aac", "-b:a", "256k"]
|
|
|
|
|
else:
|
|
|
|
|
cmd += ["-an"]
|
|
|
|
|
cmd += venc + ["-pix_fmt", "yuv420p", output_path]
|
|
|
|
|
|
|
|
|
|
print(f" 轉場拼接 {n} 段(轉場 {d:.2f}s,畫面={VIDEO_TRANSITION},"
|
|
|
|
|
f"聲音=恆定功率,編碼={'NVENC' if use_gpu else 'x264'})...")
|
|
|
|
|
print(" ※ 4K 重新編碼較久,請耐心等候。")
|
|
|
|
|
run_cmd(cmd)
|
|
|
|
|
print(f"[INFO] 完成! 輸出檔案: {output_path}")
|
|
|
|
|
|
|
|
|
|
|
2026-06-02 17:32:27 +08:00
|
|
|
# ---------- preview: 輸出 ROI 預覽圖 ----------
|
|
|
|
|
|
|
|
|
|
def save_preview(video_path: str, roi: Tuple[int, int, int, int],
|
|
|
|
|
at_seconds: float = 0.0, prefix: str = "preview") -> None:
|
|
|
|
|
"""抽一幀,畫上 ROI 紅框並另存裁切結果,方便確認/微調座標。"""
|
|
|
|
|
cap = cv2.VideoCapture(video_path)
|
|
|
|
|
if not cap.isOpened():
|
|
|
|
|
raise RuntimeError(f"無法開啟影片: {video_path}")
|
|
|
|
|
fps = cap.get(cv2.CAP_PROP_FPS) or 60.0
|
|
|
|
|
cap.set(cv2.CAP_PROP_POS_FRAMES, int(at_seconds * fps))
|
|
|
|
|
ok, frame = cap.read()
|
|
|
|
|
cap.release()
|
|
|
|
|
if not ok:
|
|
|
|
|
raise RuntimeError("無法讀取預覽影格(試試把 --preview-at 設小一點)")
|
|
|
|
|
|
|
|
|
|
x, y, w, h = roi
|
|
|
|
|
vis = frame.copy()
|
|
|
|
|
cv2.rectangle(vis, (x, y), (x + w, y + h), (0, 0, 255), 4) # 紅框
|
|
|
|
|
cv2.imwrite(f"{prefix}_roi.png", vis) # 整張畫面+紅框
|
|
|
|
|
|
|
|
|
|
crop = frame[y:y + h, x:x + w]
|
|
|
|
|
cv2.imwrite(f"{prefix}_crop.png", crop) # 裁切結果
|
|
|
|
|
cv2.imwrite(f"{prefix}_crop_processed.png", preprocess_roi(crop)) # 前處理後
|
|
|
|
|
|
|
|
|
|
print("[INFO] 已輸出預覽圖:")
|
|
|
|
|
print(f" {prefix}_roi.png <--整張畫面,確認紅框是否框住時速表")
|
|
|
|
|
print(f" {prefix}_crop.png <--裁切出的時速表區域")
|
|
|
|
|
print(f" {prefix}_crop_processed.png <--送進 OCR 前的前處理結果")
|
|
|
|
|
print(f" 來源影片 = {video_path}")
|
|
|
|
|
print(f" 目前 ROI = {roi} (x, y, w, h)")
|
|
|
|
|
print(" 若框得不準,請修改檔案最上方的 ROI 變數或用 --roi 參數覆寫。")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 小工具 ----------
|
|
|
|
|
|
|
|
|
|
def fmt_ts(seconds: float) -> str:
|
|
|
|
|
"""秒數 → HH:MM:SS 字串。"""
|
|
|
|
|
seconds = max(0, int(round(seconds)))
|
|
|
|
|
h, rem = divmod(seconds, 3600)
|
|
|
|
|
m, s = divmod(rem, 60)
|
|
|
|
|
return f"{h:02d}:{m:02d}:{s:02d}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def natural_key(name: str):
|
|
|
|
|
"""自然排序鍵: 讓 1,2,...,10 正確排序(雖然 01,02 補零時一般排序也對)。"""
|
|
|
|
|
return [int(t) if t.isdigit() else t.lower()
|
|
|
|
|
for t in re.split(r'(\d+)', name)]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def find_videos(folder: str) -> List[str]:
|
|
|
|
|
"""找出資料夾內所有影片檔(只看第一層,不遞迴),依檔名自然排序。"""
|
|
|
|
|
items = []
|
|
|
|
|
for name in os.listdir(folder):
|
|
|
|
|
full = os.path.join(folder, name)
|
|
|
|
|
if os.path.isfile(full) and os.path.splitext(name)[1].lower() in VIDEO_EXTS:
|
|
|
|
|
items.append(name)
|
|
|
|
|
items.sort(key=natural_key)
|
|
|
|
|
return [os.path.join(folder, n) for n in items]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def list_folders(base_dir: str) -> List[str]:
|
|
|
|
|
"""列出 base_dir 底下第一層的子資料夾,依名稱自然排序。"""
|
|
|
|
|
items = []
|
|
|
|
|
try:
|
|
|
|
|
for name in os.listdir(base_dir):
|
|
|
|
|
full = os.path.join(base_dir, name)
|
|
|
|
|
if os.path.isdir(full):
|
|
|
|
|
items.append(name)
|
|
|
|
|
except FileNotFoundError:
|
|
|
|
|
return []
|
|
|
|
|
items.sort(key=natural_key)
|
|
|
|
|
return [os.path.join(base_dir, n) for n in items]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 互動式鍵盤選單 (Windows;用內建 msvcrt,免裝套件) ----------
|
|
|
|
|
|
|
|
|
|
def _getkey() -> str:
|
|
|
|
|
"""讀一個按鍵並回傳語意字串(UP/DOWN/ENTER/SPACE/ESC 或單一字元)。"""
|
|
|
|
|
import msvcrt
|
|
|
|
|
ch = msvcrt.getwch()
|
|
|
|
|
if ch in ("\x00", "\xe0"): # 方向鍵等特殊鍵的前導碼
|
|
|
|
|
ch2 = msvcrt.getwch()
|
|
|
|
|
return {"H": "UP", "P": "DOWN", "K": "LEFT", "M": "RIGHT"}.get(ch2, "")
|
|
|
|
|
if ch in ("\r", "\n"):
|
|
|
|
|
return "ENTER"
|
|
|
|
|
if ch == " ":
|
|
|
|
|
return "SPACE"
|
|
|
|
|
if ch == "\x1b":
|
|
|
|
|
return "ESC"
|
|
|
|
|
if ch == "\x03": # Ctrl-C
|
|
|
|
|
raise KeyboardInterrupt
|
|
|
|
|
return ch
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def menu_select_one(title: str, labels: List[str]) -> Optional[int]:
|
|
|
|
|
"""單選選單。↑↓ 移動、Enter 或數字鍵選定、Esc 取消。回傳索引或 None。"""
|
|
|
|
|
idx = 0
|
|
|
|
|
n = len(labels)
|
|
|
|
|
hint = "操作: ↑↓ 移動 | Enter 選定 | 數字鍵直接選 | Esc 取消"
|
|
|
|
|
while True:
|
|
|
|
|
os.system("cls")
|
|
|
|
|
print("=" * 60)
|
|
|
|
|
print(title)
|
|
|
|
|
print("=" * 60)
|
|
|
|
|
print(hint + "\n")
|
|
|
|
|
for i, lab in enumerate(labels):
|
|
|
|
|
cursor = ">" if i == idx else " "
|
|
|
|
|
print(f" {cursor} {i + 1:>2}. {lab}")
|
|
|
|
|
key = _getkey()
|
|
|
|
|
if key == "UP":
|
|
|
|
|
idx = (idx - 1) % n
|
|
|
|
|
elif key == "DOWN":
|
|
|
|
|
idx = (idx + 1) % n
|
|
|
|
|
elif key == "ENTER":
|
|
|
|
|
return idx
|
|
|
|
|
elif key == "ESC":
|
|
|
|
|
return None
|
|
|
|
|
elif key.isdigit() and 1 <= int(key) <= n:
|
|
|
|
|
return int(key) - 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def menu_select_many(title: str, labels: List[str]) -> Optional[List[int]]:
|
|
|
|
|
"""
|
|
|
|
|
多選選單。↑↓ 移動、空白鍵勾選/取消、a 全選或全不選、Enter 開始、Esc 取消。
|
|
|
|
|
預設「全部勾選」(整批處理最常見)。回傳已選索引清單或 None。
|
|
|
|
|
"""
|
|
|
|
|
idx = 0
|
|
|
|
|
n = len(labels)
|
|
|
|
|
checked = set(range(n)) # 預設全選
|
|
|
|
|
hint = ("操作: ↑↓ 移動 | 空白鍵 勾選/取消 | a 全選切換 | "
|
|
|
|
|
"數字鍵 勾選/取消 | Enter 開始處理 | Esc 取消")
|
|
|
|
|
while True:
|
|
|
|
|
os.system("cls")
|
|
|
|
|
print("=" * 60)
|
|
|
|
|
print(title)
|
|
|
|
|
print("=" * 60)
|
|
|
|
|
print(hint + "\n")
|
|
|
|
|
for i, lab in enumerate(labels):
|
|
|
|
|
cursor = ">" if i == idx else " "
|
|
|
|
|
box = "[v]" if i in checked else "[ ]"
|
|
|
|
|
print(f" {cursor} {box} {i + 1:>2}. {lab}")
|
|
|
|
|
print(f"\n 已勾選 {len(checked)} / {n} 支")
|
|
|
|
|
key = _getkey()
|
|
|
|
|
if key == "UP":
|
|
|
|
|
idx = (idx - 1) % n
|
|
|
|
|
elif key == "DOWN":
|
|
|
|
|
idx = (idx + 1) % n
|
|
|
|
|
elif key == "SPACE":
|
|
|
|
|
checked.discard(idx) if idx in checked else checked.add(idx)
|
|
|
|
|
elif key in ("a", "A"):
|
|
|
|
|
checked = set() if len(checked) == n else set(range(n))
|
|
|
|
|
elif key.isdigit() and 1 <= int(key) <= n:
|
|
|
|
|
j = int(key) - 1
|
|
|
|
|
checked.discard(j) if j in checked else checked.add(j)
|
|
|
|
|
elif key == "ENTER":
|
|
|
|
|
if checked: # 至少選一支才開始,避免誤按
|
|
|
|
|
return sorted(checked)
|
|
|
|
|
elif key == "ESC":
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def run_selector(base_dir: str, out_list: str) -> int:
|
|
|
|
|
"""
|
|
|
|
|
互動選擇器(給 啟動.bat 呼叫):
|
|
|
|
|
1. 掃描 base_dir 列出資料夾 → 單選
|
|
|
|
|
2. 掃描該資料夾影片 → 空白鍵打勾多選
|
|
|
|
|
把選中的影片「絕對路徑」逐行寫到 out_list(UTF-8)。
|
|
|
|
|
取消或沒選 → 不產生 out_list,回傳非 0,bat 端據此中止。
|
|
|
|
|
回傳 0 表示成功寫出清單。
|
|
|
|
|
"""
|
|
|
|
|
# 只列出「有影片」的資料夾(略過 redlight_remover 等沒有影片的)
|
|
|
|
|
folders = [(f, vs) for f in list_folders(base_dir) if (vs := find_videos(f))]
|
|
|
|
|
if not folders:
|
|
|
|
|
print(f"[ERROR] 在 {base_dir} 底下找不到含影片的資料夾。")
|
|
|
|
|
return 2
|
|
|
|
|
|
|
|
|
|
flabels = [f"{os.path.basename(f)} ({len(vs)} 支影片)" for f, vs in folders]
|
|
|
|
|
sel = menu_select_one(f"選擇要處理的資料夾 (根目錄: {base_dir})", flabels)
|
|
|
|
|
if sel is None:
|
|
|
|
|
print("已取消。")
|
|
|
|
|
return 1
|
|
|
|
|
folder, allvids = folders[sel] # allvids 已在上面篩出,不必重抓
|
|
|
|
|
|
|
|
|
|
vlabels = []
|
|
|
|
|
for v in allvids:
|
|
|
|
|
try:
|
|
|
|
|
gb = os.path.getsize(v) / (1024 ** 3)
|
|
|
|
|
vlabels.append(f"{os.path.basename(v)} ({gb:.1f} GB)")
|
|
|
|
|
except OSError:
|
|
|
|
|
vlabels.append(os.path.basename(v))
|
|
|
|
|
picks = menu_select_many(
|
|
|
|
|
f"勾選要去紅燈的影片 (資料夾: {os.path.basename(folder)})", vlabels)
|
|
|
|
|
if picks is None:
|
|
|
|
|
print("已取消。")
|
|
|
|
|
return 1
|
|
|
|
|
|
|
|
|
|
chosen = [allvids[i] for i in picks]
|
|
|
|
|
with open(out_list, "w", encoding="utf-8") as f:
|
|
|
|
|
for p in chosen:
|
|
|
|
|
f.write(os.path.abspath(p) + "\n")
|
|
|
|
|
|
|
|
|
|
os.system("cls")
|
|
|
|
|
print(f"已選擇 {len(chosen)} 支影片(資料夾 {os.path.basename(folder)}):")
|
|
|
|
|
for p in chosen:
|
|
|
|
|
print(f" - {os.path.basename(p)}")
|
|
|
|
|
return 0
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def resolve_input(arg: str, base_dir: str) -> str:
|
|
|
|
|
"""
|
|
|
|
|
把使用者輸入解析成實際路徑:
|
|
|
|
|
1. 若 arg 本身就是存在的檔案或資料夾 → 直接用
|
|
|
|
|
2. 否則嘗試 base_dir\\arg(例如只打 20260502)
|
|
|
|
|
回傳實際存在的路徑;找不到則丟出錯誤。
|
|
|
|
|
"""
|
|
|
|
|
if os.path.exists(arg):
|
|
|
|
|
return arg
|
|
|
|
|
candidate = os.path.join(base_dir, arg)
|
|
|
|
|
if os.path.exists(candidate):
|
|
|
|
|
return candidate
|
|
|
|
|
raise FileNotFoundError(
|
|
|
|
|
f"找不到輸入: '{arg}'(也試過 '{candidate}')")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def write_csv(samples: List[Sample], path: str) -> None:
|
|
|
|
|
"""把每秒辨識結果寫成 CSV,方便檢查 OCR 準不準。"""
|
|
|
|
|
with open(path, "w", newline="", encoding="utf-8-sig") as f:
|
|
|
|
|
writer = csv.writer(f)
|
|
|
|
|
writer.writerow(["time_s", "time_hms", "speed", "confidence",
|
|
|
|
|
"raw_text", "stopped"])
|
|
|
|
|
for s in samples:
|
|
|
|
|
writer.writerow([
|
|
|
|
|
f"{s.t:.1f}", fmt_ts(s.t),
|
|
|
|
|
"" if s.speed is None else s.speed,
|
|
|
|
|
f"{s.conf:.2f}", s.raw, int(is_stopped(s)),
|
|
|
|
|
])
|
|
|
|
|
print(f"[INFO] 已輸出逐秒辨識紀錄: {path}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def check_ffmpeg() -> None:
|
|
|
|
|
"""確認 ffmpeg 可用,否則提早給出清楚的錯誤訊息。"""
|
|
|
|
|
if shutil.which("ffmpeg") is None:
|
|
|
|
|
sys.exit("[ERROR] 找不到 ffmpeg。請先安裝 FFmpeg 並加入系統 PATH,"
|
|
|
|
|
"再重新執行(可用 `ffmpeg -version` 測試是否裝好)。")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 處理單一影片 ----------
|
|
|
|
|
|
|
|
|
|
def process_one(video_path: str, roi: Tuple[int, int, int, int],
|
|
|
|
|
out_dir: str, reader, args,
|
|
|
|
|
out_file: Optional[str] = None) -> dict:
|
|
|
|
|
"""
|
|
|
|
|
對單一影片完整跑一次:辨識 → 算保留區段 → 切割拼接。
|
|
|
|
|
out_file 若指定則用該完整檔名,否則自動命名放到 out_dir(no_redlight 資料夾)。
|
|
|
|
|
回傳一個結果摘要 dict。
|
|
|
|
|
"""
|
|
|
|
|
stem = Path(video_path).stem
|
|
|
|
|
ext = Path(video_path).suffix or ".mp4"
|
|
|
|
|
if out_file:
|
|
|
|
|
out_path = out_file
|
|
|
|
|
out_dir = os.path.dirname(os.path.abspath(out_file)) or "."
|
|
|
|
|
else:
|
|
|
|
|
out_path = os.path.join(out_dir, f"{stem}{OUTPUT_SUFFIX}{ext}")
|
|
|
|
|
csv_path = os.path.join(out_dir, f"{stem}_speeds.csv")
|
|
|
|
|
|
|
|
|
|
print("\n" + "=" * 60)
|
|
|
|
|
print(f"處理: {os.path.basename(video_path)}")
|
|
|
|
|
print("=" * 60)
|
|
|
|
|
|
|
|
|
|
debug_dir = None
|
|
|
|
|
if args.debug_dir:
|
|
|
|
|
debug_dir = os.path.join(args.debug_dir, stem)
|
|
|
|
|
os.makedirs(debug_dir, exist_ok=True)
|
|
|
|
|
|
|
|
|
|
# 1. 逐秒辨識
|
|
|
|
|
samples, duration, fps = sample_speeds(
|
|
|
|
|
video_path, roi, args.interval, reader, debug_dir=debug_dir)
|
|
|
|
|
|
|
|
|
|
# 2. 輸出逐秒紀錄
|
|
|
|
|
write_csv(samples, csv_path)
|
|
|
|
|
|
|
|
|
|
# 3. 計算保留/移除區段
|
|
|
|
|
keeps, removals = find_keep_intervals(samples, duration, args.interval)
|
|
|
|
|
|
2026-06-03 21:07:03 +08:00
|
|
|
removed_total = sum(r["end"] - r["start"] for r in removals)
|
2026-06-02 17:32:27 +08:00
|
|
|
kept_total = sum(e - s for s, e in keeps)
|
|
|
|
|
print(f"[結果] 總長 {fmt_ts(duration)} | 紅燈 {len(removals)} 段(移除 "
|
|
|
|
|
f"{fmt_ts(removed_total)})| 保留 {len(keeps)} 段({fmt_ts(kept_total)})")
|
2026-06-03 21:07:03 +08:00
|
|
|
for idx, r in enumerate(removals, 1):
|
|
|
|
|
print(f" 移除#{idx}: {fmt_ts(r['start'])} ~ {fmt_ts(r['end'])} "
|
|
|
|
|
f"(剪掉 {r['end'] - r['start']:.0f}s) 原因: {r['reason']}")
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
if not keeps:
|
|
|
|
|
print("[WARN] 沒有可保留的片段,跳過此檔(請檢查 ROI 與 speeds.csv)。")
|
|
|
|
|
return {"file": stem, "status": "skipped_no_keep",
|
|
|
|
|
"removed": removed_total}
|
|
|
|
|
|
2026-06-02 18:59:35 +08:00
|
|
|
# 4. FFmpeg 切割 + 拼接
|
2026-06-02 19:29:44 +08:00
|
|
|
audio_proc = VOLUME_BOOST or HARD_LIMITER or LOWPASS
|
2026-06-02 18:59:35 +08:00
|
|
|
if TRANSITION and len(keeps) >= 1:
|
2026-06-02 19:29:44 +08:00
|
|
|
# 有轉場: xfade 淡出淡入 + acrossfade 恆定功率 + 音量處理(需重新編碼)
|
2026-06-02 18:59:35 +08:00
|
|
|
ffmpeg_cut_concat_xfade(video_path, keeps, out_path,
|
|
|
|
|
use_gpu=args.gpu, tmp_parent=out_dir)
|
|
|
|
|
else:
|
2026-06-02 19:29:44 +08:00
|
|
|
if audio_proc:
|
|
|
|
|
print("[WARN] 音量處理(音量增益/強制壓限/低通)需重新編碼,"
|
|
|
|
|
"目前 transition=false 為無損模式,音訊將維持原樣不處理。")
|
2026-06-02 18:59:35 +08:00
|
|
|
# 無轉場: 維持無損快剪
|
|
|
|
|
ffmpeg_cut_concat(video_path, keeps, out_path, args.reencode,
|
|
|
|
|
tmp_parent=out_dir)
|
2026-06-03 21:07:03 +08:00
|
|
|
|
|
|
|
|
# 5. 輸出剪輯 log: 哪些片段被剪掉、各自原因
|
|
|
|
|
log_path = os.path.join(out_dir, f"{stem}_cut_log.txt")
|
|
|
|
|
write_cut_log(log_path, video_path, out_path, duration,
|
|
|
|
|
removals, keeps, removed_total, kept_total)
|
|
|
|
|
|
2026-06-02 17:32:27 +08:00
|
|
|
return {"file": stem, "status": "ok", "output": out_path,
|
2026-06-03 21:07:03 +08:00
|
|
|
"removed": removed_total, "redlights": len(removals),
|
|
|
|
|
"log": log_path}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def write_cut_log(path: str, src: str, out: str, duration: float,
|
|
|
|
|
removals: List[dict], keeps: List[Tuple[float, float]],
|
|
|
|
|
removed_total: float, kept_total: float) -> None:
|
|
|
|
|
"""輸出一份人看得懂的剪輯紀錄: 列出每段被剪掉的時間範圍與原因。"""
|
|
|
|
|
lines = []
|
|
|
|
|
lines.append("=" * 56)
|
|
|
|
|
lines.append("去紅燈剪輯紀錄")
|
|
|
|
|
lines.append("=" * 56)
|
|
|
|
|
lines.append(f"來源影片 : {src}")
|
|
|
|
|
lines.append(f"輸出影片 : {out}")
|
|
|
|
|
lines.append(f"原始總長 : {fmt_ts(duration)} ({duration:.1f}s)")
|
|
|
|
|
lines.append(f"剪掉 : {len(removals)} 段,共 {fmt_ts(removed_total)} "
|
|
|
|
|
f"({removed_total:.1f}s)")
|
|
|
|
|
lines.append(f"保留 : {len(keeps)} 段,共 {fmt_ts(kept_total)} "
|
|
|
|
|
f"({kept_total:.1f}s)")
|
|
|
|
|
lines.append("")
|
|
|
|
|
lines.append("【被剪掉的片段與原因】")
|
|
|
|
|
if removals:
|
|
|
|
|
for idx, r in enumerate(removals, 1):
|
|
|
|
|
lines.append(
|
|
|
|
|
f" #{idx:>2} {fmt_ts(r['start'])} ~ {fmt_ts(r['end'])} "
|
|
|
|
|
f"(剪掉 {r['end'] - r['start']:.0f} 秒) → {r['reason']}")
|
|
|
|
|
else:
|
|
|
|
|
lines.append(" (無;整支影片皆保留)")
|
|
|
|
|
lines.append("")
|
|
|
|
|
lines.append("【保留下來的片段(行駛畫面)】")
|
|
|
|
|
for idx, (s, e) in enumerate(keeps, 1):
|
|
|
|
|
lines.append(f" #{idx:>2} {fmt_ts(s)} ~ {fmt_ts(e)} ({e - s:.0f} 秒)")
|
|
|
|
|
lines.append("")
|
|
|
|
|
lines.append("判定規則: 連續停止 >= 紅燈門檻,或停到影片結尾即剪掉;"
|
|
|
|
|
"兩段停車間行駛未超過起步門檻會併入一起剪;讀不到時速一律當 0。")
|
|
|
|
|
|
|
|
|
|
text = "\n".join(lines) + "\n"
|
|
|
|
|
with open(path, "w", encoding="utf-8-sig") as f:
|
|
|
|
|
f.write(text)
|
|
|
|
|
print(f"[INFO] 已輸出剪輯紀錄: {path}")
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 設定檔 (config.json) ----------
|
|
|
|
|
|
|
|
|
|
CONFIG_NAME = "config.json"
|
|
|
|
|
|
|
|
|
|
# 設定檔的 key → 全域變數名稱。值的型別由 JSON 本身決定(數字/布林/陣列)。
|
|
|
|
|
_CONFIG_KEYS = {
|
|
|
|
|
"roi": "ROI", # [x, y, w, h] 四個數字
|
|
|
|
|
"sample_interval": "SAMPLE_INTERVAL",
|
|
|
|
|
"progress_every": "PROGRESS_EVERY",
|
|
|
|
|
"stop_seconds": "STOP_SECONDS",
|
|
|
|
|
"speed_threshold": "SPEED_THRESHOLD",
|
|
|
|
|
"min_confidence": "MIN_CONFIDENCE",
|
|
|
|
|
"smooth_window": "SMOOTH_WINDOW",
|
2026-06-02 21:01:01 +08:00
|
|
|
"depart_seconds": "DEPART_SECONDS",
|
2026-06-02 18:46:33 +08:00
|
|
|
"cut_before_stop": "CUT_BEFORE_STOP",
|
|
|
|
|
"keep_after_stop": "KEEP_AFTER_STOP",
|
2026-06-02 17:32:27 +08:00
|
|
|
"min_keep": "MIN_KEEP",
|
|
|
|
|
"use_gpu": "USE_GPU",
|
|
|
|
|
"reencode": "REENCODE",
|
2026-06-02 18:59:35 +08:00
|
|
|
"transition": "TRANSITION",
|
|
|
|
|
"transition_duration": "TRANSITION_DURATION",
|
|
|
|
|
"video_transition": "VIDEO_TRANSITION",
|
2026-06-02 21:36:26 +08:00
|
|
|
"video_quality": "VIDEO_QUALITY",
|
2026-06-02 19:29:44 +08:00
|
|
|
"volume_boost": "VOLUME_BOOST",
|
|
|
|
|
"volume_boost_percent": "VOLUME_BOOST_PERCENT",
|
|
|
|
|
"hard_limiter": "HARD_LIMITER",
|
|
|
|
|
"lowpass": "LOWPASS",
|
|
|
|
|
"lowpass_hz": "LOWPASS_HZ",
|
2026-06-02 17:32:27 +08:00
|
|
|
"base_dir": "BASE_DIR",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _current_config() -> dict:
|
|
|
|
|
"""把目前的全域參數打包成 dict,用來寫出預設設定檔。"""
|
|
|
|
|
cfg = {}
|
|
|
|
|
for key, gname in _CONFIG_KEYS.items():
|
|
|
|
|
val = globals()[gname]
|
|
|
|
|
cfg[key] = list(val) if isinstance(val, tuple) else val
|
|
|
|
|
return cfg
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_config(script_dir: str) -> Optional[str]:
|
|
|
|
|
"""
|
|
|
|
|
讀取與程式同目錄的 config.json,把參數覆寫到全域變數。
|
|
|
|
|
若檔案不存在,會用目前內建預設「自動產生一份」,方便使用者直接編輯。
|
|
|
|
|
回傳設定檔路徑(讀取或產生失敗則回 None)。
|
|
|
|
|
"""
|
|
|
|
|
path = os.path.join(script_dir, CONFIG_NAME)
|
|
|
|
|
|
|
|
|
|
# 不存在 → 產生一份預設設定檔
|
|
|
|
|
if not os.path.exists(path):
|
|
|
|
|
try:
|
|
|
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
|
|
|
json.dump(_current_config(), f, ensure_ascii=False, indent=2)
|
|
|
|
|
print(f"[INFO] 已產生預設設定檔: {path}")
|
|
|
|
|
except OSError as e:
|
|
|
|
|
print(f"[WARN] 無法產生設定檔({e}),改用內建預設。")
|
|
|
|
|
return path
|
|
|
|
|
|
|
|
|
|
# 存在 → 讀取並套用
|
|
|
|
|
try:
|
|
|
|
|
with open(path, encoding="utf-8-sig") as f:
|
|
|
|
|
data = json.load(f)
|
|
|
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
|
|
|
print(f"[WARN] 設定檔讀取/解析失敗({e}),改用內建預設: {path}")
|
|
|
|
|
return None
|
|
|
|
|
if not isinstance(data, dict):
|
|
|
|
|
print(f"[WARN] 設定檔格式不是物件(應為 {{...}}),改用內建預設: {path}")
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
applied = []
|
|
|
|
|
for key, val in data.items():
|
|
|
|
|
if key not in _CONFIG_KEYS:
|
|
|
|
|
print(f"[WARN] 設定檔: 不認識的項目『{key}』,略過")
|
|
|
|
|
continue
|
|
|
|
|
gname = _CONFIG_KEYS[key]
|
|
|
|
|
try:
|
|
|
|
|
if key == "roi":
|
|
|
|
|
nums = [int(x) for x in val]
|
|
|
|
|
if len(nums) != 4:
|
|
|
|
|
raise ValueError("roi 需要 4 個數字 [x, y, w, h]")
|
|
|
|
|
globals()["ROI"] = tuple(nums)
|
|
|
|
|
else:
|
|
|
|
|
globals()[gname] = val
|
|
|
|
|
applied.append(f"{key}={val}")
|
|
|
|
|
except (TypeError, ValueError) as e:
|
|
|
|
|
print(f"[WARN] 設定檔『{key}={val}』無效: {e}(沿用預設)")
|
|
|
|
|
|
|
|
|
|
if applied:
|
|
|
|
|
print(f"[INFO] 已套用設定檔 ({CONFIG_NAME}): " + ", ".join(applied))
|
|
|
|
|
return path
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- 主程式 ----------
|
|
|
|
|
|
|
|
|
|
def main():
|
|
|
|
|
# global 宣告必須在這兩個變數第一次被使用之前(argparse default 也算使用),
|
|
|
|
|
# 否則 Python 3.14 會報 "used prior to global declaration"
|
|
|
|
|
global STOP_SECONDS, SAMPLE_INTERVAL
|
|
|
|
|
|
|
|
|
|
# 先讀設定檔(覆寫全域預設),再建 argparse —— 讓 CLI 參數仍可再覆寫設定檔
|
|
|
|
|
load_config(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
|
|
|
|
parser = argparse.ArgumentParser(
|
|
|
|
|
description="安全帽機車影片 — 自動去紅燈剪輯工具(支援資料夾批次)",
|
|
|
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
|
|
|
)
|
|
|
|
|
parser.add_argument("input", nargs="*", default=None,
|
|
|
|
|
help="影片檔(可多個)、資料夾路徑,或只打資料夾名稱(會到 BASE_DIR 底下找),"
|
|
|
|
|
"例如: 20260502。不給則進入互動選單")
|
|
|
|
|
parser.add_argument("--select", action="store_true",
|
|
|
|
|
help="啟動.bat 專用: 顯示資料夾/影片選單,把勾選的影片寫到 --out-list 後結束")
|
|
|
|
|
parser.add_argument("--out-list", default=None,
|
|
|
|
|
help="搭配 --select: 選好的影片絕對路徑會逐行寫到這個檔案")
|
|
|
|
|
parser.add_argument("-o", "--output", default=None,
|
|
|
|
|
help="單檔模式: 輸出檔路徑;資料夾模式: 輸出資料夾。"
|
|
|
|
|
"不指定則自動放到 <資料夾>\\no_redlight\\")
|
|
|
|
|
parser.add_argument("--base-dir", default=BASE_DIR,
|
|
|
|
|
help="只輸入資料夾名稱時的根目錄")
|
|
|
|
|
parser.add_argument("--roi", nargs=4, type=int, metavar=("X", "Y", "W", "H"),
|
|
|
|
|
help="覆寫 ROI 座標 (x y w h),不指定則用檔案內的 ROI 常數")
|
|
|
|
|
parser.add_argument("--interval", type=float, default=SAMPLE_INTERVAL,
|
|
|
|
|
help="每隔幾秒辨識一次")
|
|
|
|
|
parser.add_argument("--stop-seconds", type=float, default=STOP_SECONDS,
|
|
|
|
|
help="時速 0 連續超過幾秒判定為紅燈")
|
|
|
|
|
parser.add_argument("--gpu", action="store_true", default=USE_GPU,
|
|
|
|
|
help="EasyOCR 使用 GPU")
|
|
|
|
|
parser.add_argument("--reencode", action="store_true", default=REENCODE,
|
|
|
|
|
help="重新編碼(切點精準到幀,但較慢、非無損)")
|
|
|
|
|
parser.add_argument("--preview", action="store_true",
|
|
|
|
|
help="只輸出 ROI 預覽圖後結束(資料夾模式抽第一支影片)")
|
|
|
|
|
parser.add_argument("--preview-at", type=float, default=0.0,
|
|
|
|
|
help="預覽要抽第幾秒的畫面")
|
|
|
|
|
parser.add_argument("--preview-file", default=None,
|
|
|
|
|
help="資料夾模式下,指定要用哪支影片做預覽(檔名)")
|
|
|
|
|
parser.add_argument("--debug-dir", default=None,
|
|
|
|
|
help="把每個取樣點的 ROI 存到此資料夾(除錯用)")
|
|
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
|
|
|
|
# 套用全域參數覆寫 (讓 CLI 參數優先)
|
|
|
|
|
STOP_SECONDS = args.stop_seconds
|
|
|
|
|
SAMPLE_INTERVAL = args.interval
|
|
|
|
|
roi = tuple(args.roi) if args.roi else ROI
|
|
|
|
|
|
|
|
|
|
# === --select: 啟動.bat 專用選擇器,選好寫出清單檔後結束 ===
|
|
|
|
|
if args.select:
|
|
|
|
|
if not args.out_list:
|
|
|
|
|
sys.exit("[ERROR] --select 需搭配 --out-list <清單檔路徑>")
|
|
|
|
|
sys.exit(run_selector(args.base_dir, args.out_list))
|
|
|
|
|
|
|
|
|
|
inputs = list(args.input) if args.input else []
|
|
|
|
|
out_dir = None
|
|
|
|
|
single_named_output = None
|
|
|
|
|
|
|
|
|
|
if len(inputs) >= 2:
|
|
|
|
|
# === 多個影片檔: 啟動.bat 勾選好後帶參數呼叫(主要路徑)===
|
|
|
|
|
videos = []
|
|
|
|
|
for p in inputs:
|
|
|
|
|
try:
|
|
|
|
|
videos.append(resolve_input(p, args.base_dir))
|
|
|
|
|
except FileNotFoundError as e:
|
|
|
|
|
sys.exit(f"[ERROR] {e}")
|
|
|
|
|
base_folder = os.path.dirname(os.path.abspath(videos[0]))
|
|
|
|
|
out_dir = args.output if args.output else os.path.join(base_folder, OUTPUT_SUBDIR)
|
|
|
|
|
print(f"[INFO] 已指定 {len(videos)} 支影片,準備處理:")
|
|
|
|
|
for v in videos:
|
|
|
|
|
print(f" - {os.path.basename(v)}")
|
|
|
|
|
|
|
|
|
|
elif len(inputs) == 0:
|
|
|
|
|
# === 互動後備: 直接執行 py 沒帶參數時,自己跳選單 ===
|
|
|
|
|
folders = [(f, vs) for f in list_folders(args.base_dir)
|
|
|
|
|
if (vs := find_videos(f))]
|
|
|
|
|
if not folders:
|
|
|
|
|
sys.exit(f"[ERROR] 在 {args.base_dir} 底下找不到含影片的資料夾。")
|
|
|
|
|
flabels = [f"{os.path.basename(f)} ({len(vs)} 支影片)" for f, vs in folders]
|
|
|
|
|
sel = menu_select_one(f"選擇要處理的資料夾 (根目錄: {args.base_dir})", flabels)
|
|
|
|
|
if sel is None:
|
|
|
|
|
print("已取消。")
|
|
|
|
|
return
|
|
|
|
|
in_path, allvids = folders[sel]
|
|
|
|
|
vlabels = []
|
|
|
|
|
for v in allvids:
|
|
|
|
|
try:
|
|
|
|
|
gb = os.path.getsize(v) / (1024 ** 3)
|
|
|
|
|
vlabels.append(f"{os.path.basename(v)} ({gb:.1f} GB)")
|
|
|
|
|
except OSError:
|
|
|
|
|
vlabels.append(os.path.basename(v))
|
|
|
|
|
picks = menu_select_many(
|
|
|
|
|
f"勾選要去紅燈的影片 (資料夾: {os.path.basename(in_path)})", vlabels)
|
|
|
|
|
if picks is None:
|
|
|
|
|
print("已取消。")
|
|
|
|
|
return
|
|
|
|
|
videos = [allvids[i] for i in picks]
|
|
|
|
|
out_dir = os.path.join(in_path, OUTPUT_SUBDIR)
|
|
|
|
|
os.system("cls")
|
|
|
|
|
print(f"[INFO] 資料夾: {in_path}")
|
|
|
|
|
print(f"[INFO] 已選 {len(videos)} 支影片,準備處理:")
|
|
|
|
|
for v in videos:
|
|
|
|
|
print(f" - {os.path.basename(v)}")
|
|
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
# === 單一輸入: 影片檔 / 資料夾 / 只給資料夾名稱 ===
|
|
|
|
|
try:
|
|
|
|
|
in_path = resolve_input(inputs[0], args.base_dir)
|
|
|
|
|
except FileNotFoundError as e:
|
|
|
|
|
sys.exit(f"[ERROR] {e}")
|
|
|
|
|
if os.path.isdir(in_path):
|
|
|
|
|
videos = find_videos(in_path)
|
|
|
|
|
if not videos:
|
|
|
|
|
sys.exit(f"[ERROR] 資料夾內找不到影片檔: {in_path}\n"
|
|
|
|
|
f" 支援的副檔名: {sorted(VIDEO_EXTS)}")
|
|
|
|
|
print(f"[INFO] 資料夾: {in_path}")
|
|
|
|
|
print(f"[INFO] 找到 {len(videos)} 支影片:")
|
|
|
|
|
for v in videos:
|
|
|
|
|
print(f" - {os.path.basename(v)}")
|
|
|
|
|
out_dir = args.output if args.output else os.path.join(in_path, OUTPUT_SUBDIR)
|
|
|
|
|
else:
|
|
|
|
|
videos = [in_path]
|
|
|
|
|
if args.output and os.path.splitext(args.output)[1]:
|
2026-06-02 21:03:36 +08:00
|
|
|
# -o 指定了含副檔名的完整輸出檔名 → 照用
|
2026-06-02 17:32:27 +08:00
|
|
|
single_named_output = args.output
|
|
|
|
|
out_dir = os.path.dirname(os.path.abspath(args.output)) or "."
|
2026-06-02 21:03:36 +08:00
|
|
|
elif args.output:
|
|
|
|
|
# -o 指定了資料夾
|
|
|
|
|
out_dir = args.output
|
2026-06-02 17:32:27 +08:00
|
|
|
else:
|
2026-06-02 21:03:36 +08:00
|
|
|
# 預設: 與多支/整資料夾一致,放到來源同層的 no_redlight\ 子資料夾
|
|
|
|
|
out_dir = os.path.join(os.path.dirname(os.path.abspath(in_path)),
|
|
|
|
|
OUTPUT_SUBDIR)
|
2026-06-02 17:32:27 +08:00
|
|
|
|
|
|
|
|
# --- preview 模式: 確認 ROI 後就結束(僅單一輸入支援)---
|
|
|
|
|
if args.preview:
|
|
|
|
|
target = videos[0]
|
|
|
|
|
if args.preview_file:
|
|
|
|
|
p2 = [v for v in videos if os.path.basename(v) == args.preview_file]
|
|
|
|
|
if p2:
|
|
|
|
|
target = p2[0]
|
|
|
|
|
save_preview(target, roi, args.preview_at, prefix="preview")
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
check_ffmpeg()
|
|
|
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
|
|
|
if args.debug_dir:
|
|
|
|
|
os.makedirs(args.debug_dir, exist_ok=True)
|
|
|
|
|
|
|
|
|
|
# 載入 OCR 模型(只載一次,所有影片共用)
|
|
|
|
|
reader = build_reader(args.gpu)
|
|
|
|
|
|
|
|
|
|
# 逐支處理;單支失敗不影響其他支
|
|
|
|
|
results = []
|
|
|
|
|
for v in videos:
|
|
|
|
|
try:
|
|
|
|
|
res = process_one(v, roi, out_dir, reader, args,
|
|
|
|
|
out_file=single_named_output)
|
|
|
|
|
results.append(res)
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(f"[ERROR] 處理 {os.path.basename(v)} 失敗: {e}")
|
|
|
|
|
results.append({"file": Path(v).stem, "status": "error",
|
|
|
|
|
"error": str(e)})
|
|
|
|
|
|
|
|
|
|
# 總結
|
|
|
|
|
print("\n" + "#" * 60)
|
|
|
|
|
print("# 全部處理完成 — 總結")
|
|
|
|
|
print("#" * 60)
|
|
|
|
|
for r in results:
|
|
|
|
|
if r["status"] == "ok":
|
|
|
|
|
print(f" [OK] {r['file']}: 去除 {r['redlights']} 個紅燈,共 "
|
|
|
|
|
f"{fmt_ts(r['removed'])} -> {r['output']}")
|
|
|
|
|
elif r["status"] == "skipped_no_keep":
|
|
|
|
|
print(f" [SKIP] {r['file']}: 略過(沒有可保留片段)")
|
|
|
|
|
else:
|
|
|
|
|
print(f" [FAIL] {r['file']}: {r.get('error', '')}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
|
main()
|