Speakturbo Tts

name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  rust-cli:
    name: Build Rust CLI
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
        os: [macos-latest, ubuntu-latest]
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Install Rust
        uses: dtolnay/rust-action@stable
      
      - name: Build
        working-directory: speakturbo-cli
        run: cargo build --release
      
      - name: Upload binary
        uses: actions/upload-artifact@v4
        with:
          name: speakturbo-${{ matrix.os }}
          path: speakturbo-cli/target/release/speakturbo

  python-daemon:
    name: Test Python Daemon
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install pocket-tts uvicorn fastapi pytest
      
      - name: Lint
        run: |
          pip install ruff
          ruff check speakturbo/
      
      - name: Test imports
        run: |
          python -c "from speakturbo.daemon_streaming import app"
          python -c "from speakturbo.cli import main"

name: Install Test
on:
  workflow_dispatch:
  
jobs:
  install:
    runs-on: ubuntu-latest
    steps:
      - name: Install skills
        run: |
          npx -y skills add EmZod/speak -y || true
          npx -y skills add EmZod/Speak-Turbo -y || true

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
.venv/
venv/
ENV/

# Rust
target/
Cargo.lock

# IDE
.idea/
.vscode/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# Project
*.wav
*.mp3
*.log
/tmp/

#!/bin/bash
set -e

echo "Installing Speak-Turbo..."

# Cleanup on failure
trap 'echo "Install failed." >&2' ERR

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

# Install Python dependencies
# Version bounds match pyproject.toml [project.dependencies] — Keep in sync
echo "→ Installing Python dependencies..."
pip install --quiet "pocket-tts>=0.1.0,<1.0" "uvicorn>=0.20.0,<1.0" "fastapi>=0.100.0,<1.0" "python-dateutil>=2.7,<3.0"

# Create bin directory
mkdir -p ~/.local/bin

# Build Rust CLI from local source, or fall back to Python wrapper
if command -v cargo &> /dev/null && [ -d "$SCRIPT_DIR/speakturbo-cli" ]; then
    echo "→ Building Rust CLI from local source..."
    cd "$SCRIPT_DIR/speakturbo-cli"
    cargo build --release --quiet
    cp target/release/speakturbo ~/.local/bin/
else
    echo "→ Rust or local source not found. Installing Python CLI wrapper..."
    cat > ~/.local/bin/speakturbo << 'EOF'
#!/bin/bash
# Fallback wrapper - runs Python CLI
python -m speakturbo.cli "$@"
EOF
    chmod +x ~/.local/bin/speakturbo
fi

# Install Python package
echo "→ Installing daemon..."
if [ -d "$SCRIPT_DIR/speakturbo" ]; then
    pip install --quiet -e "$SCRIPT_DIR"
fi

# Add to PATH if needed
if [[ ":$PATH:" != *":$HOME/.local/bin:"* ]]; then
    echo ""
    echo "Add to your shell profile:"
    echo '  export PATH="$HOME/.local/bin:$PATH"'
    echo ""
fi

echo "✓ Speak-Turbo installed!"
echo ""
echo "Test it:"
echo "  speakturbo \"Hello world\""

MIT License

Copyright (c) 2026 EmZod

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

{
  "name": "@jay-zod/speakturbo",
  "version": "1.0.7",
  "description": "Ultra-fast TTS with ~90ms latency. Local text-to-speech for AI agents with 8 built-in voices.",
  "keywords": ["tts", "text-to-speech", "mlx", "apple-silicon", "pi-package", "speakturbo-tts", "voice", "audio", "ai-agent", "claude-code", "skills"],
  "author": "jay-zod",
  "license": "MIT",
  "repository": {
    "type": "git",
    "url": "https://github.com/EmZod/Speak-Turbo"
  },
  "pi": {
    "skills": ["SKILL.md"]
  }
}

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "speakturbo"
version = "0.1.0"
description = "Ultra-fast local TTS daemon for AI agents"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
keywords = ["tts", "text-to-speech", "ai", "agents", "voice"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
    "Topic :: Multimedia :: Sound/Audio :: Speech",
]
dependencies = [
    "pocket-tts>=0.1.0",
    "fastapi>=0.100.0",
    "uvicorn>=0.20.0",
    "python-dateutil>=2.7",  # Required by matplotlib (pocket-tts dependency)
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "pytest-asyncio>=0.20",
]

[project.scripts]
speakturbo-daemon = "speakturbo.daemon_streaming:main"

[project.urls]
Homepage = "https://github.com/EmZod/Speak-Turbo"
Repository = "https://github.com/EmZod/Speak-Turbo"
Issues = "https://github.com/EmZod/Speak-Turbo/issues"

[tool.setuptools.packages.find]
include = ["speakturbo*"]

[package]
name = "speakturbo"
version = "0.1.0"
edition = "2021"

[dependencies]
clap = { version = "4", features = ["derive"] }
ureq = "2"
rodio = { version = "0.17", default-features = false, features = ["wav"] }
anyhow = "1"
dirs = "5"

[profile.release]
lto = true
codegen-units = 1
strip = true

use anyhow::{Context, Result};
use clap::Parser;
use rodio::{OutputStream, Sink, Source};
use std::collections::VecDeque;
use std::io::Read;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

const DAEMON_URL: &str = "http://127.0.0.1:7125";
const SAMPLE_RATE: u32 = 24000;

// Buffer size: 150ms provides stable playback without perceptible latency
const MIN_BUFFER_MS: u32 = 150;
// Fade-in duration: 10ms (240 samples) eliminates startup transients
const FADE_IN_SAMPLES: usize = 240;
const MIN_BUFFER_SAMPLES: usize = (SAMPLE_RATE * MIN_BUFFER_MS / 1000) as usize;

