Passman22231's picture
Update app.py
f3c8c7f verified
Raw History Blame Contribute Delete
11.1 kB
"""
Motion-Track Shape Censor (CPU ์ „์šฉ, GPU ๋ถˆํ•„์š”)
ํด๋ฆญํ•œ ์ง€์ ์„ OpenCV ๊ด‘ํ•™ํ๋ฆ„์œผ๋กœ ์ถ”์ ํ•ด์„œ ํฐ ์›/๋„ค๋ชจ๋ฅผ ์”Œ์šฐ๋Š” ๊ฒฝ๋Ÿ‰ ๊ฒ€์—ด ๋„๊ตฌ.
SAM ๊ณ„์—ด์˜ ํ˜•ํƒœ ๋ฐฑ์ƒ‰ ๋งˆ์Šคํ‚น์ด ๋ชป ์žก๋Š” ๋ถ€๋ถ„์„ '๋„ํ˜•'์œผ๋กœ ๋ณด์กฐ ๊ฒ€์—ดํ•˜๋Š” ์šฉ๋„.
ํ—ˆ๊น…ํŽ˜์ด์Šค ๋ฌด๋ฃŒ CPU Basic Space์— ์˜ฌ๋ฆฌ๋ฉด ์ƒ์‹œ ์š”๊ธˆ 0์›์œผ๋กœ ํ•ญ์ƒ ์‚ฌ์šฉ ๊ฐ€๋Šฅ.
"""
import os
import tempfile
import cv2
import numpy as np
import gradio as gr
MAX_SECONDS = 12 # ์ฒ˜๋ฆฌ ์ƒํ•œ (์งง์€ ํด๋ฆฝ ์ „์šฉ)
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# ์ƒํƒœ
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
STATE = {
"video": None,
"frames": [], # RGB ํ”„๋ ˆ์ž„ ๋ฆฌ์ŠคํŠธ
"gray": [], # ๊ทธ๋ ˆ์ด์Šค์ผ€์ผ(๊ด‘ํ•™ํ๋ฆ„์šฉ)
"fps": 24.0,
"w": 0,
"h": 0,
}
def _load_video(video_path):
if not video_path:
return None, gr.update(maximum=1, value=0), [], gr.update(open=True)
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS) or 24.0
frames, grays = [], []
max_frames = int(fps * MAX_SECONDS)
while len(frames) < max_frames:
ok, fr = cap.read()
if not ok:
break
rgb = cv2.cvtColor(fr, cv2.COLOR_BGR2RGB)
frames.append(rgb)
grays.append(cv2.cvtColor(fr, cv2.COLOR_BGR2GRAY))
cap.release()
if not frames:
return None, gr.update(maximum=1, value=0), [], gr.update(open=True)
STATE.update(video=video_path, frames=frames, gray=grays, fps=fps,
w=frames[0].shape[1], h=frames[0].shape[0])
# ์˜์ƒ ๋กœ๋“œ๋˜๋ฉด ์—…๋กœ๋“œ ํŒจ๋„ ์ ‘๊ธฐ
return frames[0], gr.update(maximum=len(frames) - 1, value=0), [], gr.update(open=False)
def _draw_points(idx, clicks):
"""ํ˜„์žฌ ํ”„๋ ˆ์ž„์— ํด๋ฆญ ์ง€์  ํ‘œ์‹œ."""
if not STATE["frames"]:
return None
idx = int(np.clip(idx, 0, len(STATE["frames"]) - 1))
img = STATE["frames"][idx].copy()
for (x, y) in (clicks or []):
cv2.circle(img, (int(x), int(y)), 6, (0, 200, 0), -1)
cv2.circle(img, (int(x), int(y)), 6, (255, 255, 255), 1)
return img
def _click(idx, clicks, evt: gr.SelectData):
x, y = evt.index[0], evt.index[1]
clicks = (clicks or []) + [[x, y]]
return _draw_points(idx, clicks), clicks
def _reset(idx):
return _draw_points(idx, []), []
def _preview(idx, clicks, mask_mode, shape_size):
"""ํ˜„์žฌ ํ”„๋ ˆ์ž„์— ๋„ํ˜•์„ ์”Œ์šด ๋ฏธ๋ฆฌ๋ณด๊ธฐ."""
if not STATE["frames"]:
return None, "์˜์ƒ์„ ์—…๋กœ๋“œํ•˜์„ธ์š”."
if not clicks:
return None, "๊ฐ€๋ฆด ์ง€์ ์„ ํด๋ฆญํ•˜์„ธ์š”."
idx = int(np.clip(idx, 0, len(STATE["frames"]) - 1))
img = STATE["frames"][idx].copy()
s = int(shape_size)
for (x, y) in clicks:
x, y = int(x), int(y)
if mask_mode == "ํฐ ์›":
cv2.circle(img, (x, y), s, (255, 255, 255), -1)
else:
cv2.rectangle(img, (x - s, y - s), (x + s, y + s), (255, 255, 255), -1)
return img, "ํฐ ๋„ํ˜•์ด ๊ฐ€๋ ค์งˆ ์œ„์น˜์ž…๋‹ˆ๋‹ค. ์œ„์น˜ยทํฌ๊ธฐ๋ฅผ ํ™•์ธํ•˜๊ณ  ์ฒ˜๋ฆฌ ์‹คํ–‰์„ ๋ˆ„๋ฅด์„ธ์š”."
def _track_points(start_idx, points, error_threshold=30.0):
"""
start_idx ํ”„๋ ˆ์ž„์˜ points๋ฅผ ๊ด‘ํ•™ํ๋ฆ„์œผ๋กœ ์ „์ฒด ํ”„๋ ˆ์ž„์— ๋Œ€ํ•ด ์ถ”์ .
๋ฐ˜ํ™˜: {frame_idx: [(x, y, visible), ...]} โ€” ๊ฐ ํ”„๋ ˆ์ž„์—์„œ์˜ ์  ์œ„์น˜ + ๊ฐ€์‹œ์„ฑ.
์•ž์œผ๋กœ(โ†’๋) + ๋’ค๋กœ(โ†’์ฒ˜์Œ) ์–‘๋ฐฉํ–ฅ ์ถ”์ .
visible=False๋ฉด ๊ทธ ํ”„๋ ˆ์ž„์—์„œ ์ถ”์ ์„ ๋†“์นœ ๊ฒƒ(๋ถ€์œ„๊ฐ€ ๊ฐ€๋ ค์ง/์‚ฌ๋ผ์ง) โ†’ ๋„ํ˜• ์ˆจ๊น€.
"""
grays = STATE["gray"]
n = len(grays)
npts = len(points)
tracked = {start_idx: [(float(p[0]), float(p[1]), True) for p in points]}
lk_params = dict(winSize=(21, 21), maxLevel=3,
criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 20, 0.03))
def run(direction):
pts = np.array(points, dtype=np.float32).reshape(-1, 1, 2)
rng = range(start_idx + 1, n) if direction > 0 else range(start_idx - 1, -1, -1)
prev_pts = pts
prev_idx = start_idx
# ํ•œ ๋ฒˆ ๋†“์นœ ์ ์€ ๊ทธ ๋ฐฉํ–ฅ์œผ๋กœ ๊ณ„์† ์ˆจ๊น€ (ํƒ ์‚ฌ๋ผ์ง„ ๋’ค ์œ ์ง€)
lost = [False] * npts
for f in rng:
new_pts, st, err = cv2.calcOpticalFlowPyrLK(
grays[prev_idx], grays[f], prev_pts, None, **lk_params)
if new_pts is None:
new_pts = prev_pts
st = np.zeros((npts, 1), dtype=np.uint8)
err = np.full((npts, 1), 999.0, dtype=np.float32)
row = []
for i in range(npts):
ok = (st is not None and int(st[i][0]) == 1)
e = float(err[i][0]) if err is not None else 999.0
# ์ถ”์  ์‹คํŒจ ๋˜๋Š” ์˜ค์ฐจ๊ฐ€ ์ž„๊ณ„๊ฐ’ ์ดˆ๊ณผ โ†’ ์ด ์ ์€ ์‚ฌ๋ผ์ง„ ๊ฒƒ์œผ๋กœ ๊ฐ„์ฃผ
if (not ok) or (e > error_threshold):
lost[i] = True
x = float(np.clip(new_pts[i, 0, 0], 0, STATE["w"] - 1))
y = float(np.clip(new_pts[i, 0, 1], 0, STATE["h"] - 1))
visible = not lost[i]
row.append((x, y, visible))
new_pts[i, 0, 0] = x
new_pts[i, 0, 1] = y
# ๊ธฐ์กด ๊ฐ’์ด ์žˆ์œผ๋ฉด(๋ฐ˜๋Œ€ ๋ฐฉํ–ฅ์—์„œ ์ฑ„์›Œ์ง„ ๊ฒฝ์šฐ) ๊ฐ€์‹œ์„ฑ์€ OR๋กœ ํ•ฉ์นจ
if f in tracked:
merged = []
for a, b in zip(tracked[f], row):
merged.append((b[0], b[1], a[2] or b[2]))
tracked[f] = merged
else:
tracked[f] = row
prev_pts = new_pts
prev_idx = f
run(+1) # ์•ž์œผ๋กœ
run(-1) # ๋’ค๋กœ (์–‘๋ฐฉํ–ฅ)
return tracked
def _render(start_idx, clicks, mask_mode, shape_size, sensitivity):
if not STATE["frames"]:
return None, "์˜์ƒ์„ ์—…๋กœ๋“œํ•˜์„ธ์š”."
if not clicks:
return None, "๊ฐ€๋ฆด ์ง€์ ์„ ํด๋ฆญํ•˜์„ธ์š”."
start_idx = int(np.clip(start_idx, 0, len(STATE["frames"]) - 1))
points = [[float(x), float(y)] for (x, y) in clicks]
# ๋ฏผ๊ฐ๋„ โ†’ ์˜ค์ฐจ ์ž„๊ณ„๊ฐ’ ๋ณ€ํ™˜ (๋ฏผ๊ฐ๋„ ๋†’์„์ˆ˜๋ก ์‰ฝ๊ฒŒ ์‚ฌ๋ผ์ง = ์ž„๊ณ„๊ฐ’ ๋‚ฎ์Œ)
# ์Šฌ๋ผ์ด๋” 1(๋‘”๊ฐ)~100(๋ฏผ๊ฐ) โ†’ threshold 60~8
error_threshold = 60.0 - (float(sensitivity) - 1) * (52.0 / 99.0)
# ์  ์ถ”์  (๊ฐ€์‹œ์„ฑ ํฌํ•จ)
tracked = _track_points(start_idx, points, error_threshold=error_threshold)
# ๋ Œ๋”
tmp = tempfile.mkdtemp()
# ๊ณ ์œ  ํŒŒ์ผ๋ช…: ํƒ€์ž„์Šคํƒฌํ”„ + ์งง์€ ๋žœ๋ค (๋ถ„๋ฅ˜/๊ฒน์นจ๋ฐฉ์ง€ ํŽธ์˜)
import datetime, uuid
stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
rand = uuid.uuid4().hex[:4]
out_path = os.path.join(tmp, f"censored_{stamp}_{rand}.mp4")
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
vw = cv2.VideoWriter(out_path, fourcc, STATE["fps"], (STATE["w"], STATE["h"]))
s = int(shape_size)
for f in range(len(STATE["frames"])):
# BGR๋กœ ๋ณ€ํ™˜ํ•ด์„œ ๊ธฐ๋ก
frame_bgr = cv2.cvtColor(STATE["frames"][f], cv2.COLOR_RGB2BGR).copy()
pts = tracked.get(f, [])
for pt in pts:
x, y, visible = int(pt[0]), int(pt[1]), pt[2]
if not visible:
continue # ์ถ”์  ๋†“์นœ ํ”„๋ ˆ์ž„ โ†’ ๋„ํ˜• ์ˆจ๊น€ (ํƒ ์‚ฌ๋ผ์ง)
if mask_mode == "ํฐ ์›":
cv2.circle(frame_bgr, (x, y), s, (255, 255, 255), -1)
else:
cv2.rectangle(frame_bgr, (x - s, y - s), (x + s, y + s), (255, 255, 255), -1)
vw.write(frame_bgr)
vw.release()
return out_path, f"์™„๋ฃŒ! {len(clicks)}๊ฐœ ์ง€์  ์ถ”์ . ๊ฐ€๋ ค์ง€๋ฉด ๋„ํ˜•์ด ์ž๋™์œผ๋กœ ์‚ฌ๋ผ์ง‘๋‹ˆ๋‹ค."
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# UI
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
with gr.Blocks(title="Motion-Track Shape Censor") as demo:
gr.Markdown("""
### Motion-Track Shape Censor (๋„ํ˜• ๊ฒ€์—ด ยท CPU ์ „์šฉ)
ํด๋ฆญํ•œ ์ง€์ ์„ ๊ด‘ํ•™ํ๋ฆ„์œผ๋กœ ์ถ”์ ํ•ด์„œ ํฐ ์›/๋„ค๋ชจ๋ฅผ ์”Œ์›๋‹ˆ๋‹ค.
ํ˜•ํƒœ ๋ฐฑ์ƒ‰ ๋งˆ์Šคํ‚น์ด ๋ชป ์žก๋Š” ๋ถ€๋ถ„์„ ๋„ํ˜•์œผ๋กœ ๋ณด์กฐ ๊ฒ€์—ดํ•˜๋Š” ์šฉ๋„. **GPU ๋ถˆํ•„์š”.**
""")
clicks_st = gr.State([])
with gr.Row():
with gr.Column():
with gr.Accordion("์ž…๋ ฅ ์˜์ƒ (์—…๋กœ๋“œ/๊ต์ฒดํ•˜๋ ค๋ฉด ํŽผ์น˜๊ธฐ)", open=True) as upload_acc:
vid = gr.Video(label="์ž…๋ ฅ ์˜์ƒ (mp4)")
frame_img = gr.Image(label="ํ”„๋ ˆ์ž„ โ€” ๊ฐ€๋ฆด ์ง€์  ํด๋ฆญ (์—ฌ๋Ÿฌ ๊ฐœ ๊ฐ€๋Šฅ)", interactive=True)
frame_slider = gr.Slider(0, 1, value=0, step=1,
label="ํ”„๋ ˆ์ž„ ์„ ํƒ (๋Œ€์ƒ์ด ์ž˜ ๋ณด์ด๋Š” ์ง€์ )")
reset_btn = gr.Button("์  ์ดˆ๊ธฐํ™”")
with gr.Column():
preview_img = gr.Image(label="๋ฏธ๋ฆฌ๋ณด๊ธฐ (ํฐ ๋„ํ˜• = ๊ฐ€๋ ค์งˆ ์œ„์น˜)")
mask_mode = gr.Radio(["ํฐ ์›", "ํฐ ๋„ค๋ชจ"], value="ํฐ ์›", label="๋„ํ˜• ์ข…๋ฅ˜")
shape_size = gr.Slider(10, 200, value=50, step=5,
label="๋„ํ˜• ํฌ๊ธฐ (์›=๋ฐ˜์ง€๋ฆ„ / ๋„ค๋ชจ=๋ฐ˜์ชฝ ๋ณ€, px)")
sensitivity = gr.Slider(1, 100, value=50, step=1,
label="์‚ฌ๋ผ์ง ๋ฏผ๊ฐ๋„ (๋†’์„์ˆ˜๋ก ๊ฐ€๋ ค์งˆ ๋•Œ ๋นจ๋ฆฌ ์‚ฌ๋ผ์ง)")
with gr.Row():
preview_btn = gr.Button("๋ฏธ๋ฆฌ๋ณด๊ธฐ")
go_btn = gr.Button("์ฒ˜๋ฆฌ ์‹คํ–‰", variant="primary")
status = gr.Textbox(label="์ƒํƒœ", interactive=False)
result_video = gr.Video(label="๊ฒฐ๊ณผ ์˜์ƒ (์›๋ณธ + ํฐ ๋„ํ˜•)", height=280)
vid.change(_load_video, inputs=vid,
outputs=[frame_img, frame_slider, clicks_st, upload_acc])
frame_slider.change(_draw_points, inputs=[frame_slider, clicks_st], outputs=frame_img)
frame_img.select(_click, inputs=[frame_slider, clicks_st], outputs=[frame_img, clicks_st])
reset_btn.click(_reset, inputs=frame_slider, outputs=[frame_img, clicks_st])
preview_btn.click(_preview, inputs=[frame_slider, clicks_st, mask_mode, shape_size],
outputs=[preview_img, status])
go_btn.click(_render, inputs=[frame_slider, clicks_st, mask_mode, shape_size, sensitivity],
outputs=[result_video, status])
gr.Markdown("""
**์‚ฌ์šฉ๋ฒ•**: ์˜์ƒ ์—…๋กœ๋“œ โ†’ ํ”„๋ ˆ์ž„ ์ด๋™ โ†’ ๊ฐ€๋ฆด ์ง€์  ํด๋ฆญ(์—ฌ๋Ÿฌ ๊ฐœ OK) โ†’ ๋ฏธ๋ฆฌ๋ณด๊ธฐ๋กœ ํ™•์ธ โ†’ ์ฒ˜๋ฆฌ ์‹คํ–‰
ยท ๊ฐ ์ ์„ ์•ž๋’ค๋กœ ์ถ”์ ํ•ด์„œ ๋„ํ˜•์ด ๋Œ€์ƒ์„ ๋”ฐ๋ผ๊ฐ‘๋‹ˆ๋‹ค.
ยท **๋ถ€์œ„๊ฐ€ ์† ๋“ฑ์œผ๋กœ ๊ฐ€๋ ค์ง€๋ฉด ๋„ํ˜•์ด ํƒ ์‚ฌ๋ผ์ง‘๋‹ˆ๋‹ค** โ€” ์ž˜ ์•ˆ ๋˜๋ฉด '์‚ฌ๋ผ์ง ๋ฏผ๊ฐ๋„'๋ฅผ ์กฐ์ ˆํ•˜์„ธ์š”.
ยท ๋Œ€์ƒ์ด ๊ธ‰ํšŒ์ „ํ•˜๊ฑฐ๋‚˜ ํ™”๋ฉด์„ ๋ฒ—์–ด๋‚˜๋ฉด ์ถ”์ ์ด ํ”๋“ค๋ฆด ์ˆ˜ ์žˆ์–ด์š”(์งง์€ ํด๋ฆฝ ๊ถŒ์žฅ).
""")
if __name__ == "__main__":
demo.queue(max_size=10).launch()