ff

app.py

@@ -0,0 +1,134 @@
+# import base64
+# import io
+# import logging
+# from typing import List, Optional
+
+# import torch
+# import torchaudio
+# import uvicorn
+# from fastapi import FastAPI, HTTPException
+# from fastapi.middleware.cors import CORSMiddleware
+# from pydantic import BaseModel
+
+# from generator import load_csm_1b, Segment
+# import gradio as gr
+
+# logging.basicConfig(level=logging.INFO)
+# logger = logging.getLogger(__name__)
+
+# app = FastAPI(
+#     title="CSM 1B API",
+#     description="API for Sesame's Conversational Speech Model",
+#     version="1.0.0",
+# )
+
+# app.add_middleware(
+#     CORSMiddleware,
+#     allow_origins=["*"],
+#     allow_credentials=True,
+#     allow_methods=["*"],
+#     allow_headers=["*"],
+# )
+
+# generator = None
+
+# class SegmentRequest(BaseModel):
+#     speaker: int
+#     text: str
+#     audio_base64: Optional[str] = None
+
+# class GenerateAudioRequest(BaseModel):
+#     text: str
+#     speaker: int
+#     context: List[SegmentRequest] = []
+#     max_audio_length_ms: float = 10000
+#     temperature: float = 0.9
+#     topk: int = 50
+
+# class AudioResponse(BaseModel):
+#     audio_base64: str
+#     sample_rate: int
+
+# @app.on_event("startup")
+# async def startup_event():
+#     global generator
+#     logger.info("Loading CSM 1B model...")
+    
+#     device = "cuda" if torch.cuda.is_available() else "cpu"
+#     if device == "cpu":
+#         logger.info("Loading CSM 1B model...")
+#         logger.warning("GPU not available. Using CPU, performance may be slow!")
+#     logger.info(f"Using device: {device}")
+#     try:
+#         generator = load_csm_1b(device=device)
+#         logger.info(f"Model loaded successfully on device: {device}")
+#     except Exception as e:
+#         logger.error(f"Could not load model: {str(e)}")
+#         raise e
+
+# @app.post("/generate-audio", response_model=AudioResponse)
+# async def generate_audio(request: GenerateAudioRequest):
+#     global generator
+    
+#     if generator is None:
+#         raise HTTPException(status_code=503, detail="Model not loaded. Please try again later.")
+    
+#     try:
+#         context_segments = []
+#         for segment in request.context:
+#             if segment.audio_base64:
+#                 audio_bytes = base64.b64decode(segment.audio_base64)
+#                 audio_buffer = io.BytesIO(audio_bytes)
+                
+#                 audio_tensor, sample_rate = torchaudio.load(audio_buffer)
+#                 audio_tensor = torchaudio.functional.resample(
+#                     audio_tensor.squeeze(0), 
+#                     orig_freq=sample_rate, 
+#                     new_freq=generator.sample_rate
+#                 )
+#             else:
+#                 audio_tensor = torch.zeros(0, dtype=torch.float32)
+            
+#             context_segments.append(
+#                 Segment(text=segment.text, speaker=segment.speaker, audio=audio_tensor)
+#             )
+        
+#         audio = generator.generate(
+#             text=request.text,
+#             speaker=request.speaker,
+#             context=context_segments,
+#             max_audio_length_ms=request.max_audio_length_ms,
+#             temperature=request.temperature,
+#             topk=request.topk,
+#         )
+        
+#         buffer = io.BytesIO()
+#         torchaudio.save(buffer, audio.unsqueeze(0).cpu(), generator.sample_rate, format="wav")
+#         # torchaudio.save("audio.wav", audio.unsqueeze(0).cpu(), generator.sample_rate)
+#         buffer.seek(0)
+#         # audio_base64 = base64.b64encode(buffer.read()).decode("utf-8")
+        
+#         return AudioResponse(
+#             content=buffer.read(),
+#             media_type="audio/wav",
+#             headers={"Content-Disposition": "attachment; filename=audio.wav"}
+#         )
+        
+#     except Exception as e:
+#         logger.error(f"error when building audio: {str(e)}")
+#         raise HTTPException(status_code=500, detail=f"error when building audio: {str(e)}")
+
+# @app.get("/health")
+# async def health_check():
+#     if generator is None:
+#         return {"status": "not_ready", "message": "Model is loading"}
+#     return {"status": "ready", "message": "API is ready to serve"}
+
+
+import gradio as gr
+
+def greet(name):
+    return "Hello " + name + "!!"
+
+demo = gr.Interface(fn=greet, inputs="text", outputs="text")
+demo.launch()