scripts/omni_o_call.py

import argparse, os, sys, json, time, math, torch, threading, queue, base64, io, logging, contextlib, warnings
import numpy as np
from PIL import Image
warnings.filterwarnings('ignore')
logging.getLogger().setLevel(logging.ERROR)

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from models.vam import VAM, VAMConfig
from serve.realtime import RealtimeSession

M = {}
V = {}  # voice_name -> {ref_codes, spk_emb}
MODEL_LOCK = threading.Lock()
VOICES_BUILTIN, VOICES_UNSEEN, VOICES_MANUAL = [], [], []
SAMPLES_PER_FRAME = 1920
REF_FRAMES = 300
CLONE_VOICE = 'voice_clone'
CLONE_FILE = 'voice_clone.pt'

def sse(d): return f"data: {json.dumps(d)}\n\n"

def asr_run(samples):
    from funasr.utils.postprocess_utils import rich_transcription_postprocess
    r = M['asr'].generate(input=samples, cache={}, language='auto', use_itn=True)
    return rich_transcription_postprocess(r[0]['text']).strip() if r else ''

def prep_audio(samples):
    m, dev = M['model'], M['device']
    proc = m.audio_processor(samples, sampling_rate=16000, return_tensors="pt", return_attention_mask=True)
    mel = proc.input_features.squeeze(0).unsqueeze(0).to(dev)
    vlen = proc.attention_mask.sum().item()
    return mel, torch.tensor([vlen], device=dev), m.config.audio_special_token * (vlen or 1)

def prep_image(b64):
    img = Image.open(io.BytesIO(base64.b64decode(b64))).convert('RGB')
    return {k: v.to(M['device']) for k, v in M['model'].vision_processor(images=img, return_tensors="pt").items()}

def build_ids(prompt, history):
    tok, dev = M['tokenizer'], M['device']
    cfg = M['cfg']
    hist = history[-cfg.max_history_turns:] if cfg.max_history_turns > 0 else []
    msgs = hist + [{"role": "user", "content": prompt}]
    t = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
    return torch.tensor(tok(t)['input_ids'], dtype=torch.long, device=dev)[None, ...]

def _mimi_decode(frames):
    codes = [f for f in frames if f and len(f) == 8]
    if not codes or not M['mimi']: return None
    mc = torch.tensor(codes, dtype=torch.long, device=M['device']).T.unsqueeze(0)
    mc = torch.where(mc >= 2049, torch.zeros_like(mc), mc)
    with torch.no_grad():
        au = M['mimi'].decode(mc).audio_values.squeeze().cpu().numpy()
    return au, mc.shape[-1]

def pcm_bytes(frames, ov):
    r = _mimi_decode(frames)
    if r is None: return None
    au, T = r
    if ov > 0: au = au[int(ov * len(au) / T):]
    return (au * 32767).astype('int16').tobytes()

def stream_pcm(frames, flush=False):
    if not M['mimi']: return
    cf, ov_max, n = M['cfg'].audio_chunk_frames, M['cfg'].audio_overlap, len(frames)
    if not flush and n >= cf and n % cf == 0:
        ov = min(ov_max, n - cf)
        p = pcm_bytes(frames[-(cf + ov):], ov)
        if p: yield p
    elif flush:
        rem = n % cf
        if rem:
            ov = min(ov_max, n - rem)
            p = pcm_bytes(frames[-(rem + ov):], ov)
            if p: yield p

def clone_voice_path():
    p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'model', 'speaker', CLONE_FILE)
    os.makedirs(os.path.dirname(p), exist_ok=True)
    return p

def delete_manual_voice(name):
    if name not in VOICES_MANUAL:
        raise RuntimeError('can only delete manually cloned voices')
    out_path = clone_voice_path()
    saved = torch.load(out_path, map_location='cpu') if os.path.exists(out_path) else {}
    if name in saved:
        saved.pop(name)
        torch.save(saved, out_path)
    V.pop(name, None)
    if name in VOICES_MANUAL:
        VOICES_MANUAL.remove(name)

def normalize_voice_name(name):
    name = ' '.join(str(name or '').split())
    if not name:
        name = CLONE_VOICE
    if len(name) > 24:
        raise RuntimeError('voice name too long (max 24 chars)')
    if name.lower() == 'default':
        raise RuntimeError('default is reserved')
    if name in VOICES_BUILTIN or name in VOICES_UNSEEN:
        raise RuntimeError('name already taken by an existing voice')
    return name

