""" NextJump Broadcast — Hugging Face Space. Upload a jump-rope video and get back a broadcast-ready overlay video (vertical 9:16 or horizontal 16:9) with the AI jump count, speed gauge, splits, rhythm and — optionally — official start/end beep detection. All of the work is done by claude_anim_v2.main(); this file is only the Gradio front end. Layout expected next to this file (run sync_assets.py from the repo to populate it): claude_anim_v2.py, beep.WAV, elitejumps_logo.png, nextjump_logo.png, graphics/art/SalmaproMedium.otf, movenet-pytorch/{movenet/, _models/movenet_thunder.pth} When run from inside the repo without syncing, the repo root is used instead. The counting checkpoint is private and is NOT part of the Space: it is downloaded at startup from the private model repo (like NextJump-Tournament/app.py), authenticated with the DATASET_SECRET Space secret. NEXTJUMP_MODEL_REPO model repo id (default lumos-motion/nextjump) NEXTJUMP_MODEL_FILE ONNX file in that repo (default nextjump_all_128.onnx) DATASET_SECRET / HF_TOKEN read token for the private repo (locally, a `huggingface-cli login` also works) NEXTJUMP_ONNX optional local path that skips the download """ import os import sys import time import uuid import base64 import tempfile import cv2 import gradio as gr from huggingface_hub import hf_hub_download HERE = os.path.dirname(os.path.abspath(__file__)) for _root in (HERE, os.path.dirname(HERE)): if os.path.isfile(os.path.join(_root, 'claude_anim_v2.py')): sys.path.insert(0, _root) break import claude_anim_v2 as nj # noqa: E402 MODEL_REPO = os.environ.get('NEXTJUMP_MODEL_REPO', 'lumos-motion/nextjump') MODEL_FILE = os.environ.get('NEXTJUMP_MODEL_FILE', 'nextjump_all_128.onnx') HF_TOKEN = os.environ.get('DATASET_SECRET') or os.environ.get('HF_TOKEN') # None -> cached CLI login def resolve_model(): """Path to the counting checkpoint, downloaded (and cached) from the private model repo.""" override = os.environ.get('NEXTJUMP_ONNX') if override: return override try: path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE, repo_type='model', token=HF_TOKEN) print(f'Counting model: {MODEL_REPO}/{MODEL_FILE} -> {path}') return path except Exception as e: local = os.path.join(nj.ROOT, 'checkpoint', 'onnx', MODEL_FILE) # local dev inside the repo if os.path.isfile(local): print(f'Could not download {MODEL_REPO}/{MODEL_FILE} ({type(e).__name__}); using local {local}') return local raise RuntimeError(f'Could not download {MODEL_FILE} from the private repo {MODEL_REPO} ' f'({type(e).__name__}: {e}). If this is an auth error, add a read token for the repo ' 'as the DATASET_SECRET secret in the Space settings.') from e # Render worker processes (multiprocessing 'spawn') re-import this file as __mp_main__: they must not # download models or build the UI again, so all of that is skipped there. IS_RENDER_WORKER = __name__ == '__mp_main__' ONNX_PATH = None if IS_RENDER_WORKER else resolve_model() # at startup, so the first request doesn't wait if not IS_RENDER_WORKER: try: # also warm the torchvision person-detector weights used by smart framing (public, ~75 MB) nj.load_person_detector('coco', 'cpu') except Exception as e: print(f'Could not pre-load the person detector ({e}); it will be downloaded on first use') MAX_SECONDS = float(os.environ.get('NEXTJUMP_MAX_SECONDS', 200)) # longest accepted upload OUT_DIR = os.path.join(tempfile.gettempdir(), 'nextjump_space') KEEP_OUTPUTS_S = 2 * 3600 FORMATS = {'Social (TikTok / Reels / Shorts, full screen)': 'social', 'Broadcast vertical 9:16 (full stats panel)': 'vertical', 'Broadcast horizontal 16:9 (YouTube / TV)': 'horizontal'} DEFAULT_FILL = {'social': 0.25, 'vertical': 0.50, 'horizontal': 0.60} FRAMING = {'Smart: find the most active jumper(s)': 'smart', 'Pose: follow the first person found': 'pose', 'Off: fixed centre crop': 'off'} def probe(path): cap = cv2.VideoCapture(path) fps = cap.get(cv2.CAP_PROP_FPS) or 0.0 frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) w, h = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)), int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) ok = cap.isOpened() and cap.read()[0] cap.release() return dict(ok=ok, fps=fps, frames=frames, w=w, h=h, duration=frames / fps if fps else 0.0) def cleanup_old_outputs(): if not os.path.isdir(OUT_DIR): return now = time.time() for f in os.listdir(OUT_DIR): p = os.path.join(OUT_DIR, f) try: if now - os.path.getmtime(p) > KEEP_OUTPUTS_S: os.remove(p) except OSError: pass def summary_markdown(s, info): lines = [f'## {s["final"]} jumps'] if s['event'] is not None: t0, t1 = s['event'] lines.append(f'Counted between the beeps: start **{t0:.2f} s**, end **{t1:.2f} s** ' f'({t1 - t0:.2f} s event). Nothing before the start beep or after the end beep is counted.') elif s['beep_requested']: lines.append('⚠️ Beep detection was on, but no start/end beep pair matching the event length was found, ' 'so the whole video was counted.') else: lines.append(f'Counted over the detected jumping window ({s["duration"]:.1f} s).') if s.get('misses'): lines.append(f'\n**{s["misses"]}** miss{"es" if s["misses"] != 1 else ""} detected' + (f'; the first jump after each miss was not counted (rulebook), ' f'**{s["skipped"]}** jump{"s" if s["skipped"] != 1 else ""} removed.' if s.get('skipped') else '.')) lines.append(f'\nAverage speed **{s["avg_jps"]:.2f}** jumps/s · peak **{s["peak_jps"]:.2f}** jumps/s') if s['splits']: lines.append(f'\n| {s["split_seconds"]} s split | ' + ' | '.join(str(k + 1) for k in range(len(s['splits']))) + ' |') lines.append('|---|' + '---|' * len(s['splits'])) lines.append('| jumps | ' + ' | '.join(str(v) for v in s['splits']) + ' |') lines.append(f'\nOutput {s["size"][0]}×{s["size"][1]} @ {s["fps"]:.2f} fps, {s["frames"]} frames · ' f'source {info["w"]}×{info["h"]} @ {info["fps"]:.2f} fps') return '\n'.join(lines) def generate(video, fmt_label, title, theme_label, turbo, beep, event_length, framing_label, jumpers, skip_after_miss, count_on_crop, out_fps, fill, smooth, max_zoom, foot_rings, skeleton, show_pace, mirror, progress=gr.Progress()): if not video: raise gr.Error('Please upload a video first.') info = probe(video) if not info['ok']: raise gr.Error('Could not read that video file. Please upload an MP4 or MOV.') if info['duration'] > MAX_SECONDS: raise gr.Error(f'That video is {info["duration"]:.0f} s long; the limit here is {MAX_SECONDS:.0f} s. ' 'Please trim it around the event.') if info['fps'] < 45: gr.Warning(f'This video is {info["fps"]:.0f} fps. The counting model was trained on 60 fps footage, ' 'so counts may be less accurate.') if beep and not event_length: raise gr.Error('Please enter the event length (seconds between the beeps).') cleanup_old_outputs() os.makedirs(OUT_DIR, exist_ok=True) fmt = fmt_label if fmt_label in FORMATS.values() else FORMATS.get(fmt_label, 'social') base = os.path.splitext(os.path.basename(video))[0][:40] or 'video' output = os.path.join(OUT_DIR, f'{base}_nextjump_{fmt}_{uuid.uuid4().hex[:6]}.mp4') argv = ['--video', video, '--output', output, '--onnx', ONNX_PATH, '--format', fmt, '--title', title or '', '--theme', theme_label if theme_label in ('elitejumps', 'nextjump') else 'elitejumps', '--fill', str(fill), '--smooth', str(smooth), '--max_zoom', str(max_zoom)] if out_fps != 'Auto': argv += ['--out_fps', str(out_fps)] if turbo: argv.append('--turbo') framing = framing_label if framing_label in FRAMING.values() else FRAMING.get(framing_label, 'smart') argv += ['--framing', framing, '--jumpers', str(int(jumpers or 1))] if count_on_crop and framing == 'smart': argv.append('--count_on_crop') if skip_after_miss: argv.append('--skip_after_miss') if beep: argv += ['--beep', '--event_length', str(float(event_length))] if foot_rings: argv.append('--fx') if skeleton: argv.append('--skeleton') if not show_pace: argv.append('--no_pace') if mirror: argv.append('--mirror') progress(0, desc='Starting') try: s = nj.main(argv, progress=lambda f, desc: progress(f, desc=desc)) except Exception as e: # surface a readable message in the UI; full trace goes to the logs import traceback traceback.print_exc() raise gr.Error(f'Processing failed: {e}') return s['output'], s['output'], summary_markdown(s, info) def _logo_html(): path = os.path.join(nj.ROOT, nj.THEMES[nj.DEFAULT_THEME]['logo']) logo = '' if os.path.isfile(path): with open(path, 'rb') as f: logo = f'NextJump' return f"""
{logo or f'{nj.THEMES[nj.DEFAULT_THEME]["wordmark"]}'}