#[derive(Parser)]
#[command(name = "speakturbo")]
#[command(about = "Ultra-fast TTS CLI")]
#[command(version)]
struct Args {
    /// Text to speak
    text: Option<String>,

    #[arg(short, long, default_value = "alba")]
    voice: String,

    #[arg(short, long)]
    output: Option<String>,

    /// Allow output to this directory (repeatable)
    #[arg(long)]
    allow_dir: Vec<String>,

    #[arg(long)]
    list_voices: bool,
    
    /// Quiet mode - minimal output
    #[arg(short, long)]
    quiet: bool,
}

/// Resolve a path like Python's os.path.realpath: canonicalize the deepest
/// existing ancestor, keep non-existent tail components as-is.
/// This avoids the std::fs::canonicalize pitfall (fails if file doesn't exist)
/// while still resolving symlinks (critical: macOS /tmp -> /private/tmp).
fn resolve_path(raw: &str) -> std::path::PathBuf {
    use std::path::{Path, PathBuf};

    let path = if Path::new(raw).is_relative() {
        std::env::current_dir().unwrap_or_default().join(raw)
    } else {
        PathBuf::from(raw)
    };

    // Fast path: file already exists, canonicalize the whole thing
    if let Ok(p) = std::fs::canonicalize(&path) {
        return p;
    }

    // Walk up to find the deepest existing ancestor
    let mut tail: Vec<std::ffi::OsString> = Vec::new();
    let mut ancestor = path.clone();
    loop {
        if ancestor.exists() {
            let base = std::fs::canonicalize(&ancestor).unwrap_or(ancestor);
            let mut result = base;
            for component in tail.into_iter().rev() {
                result = result.join(component);
            }
            return result;
        }
        match ancestor.file_name() {
            Some(name) => {
                tail.push(name.to_os_string());
                ancestor = ancestor
                    .parent()
                    .map(|p| p.to_path_buf())
                    .unwrap_or(ancestor);
            }
            None => break, // At root
        }
    }

    path // Fallback (shouldn't happen — root always exists)
}

fn load_allowed_dirs() -> Vec<std::path::PathBuf> {
    let Some(home) = dirs::home_dir() else {
        return vec![];
    };
    let config_path = home.join(".speakturbo").join("config");
    let Ok(contents) = std::fs::read_to_string(&config_path) else {
        return vec![];
    };

    contents
        .lines()
        .map(|l| l.trim())
        .filter(|l| !l.is_empty() && !l.starts_with('#'))
        .filter_map(|l| {
            // Expand ~ to home dir (like Python's os.path.expanduser)
            let expanded = if l.starts_with("~/") {
                home.join(&l[2..])
            } else if l == "~" {
                home.clone()
            } else {
                std::path::PathBuf::from(l)
            };
            if expanded.is_absolute() {
                Some(expanded)
            } else {
                None
            }
        })
        .collect()
}

fn validate_output_path(output: &str, extra_allowed: &[String]) -> Result<std::path::PathBuf> {
    use std::path::{Path, PathBuf};

    let resolved = resolve_path(output);

    let mut allowed: Vec<PathBuf> = vec![
        PathBuf::from("/tmp"),
        PathBuf::from("/var/tmp"),
        std::env::temp_dir(),
    ];

    if let Ok(cwd) = std::env::current_dir() {
        allowed.push(cwd);
    }
    if let Some(home) = dirs::home_dir() {
        allowed.push(home.join(".speakturbo"));
    }

    allowed.extend(load_allowed_dirs());

    for dir in extra_allowed {
        allowed.push(resolve_path(dir));
    }

    // Canonicalize ALL allowlist entries (critical for macOS: /tmp -> /private/tmp)
    let allowed: Vec<PathBuf> = allowed
        .into_iter()
        .map(|d| std::fs::canonicalize(&d).unwrap_or(d))
        .collect();

    for allowed_dir in &allowed {
        // PathBuf::starts_with is component-aware: /tmp won't match /tmpevil
        if resolved.starts_with(allowed_dir) {
            return Ok(resolved);
        }
    }

    let parent_dir = resolved.parent().unwrap_or(Path::new("."));
    let allowed_display: Vec<String> = allowed
        .iter()
        .map(|p| format!("    {}", p.display()))
        .collect::<std::collections::BTreeSet<_>>()
        .into_iter()
        .collect();

    eprintln!("Error: Output path is outside allowed directories.\n");
    eprintln!("  Path: {}\n", resolved.display());
    eprintln!("Allowed directories:");
    for line in &allowed_display {
        eprintln!("{}", line);
    }
    eprintln!();
    eprintln!("To allow this directory for this command:");
    eprintln!(
        "  speakturbo \"text\" -o {} --allow-dir {}\n",
        output,
        parent_dir.display()
    );
    eprintln!("To allow it permanently, add to ~/.speakturbo/config:");
    eprintln!(
        "  mkdir -p ~/.speakturbo && echo \"{}\" >> ~/.speakturbo/config",
        parent_dir.display()
    );

    std::process::exit(1);
}