def validate_clone_audio(w16):
    if w16.numel() < int(16000 * 1.8):
        raise RuntimeError('audio too short, read the full sentence')
    peak = w16.abs().max().item()
    frame, hop = 800, 400
    if w16.numel() >= frame:
        rms = w16.unfold(0, frame, hop).pow(2).mean(dim=1).sqrt().cpu().numpy()
    else:
        rms = np.array([w16.pow(2).mean().sqrt().item()])
    hi = float(np.quantile(rms, 0.95))
    lo = float(np.quantile(rms, 0.2))
    if hi < 0.008:
        raise RuntimeError('audio too quiet, move closer to mic')
    if hi > 0 and lo / hi > 0.45:
        raise RuntimeError('too much background noise')
    if peak > 0.995:
        raise RuntimeError('audio clipped, move away from mic')

def build_clone_voice(audio_b64):
    if M.get('mimi') is None or M.get('campplus') is None or M.get('mel_fn') is None:
        raise RuntimeError('Mimi or CAM++ not loaded')
    from pydub import AudioSegment
    seg = AudioSegment.from_file(io.BytesIO(base64.b64decode(audio_b64))).set_channels(1).set_sample_width(2)
    if len(seg) < 1000:
        raise RuntimeError('audio too short, record at least 1 second')
    try:
        seg = seg.speedup(playback_speed=1.5, chunk_size=150, crossfade=25)
    except Exception:
        seg = seg.speedup(playback_speed=1.5)
    seg24 = seg.set_frame_rate(24000)
    seg16 = seg.set_frame_rate(16000)
    w24 = torch.tensor(np.frombuffer(seg24.raw_data, dtype=np.int16).astype(np.float32) / 32768.0)
    w16 = torch.tensor(np.frombuffer(seg16.raw_data, dtype=np.int16).astype(np.float32) / 32768.0)
    validate_clone_audio(w16)
    mimi_dev = next(M['mimi'].parameters()).device
    mimi_dtype = torch.float16 if mimi_dev.type != 'cpu' else torch.float32
    with torch.inference_mode():
        t = w24.unsqueeze(0).unsqueeze(0).to(device=mimi_dev, dtype=mimi_dtype)
        codes = M['mimi'].encode(t).audio_codes
        nf = math.ceil(w24.shape[-1] / SAMPLES_PER_FRAME)
        ref_codes = codes[0, :8, :nf].cpu()[:, :min(nf, REF_FRAMES)]
    with torch.no_grad():
        mel = M['mel_fn'](w16.unsqueeze(0).to(M['device']))
        feat = mel.clamp(min=1e-10).log().transpose(1, 2)
        feat = feat - feat.mean(dim=1, keepdim=True)
        spk_emb = M['campplus'](feat).squeeze(0).cpu()
    return {'ref_codes': ref_codes, 'spk_emb': spk_emb}

def register_voice(name, value, group='manual'):
    V[name] = value
    groups = {'builtin': VOICES_BUILTIN, 'unseen': VOICES_UNSEEN, 'manual': VOICES_MANUAL}
    dst = groups[group]
    if name not in dst: dst.append(name)
    for k, lst in groups.items():
        if k != group and name in lst: lst.remove(name)

def voice_args(name):
    if name and name != 'default' and name in V:
        v = V[name]
        dev = M['device']
        rc = v['ref_codes'].unsqueeze(0).to(dev)
        se = v['spk_emb'].half().unsqueeze(0).to(dev) if 'spk_emb' in v else None
        return {'ref_codes': rc, 'spk_emb': se}
    return {}

def run_generate(x, audio_inputs, audio_lens, pixel_values, **kw):
    with MODEL_LOCK, torch.no_grad():
        yield from M['model'].generate(
            x, M['tokenizer'].eos_token_id, stream=True, return_audio_codes=True,
            audio_inputs=audio_inputs, audio_lens=audio_lens, pixel_values=pixel_values, **kw)

