# voice.py """ Voice channel commands for Miku Discord bot. Handles joining, leaving, and status commands for voice chat sessions. """ import discord from utils.voice_manager import voice_manager from utils.logger import get_logger logger = get_logger('voice_commands') async def handle_voice_command(message, cmd, args): """ Handle voice-related commands. Args: message: Discord message object cmd: Command name (join, leave, voice-status, test) args: Command arguments """ if cmd == 'join': await _handle_join(message, args) elif cmd == 'leave': await _handle_leave(message) elif cmd == 'voice-status': await _handle_status(message) elif cmd == 'test': await _handle_test(message, args) else: await message.channel.send(f"❌ Unknown voice command: `{cmd}`") async def _handle_join(message, args): """ Handle !miku join command. Join voice channel and start session with resource locks. """ # Get voice channel voice_channel = None if args and args[0].startswith('<#'): # Channel mentioned (e.g., !miku join #voice-chat) try: channel_id = int(args[0][2:-1]) voice_channel = message.guild.get_channel(channel_id) if not isinstance(voice_channel, discord.VoiceChannel): await message.channel.send("❌ That's not a voice channel!") return except (ValueError, AttributeError): await message.channel.send("❌ Invalid channel!") return else: # Use user's current voice channel if message.author.voice and message.author.voice.channel: voice_channel = message.author.voice.channel else: await message.channel.send( "❌ You must be in a voice channel! " "Or mention a voice channel like `!miku join #voice-chat`" ) return # Check permissions if not voice_channel.permissions_for(message.guild.me).connect: await message.channel.send(f"❌ I don't have permission to join {voice_channel.mention}!") return if not voice_channel.permissions_for(message.guild.me).speak: await message.channel.send(f"❌ I don't have permission to speak in {voice_channel.mention}!") return # Start session try: await message.channel.send(f"🎤 Joining {voice_channel.mention}...") await voice_manager.start_session( message.guild.id, voice_channel, message.channel # Use current text channel for prompts ) embed = discord.Embed( title="🎤 Voice Chat Active", description=f"I've joined {voice_channel.mention}!", color=discord.Color.from_rgb(134, 206, 203) # Miku teal ) embed.add_field( name="How to use", value=f"Send messages in {message.channel.mention} to make me speak!", inline=False ) embed.add_field( name="⚠️ Resource Mode", value=( "• Text inference on AMD GPU only\n" "• Vision model disabled\n" "• Image generation disabled\n" "• Other text channels paused" ), inline=False ) embed.set_footer(text="Use !miku leave to end the session") await message.channel.send(embed=embed) logger.info(f"Voice session started by {message.author} in {voice_channel.name}") except Exception as e: await message.channel.send(f"❌ Failed to join voice: {str(e)}") logger.error(f"Failed to start voice session: {e}", exc_info=True) async def _handle_leave(message): """ Handle !miku leave command. Leave voice channel and release all resources. """ if not voice_manager.active_session: await message.channel.send("❌ I'm not in a voice channel!") return # Check if user is in the same guild as the active session if voice_manager.active_session.guild_id != message.guild.id: await message.channel.send("❌ I'm in a voice channel in a different server!") return try: voice_channel_name = voice_manager.active_session.voice_channel.name await message.channel.send("👋 Leaving voice channel...") await voice_manager.end_session() embed = discord.Embed( title="👋 Voice Chat Ended", description=f"Left {voice_channel_name}", color=discord.Color.from_rgb(134, 206, 203) ) embed.add_field( name="✅ Resources Released", value=( "• Vision model available\n" "• Image generation available\n" "• Text channels resumed\n" "• All features restored" ), inline=False ) await message.channel.send(embed=embed) logger.info(f"Voice session ended by {message.author}") except Exception as e: await message.channel.send(f"⚠️ Error leaving voice: {str(e)}") logger.error(f"Failed to end voice session: {e}", exc_info=True) async def _handle_status(message): """ Handle !miku voice-status command. Show current voice session status. """ if not voice_manager.active_session: embed = discord.Embed( title="🔇 No Active Voice Session", description="I'm not currently in a voice channel.", color=discord.Color.greyple() ) embed.add_field( name="To start", value="Use `!miku join` while in a voice channel", inline=False ) await message.channel.send(embed=embed) return session = voice_manager.active_session # Check if in same guild if session.guild_id != message.guild.id: await message.channel.send("ℹ️ I'm in a voice channel in a different server.") return embed = discord.Embed( title="🎤 Voice Session Active", description=f"Currently in voice chat", color=discord.Color.from_rgb(134, 206, 203) ) embed.add_field( name="Voice Channel", value=session.voice_channel.mention, inline=True ) embed.add_field( name="Prompt Channel", value=session.text_channel.mention, inline=True ) embed.add_field( name="📊 Resource Allocation", value=( "**GPU Usage:**\n" "• AMD RX 6800: Text model + RVC\n" "• GTX 1660: Soprano TTS only\n\n" "**Blocked Features:**\n" "• ❌ Vision model\n" "• ❌ Image generation\n" "• ❌ Bipolar mode\n" "• ❌ Profile picture changes\n" "• ⏸️ Autonomous engine\n" "• ⏸️ Scheduled events\n" "• 📦 Other text channels (queued)" ), inline=False ) embed.set_footer(text="Use !miku leave to end the session") await message.channel.send(embed=embed) async def _handle_test(message, args): """ Handle !miku test command. Test TTS audio playback in the current voice session. """ session = voice_manager.active_session if not session: await message.channel.send("❌ No active voice session! Use `!miku join` first.") return if not session.audio_source: await message.channel.send("❌ Audio source not connected!") return # Get test text from args or use default test_text = " ".join(args) if args else "Hello! This is a test of my voice chat system." try: await message.channel.send(f"🎤 Speaking: *\"{test_text}\"*") logger.info(f"Testing voice playback: {test_text}") # Stream text to TTS via the audio source await session.audio_source.stream_text(test_text) await message.add_reaction("✅") logger.info("✓ Test audio sent to TTS") except Exception as e: logger.error(f"Failed to test voice playback: {e}", exc_info=True) await message.channel.send(f"❌ Error testing voice: {e}")