fn main() -> Result<()> {
    let args = Args::parse();
    let start = Instant::now();

    if args.list_voices {
        println!("Voices: alba, marius, javert, jean, fantine, cosette, eponine, azelma");
        return Ok(());
    }

    let text = match args.text {
        Some(t) => t,
        None => {
            let mut buf = String::new();
            std::io::stdin().read_to_string(&mut buf)?;
            buf
        }
    };

    if text.trim().is_empty() {
        eprintln!("Error: No text");
        std::process::exit(1);
    }

    // Validate output path BEFORE HTTP request (fail fast)
    let resolved_output = if let Some(ref output_path) = args.output {
        Some(validate_output_path(output_path, &args.allow_dir)?)
    } else {
        None
    };

    let url = format!("{}/tts?text={}&voice={}", 
        DAEMON_URL, 
        urlencoding::encode(&text),
        urlencoding::encode(&args.voice)
    );

    // Fast HTTP request
    let response = ureq::get(&url)
        .call()
        .context("Daemon not running?")?;

    if let Some(resolved) = resolved_output {
        let mut file = std::fs::File::create(&resolved)?;
        std::io::copy(&mut response.into_reader(), &mut file)?;
        if !args.quiet {
            eprintln!("Saved: {}", resolved.display());
        }
    } else {
        stream_audio(response, start, args.quiet)?;
    }

    Ok(())
}

fn stream_audio(response: ureq::Response, start: Instant, quiet: bool) -> Result<()> {
    let (_stream, stream_handle) = OutputStream::try_default()
        .context("No audio output")?;
    let sink = Sink::try_new(&stream_handle)?;

    // Lock-free-ish shared state
    let buffer = Arc::new(LockFreeBuffer::new());
    let buffer_clone = Arc::clone(&buffer);

    // Skip WAV header
    let mut reader = response.into_reader();
    let mut header = [0u8; 44];
    reader.read_exact(&mut header)?;

    // Network reader thread - HIGH PRIORITY
    let start_clone = start;
    std::thread::Builder::new()
        .name("net-reader".into())
        .spawn(move || {
            let mut chunk_buf = [0u8; 4096];
            let mut first = true;
            
            loop {
                match reader.read(&mut chunk_buf) {
                    Ok(0) => {
                        buffer_clone.set_done();
                        break;
                    }
                    Ok(n) => {
                        if first && !quiet {
                            eprintln!("⚡ {}ms", start_clone.elapsed().as_millis());
                            first = false;
                        }
                        
                        // Direct byte-to-sample conversion, no allocation
                        for chunk in chunk_buf[..n].chunks_exact(2) {
                            let sample = i16::from_le_bytes([chunk[0], chunk[1]]);
                            buffer_clone.push(sample);
                        }
                    }
                    Err(_) => {
                        buffer_clone.set_done();
                        break;
                    }
                }
            }
        })?;

    // Wait for minimal buffer
    while buffer.len() < MIN_BUFFER_SAMPLES && !buffer.is_done() {
        std::thread::sleep(Duration::from_micros(500)); // 0.5ms polling
    }

    if !quiet {
        eprintln!("▶ {}ms", start.elapsed().as_millis());
    }

    // Play!
    let source = StreamSource { buffer, samples_emitted: 0 };
    sink.append(source);
    sink.sleep_until_end();

    if !quiet {
        eprintln!("✓ {}ms", start.elapsed().as_millis());
    }

    Ok(())
}

/// Simple lock-free-ish ring buffer using atomic operations
struct LockFreeBuffer {
    data: std::sync::Mutex<VecDeque<i16>>,
    len: AtomicUsize,
    done: AtomicBool,
}

impl LockFreeBuffer {
    fn new() -> Self {
        Self {
            data: std::sync::Mutex::new(VecDeque::with_capacity(SAMPLE_RATE as usize)),
            len: AtomicUsize::new(0),
            done: AtomicBool::new(false),
        }
    }

    fn push(&self, sample: i16) {
        self.data.lock().unwrap().push_back(sample);
        self.len.fetch_add(1, Ordering::Release);
    }

    fn pop(&self) -> Option<i16> {
        let mut data = self.data.lock().unwrap();
        if let Some(s) = data.pop_front() {
            self.len.fetch_sub(1, Ordering::Release);
            Some(s)
        } else {
            None
        }
    }

    fn len(&self) -> usize {
        self.len.load(Ordering::Acquire)
    }

    fn is_done(&self) -> bool {
        self.done.load(Ordering::Acquire)
    }

    fn set_done(&self) {
        self.done.store(true, Ordering::Release);
    }
}

struct StreamSource {
    buffer: Arc<LockFreeBuffer>,
    samples_emitted: usize,
}

impl Iterator for StreamSource {
    type Item = i16;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if let Some(sample) = self.buffer.pop() {
                // Apply fade-in to first FADE_IN_SAMPLES to eliminate startup transients
                let output = if self.samples_emitted < FADE_IN_SAMPLES {
                    let factor = self.samples_emitted as f32 / FADE_IN_SAMPLES as f32;
                    (sample as f32 * factor) as i16
                } else {
                    sample
                };
                self.samples_emitted += 1;
                return Some(output);
            }
            
            if self.buffer.is_done() {
                return None;
            }
            
            // Spin-wait (aggressive but low latency)
            std::hint::spin_loop();
        }
    }
}

impl Source for StreamSource {
    fn current_frame_len(&self) -> Option<usize> { None }
    fn channels(&self) -> u16 { 1 }
    fn sample_rate(&self) -> u32 { SAMPLE_RATE }
    fn total_duration(&self) -> Option<Duration> { None }
}

mod urlencoding {
    pub fn encode(s: &str) -> String {
        let mut r = String::with_capacity(s.len() * 2);
        for c in s.chars() {
            match c {
                'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' | '.' | '~' => r.push(c),
                ' ' => r.push_str("%20"),
                _ => {
                    for b in c.to_string().as_bytes() {
                        r.push_str(&format!("%{:02X}", b));
                    }
                }
            }
        }
        r
    }
}

# speakturbo - Ultra-fast TTS