def prepare_turn(text, samples, image_b64, do_asr_for_image):
    audio_inputs = audio_lens = pixel_values = None
    prompt = text or ''
    user_text = text or ''
    asr_thread, asr_result = None, [None]
    if samples is not None:
        if image_b64 and do_asr_for_image:
            user_text = asr_run(samples)
            prompt = user_text
        else:
            audio_inputs, audio_lens, prompt = prep_audio(samples)
            if M['cfg'].max_history_turns > 0:
                sa = samples.copy()
                def _a(): asr_result[0] = asr_run(sa)
                asr_thread = threading.Thread(target=_a); asr_thread.start()
    if image_b64:
        pixel_values = prep_image(image_b64)
        m = M['model']
        prompt = (prompt + "\n\n" if prompt else "") + m.config.image_special_token * m.config.image_token_len
    return audio_inputs, audio_lens, pixel_values, prompt, user_text, asr_thread, asr_result


def init_web_app():
    from flask import Flask, request, Response, send_from_directory
    from flask_cors import CORS
    from flask_sock import Sock

    app = Flask(__name__, static_folder='.')
    CORS(app)
    sock = Sock(app)

    @app.route('/')
    def index(): return send_from_directory('.', 'omni_o_web.html')
    @app.route('/call')
    def call_page(): return send_from_directory('.', 'omni_o_web.html')

    @app.route('/voices')
    def get_voices():
        return json.dumps({'builtin': sorted(VOICES_BUILTIN), 'unseen': sorted(VOICES_UNSEEN), 'manual': sorted(VOICES_MANUAL)})

    @app.route('/models')
    def get_models():
        return json.dumps({'models': [M.get('model_name', 'omni-o')], 'current': M.get('model_name', 'omni-o')})

    @app.route('/clone_voice', methods=['POST'])
    def clone_voice():
        d = request.json or {}
        if not d.get('audio'):
            return Response(json.dumps({'ok': False, 'error': 'missing audio'}), status=400, mimetype='application/json')
        try:
            name = normalize_voice_name(d.get('name'))
            value = build_clone_voice(d['audio'])
            out_path = clone_voice_path()
            saved = torch.load(out_path, map_location='cpu') if os.path.exists(out_path) else {}
            saved[name] = value
            torch.save(saved, out_path)
            register_voice(name, value, group='manual')
            return Response(json.dumps({'ok': True, 'voice': name, 'path': './model/speaker/' + CLONE_FILE}), mimetype='application/json')
        except Exception as e:
            return Response(json.dumps({'ok': False, 'error': str(e)}), status=500, mimetype='application/json')

    @app.route('/delete_voice', methods=['POST'])
    def delete_voice():
        d = request.json or {}
        name = ' '.join(str(d.get('name') or '').split())
        if not name:
            return Response(json.dumps({'ok': False, 'error': 'missing name'}), status=400, mimetype='application/json')
        try:
            delete_manual_voice(name)
            return Response(json.dumps({'ok': True, 'voice': name}), mimetype='application/json')
        except Exception as e:
            return Response(json.dumps({'ok': False, 'error': str(e)}), status=500, mimetype='application/json')

    @app.route('/chat', methods=['POST'])
    def chat():
        d = request.json
        history = d.get('history', [])
        samples = None
        if d.get('audio'):
            from pydub import AudioSegment
            seg = AudioSegment.from_file(io.BytesIO(base64.b64decode(d['audio']))).set_frame_rate(16000).set_channels(1).set_sample_width(2)
            samples = np.frombuffer(seg.raw_data, dtype=np.int16).astype(np.float32) / 32768.0
        va = voice_args(d.get('voice', 'default'))

        def gen():
            audio_inputs, audio_lens, pixel_values, prompt, user_text, asr_th, asr_res = prepare_turn(
                d.get('text', ''), samples, d.get('image'), do_asr_for_image=True)
            if torch.cuda.is_available():
                torch.cuda.synchronize()
            x = build_ids(prompt, history)
            asr_sent = False
            if user_text and samples is not None and d.get('image'):
                yield sse({'type': 'user_prompt', 'content': user_text}); asr_sent = True
            frames, text_ttft, audio_ttft = [], None, None
            t0 = time.time(); hi = 0
            for y, af in run_generate(x, audio_inputs, audio_lens, pixel_values,
                                       max_new_tokens=d.get('max_tokens', 512),
                                       temperature=d.get('temperature', 1), top_p=0.85, **va):
                if not asr_sent and asr_th and not asr_th.is_alive():
                    asr_th.join()
                    if asr_res[0]: yield sse({'type': 'user_prompt', 'content': asr_res[0]})
                    asr_sent = True
                if y is not None:
                    if text_ttft is None:
                        text_ttft = (time.time() - t0) * 1000
                        yield sse({'type': 'ttft', 'text_ttft': round(text_ttft, 1)})
                    ans = M['tokenizer'].decode(y[0].tolist(), skip_special_tokens=True)
                    if ans and ans[-1] != '\ufffd' and len(ans) > hi:
                        yield sse({'type': 'text', 'content': ans[hi:]}); hi = len(ans)
                if af:
                    if audio_ttft is None:
                        audio_ttft = (time.time() - t0) * 1000
                        yield sse({'type': 'ttft', 'audio_ttft': round(audio_ttft, 1)})
                    frames.append(af)
                    for pcm in stream_pcm(frames):
                        b64 = base64.b64encode(pcm).decode()
                        for i in range(0, len(b64), 2000):
                            yield sse({'type': 'pcm', 'c': b64[i:i+2000], 'd': i+2000 >= len(b64)})
            for pcm in stream_pcm(frames, flush=True):
                b64 = base64.b64encode(pcm).decode()
                for i in range(0, len(b64), 2000):
                    yield sse({'type': 'pcm', 'c': b64[i:i+2000], 'd': i+2000 >= len(b64)})
            if not asr_sent:
                if asr_th:
                    asr_th.join()
                    if asr_res[0]: yield sse({'type': 'user_prompt', 'content': asr_res[0]})
                else:
                    yield sse({'type': 'user_prompt', 'content': prompt})
            yield sse({'type': 'done'})
        return Response(gen(), mimetype='text/event-stream', headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})

    @sock.route('/ws/realtime')
    def realtime(ws):
        session = RealtimeSession(M['vad_path'])
        q = queue.Queue(); alive = [True]; state = {'history': [], 'image': None}
        n_hist = M['cfg'].max_history_turns

        def push_audio(data):
            return session.push_chunk(np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0)

        def set_ctx(msg):
            h = msg.get('history') or []
            state['history'] = h[-n_hist:] if n_hist > 0 else []
            if 'image' in msg: state['image'] = msg.get('image')
            if 'voice' in msg: state['voice'] = msg.get('voice', 'default')

        def poll_interrupt():
            while True:
                try: data = q.get_nowait()
                except queue.Empty: return False
                if isinstance(data, bytes):
                    if push_audio(data) == 'interrupt': return True
                    ws.send(json.dumps({'type': 'vad', 'speaking': session.speaking}))
                else:
                    m = json.loads(data)
                    if m.get('type') == 'context': set_ctx(m)
                    elif m.get('type') in ('stop', 'end'):
                        if m['type'] == 'end': alive[0] = False
                        session.interrupt = True; return True

        def recv_loop():
            while alive[0]:
                try:
                    data = ws.receive(timeout=1)
                    if data is None: alive[0] = False; break
                    q.put(data)
                except: alive[0] = False; break

        threading.Thread(target=recv_loop, daemon=True).start()
        try:
            while alive[0]:
                try: data = q.get(timeout=0.05)
                except queue.Empty: continue
                if isinstance(data, str):
                    m = json.loads(data)
                    if m.get('type') == 'context': set_ctx(m)
                    elif m.get('type') == 'stop': session.interrupt = True
                    elif m.get('type') == 'end': break
                    continue
                if session.generating:
                    push_audio(data); ws.send(json.dumps({'type': 'vad', 'speaking': session.speaking})); continue
                status = push_audio(data)
                ws.send(json.dumps({'type': 'vad', 'speaking': session.speaking}))
                if status != 'speech_end': continue

                session.generating = True
                audio = session.get_audio()
                ws.send(json.dumps({'type': 'generating'}))
                audio_inputs, audio_lens, pixel_values, prompt, user_text, asr_th, asr_res = prepare_turn(
                    '', audio, state['image'], do_asr_for_image=True)
                if state['image']: state['image'] = None
                if torch.cuda.is_available():
                    torch.cuda.synchronize()
                x = build_ids(prompt, state['history'])
                va_rt = voice_args(state.get('voice', 'default'))

                frames, full_text, interrupted = [], '', False
                for y, af in run_generate(x, audio_inputs, audio_lens, pixel_values,
                                           max_new_tokens=512, temperature=0.7, **va_rt):
                    if poll_interrupt() or session.interrupt: interrupted = True; break
                    if y is not None:
                        ans = M['tokenizer'].decode(y[0].tolist(), skip_special_tokens=True)
                        if ans and ans[-1] != '\ufffd' and len(ans) > len(full_text):
                            ws.send(json.dumps({'type': 'text', 'content': ans[len(full_text):]})); full_text = ans
                    if af:
                        frames.append(af)
                        for pcm in stream_pcm(frames):
                            ws.send(json.dumps({'type': 'pcm', 'data': base64.b64encode(pcm).decode()}))
                if not interrupted:
                    for pcm in stream_pcm(frames, flush=True):
                        ws.send(json.dumps({'type': 'pcm', 'data': base64.b64encode(pcm).decode()}))
                if asr_th:
                    asr_th.join(); user_text = asr_res[0] or user_text
                if n_hist > 0:
                    if user_text: state['history'].append({'role': 'user', 'content': user_text})
                    if full_text: state['history'].append({'role': 'assistant', 'content': full_text})
                    state['history'] = state['history'][-n_hist:]
                ws.send(json.dumps({'type': 'done', 'interrupted': interrupted or session.interrupt}))
                session.generating = False; session.interrupt = False
        finally:
            alive[0] = False
    return app


