spaces_client.py

# spaces_client.py - Client library for your Spaces deployment
import asyncio
import aiohttp
import requests
from typing import Optional, Union
import io
import tempfile
import logging

logger = logging.getLogger(__name__)

class SpacesPiperTTS:
    """Client for Piper TTS deployed on Hugging Face Spaces"""
    
    def __init__(self, space_name: str, hf_token: Optional[str] = None):
        """
        Initialize client for Spaces deployment
        
        Args:
            space_name: Your Space name (format: "username/space-name")
            hf_token: Optional Hugging Face token for private spaces
        """
        self.space_name = space_name
        self.base_url = f"https://{space_name.replace('/', '-')}.hf.space"
        self.api_url = f"{self.base_url}/api"
        self.hf_token = hf_token
        
        # Headers for authenticated requests
        self.headers = {}
        if hf_token:
            self.headers["Authorization"] = f"Bearer {hf_token}"
    
    def synthesize(self, text: str) -> bytes:
        """
        Synchronous TTS synthesis
        
        Args:
            text: Text to convert to speech
            
        Returns:
            WAV audio data as bytes
        """
        try:
            response = requests.post(
                f"{self.api_url}/tts",
                json={"text": text},
                headers=self.headers,
                timeout=30
            )
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            logger.error(f"TTS request failed: {e}")
            raise
    
    async def synthesize_async(self, text: str) -> bytes:
        """
        Asynchronous TTS synthesis
        
        Args:
            text: Text to convert to speech
            
        Returns:
            WAV audio data as bytes
        """
        async with aiohttp.ClientSession() as session:
            try:
                async with session.post(
                    f"{self.api_url}/tts",
                    json={"text": text},
                    headers=self.headers,
                    timeout=aiohttp.ClientTimeout(total=30)
                ) as response:
                    response.raise_for_status()
                    return await response.read()
            except aiohttp.ClientError as e:
                logger.error(f"Async TTS request failed: {e}")
                raise
    
    def synthesize_to_file(self, text: str, filename: str) -> str:
        """
        Synthesize and save to file
        
        Args:
            text: Text to convert to speech
            filename: Output filename
            
        Returns:
            Path to saved file
        """
        audio_data = self.synthesize(text)
        with open(filename, 'wb') as f:
            f.write(audio_data)
        return filename
    
    async def synthesize_to_file_async(self, text: str, filename: str) -> str:
        """
        Async synthesize and save to file
        
        Args:
            text: Text to convert to speech
            filename: Output filename
            
        Returns:
            Path to saved file
        """
        audio_data = await self.synthesize_async(text)
        with open(filename, 'wb') as f:
            f.write(audio_data)
        return filename
    
    def health_check(self) -> dict:
        """Check if the service is healthy"""
        try:
            response = requests.get(f"{self.api_url}/health", headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f"Health check failed: {e}")
            raise
    
    async def health_check_async(self) -> dict:
        """Async health check"""
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(f"{self.api_url}/health", headers=self.headers) as response:
                    response.raise_for_status()
                    return await response.json()
            except aiohttp.ClientError as e:
                logger.error(f"Async health check failed: {e}")
                raise

class DigitalCompanionSpaces:
    """High-level wrapper for digital companion integration with Spaces"""
    
    def __init__(self, space_name: str, hf_token: Optional[str] = None):
        self.tts = SpacesPiperTTS(space_name, hf_token)
        self.conversation_history = []
    
    def speak(self, text: str, save_to_file: Optional[str] = None) -> Union[bytes, str]:
        """
        Generate speech for the companion
        
        Args:
            text: Text to speak
            save_to_file: Optional filename to save audio
            
        Returns:
            Audio bytes or filename if saved
        """
        # Log conversation
        self.conversation_history.append({
            "role": "assistant", 
            "text": text,
            "timestamp": asyncio.get_event_loop().time()
        })
        
        # Generate speech
        if save_to_file:
            return self.tts.synthesize_to_file(text, save_to_file)
        else:
            return self.tts.synthesize(text)
    
    async def speak_async(self, text: str, save_to_file: Optional[str] = None) -> Union[bytes, str]:
        """Async version of speak"""
        # Log conversation
        self.conversation_history.append({
            "role": "assistant", 
            "text": text,
            "timestamp": asyncio.get_event_loop().time()
        })
        
        # Generate speech
        if save_to_file:
            return await self.tts.synthesize_to_file_async(text, save_to_file)
        else:
            return await self.tts.synthesize_async(text)
    
    def get_conversation_history(self) -> list:
        """Get conversation history"""
        return self.conversation_history.copy()
    
    def clear_history(self):
        """Clear conversation history"""
        self.conversation_history.clear()