#!/usr/bin/env python3
"""
speakturbo CLI - Ultra-fast text-to-speech

Usage:
    speakturbo "Hello world"              # Play audio
    speakturbo "Hello" -o output.wav      # Save to file
    speakturbo "Hello" -v marius          # Use different voice
    speakturbo --list-voices              # List available voices
    echo "Hello" | speakturbo             # Read from stdin
"""

import argparse
import os
import signal
import subprocess
import sys
import tempfile
import time
from pathlib import Path

import requests

__version__ = "0.1.0"

# Daemon configuration
DAEMON_HOST = "127.0.0.1"
DAEMON_PORT = 7125
DAEMON_URL = f"http://{DAEMON_HOST}:{DAEMON_PORT}"
PID_FILE = Path.home() / ".speakturbo" / "daemon.pid"

VOICES = ["alba", "marius", "javert", "jean", "fantine", "cosette", "eponine", "azelma"]

# Default directories where -o output is allowed without --allow-dir
DEFAULT_ALLOWED_DIRS = [
    "/tmp",
    "/var/tmp",
    tempfile.gettempdir(),
]


def is_daemon_running() -> bool:
    """Check if daemon is running."""
    try:
        response = requests.get(f"{DAEMON_URL}/health", timeout=1)
        return response.status_code == 200
    except requests.exceptions.RequestException:
        return False


def start_daemon():
    """Start the daemon in background."""
    if is_daemon_running():
        return True
    
    print("Starting speakturbo daemon...", file=sys.stderr)
    
    # Start streaming daemon as subprocess
    daemon_script = Path(__file__).parent / "daemon_streaming.py"
    
    # Create log directory
    log_dir = Path("/tmp")
    log_file = log_dir / "speakturbo.log"
    
    with open(log_file, "a") as log:
        process = subprocess.Popen(
            [sys.executable, str(daemon_script)],
            stdout=log,
            stderr=log,
            start_new_session=True,
        )
    
    # Save PID
    PID_FILE.parent.mkdir(parents=True, exist_ok=True)
    PID_FILE.write_text(str(process.pid))
    
    # Wait for daemon to be ready (can take 2-5s for model loading)
    for _ in range(100):  # 10 seconds max
        if is_daemon_running():
            print("Daemon ready.", file=sys.stderr)
            return True
        time.sleep(0.1)
    
    print("Warning: Daemon may not have started correctly.", file=sys.stderr)
    return False


def stop_daemon():
    """Stop the daemon."""
    if PID_FILE.exists():
        try:
            pid = int(PID_FILE.read_text().strip())
            os.kill(pid, signal.SIGTERM)
            PID_FILE.unlink()
            print("Daemon stopped.")
        except (ValueError, ProcessLookupError):
            PID_FILE.unlink()
            print("Daemon was not running.")
    else:
        print("Daemon is not running.")


def daemon_status():
    """Check daemon status."""
    if is_daemon_running():
        try:
            response = requests.get(f"{DAEMON_URL}/health", timeout=1)
            data = response.json()
            print(f"Daemon: running")
            print(f"Voices: {', '.join(data.get('voices', []))}")
        except Exception as e:
            print(f"Daemon: running (but health check failed: {e})")
    else:
        print("Daemon: not running")


def list_voices():
    """List available voices."""
    print("Available voices:")
    for voice in VOICES:
        print(f"  - {voice}")


def load_allowed_dirs() -> list[str]:
    """Load user-configured allowed directories from ~/.speakturbo/config."""
    config_file = Path.home() / ".speakturbo" / "config"
    custom_dirs = []
    if config_file.exists():
        for line in config_file.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                expanded = os.path.expanduser(line)
                if os.path.isabs(expanded):
                    custom_dirs.append(expanded)
    return custom_dirs


def validate_output_path(output: str, extra_allowed: list[str] | None = None) -> str:
    """Validate output path is within allowed directories.

    Returns the resolved absolute path if allowed.
    Exits with a clear error message if not.
    """
    resolved = os.path.realpath(os.path.expanduser(output))

    allowed = list(DEFAULT_ALLOWED_DIRS)
    allowed.append(os.getcwd())
    allowed.append(str(Path.home() / ".speakturbo"))
    allowed.extend(load_allowed_dirs())
    if extra_allowed:
        allowed.extend(extra_allowed)

    # Normalize ALL paths once (realpath resolves symlinks + makes absolute)
    allowed = [os.path.realpath(os.path.expanduser(d)) for d in allowed]

    for allowed_dir in allowed:
        if resolved.startswith(allowed_dir + os.sep) or resolved == allowed_dir:
            return resolved

    # Build error message that tells the agent exactly what to do
    allowed_display = "\n".join(f"    {d}" for d in sorted(set(allowed)))
    parent_dir = os.path.dirname(resolved)
    print(
        f"Error: Output path is outside allowed directories.\n"
        f"\n"
        f"  Path: {resolved}\n"
        f"\n"
        f"Allowed directories:\n"
        f"{allowed_display}\n"
        f"\n"
        f"To allow this directory for this command:\n"
        f"  speakturbo \"text\" -o {output} --allow-dir {parent_dir}\n"
        f"\n"
        f"To allow it permanently, add to ~/.speakturbo/config:\n"
        f"  mkdir -p ~/.speakturbo && echo \"{parent_dir}\" >> ~/.speakturbo/config",
        file=sys.stderr,
    )
    sys.exit(1)


