Python-Discord-Audio-Bot/main.py

# Python client file (client.py)
import argparse
import logging
import os
import platform
import socketio
import asyncio
from discord import Intents, opus
from discord.ext import commands
from NoiseGatev2 import NoiseGate

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

sio = socketio.AsyncClient()
client = None
device_id = None
ng_threshold = None


### Core functions
def load_opus():
    processor = platform.machine()
    logger.info(f"Processor: {processor}")

    if os.name == 'nt':
        opus_path = './opus/libopus_amd64.dll' if processor == "AMD64" else None
    else:
        opus_path = './opus/libopus_aarcch64.so' if processor == "aarch64" else './opus/libopus_armv7l.so'

    if opus_path:
        opus.load_opus(opus_path)
        logger.info(f"Loaded OPUS library from {opus_path}")
        return True
    else:
        logger.error("Unsupported architecture or OS.")
        return False


async def join_voice_channel(channel_id):
    global device_id, ng_threshold
    channel = client.get_channel(int(channel_id))
    logger.info(f"Joining voice channel {channel}")
    voice_connection = await channel.connect(timeout=60.0, reconnect=True)

    if opus.is_loaded():
        logger.info("OPUS library loaded successfully")
        stream_handler = NoiseGate(
            _input_device_index=device_id,
            _voice_connection=voice_connection,
            _noise_gate_threshold=ng_threshold
        )
        stream_handler.run()
        logger.info("Audio stream started")
        return True
    else:
        logger.error("Failed to load OPUS library")
        return False


async def leave_voice_channel(guild_id):
    guild = await client.fetch_guild(guild_id)
    logger.info(f"Leaving voice channel in guild {guild}")
    voice_client = guild.voice_client
    if voice_client:
        await voice_client.disconnect()
        logger.info("Disconnected from voice channel")
    else:
        logger.info("Not connected to any voice channel in the guild")
    
    # Check if the client needs to open
    if len(check_for_open_vc_connections()) <= 0:
        # Tell the server the client is going to close
        return False

    return True


def check_for_open_vc_connections():
    return client.voice_clients


def check_if_discord_vc_connected(guild_id):
    if client:
        return any(int(vc.guild.id) == int(guild_id) for vc in client.voice_clients)
    
    return False


async def get_discord_username(guild_id):
    try:
        guild = await client.fetch_guild(guild_id)
        member = await guild.fetch_member(get_discord_id())

        if member.nick:
            print(f"Username: {member.nick}")
            return member.nick
        
        print(f"Username: {client.user.name if client.user else None}")
        return client.user.name if client.user else None
    except Exception as e:
        logging.warning(e)
    
    return None


def check_if_client_is_open():
    if client:
        return client.is_ready()
    
    return False


def get_discord_id():
    print(f"ID: {client.user.id if client.user else None}")
    return int(client.user.id) if client.user else None


async def on_connect():
    logger.info("Connected to WebSocket server")


async def on_disconnect():
    logger.info("Disconnected from WebSocket server")


### Socket Events
@sio.event
async def connect_error():
    logger.error("Connection to WebSocket server failed")


@sio.event
async def join_server(data):
    logger.info(f"Received command to join server: {data['channelId']}")
    return await join_voice_channel(data['channelId'])    


@sio.event
async def leave_server(data):
    logger.info(f"Received command to leave server: {data['guild_id']}")
    return await leave_voice_channel(data['guild_id'])


@sio.event
async def check_discord_vc_connected(data):
    return check_if_discord_vc_connected(data['guild_id'])


@sio.event
async def request_discord_username(data):
    return await get_discord_username(data['guild_id'])


@sio.event
async def check_client_is_open():
    return check_if_client_is_open()


@sio.event
async def request_discord_id():
    return get_discord_id()


async def on_ready():
    logger.info(f"We have logged in as {client.user}")
    logger.info("Loading OPUS library")
    if not load_opus():
        return

    # Send update to socket server
    try:
        logger.info('Emitting to the server')
        await sio.emit('discord_ready', {'message': 'Discord bot is ready'})
    except Exception as e:
        logger.error(f"Error emitting to the server: {e}")
        logger.info('Server not ready yet')


async def main(args):
    global client, device_id, ng_threshold

    # Connect to the WebSocket server
    await sio.connect('http://127.0.0.1:{}'.format(args.websocket_port), namespaces=['/'])
    logger.info("Connecting to WebSocket server...")

    intents = Intents.default()
    client = commands.Bot(command_prefix='!', intents=intents)

    device_id = args.device_id
    ng_threshold = args.ng_threshold

    client.add_listener(on_connect)
    client.add_listener(on_disconnect)
    client.add_listener(on_ready)

    await client.start(args.client_id)


def parse_arguments():
    parser = argparse.ArgumentParser()
    parser.add_argument("device_id", type=int, help="The ID of the audio device to use")
    parser.add_argument("client_id", type=str, help="The Discord client ID")
    parser.add_argument("websocket_port", type=int, help="The port of the WebSocket server")
    parser.add_argument("-n", "--ng_threshold", type=int, default=50,
                        help="Change the noise gate threshold. Defaults to 50")
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_arguments()
    logger.info("Arguments: %s", args)
    asyncio.run(main(args))