jiba21/orfeo
0
1import gradio as gr2import subprocess3import os4import tempfile5import shutil6import base647import mimetypes8from typing import List, Dict, Any, Optional9from fastapi import FastAPI, File, UploadFile, HTTPException10from fastapi.responses import FileResponse11from fastapi.middleware.cors import CORSMiddleware12import uvicorn13import threading14from pathlib import Path15from datetime import datetime16import json17 18# MCP Server Implementation19class MusicRecognitionMCPServer:20 def __init__(self, allowed_directories: List[str] = None):21 """Initialize MCP Server with configurable file access"""22 self.allowed_directories = allowed_directories or ["/tmp", "uploads", "output"]23 self.processed_files = {} # Cache of processed results24 25 # Convert relative paths to absolute and check access26 abs_directories = []27 for directory in self.allowed_directories:28 # Convert relative paths to absolute29 if not os.path.isabs(directory):30 directory = os.path.abspath(directory)31 32 abs_directories.append(directory)33 print(f"Checking directory: {directory}")34 35 # Only try to create if it doesn't exist and we have permission36 if not os.path.exists(directory):37 try:38 os.makedirs(directory, exist_ok=True)39 os.chmod(directory, 0o755)40 print(f"✅ Directory created: {directory}")41 except PermissionError:42 print(f"⚠️ Directory doesn't exist but can't create: {directory}")43 print(f" This is normal for system directories like /tmp")44 except Exception as e:45 print(f"⚠️ Warning creating {directory}: {e}")46 else:47 print(f"✅ Directory exists: {directory}")48 49 self.allowed_directories = abs_directories50 print(f"Final allowed directories: {self.allowed_directories}")51 52 # Test Audiveris installation53 self._test_audiveris()54 55 def list_resources(self) -> List[Dict[str, Any]]:56 """List available resources following MCP patterns"""57 resources = []58 59 # Add processed files as resources60 for file_id, file_info in self.processed_files.items():61 resources.append({62 "uri": f"musicxml://{file_id}",63 "name": file_info["original_name"],64 "description": f"MusicXML file converted from {file_info['original_name']} on {file_info['processed_at']}",65 "mimeType": "application/vnd.recordare.musicxml+xml"66 })67 68 # Add available PDF files in allowed directories69 for directory in self.allowed_directories:70 if os.path.exists(directory):71 for file_path in Path(directory).rglob("*.pdf"):72 if self._is_file_accessible(str(file_path)):73 resources.append({74 "uri": f"file://{file_path}",75 "name": file_path.name,76 "description": f"PDF music score available for processing: {file_path.name}",77 "mimeType": "application/pdf"78 })79 80 return resources81 82 def read_resource(self, uri: str) -> Dict[str, Any]:83 """Read resource content following MCP patterns"""84 if uri.startswith("musicxml://"):85 # Return processed MusicXML file86 file_id = uri.replace("musicxml://", "")87 if file_id in self.processed_files:88 file_info = self.processed_files[file_id]89 try:90 with open(file_info["output_path"], "rb") as f:91 content = base64.b64encode(f.read()).decode()92 93 return {94 "contents": [{95 "type": "resource",96 "resource": {97 "uri": uri,98 "text": content,99 "mimeType": "application/vnd.recordare.musicxml+xml"100 }101 }]102 }103 except FileNotFoundError:104 raise Exception(f"MusicXML file not found: {file_info['output_path']}")105 else:106 raise Exception(f"Resource not found: {uri}")107 108 elif uri.startswith("file://"):109 # Return PDF file content110 file_path = uri.replace("file://", "")111 if not self._is_file_accessible(file_path):112 raise Exception(f"File access denied: {file_path}")113 114 try:115 with open(file_path, "rb") as f:116 content = base64.b64encode(f.read()).decode()117 118 return {119 "contents": [{120 "type": "resource", 121 "resource": {122 "uri": uri,123 "text": content,124 "mimeType": "application/pdf"125 }126 }]127 }128 except FileNotFoundError:129 raise Exception(f"File not found: {file_path}")130 131 else:132 raise Exception(f"Unsupported URI scheme: {uri}")133 134 def _is_file_accessible(self, file_path: str) -> bool:135 """Check if file is within allowed directories"""136 abs_path = os.path.abspath(file_path)137 return any(abs_path.startswith(os.path.abspath(d)) for d in self.allowed_directories)138 139 def recognize_music_tool(self, pdf_uri: str, output_dir: str = None) -> Dict[str, Any]:140 """Tool for music recognition following MCP patterns"""141 # Handle different URI formats142 if pdf_uri.startswith("file://"):143 pdf_path = pdf_uri.replace("file://", "")144 elif pdf_uri.startswith("data:"):145 # Handle data URIs (base64 encoded)146 return self._process_data_uri(pdf_uri, output_dir)147 else:148 # Assume it's a direct file path149 pdf_path = pdf_uri150 151 if not self._is_file_accessible(pdf_path):152 raise Exception(f"File access denied: {pdf_path}")153 154 if not os.path.exists(pdf_path):155 raise Exception(f"PDF file not found: {pdf_path}")156 157 try:158 output_file = self._recognize_music_core(pdf_path, output_dir)159 160 # Store in processed files cache161 file_id = f"music_{len(self.processed_files) + 1}_{int(datetime.now().timestamp())}"162 self.processed_files[file_id] = {163 "original_name": os.path.basename(pdf_path),164 "original_path": pdf_path,165 "output_path": output_file,166 "processed_at": datetime.now().isoformat(),167 "file_id": file_id168 }169 170 # Return MCP-compliant response171 return {172 "content": [{173 "type": "text",174 "text": f"✅ Successfully converted '{os.path.basename(pdf_path)}' to MusicXML.\n\n"175 f"📁 Output file: {output_file}\n"176 f"🔗 Resource URI: musicxml://{file_id}\n"177 f"📊 File size: {os.path.getsize(output_file)} bytes\n\n"178 f"You can now access this MusicXML file as a resource using the URI: `musicxml://{file_id}`"179 }],180 "isError": False181 }182 183 except Exception as e:184 return {185 "content": [{186 "type": "text", 187 "text": f"❌ Music recognition failed: {str(e)}"188 }],189 "isError": True190 }191 192 def _process_data_uri(self, data_uri: str, output_dir: str = None) -> Dict[str, Any]:193 """Process base64 encoded data URI"""194 try:195 # Parse data URI: data:application/pdf;base64,<data>196 header, data = data_uri.split(',', 1)197 mime_type = header.split(';')[0].replace('data:', '')198 199 if mime_type != 'application/pdf':200 raise Exception(f"Unsupported MIME type: {mime_type}")201 202 # Fix base64 padding if needed203 data = self._fix_base64_padding(data)204 205 # Decode base64 data206 pdf_data = base64.b64decode(data)207 208 # Save to temporary file209 temp_dir = output_dir or "/tmp"210 temp_pdf = os.path.join(temp_dir, f"temp_{int(datetime.now().timestamp())}.pdf")211 212 with open(temp_pdf, 'wb') as f:213 f.write(pdf_data)214 215 # Process the file216 return self.recognize_music_tool(f"file://{temp_pdf}", output_dir)217 218 except Exception as e:219 raise Exception(f"Failed to process data URI: {str(e)}")220 221 def _fix_base64_padding(self, data: str) -> str:222 """Fix base64 padding to make it valid"""223 # Remove any whitespace224 data = data.strip().replace('\n', '').replace('\r', '').replace(' ', '')225 226 # Add padding if needed227 missing_padding = len(data) % 4228 if missing_padding:229 data += '=' * (4 - missing_padding)230 231 return data232 233 def _recognize_music_core(self, pdf_file_path: str, output_dir: str = None) -> str:234 """Core music recognition function"""235 audiveris = "/opt/audiveris/bin/Audiveris"236 237 if output_dir is None:238 output_dir = "/tmp/output"239 240 # Ensure output directory exists with proper permissions241 os.makedirs(output_dir, exist_ok=True)242 try:243 os.chmod(output_dir, 0o755)244 except Exception as e:245 print(f"Warning: Could not set permissions for {output_dir}: {e}")246 247 if not self._is_file_accessible(output_dir):248 raise Exception(f"Output directory access denied: {output_dir}")249 250 # Verify input file exists251 if not os.path.exists(pdf_file_path):252 raise Exception(f"Input PDF file not found: {pdf_file_path}")253 254 pdf_file_name = os.path.basename(pdf_file_path)255 pdf_name_without_ext = os.path.splitext(pdf_file_name)[0]256 257 # Try both possible extensions258 possible_extensions = [".mxl", ".xml", ".musicxml"]259 output_files = [os.path.join(output_dir, f"{pdf_name_without_ext}{ext}") for ext in possible_extensions]260 261 cmd = [262 audiveris, "-batch", "-export", "-output", output_dir, pdf_file_path263 ]264 265 print(f"Running Audiveris command: {' '.join(cmd)}")266 result = subprocess.run(cmd, capture_output=True, text=True)267 268 print(f"Audiveris stdout: {result.stdout}")269 print(f"Audiveris stderr: {result.stderr}")270 print(f"Audiveris return code: {result.returncode}")271 272 # List files in output directory for debugging273 if os.path.exists(output_dir):274 files_in_output = os.listdir(output_dir)275 print(f"Files in output directory: {files_in_output}")276 277 # Check if any of the possible output files exist278 existing_output = None279 for output_file in output_files:280 if os.path.exists(output_file):281 existing_output = output_file282 break283 284 if existing_output:285 print(f"Found output file: {existing_output}")286 return existing_output287 288 # If no output file found, provide detailed error289 error_msg = f"Audiveris processing failed.\n"290 error_msg += f"Return code: {result.returncode}\n"291 error_msg += f"Stdout: {result.stdout}\n"292 error_msg += f"Stderr: {result.stderr}\n"293 error_msg += f"Expected files: {output_files}\n"294 error_msg += f"Files in output dir: {os.listdir(output_dir) if os.path.exists(output_dir) else 'Directory does not exist'}\n"295 296 raise Exception(error_msg)297 298 def _test_audiveris(self):299 """Test if Audiveris is properly installed"""300 audiveris = "/opt/audiveris/bin/Audiveris"301 302 if not os.path.exists(audiveris):303 print(f"⚠️ Warning: Audiveris not found at {audiveris}")304 return False305 306 try:307 # Test Audiveris with help command308 result = subprocess.run([audiveris, "-help"], capture_output=True, text=True, timeout=10)309 if "Audiveris" in result.stdout or "Audiveris" in result.stderr:310 print("✅ Audiveris installation verified")311 return True312 else:313 print(f"⚠️ Warning: Audiveris may not be working properly")314 print(f"Output: {result.stdout}")315 print(f"Error: {result.stderr}")316 return False317 except Exception as e:318 print(f"⚠️ Warning: Could not test Audiveris: {e}")319 return False320 321# Initialize MCP Server322mcp_server = MusicRecognitionMCPServer(["/tmp", "uploads", "output"])323 324# Create FastAPI app325app = FastAPI(title="Music Recognition API", version="1.0.0")326 327app.add_middleware(328 CORSMiddleware,329 allow_origins=["*"],330 allow_credentials=True,331 allow_methods=["*"],332 allow_headers=["*"],333)334 335# MCP-compliant endpoints336@app.get("/mcp/resources")337async def list_mcp_resources():338 """List available MCP resources"""339 try:340 resources = mcp_server.list_resources()341 return {"resources": resources}342 except Exception as e:343 raise HTTPException(status_code=500, detail=str(e))344 345@app.post("/mcp/resources/read")346async def read_mcp_resource(request: dict):347 """Read MCP resource content"""348 try:349 uri = request.get("uri")350 if not uri:351 raise HTTPException(status_code=400, detail="URI is required")352 353 content = mcp_server.read_resource(uri)354 return content355 except Exception as e:356 raise HTTPException(status_code=500, detail=str(e))357 358@app.post("/mcp/tools/recognize_music")359async def mcp_recognize_music_tool(request: dict):360 """MCP tool for music recognition"""361 try:362 pdf_uri = request.get("pdf_uri")363 output_dir = request.get("output_dir")364 365 if not pdf_uri:366 raise HTTPException(status_code=400, detail="pdf_uri is required")367 368 result = mcp_server.recognize_music_tool(pdf_uri, output_dir)369 return result370 except Exception as e:371 raise HTTPException(status_code=500, detail=str(e))372 373# Original FastAPI endpoints for backward compatibility374@app.post("/api/recognize-music")375async def recognize_music_api(file: UploadFile = File(...)):376 """FastAPI endpoint for music recognition"""377 if not file.filename.lower().endswith('.pdf'):378 raise HTTPException(status_code=400, detail="Only PDF files are allowed")379 380 with tempfile.TemporaryDirectory() as temp_dir:381 input_path = os.path.join(temp_dir, file.filename)382 with open(input_path, "wb") as buffer:383 shutil.copyfileobj(file.file, buffer)384 385 try:386 output_file = mcp_server._recognize_music_core(input_path, temp_dir)387 return FileResponse(388 path=output_file,389 filename=os.path.basename(output_file),390 media_type='application/octet-stream'391 )392 except Exception as e:393 raise HTTPException(status_code=500, detail=str(e))394 395@app.get("/api/health")396async def health_check():397 """Health check endpoint"""398 return {"status": "healthy", "service": "music-recognition"}399 400@app.get("/api/info")401async def get_info():402 """Get API information"""403 return {404 "name": "Music Recognition API",405 "version": "1.0.0",406 "description": "Convert PDF music scores to MusicXML using Audiveris",407 "mcp_compliant": True,408 "endpoints": {409 "GET /mcp/resources": "List MCP resources",410 "POST /mcp/resources/read": "Read MCP resource content",411 "POST /mcp/tools/recognize_music": "MCP music recognition tool",412 "POST /api/recognize-music": "Upload PDF and get MusicXML",413 "GET /api/health": "Health check",414 "GET /api/info": "API information"415 }416 }417 418# Gradio interface function419def recognize_music_gradio(pdf_file):420 """Gradio wrapper for music recognition"""421 try:422 print(f"Processing file: {pdf_file.name}")423 result = mcp_server.recognize_music_tool(f"file://{pdf_file.name}")424 425 if result.get("isError"):426 error_msg = result["content"][0]["text"]427 print(f"Error in music recognition: {error_msg}")428 return None429 430 # Extract file ID from the response431 response_text = result["content"][0]["text"]432 print(f"Response text: {response_text}")433 434 if "musicxml://" in response_text:435 file_id = response_text.split("musicxml://")[1].split("`")[0]436 print(f"Extracted file ID: {file_id}")437 438 if file_id in mcp_server.processed_files:439 file_info = mcp_server.processed_files[file_id]440 output_path = file_info["output_path"]441 print(f"Output path from cache: {output_path}")442 443 if os.path.exists(output_path):444 print(f"✅ File exists: {output_path}")445 return output_path446 else:447 print(f"❌ File not found: {output_path}")448 449 # If the above doesn't work, try to find the file directly450 pdf_basename = os.path.splitext(os.path.basename(pdf_file.name))[0]451 possible_files = [452 f"/tmp/output/{pdf_basename}.mxl",453 f"/tmp/output/{pdf_basename}.xml", 454 f"/tmp/output/{pdf_basename}.musicxml"455 ]456 457 for file_path in possible_files:458 print(f"Checking: {file_path}")459 if os.path.exists(file_path):460 print(f"✅ Found file: {file_path}")461 return file_path462 463 print("❌ No output file found in any expected location")464 print(f"Files in /tmp/output: {os.listdir('/tmp/output') if os.path.exists('/tmp/output') else 'Directory not found'}")465 return None466 467 except Exception as e:468 print(f"Exception in Gradio wrapper: {str(e)}")469 import traceback470 traceback.print_exc()471 return None472 473# Create Gradio interface474gradio_interface = gr.Interface(475 fn=recognize_music_gradio,476 inputs=gr.File(file_types=[".pdf"], label="Upload PDF music score"),477 outputs=gr.File(label="Download MusicXML file"),478 title="Music Score Recognition",479 description="Upload a PDF music score and create a MusicXML file from it.",480)481 482def run_gradio():483 """Run Gradio in a separate thread"""484 gradio_interface.launch(485 server_name="0.0.0.0", 486 server_port=7860, 487 share=True, 488 mcp_server=True,489 prevent_thread_lock=True490 )491 492def run_fastapi():493 """Run FastAPI server"""494 uvicorn.run(495 app, 496 host="0.0.0.0", 497 port=8000,498 log_level="info"499 )500 501if __name__ == "__main__":502 # Start Gradio in a separate thread503 gradio_thread = threading.Thread(target=run_gradio, daemon=True)504 gradio_thread.start()505 506 print("🎵 MCP-Compliant Music Recognition Service Starting...")507 print("📱 Gradio UI: http://localhost:7860")508 print("🚀 FastAPI: http://localhost:8000")509 print("📋 API Docs: http://localhost:8000/docs")510 print("🔗 MCP Resources: http://localhost:8000/mcp/resources")511 512 # Run FastAPI in the main thread513 run_fastapi()514