def generate_speech(text: str, voice: str = "alba", output: str = None,
                    play: bool = True, allowed_dirs: list[str] | None = None):
    """Generate speech from text."""
    # Ensure daemon is running
    if not is_daemon_running():
        if not start_daemon():
            print("Error: Could not start daemon.", file=sys.stderr)
            sys.exit(1)
    
    # Validate voice
    if voice not in VOICES:
        print(f"Error: Invalid voice '{voice}'. Available: {', '.join(VOICES)}", file=sys.stderr)
        sys.exit(1)
    
    # Validate text
    if not text or not text.strip():
        print("Error: Text cannot be empty.", file=sys.stderr)
        sys.exit(1)
    
    # Validate output path BEFORE HTTP request (fail fast)
    if output:
        output = validate_output_path(output, extra_allowed=allowed_dirs)
    
    # Generate audio
    try:
        response = requests.get(
            f"{DAEMON_URL}/tts",
            params={"text": text.strip(), "voice": voice},
            stream=True,
            timeout=60,
        )
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print(f"Error: {e}", file=sys.stderr)
        if response.status_code == 400:
            try:
                detail = response.json().get("detail", "Unknown error")
                print(f"  {detail}", file=sys.stderr)
            except:
                pass
        sys.exit(1)
    except requests.exceptions.RequestException as e:
        print(f"Error: Could not connect to daemon: {e}", file=sys.stderr)
        sys.exit(1)
    
    # Handle output
    if output:
        # Save to file
        with open(output, "wb") as f:
            for chunk in response.iter_content(chunk_size=4096):
                f.write(chunk)
        print(f"Saved to {output}", file=sys.stderr)
    elif play:
        # Play audio
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            temp_path = f.name
            for chunk in response.iter_content(chunk_size=4096):
                f.write(chunk)
        
        try:
            # Use afplay on macOS
            subprocess.run(["afplay", temp_path], check=True)
        except FileNotFoundError:
            # Try aplay on Linux
            try:
                subprocess.run(["aplay", temp_path], check=True)
            except FileNotFoundError:
                print(f"Audio saved to: {temp_path}", file=sys.stderr)
                print("Install afplay (macOS) or aplay (Linux) to play audio.", file=sys.stderr)
        finally:
            try:
                os.unlink(temp_path)
            except:
                pass


def main():
    parser = argparse.ArgumentParser(
        description="speakturbo - Ultra-fast text-to-speech",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  speakturbo "Hello world"              # Play audio
  speakturbo "Hello" -o output.wav      # Save to file
  speakturbo "Hello" -v marius          # Use different voice
  speakturbo --list-voices              # List available voices
  echo "Hello" | speakturbo             # Read from stdin
""",
    )
    
    parser.add_argument("text", nargs="?", help="Text to speak")
    parser.add_argument("-v", "--voice", default="alba", help="Voice to use (default: alba)")
    parser.add_argument("-o", "--output", help="Output WAV file (default: play audio)")
    parser.add_argument("--allow-dir", action="append", default=None,
                        help="Allow output to this directory (repeatable)")
    parser.add_argument("--no-play", action="store_true", help="Don't play audio (only with -o)")
    parser.add_argument("--list-voices", action="store_true", help="List available voices")
    parser.add_argument("--version", action="version", version=f"speakturbo {__version__}")
    
    # Daemon management
    parser.add_argument("--daemon-start", action="store_true", help="Start daemon")
    parser.add_argument("--daemon-stop", action="store_true", help="Stop daemon")
    parser.add_argument("--daemon-status", action="store_true", help="Check daemon status")
    
    args = parser.parse_args()
    
    # Handle daemon commands
    if args.daemon_start:
        start_daemon()
        return
    
    if args.daemon_stop:
        stop_daemon()
        return
    
    if args.daemon_status:
        daemon_status()
        return
    
    if args.list_voices:
        list_voices()
        return
    
    # Get text from argument or stdin
    text = args.text
    if text is None:
        if not sys.stdin.isatty():
            text = sys.stdin.read()
        else:
            parser.print_help()
            sys.exit(1)
    
    # Generate speech
    generate_speech(
        text=text,
        voice=args.voice,
        output=args.output,
        play=not args.no_play,
        allowed_dirs=args.allow_dir,
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
speakturbo daemon - Ultra-fast TTS with TRUE streaming.
Auto-shuts down after 1 hour idle.
"""

import asyncio
import struct
import threading
import time
from typing import Optional

import uvicorn
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse, JSONResponse

from pocket_tts import TTSModel

# High-quality built-in voices only
VOICES = ["alba", "marius", "javert", "jean", "fantine", "cosette", "eponine", "azelma"]

# Auto-shutdown after 1 hour idle
IDLE_TIMEOUT_SECONDS = 3600

_model: Optional[TTSModel] = None
_voice_states: dict = {}
_last_request_time: float = time.time()


def get_model() -> TTSModel:
    global _model
    if _model is None:
        print("Loading TTS model...")
        _model = TTSModel.load_model()
    return _model


def get_voice_state(voice: str) -> dict:
    global _voice_states
    if voice not in _voice_states:
        print(f"Loading voice: {voice}")
        _voice_states[voice] = get_model().get_state_for_audio_prompt(voice)
    return _voice_states[voice]


def wav_header(sample_rate: int) -> bytes:
    """Streaming WAV header with max size."""
    return struct.pack(
        '<4sI4s4sIHHIIHH4sI',
        b'RIFF', 0x7FFFFFFF + 36, b'WAVE', b'fmt ', 16, 1, 1,
        sample_rate, sample_rate * 2, 2, 16, b'data', 0x7FFFFFFF,
    )


def idle_monitor():
    """Background thread that shuts down after idle timeout."""
    import os
    import signal
    
    while True:
        time.sleep(60)  # Check every minute
        idle_time = time.time() - _last_request_time
        
        if idle_time > IDLE_TIMEOUT_SECONDS:
            print(f"\nIdle for {idle_time/60:.0f} minutes. Shutting down...")
            os.kill(os.getpid(), signal.SIGTERM)
            break


app = FastAPI(title="speakturbo")


# DNS rebinding protection - only allow localhost
@app.middleware("http")
async def validate_host(request: Request, call_next):
    host = request.headers.get("host", "").split(":")[0]
    if host not in {"127.0.0.1", "localhost"}:
        return JSONResponse(status_code=403, content={"detail": "Forbidden"})
    return await call_next(request)


@app.get("/health")
async def health():
    global _last_request_time
    _last_request_time = time.time()
    
    idle_mins = (time.time() - _last_request_time) / 60
    return {
        "status": "ready",
        "voices": VOICES,
        "idle_timeout_mins": IDLE_TIMEOUT_SECONDS / 60,
    }


@app.get("/tts")
async def tts(text: str, voice: str = "alba"):
    """Ultra-fast streaming TTS."""
    global _last_request_time
    _last_request_time = time.time()
    
    if not text or not text.strip():
        raise HTTPException(status_code=400, detail="Text cannot be empty")
    
    if voice not in VOICES:
        raise HTTPException(status_code=400, detail=f"Voice must be one of: {VOICES}")
    
    model = get_model()
    voice_state = get_voice_state(voice)
    
    async def generate():
        yield wav_header(model.sample_rate)
        for chunk in model.generate_audio_stream(voice_state, text.strip()):
            yield (chunk.clamp(-1, 1) * 32767).short().numpy().tobytes()
            await asyncio.sleep(0)
        yield bytes(int(model.sample_rate * 0.15) * 2)  # Trailing silence
    
    return StreamingResponse(generate(), media_type="audio/wav")


def main():
    """Start the speakturbo daemon."""
    get_model()
    get_voice_state("alba")  # Pre-warm default
    
    # Start idle monitor thread
    monitor = threading.Thread(target=idle_monitor, daemon=True)
    monitor.start()
    
    print(f"Voices: {VOICES}")
    print(f"Auto-shutdown after {IDLE_TIMEOUT_SECONDS/60:.0f} min idle")
    print("Starting on :7125")
    uvicorn.run(app, host="127.0.0.1", port=7125, log_level="warning")


if __name__ == "__main__":
    main()

# tests

"""
TDD Tests for speakturbo CLI

Run with: uv run pytest speakturbo/tests/test_cli.py -v
"""

import os
import subprocess
import tempfile
import time
import wave

import pytest


# Path to CLI
CLI_PATH = os.path.join(os.path.dirname(__file__), "..", "cli.py")


def run_cli(*args, input_text=None, timeout=30):
    """Run the CLI with given arguments."""
    cmd = ["python", CLI_PATH] + list(args)
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        input=input_text,
        timeout=timeout,
        cwd=os.path.dirname(CLI_PATH),
    )
    return result


