131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
import os
|
|
import subprocess
|
|
import discord
|
|
import sound
|
|
from discord.ext import commands
|
|
from main import write_config_file
|
|
|
|
|
|
class Bot(commands.Bot):
|
|
def __init__(self, **kwargs): # bot_token, device_id, device_name, command_prefix='>!'):
|
|
if 'command_prefix' not in kwargs.keys():
|
|
kwargs['command_prefix'] = '>!'
|
|
commands.Bot.__init__(self, command_prefix=commands.when_mentioned_or(kwargs['command_prefix']),
|
|
activity=discord.Game(name=f"@me"), status=discord.Status.idle)
|
|
self.DEVICE_ID = kwargs['Device_ID']
|
|
self.DEVICE_NAME = kwargs['Device_Name']
|
|
self.BOT_TOKEN = kwargs['Token']
|
|
self.Default_Channel_ID = kwargs['Channel_ID']
|
|
self.Default_Mention_Group = kwargs['Mention_Group']
|
|
self.Devices_List = sound.query_devices().items()
|
|
|
|
self.sdr_process = None
|
|
self.system_os_type = None
|
|
|
|
# Set linux or windows
|
|
self.check_os_type()
|
|
|
|
self.add_commands()
|
|
|
|
self.check_for_modules()
|
|
|
|
def start_bot(self):
|
|
self.run(self.BOT_TOKEN)
|
|
|
|
def add_commands(self):
|
|
@self.command(help="Use this to test if the bot is alive", brief="Sends a 'pong' in response")
|
|
async def ping(ctx):
|
|
await ctx.send('pong')
|
|
|
|
@self.command(help="Use this command to join the bot to your channel",
|
|
brief="Joins the voice channel that the caller is in")
|
|
async def join(ctx, *, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
await self.wait_until_ready()
|
|
discord.opus.load_opus('./opus/libopus.so')
|
|
|
|
if discord.opus.is_loaded():
|
|
stream = sound.PCMStream()
|
|
channel = ctx.author.voice.channel
|
|
await ctx.send(f"Ok {member}, I'm joining {channel}")
|
|
|
|
stream.change_device(self.DEVICE_ID)
|
|
|
|
voice_connection = await channel.connect()
|
|
voice_connection.play(discord.PCMAudio(stream))
|
|
|
|
self.start_sdr()
|
|
|
|
await self.change_presence(activity=discord.Activity(type=discord.ActivityType.listening,
|
|
name="the airwaves"),
|
|
status=discord.Status.online)
|
|
else:
|
|
await ctx.send("Opus won't load")
|
|
|
|
@self.command(help="Use this command to have the bot leave your channel",
|
|
brief="Leaves the current voice channel")
|
|
async def leave(ctx):
|
|
await ctx.voice_client.disconnect()
|
|
await self.change_presence(activity=discord.Game(name=f"@me"), status=discord.Status.idle)
|
|
|
|
@self.command(name='reload', hidden=True)
|
|
async def _reload(ctx, module: str, member: discord.Member = None):
|
|
"""Reloads a module."""
|
|
member = member or ctx.author.display_name
|
|
if self.reload_modules(module):
|
|
await ctx.send(f"Ok {member}, I reloaded {module}")
|
|
else:
|
|
await ctx.send(f"{member}, something went wrong. Please check the console")
|
|
|
|
@self.command(name='startsdr', hidden=True)
|
|
async def _startsdr(ctx, input_freq: str, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.start_sdr(freq=input_freq)
|
|
|
|
def check_device(self):
|
|
for device, index in self.Devices_List:
|
|
if int(index) == self.DEVICE_ID and str(device) == self.DEVICE_NAME:
|
|
return True
|
|
|
|
for device, index in self.Devices_List:
|
|
if str(device) == self.DEVICE_NAME:
|
|
self.DEVICE_ID = int(index)
|
|
return True
|
|
|
|
else:
|
|
return False
|
|
|
|
def check_for_modules(self):
|
|
for folder_name in os.listdir("modules"):
|
|
if str(folder_name)[0] == '.':
|
|
continue
|
|
elif os.path.exists(os.path.join("modules", folder_name, "cog.py")):
|
|
print(f"Loaded extension: {folder_name}")
|
|
self.load_extension(f"modules.{folder_name}.cog")
|
|
|
|
def reload_modules(self, module):
|
|
try:
|
|
self.unload_extension(f"modules.{module}.cog")
|
|
print(f"Unloaded {module}")
|
|
self.load_extension(f"modules.{module}.cog")
|
|
print(f"Loaded {module}")
|
|
return True
|
|
except Exception as e:
|
|
print(e)
|
|
return False
|
|
|
|
def check_os_type(self):
|
|
if os.name == 'nt':
|
|
self.system_os_type = 'Windows'
|
|
else:
|
|
self.system_os_type = 'Linux'
|
|
|
|
def start_sdr(self, freq="104.7M", gain=40, squelch=0, sample_rate="200k", mode="wbfm"):
|
|
possible_modes = ['wbfm', 'fm']
|
|
if type(freq) == str:
|
|
# Single freq sent
|
|
if self.sdr_process:
|
|
self.sdr_process.terminate()
|
|
|
|
self.sdr_process = subprocess.Popen(["rtl_fm", f"-M {mode} -f {freq} -g {gain} -l {squelch} -s {sample_rate} | play -t raw -r 32k -es -b 16 -c 1 -V1 -"], shell=True)
|