Upload a jump-rope video and get a broadcast-ready graphic: AI jump count, speed gauge, splits, rhythm and auto-framing on the jumper. Turn on beep detection for official timed events.

""" CSS = """ #nj-header { background:#111113; border-radius:14px; padding:22px 26px; margin-bottom:6px; border-left:6px solid #E52421; } #nj-header img { height:64px; display:block; margin-bottom:12px; } #nj-header .nj-word { color:#E52421; font-weight:800; font-size:30px; letter-spacing:1px; } #nj-header p { color:#D4D4D8; margin:0; font-size:15px; line-height:1.45; } """ def build_demo(): with gr.Blocks(title='NextJump Broadcast', css=CSS, theme=gr.themes.Soft(primary_hue='red'), delete_cache=(3600, KEEP_OUTPUTS_S)) as demo: gr.HTML(_logo_html()) with gr.Row(equal_height=False): with gr.Column(scale=1): video_in = gr.Video(label='Jump-rope video', sources=['upload']) fmt_in = gr.Radio([(k, v) for k, v in FORMATS.items()], value='social', label='Output format') title_in = gr.Textbox(label='Title', placeholder='e.g. Ali | 30s Speed', max_lines=1) theme_in = gr.Radio([('Elite Jumps', 'elitejumps'), ('NextJump', 'nextjump')], value='elitejumps', label='Graphics theme') turbo_in = gr.Checkbox(label='Turbo mode (instant replay)', value=True, info='About 2x faster: analyses and renders at 30 fps with lighter person detection.') with gr.Group(): beep_in = gr.Checkbox(label='Detect official start / end beeps', value=False, info='Counts only between the start beep and the end beep.') event_len_in = gr.Number(label='Event length (seconds between beeps)', value=30, minimum=5, maximum=600, visible=False) with gr.Group(): framing_in = gr.Radio([(k, v) for k, v in FRAMING.items()], value='smart', label='Framing', info='Smart framing ignores judges and bystanders by checking who is actually jumping.') jumpers_in = gr.Slider(1, 6, value=1, step=1, label='People to frame', info='1 for single rope; 2 for pairs; 3+ for double dutch (turners + jumpers).') skip_in = gr.Checkbox(label='Rulebook miss rule: don\'t count the first jump after a miss', value=False) with gr.Accordion('Advanced', open=False): crop_count_in = gr.Checkbox(label='Count from a crop around the selected jumper(s)', value=False, info='Smart framing only. Can help when other people in frame are moving.') fps_in = gr.Radio(['Auto', 30, 60], value='Auto', label='Output frame rate', info='Auto = 60 fps (30 fps in turbo mode). 30 fps renders about twice as fast.') fill_in = gr.Slider(0.15, 0.85, value=DEFAULT_FILL['social'], step=0.01, label='Jumper size (fraction of frame height)', info='Smaller = wider shot. Capped at the widest crop the video allows.') smooth_in = gr.Slider(0.2, 2.0, value=0.8, step=0.1, label='Camera smoothing (s)') zoom_in = gr.Slider(1.0, 4.0, value=2.5, step=0.1, label='Max zoom') fx_in = gr.Checkbox(label='Foot-strike rings', value=True, info='A ring on the floor at every landing (alternating feet when stepping), during the event.') skeleton_in = gr.Checkbox(label='Show limb tracking (arms & legs)', value=False, info='Shown only during the event. Slower: runs the pose model on every frame.') pace_in = gr.Checkbox(label='Show start-pace comparison', value=True) mirror_in = gr.Checkbox(label='Mirror video', value=False) run_btn = gr.Button('Generate video', variant='primary', size='lg') gr.Markdown('Processing runs on the Space’s hardware and can take several minutes for a ' '30 s clip on CPU. Best results: 60 fps, whole body in frame, one jumper.') with gr.Column(scale=1): video_out = gr.Video(label='NextJump broadcast video', interactive=False) file_out = gr.File(label='Download') summary_out = gr.Markdown() beep_in.change(lambda on: gr.update(visible=on), beep_in, event_len_in) fmt_in.change(lambda fmt: gr.update(value=DEFAULT_FILL.get(fmt, 0.5)), fmt_in, fill_in) framing_in.change(lambda fr: gr.update(visible=fr == 'smart'), framing_in, jumpers_in) run_btn.click(generate, inputs=[video_in, fmt_in, title_in, theme_in, turbo_in, beep_in, event_len_in, framing_in, jumpers_in, skip_in, crop_count_in, fps_in, fill_in, smooth_in, zoom_in, fx_in, skeleton_in, pace_in, mirror_in], outputs=[video_out, file_out, summary_out], concurrency_limit=1) return demo if not IS_RENDER_WORKER: demo = build_demo() demo.queue(max_size=16) if __name__ == '__main__': demo.launch()