class TestCLIBasic:
    """Basic CLI functionality."""
    
    def test_help_works(self):
        result = run_cli("--help")
        assert result.returncode == 0
        assert "speakturbo" in result.stdout.lower() or "usage" in result.stdout.lower()
    
    def test_version_works(self):
        result = run_cli("--version")
        assert result.returncode == 0
    
    def test_list_voices(self):
        result = run_cli("--list-voices")
        assert result.returncode == 0
        assert "alba" in result.stdout
        assert "marius" in result.stdout


class TestCLIGeneration:
    """Audio generation via CLI."""
    
    def test_generate_to_file(self):
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            output_path = f.name
        
        try:
            result = run_cli("Hello world", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
            
            # Verify it's a valid WAV
            with wave.open(output_path, 'rb') as wav:
                assert wav.getnchannels() == 1
                assert wav.getframerate() == 24000
                assert wav.getnframes() > 0
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)
    
    def test_generate_with_voice(self):
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            output_path = f.name
        
        try:
            result = run_cli("Hello", "-v", "marius", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)
    
    def test_read_from_stdin(self):
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            output_path = f.name
        
        try:
            result = run_cli("-o", output_path, input_text="Hello from stdin")
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)


class TestCLIValidation:
    """Input validation."""
    
    def test_invalid_voice_shows_error(self):
        result = run_cli("Hello", "-v", "nonexistent", "-o", "/tmp/test.wav")
        assert result.returncode != 0
        assert "invalid" in result.stderr.lower() or "error" in result.stderr.lower()
    
    def test_empty_text_shows_error(self):
        result = run_cli("", "-o", "/tmp/test.wav")
        assert result.returncode != 0


class TestCLIDaemon:
    """Daemon management."""
    
    def test_daemon_status(self):
        result = run_cli("--daemon-status")
        # Should work whether daemon is running or not
        assert result.returncode == 0