def init_model(args):
    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    M['cfg'] = args
    M['device'] = args.device

    with contextlib.redirect_stdout(io.StringIO()):
        from funasr import AutoModel
        M['asr'] = AutoModel(model=os.path.join(root, args.sensevoice_dir), trust_remote_code=True, device=args.device, disable_update=True)

    ckpt_dir = os.path.join(root, args.load_from)
    is_hf = os.path.exists(os.path.join(ckpt_dir, 'config.json')) and \
            (os.path.exists(os.path.join(ckpt_dir, 'model.safetensors')) or
             os.path.exists(os.path.join(ckpt_dir, 'pytorch_model.bin')))

    if is_hf:
        model = VAM.from_pretrained(
            ckpt_dir,
            audio_encoder_path=os.path.join(root, args.sensevoice_dir),
            vision_model_path=os.path.join(root, args.siglip_dir),
        )
    else:
        config = VAMConfig(
            hidden_size=args.hidden_size,
            num_hidden_layers=args.num_hidden_layers,
            num_attention_heads=args.hidden_size // 96,
            num_key_value_heads=args.hidden_size // 192,
            use_moe=args.use_moe,
        )
        weight = args.weight
        if not weight.endswith('.pth'):
            if args.use_moe and not weight.endswith('_moe'):
                weight = f'{weight}_moe.pth'
            else:
                weight = f'{weight}.pth'
        ckpt_path = os.path.join(ckpt_dir, weight)
        model = VAM(config,
                    audio_encoder_path=os.path.join(root, args.sensevoice_dir),
                    vision_model_path=os.path.join(root, args.siglip_dir))
        state = torch.load(ckpt_path, map_location='cpu', weights_only=True)
        missing, unexpected = model.load_state_dict(state, strict=False)
        if missing:
            print(f'  Missing keys (expected for encoders): {len(missing)}')
        if unexpected:
            print(f'  Unexpected keys: {len(unexpected)}')

    M['model'] = model.half().eval().to(args.device)
    if model.audio_encoder is not None:
        model.audio_encoder.to(args.device)
    if model.vision_encoder is not None:
        model.vision_encoder.to(args.device)

    tok_dir = os.path.join(root, args.tokenizer_dir)
    from transformers import AutoTokenizer
    M['tokenizer'] = AutoTokenizer.from_pretrained(tok_dir)

    model_name = args.weight or os.path.basename(args.load_from.rstrip('/'))
    M['model_name'] = model_name
    params = sum(p.numel() for p in model.parameters()) / 1e6
    print(f'Loaded {model_name}: {params:.2f}M')

    try:
        from transformers import MimiModel
        mimi_path = os.path.join(root, args.mimi_dir)
        M['mimi'] = MimiModel.from_pretrained(mimi_path).eval().to(args.device)
        if args.device != 'cpu':
            M['mimi'] = M['mimi'].half()
        print('Mimi loaded')
    except Exception as e:
        M['mimi'] = None
        print(f'Mimi load failed: {e}')

    try:
        from modelscope.models.audio.sv.DTDNN import CAMPPlus
        import torchaudio
        M['campplus'] = CAMPPlus(feat_dim=80, embedding_size=192, growth_rate=32, bn_size=4,
                                 init_channels=128, config_str='batchnorm-relu', memory_efficient=True)
        camp_path = os.path.join(root, 'checkpoint/campplus/campplus_cn_common.pt')
        sd = torch.load(camp_path, map_location='cpu')
        M['campplus'].load_state_dict({k: v.float() for k, v in sd.items()})
        M['campplus'] = M['campplus'].eval().to(args.device)
        M['mel_fn'] = torchaudio.transforms.MelSpectrogram(
            sample_rate=16000, n_fft=512, win_length=400, hop_length=160,
            n_mels=80, f_min=20, f_max=7600, norm='slaney', mel_scale='slaney',
        ).to(args.device)
        print('CAM++ loaded')
    except Exception as e:
        M['campplus'] = M['mel_fn'] = None
        print(f'CAM++ load failed (voice clone will be unavailable): {e}')

    M['vad_path'] = os.path.join(root, args.vad_dir, 'silero_vad.onnx')

    spk_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'model', 'speaker')
    os.makedirs(spk_dir, exist_ok=True)
    for fn, group in [('voices.pt', 'builtin'), ('voices_unseen.pt', 'unseen'), (CLONE_FILE, 'manual')]:
        fp = os.path.join(spk_dir, fn)
        if os.path.exists(fp):
            for speaker, v in torch.load(fp, map_location='cpu').items():
                if speaker not in V or fn == CLONE_FILE:
                    register_voice(speaker, v, group=group)
    if V: print(f'Loaded {len(V)} voices')
    if not V:
        print('No voices loaded. Use Voice Clone to enable audio output.')

    print('Warmup...')
    with torch.no_grad():
        ids = torch.tensor([[1, 2, 3]], device=args.device)
        au = torch.full((1, 8, 3), 2049, dtype=torch.long, device=args.device)
        M['model'].forward(torch.cat((au, ids.unsqueeze(1)), dim=1))
        if M['model'].audio_encoder is not None:
            try:
                M['model'].audio_encoder.model(torch.zeros(1, 100, 560, device=args.device), torch.tensor([100], device=args.device))
            except Exception:
                pass
        if M['mimi']:
            M['mimi'].decode(torch.zeros(1, 8, 1, dtype=torch.long, device=args.device))
    print('Warmup done! Ready.')


