#!/usr/bin/env python3 """Start mynah and open it. The preflight below exists because the failure it catches is silent otherwise: the banner prints, uvicorn imports the app, pydantic chokes on modern type syntax thirty frames deep, and the browser opens on a refused connection. The message you get instead names the interpreter that is actually running. """ import argparse import glob import os import shutil import socket # Set before anything from the model stack is imported; see mynah/__init__.py. os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1") os.environ.setdefault("TQDM_DISABLE", "1") import subprocess import sys import threading import webbrowser from pathlib import Path MINIMUM = (3, 10) def find_working_interpreter() -> str: """Look for a Python on this machine that could actually run mynah. Only called when the current one cannot. `python` on a pyenv machine is whatever the shim points at, which is often not the environment the model was installed into — so pointing at a specific working binary is far more use than repeating the install instructions. """ candidates: list[str] = [] for name in ("python3.12", "python3.11", "python3.13", "python3.10", "python3"): found = shutil.which(name) if found: candidates.append(found) candidates += sorted(glob.glob( os.path.expanduser("~/.pyenv/versions/3.1[0-3]*/bin/python")), reverse=True) candidates += sorted(glob.glob(str(Path(__file__).resolve().parent / ".venv/bin/python"))) probe = "import fastapi, uvicorn, chatterbox.tts_turbo" for path in dict.fromkeys(candidates): # de-duplicate, keep order try: done = subprocess.run([path, "-c", probe], capture_output=True, timeout=60) except Exception: # noqa: BLE001 - a broken candidate is just not the answer continue if done.returncode == 0: return path return "" def preflight() -> str: """Return a problem to report, or "" if the environment can run mynah.""" if sys.version_info < MINIMUM: want = ".".join(str(n) for n in MINIMUM) message = ( f"mynah needs Python {want} or newer, but this is " f"{sys.version.split()[0]}:\n {sys.executable}\n" ) print(f"\n{message}\nlooking for an interpreter that works...", file=sys.stderr) working = find_working_interpreter() if working: return message + f"\nThis one has everything mynah needs:\n {working} run.py" return message + ( "\nNo suitable interpreter found. From the project root:\n" " python3.12 -m venv .venv\n" " source .venv/bin/activate\n" " pip install -r requirements.txt\n" " python run.py" ) missing = [] for module, package in (("fastapi", "fastapi"), ("uvicorn", "uvicorn"), ("soundfile", "soundfile"), ("multipart", "python-multipart")): try: __import__(module) except ImportError: missing.append(package) if missing: return ("Missing dependencies: " + ", ".join(missing) + "\n" " pip install -r requirements.txt") try: __import__("chatterbox.tts_turbo") except ImportError as error: # Not fatal: the server still runs and reports this in the UI. But # saying it here saves a confused trip through the browser. print(f"warning: the TTS model is not importable, so nothing will " f"generate.\n {error}\n pip install -r requirements.txt\n", file=sys.stderr) return "" def port_in_use(host: str, port: int) -> bool: """Check before starting rather than after. uvicorn catches the bind failure itself and exits, so an OSError never reaches this file — and the banner would already have printed, making a dead start look like a live one. """ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: probe.bind((host, port)) return False except OSError: return True def main() -> int: parser = argparse.ArgumentParser(description="self-hosted voice cloning TTS") parser.add_argument("--host", default="127.0.0.1", help="bind address; the default keeps it off the network") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--no-open", action="store_true") parser.add_argument("--download", action="store_true", help="fetch the model weights now and exit; the server " "does this on first start otherwise") args = parser.parse_args() problem = preflight() if problem: print(f"\n{problem}\n", file=sys.stderr) return 1 os.chdir(Path(__file__).resolve().parent) from mynah.engine import ENGINE, MODEL_DIR_ENV if args.download: cached = ENGINE.cached() if cached: print(f"weights already here: {cached[0]} ({cached[1] / 1e9:.2f} GB)") return 0 print("downloading weights (~3 GB, anonymous, from huggingface.co)...") path = ENGINE.download(log=print) size = ENGINE.cached() print(f"done: {path} ({(size[1] if size else 0) / 1e9:.2f} GB)") return 0 # Only a server needs the port; --download above must work while one runs. if port_in_use(args.host, args.port): print(f"\nPort {args.port} is already in use - mynah may already be " f"running at\n http://localhost:{args.port}\n\n" f"Open that, or start another copy with --port {args.port + 1}.\n", file=sys.stderr) return 1 cached = ENGINE.cached() if cached: origin = "MYNAH_MODEL_DIR" if os.environ.get(MODEL_DIR_ENV) else "cache" print(f"weights: {cached[1] / 1e9:.2f} GB in {origin} | device: {ENGINE.device}", flush=True) else: print("weights: not downloaded yet - first start fetches ~3 GB " "(or run `python run.py --download` first)", flush=True) import uvicorn url = f"http://{'localhost' if args.host == '127.0.0.1' else args.host}:{args.port}" print(f"mynah -> {url}", flush=True) if not args.no_open: threading.Timer(1.5, webbrowser.open, [url]).start() uvicorn.run("mynah.server:app", host=args.host, port=args.port, log_level="warning") return 0 if __name__ == "__main__": raise SystemExit(main())