class TestCLIOutputPathAllowlist:
    """Output path allowlist behavior.

    Tests verify the behavioral contract:
    - Default-allowed directories work without --allow-dir
    - Non-allowed directories are blocked with a clear error
    - --allow-dir overrides the block for that invocation
    - ~/.speakturbo/config adds permanent entries
    - Error messages contain actionable fix instructions
    """

    def test_tmp_is_allowed_by_default(self):
        """Writing to /tmp should work without any flags."""
        output_path = "/tmp/speakturbo_test_allowlist.wav"
        try:
            result = run_cli("test", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)

    def test_system_tempdir_is_allowed_by_default(self):
        """Writing to tempfile.gettempdir() should work (used by NamedTemporaryFile)."""
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            output_path = f.name
        try:
            result = run_cli("test", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)

    def test_cwd_is_allowed_by_default(self):
        """Writing to a relative path (CWD) should work."""
        result = run_cli("test", "-o", "speakturbo_test_cwd.wav")
        assert result.returncode == 0
        # Clean up — file is in CWD of the CLI process (speakturbo/ dir)
        cwd_file = os.path.join(os.path.dirname(CLI_PATH), "speakturbo_test_cwd.wav")
        if os.path.exists(cwd_file):
            os.unlink(cwd_file)

    def test_speakturbo_dir_is_allowed_by_default(self):
        """Writing to ~/.speakturbo/ should work."""
        output_path = os.path.expanduser("~/.speakturbo/test_allowlist.wav")
        try:
            result = run_cli("test", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)

    def test_arbitrary_path_is_blocked(self):
        """Writing to a non-allowed path should fail with exit code 1."""
        result = run_cli("test", "-o", "/usr/share/speakturbo_test.wav")
        assert result.returncode != 0

    def test_blocked_path_error_shows_path(self):
        """Error message should show the rejected path."""
        result = run_cli("test", "-o", "/usr/share/speakturbo_test.wav")
        assert "/usr/share/speakturbo_test.wav" in result.stderr

    def test_blocked_path_error_shows_allowed_dirs(self):
        """Error message should list the directories that ARE allowed."""
        result = run_cli("test", "-o", "/usr/share/speakturbo_test.wav")
        assert "Allowed directories" in result.stderr

    def test_blocked_path_error_shows_allow_dir_fix(self):
        """Error message should show the exact --allow-dir command to fix it."""
        result = run_cli("test", "-o", "/usr/share/speakturbo_test.wav")
        assert "--allow-dir" in result.stderr

    def test_blocked_path_error_shows_config_fix(self):
        """Error message should show how to permanently allow via config."""
        result = run_cli("test", "-o", "/usr/share/speakturbo_test.wav")
        assert "~/.speakturbo/config" in result.stderr
        assert "mkdir -p" in result.stderr

    def test_allow_dir_flag_overrides_block(self):
        """--allow-dir should let you write to an otherwise-blocked path."""
        # Use a dir in HOME root — not under /tmp, gettempdir(), or ~/.speakturbo/
        test_dir = os.path.expanduser("~/speakturbo_allow_test_tmp")
        os.makedirs(test_dir, exist_ok=True)
        output_path = os.path.join(test_dir, "test.wav")
        try:
            # Without --allow-dir, this should fail
            result_blocked = run_cli("test", "-o", output_path)
            assert result_blocked.returncode != 0, "Should be blocked without --allow-dir"

            # With --allow-dir, this should work
            result = run_cli("test", "-o", output_path, "--allow-dir", test_dir)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            if os.path.exists(output_path):
                os.unlink(output_path)
            if os.path.exists(test_dir):
                os.rmdir(test_dir)

    def test_config_file_adds_permanent_entry(self):
        """Dirs listed in ~/.speakturbo/config should be allowed without --allow-dir."""
        config_file = os.path.expanduser("~/.speakturbo/config")
        # Use home root dir — not in default allowlist
        test_dir = os.path.expanduser("~/speakturbo_config_test_tmp")
        os.makedirs(test_dir, exist_ok=True)
        output_path = os.path.join(test_dir, "test.wav")

        # Save existing config if any
        had_config = os.path.exists(config_file)
        old_content = open(config_file).read() if had_config else None

        try:
            # Verify it's blocked first
            result_blocked = run_cli("test", "-o", output_path)
            assert result_blocked.returncode != 0, "Should be blocked before config change"

            # Write test dir to config
            os.makedirs(os.path.dirname(config_file), exist_ok=True)
            with open(config_file, "a") as f:
                f.write(f"\n{test_dir}\n")

            # Now it should work
            result = run_cli("test", "-o", output_path)
            assert result.returncode == 0
            assert os.path.exists(output_path)
        finally:
            # Restore original config
            if had_config:
                with open(config_file, "w") as f:
                    f.write(old_content)
            elif os.path.exists(config_file):
                os.unlink(config_file)
            if os.path.exists(output_path):
                os.unlink(output_path)
            if os.path.exists(test_dir):
                os.rmdir(test_dir)

    def test_symlink_escape_is_blocked(self):
        """A symlink in /tmp pointing outside allowed dirs should be blocked."""
        # Create symlink: /tmp/speakturbo_escape_test -> /usr/share/
        link_path = "/tmp/speakturbo_escape_test"
        target_file = link_path + "/evil.wav"

        try:
            if os.path.islink(link_path):
                os.unlink(link_path)
            os.symlink("/usr/share", link_path)

            result = run_cli("test", "-o", target_file)
            assert result.returncode != 0, "Symlink escape should be blocked"
        finally:
            if os.path.islink(link_path):
                os.unlink(link_path)

    def test_path_traversal_is_blocked(self):
        """Path with .. that escapes an allowed dir should be blocked."""
        result = run_cli("test", "-o", "/tmp/../etc/test.wav")
        assert result.returncode != 0

    def test_no_output_flag_skips_validation(self):
        """When -o is not specified, no path validation occurs."""
        # Without -o, the CLI either plays audio or (with --no-play) does nothing.
        # Either way, the allowlist error should never appear.
        result = run_cli("test", "--no-play")
        # May get daemon connection error if daemon isn't running — that's fine.
        # The key assertion: no allowlist error.
        assert "outside allowed directories" not in result.stderr

"""
TDD Tests for speakturbo daemon (daemon_streaming.py)

Tests the PRODUCTION daemon — daemon_streaming.py with GET /tts on port 7125.

Run with: cd speakturbo && uv run pytest tests/test_daemon.py -v
"""

import struct
import pytest
from fastapi.testclient import TestClient