if __name__ == '__main__':
    p = argparse.ArgumentParser(description='Omni-O Real-time Voice Call')
    p.add_argument('--load_from', default='checkpoint/omni-o/omni-o-hf', help='HF 模型权重目录（自动检测 .pth 目录兼容）')
    p.add_argument('--weight', default='', help='权重文件名（仅 .pth 模式，不含后缀）')
    p.add_argument('--tokenizer_dir', default='checkpoint/omni/native_hf', help='tokenizer目录')
    p.add_argument('--sensevoice_dir', default='checkpoint/sensevoice', help='SenseVoice ASR目录')
    p.add_argument('--siglip_dir', default='checkpoint/siglip', help='SigLIP视觉编码器目录')
    p.add_argument('--mimi_dir', default='checkpoint/mimi', help='Mimi解码器目录')
    p.add_argument('--vad_dir', default='checkpoint/vad', help='VAD模型目录')
    p.add_argument('--hidden_size', default=768, type=int)
    p.add_argument('--num_hidden_layers', default=8, type=int)
    p.add_argument('--use_moe', default=0, type=int, choices=[0, 1])
    p.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu')
    p.add_argument('--port', default=7860, type=int)
    p.add_argument('--audio_chunk_frames', default=4, type=int)
    p.add_argument('--audio_overlap', default=2, type=int)
    p.add_argument('--max_history_turns', default=0, type=int)
    args = p.parse_args()

    init_model(args)
    app = init_web_app()
    print(f'Omni-O Call server started at http://0.0.0.0:{args.port}/')
    print(f'Open http://127.0.0.1:{args.port} in Firefox for mic/camera access')
    app.run(host='0.0.0.0', port=args.port, threaded=True)