# deploy_to_spaces.py - Deployment helper script
import os
import subprocess
import sys
from pathlib import Path

def deploy_to_spaces():
    """Deploy the TTS system to Hugging Face Spaces"""
    
    print("🚀 Deploying Piper TTS to Hugging Face Spaces")
    print("=" * 50)
    
    # Check if we're in the right directory
    required_files = ["app.py", "requirements.txt"]
    for file in required_files:
        if not Path(file).exists():
            print(f"❌ Missing required file: {file}")
            return False
    
    # Check if git is initialized
    if not Path(".git").exists():
        print("📁 Initializing git repository...")
        subprocess.run(["git", "init"], check=True)
    
    # Check if huggingface_hub is installed
    try:
        import huggingface_hub
    except ImportError:
        print("📦 Installing huggingface_hub...")
        subprocess.run([sys.executable, "-m", "pip", "install", "huggingface_hub[cli]"], check=True)
    
    # Get Space details from user
    space_name = input("Enter your Space name (username/space-name): ").strip()
    if not space_name or "/" not in space_name:
        print("❌ Invalid Space name format. Use: username/space-name")
        return False
    
    # Check for HF token
    hf_token = os.getenv("HF_TOKEN")
    if not hf_token:
        hf_token = input("Enter your Hugging Face token (or set HF_TOKEN env var): ").strip()
        if not hf_token:
            print("❌ Hugging Face token is required")
            return False
    
    try:
        # Login to Hugging Face
        print("🔐 Logging into Hugging Face...")
        subprocess.run(["huggingface-cli", "login", "--token", hf_token], check=True)
        
        # Create Space
        print(f"🏗️ Creating Space: {space_name}")
        subprocess.run([
            "huggingface-cli", "repo", "create", 
            space_name, 
            "--type", "space", 
            "--space_sdk", "gradio"
        ], check=True)
        
        # Add git remote
        repo_url = f"https://huggingface.co/spaces/{space_name}"
        print(f"🔗 Adding remote: {repo_url}")
        
        # Remove existing remote if exists
        subprocess.run(["git", "remote", "remove", "origin"], check=False)
        subprocess.run(["git", "remote", "add", "origin", repo_url], check=True)
        
        # Prepare files for commit
        print("📝 Preparing files...")
        subprocess.run(["git", "add", "."], check=True)
        subprocess.run(["git", "commit", "-m", "Initial commit: Piper TTS for digital companions"], check=True)
        
        # Push to Spaces
        print("⬆️ Pushing to Spaces...")
        subprocess.run(["git", "push", "origin", "main"], check=True)
        
        print(f"✅ Successfully deployed to: {repo_url}")
        print(f"🌐 Your TTS API will be available at: https://{space_name.replace('/', '-')}.hf.space/api")
        
        return True
        
    except subprocess.CalledProcessError as e:
        print(f"❌ Deployment failed: {e}")
        return False
    except KeyboardInterrupt:
        print("\n⚠️ Deployment cancelled by user")
        return False

# example_usage.py - Usage examples
async def example_basic_usage():
    """Basic usage example"""
    print("🤖 Basic TTS Usage Example")
    
    # Replace with your actual Space name
    tts = SpacesPiperTTS("your-username/your-space-name")
    
    # Check if service is healthy
    try:
        health = tts.health_check()
        print(f"✅ Service healthy: {health}")
    except Exception as e:
        print(f"❌ Service unavailable: {e}")
        return
    
    # Generate speech
    text = "Hello! I'm your digital companion powered by Piper TTS on Hugging Face Spaces."
    
    print(f"🎵 Generating speech for: '{text}'")
    try:
        audio_data = tts.synthesize(text)
        print(f"✅ Generated {len(audio_data)} bytes of audio")
        
        # Save to file
        filename = "companion_speech.wav"
        tts.synthesize_to_file(text, filename)
        print(f"💾 Saved audio to: {filename}")
        
    except Exception as e:
        print(f"❌ TTS failed: {e}")

async def example_digital_companion():
    """Digital companion integration example"""
    print("🤖 Digital Companion Example")
    
    # Initialize companion
    companion = DigitalCompanionSpaces("your-username/your-space-name")
    
    # Conversation flow
    responses = [
        "Hello! I'm your AI companion. How are you feeling today?",
        "I understand. Let me help you with that.",
        "That's a great question! Let me think about it.",
        "I hope that was helpful. Is there anything else you'd like to know?"
    ]
    
    for i, response in enumerate(responses,