@pytest.fixture(scope="module")
def client():
    """Create test client — imports the PRODUCTION streaming daemon.
    
    base_url must be set to localhost so the DNS rebinding middleware
    (which rejects any Host header not in {127.0.0.1, localhost}) allows
    requests through. TestClient defaults to base_url="http://testserver"
    which would cause every request to get 403.
    """
    from speakturbo.daemon_streaming import app
    return TestClient(app, base_url="http://127.0.0.1:7125")


class TestHealthEndpoint:
    """Health check must be fast and informative."""
    
    def test_health_returns_200(self, client):
        response = client.get("/health")
        assert response.status_code == 200
    
    def test_health_returns_ready_status(self, client):
        response = client.get("/health")
        data = response.json()
        assert data["status"] == "ready"
    
    def test_health_lists_voices(self, client):
        response = client.get("/health")
        data = response.json()
        assert "voices" in data
        assert "alba" in data["voices"]
        assert len(data["voices"]) >= 8


class TestTTSEndpoint:
    """Core TTS functionality — uses GET with query params (matching Rust CLI and Python CLI)."""
    
    def test_tts_returns_audio(self, client):
        response = client.get("/tts", params={"text": "Hello"})
        assert response.status_code == 200
        assert response.headers["content-type"] == "audio/wav"
    
    def test_tts_returns_valid_wav(self, client):
        response = client.get("/tts", params={"text": "Hello world"})
        # Check WAV header
        assert response.content[:4] == b"RIFF"
        assert response.content[8:12] == b"WAVE"
    
    def test_tts_wav_is_playable(self, client):
        response = client.get("/tts", params={"text": "Test"})
        # Streaming WAV uses 0x7FFFFFFF for data/file sizes — wave.open() can't parse it.
        # Validate format from raw header bytes instead (maintains same coverage as original).
        content = response.content
        assert content[:4] == b"RIFF"
        assert content[8:12] == b"WAVE"
        # Verify PCM format details from WAV header
        assert struct.unpack_from('<H', content, 20)[0] == 1      # PCM format
        assert struct.unpack_from('<H', content, 22)[0] == 1      # Mono
        assert struct.unpack_from('<I', content, 24)[0] == 24000  # 24kHz
        assert struct.unpack_from('<H', content, 34)[0] == 16     # 16-bit
        assert len(content) > 44  # Has audio data beyond header
    
    def test_tts_with_voice_parameter(self, client):
        response = client.get("/tts", params={"text": "Hello", "voice": "marius"})
        assert response.status_code == 200
        assert len(response.content) > 44  # More than just WAV header
    
    def test_tts_all_voices_work(self, client):
        voices = ["alba", "marius", "javert", "jean", "fantine", "cosette", "eponine", "azelma"]
        for voice in voices:
            response = client.get("/tts", params={"text": "Test", "voice": voice})
            assert response.status_code == 200, f"Voice {voice} failed"
            assert len(response.content) > 100, f"Voice {voice} returned too little audio"


class TestTTSValidation:
    """Input validation — GET /tts with query params."""
    
    def test_empty_text_returns_400(self, client):
        # GET /tts?text= passes empty string to handler, which returns 400
        response = client.get("/tts", params={"text": ""})
        assert response.status_code == 400
    
    def test_whitespace_only_returns_400(self, client):
        response = client.get("/tts", params={"text": "   "})
        assert response.status_code == 400
    
    def test_missing_text_returns_422(self, client):
        response = client.get("/tts")
        assert response.status_code == 422
    
    def test_invalid_voice_returns_400(self, client):
        response = client.get("/tts", params={"text": "Hello", "voice": "nonexistent"})
        assert response.status_code == 400


class TestTTSStreaming:
    """Streaming must work for low latency."""
    
    def test_response_is_streamed(self, client):
        with client.stream("GET", "/tts", params={"text": "Hello world this is a test"}) as response:
            assert response.status_code == 200
            chunks = list(response.iter_bytes(chunk_size=1024))
            # Should have multiple chunks for streaming
            assert len(chunks) >= 1


class TestDNSRebindingProtection:
    """Verify localhost-only access via Host header validation middleware."""
    
    def test_localhost_allowed(self, client):
        response = client.get("/health", headers={"host": "127.0.0.1:7125"})
        assert response.status_code == 200
    
    def test_external_host_rejected(self, client):
        response = client.get("/health", headers={"host": "evil.com"})
        assert response.status_code == 403


class TestPerformance:
    """Performance requirements."""
    
    def test_ttfc_under_500ms_warm(self, client):
        """Time to first chunk should be under 500ms when warm."""
        import time
        
        # Warm up
        client.get("/tts", params={"text": "warmup"})
        
        # Measure
        start = time.perf_counter()
        with client.stream("GET", "/tts", params={"text": "Hello world"}) as response:
            next(response.iter_bytes(chunk_size=1024))  # trigger first chunk
            ttfc = (time.perf_counter() - start) * 1000
        
        assert ttfc < 500, f"TTFC was {ttfc:.0f}ms, expected < 500ms"
    
    def test_generation_faster_than_realtime(self, client):
        """Should generate faster than real-time (RTF > 1)."""
        import time
        
        text = "The quick brown fox jumps over the lazy dog."
        
        start = time.perf_counter()
        response = client.get("/tts", params={"text": text})
        generation_time = time.perf_counter() - start
        
        # Streaming WAV can't use wave.open() — calculate from byte length
        audio_bytes = len(response.content) - 44  # minus WAV header
        audio_samples = audio_bytes / 2  # 16-bit = 2 bytes per sample
        audio_duration = audio_samples / 24000  # 24kHz sample rate
        
        rtf = audio_duration / generation_time
        assert rtf > 1.0, f"RTF was {rtf:.1f}x, expected > 1.0x"