342 lines
15 KiB
Python
342 lines
15 KiB
Python
import os
|
|
import re
|
|
import time
|
|
import discord
|
|
import sound
|
|
import configparser
|
|
from discord.ext import commands
|
|
from subprocess import Popen, PIPE
|
|
|
|
|
|
# Init class for bot
|
|
class Bot(commands.Bot):
|
|
def __init__(self, **kwargs):
|
|
# If there is no custom command prefix (!help, ?help, etc.), use '>!' but also accept @ mentions
|
|
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)
|
|
|
|
# Init the core bot variables
|
|
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']
|
|
|
|
# Init the audio devices list
|
|
self.Devices_List = sound.query_devices().items()
|
|
|
|
# Init radio parameters
|
|
self.freq = "104.7M"
|
|
self.freq_re = re.compile('(\d+\.?\d*M)')
|
|
self.mode = "wbfm"
|
|
self.gain = 40
|
|
self.squelch = 0
|
|
self.sample_rate = "200k"
|
|
self.play_sample_rate = '32k'
|
|
self.sample_rate_re = re.compile('(\d+\.?\d*k)')
|
|
|
|
# Init SDR Variables
|
|
self.sdr_process = None
|
|
self.system_os_type = None
|
|
self.sdr_output_process = None
|
|
self.sdr_started = False
|
|
|
|
# Set linux or windows
|
|
self.check_os_type()
|
|
|
|
# Add discord commands to the bot
|
|
self.add_commands()
|
|
|
|
# Check the ./modules folder for any modules (cog.py)
|
|
self.check_for_modules()
|
|
|
|
# Start the bot
|
|
def start_bot(self):
|
|
self.run(self.BOT_TOKEN)
|
|
|
|
# Add discord commands to the bot
|
|
def add_commands(self):
|
|
# Test command to see if the bot is on (help command can also be used)
|
|
@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')
|
|
|
|
# Command to join the bot the voice channel the user who called the command is in
|
|
@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
|
|
|
|
# Wait for the bot to be ready to connect
|
|
await self.wait_until_ready()
|
|
|
|
# Load respective opus library
|
|
self.load_opus()
|
|
|
|
if discord.opus.is_loaded():
|
|
# Create an audio stream from selected device
|
|
stream = sound.PCMStream()
|
|
channel = ctx.author.voice.channel
|
|
await ctx.send(f"Ok {member}, I'm joining {channel}")
|
|
|
|
# Ensure the selected device is available and start the audio stream
|
|
stream.change_device(self.DEVICE_ID)
|
|
|
|
# Join the voice channel with the audio stream
|
|
voice_connection = await channel.connect()
|
|
voice_connection.play(discord.PCMAudio(stream))
|
|
|
|
# Start the SDR and begin playing to the audio stream
|
|
self.start_sdr()
|
|
|
|
# Change the activity to the channel and band-type being used
|
|
await self.change_presence(activity=discord.Activity(type=discord.ActivityType.listening,
|
|
name=f"{self.freq[:-1]}"
|
|
f" {str(self.mode).upper()}"),
|
|
status=discord.Status.online)
|
|
else:
|
|
# Return that the opus library would not load
|
|
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):
|
|
# Disconnect the client from the voice channel
|
|
await ctx.voice_client.disconnect()
|
|
# Change the presence to away and '@ me'
|
|
await self.change_presence(activity=discord.Game(name=f"@ me"), status=discord.Status.idle)
|
|
# Stop the SDR so it can cool off
|
|
self.stop_sdr()
|
|
|
|
@self.command(name='chfreq', help="Use this command to change the frequency the bot is listening to. Note: 'M'"
|
|
"is required\nExample command: '@ chfreq wbfm 104.7M\n"
|
|
"Example command: '@ chfreq fm 154.785M",
|
|
brief="Changes radio frequency")
|
|
async def chfreq(ctx, mode: str, freq: str, member: discord.Member = None):
|
|
# Possible band-types that can be used
|
|
possible_modes = ['wbfm', 'fm']
|
|
member = member or ctx.author.display_name
|
|
|
|
# Check to make sure the frequency input matches the syntax needed
|
|
if self.freq_re.search(freq):
|
|
self.freq = freq
|
|
# Check to make sure the selected mode is valid
|
|
if mode in possible_modes:
|
|
self.mode = mode
|
|
await ctx.send(f"Ok {member}, I'm changing the mode to {str(self.mode).upper()} and frequency to"
|
|
f" {self.freq}")
|
|
# If the SDR is started, restart it with the updates
|
|
if self.sdr_started:
|
|
self.start_sdr()
|
|
await self.change_presence(activity=discord.Activity(type=discord.ActivityType.listening,
|
|
name=f"{self.freq[:-1]}"
|
|
f" {str(self.mode).upper()}"),
|
|
status=discord.Status.online)
|
|
else:
|
|
await ctx.send(f"{member}, {mode} is not valid. You may only enter 'fm' or 'wbfm'")
|
|
else:
|
|
await ctx.send(f"{member}, {freq} is not valid. please refer to the help page '@ help chfreq'")
|
|
|
|
@self.command(name='chsquelch', help="Use this command to change the squelch for the frequency"
|
|
"the bot is listening to",
|
|
brief="Changes radio squelch")
|
|
async def chsquelch(ctx, squelch: int, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.squelch = squelch
|
|
await ctx.send(f"Ok {member}, I'm changing the squelch to {self.squelch}")
|
|
|
|
# If the SDR is started, restart it with the updates
|
|
if self.sdr_started:
|
|
self.start_sdr()
|
|
|
|
@self.command(name='chgain', help="Use this command to change the radio gain for the bot",
|
|
brief="Changes radio gain")
|
|
async def chgain(ctx, gain: int, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.gain = gain
|
|
await ctx.send(f"Ok {member}, I have changed the gain to {self.gain}")
|
|
|
|
# If the SDR is started, restart it with the updates
|
|
if self.sdr_started:
|
|
self.start_sdr()
|
|
|
|
@self.command(name='chbw', help="Use this command to change the center frequency bandwidth for the bot."
|
|
"Note: 'k' is required\nExample Command: '@ chbw <xxx.xxk'",
|
|
brief="Changes radio bandwidth")
|
|
async def chbw(ctx, sample_rate: str, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
if self.sample_rate_re.search(sample_rate):
|
|
|
|
# Check to see if sampling rate of player needs to be changed
|
|
if '.' in sample_rate:
|
|
if len(sample_rate.split('.')[0]) <= 2:
|
|
self.sample_rate = sample_rate
|
|
self.play_sample_rate = "12k"
|
|
else:
|
|
self.sample_rate = sample_rate
|
|
self.play_sample_rate = "32k"
|
|
else:
|
|
if len(sample_rate.split('k')[0]) <= 2:
|
|
self.sample_rate = sample_rate
|
|
self.play_sample_rate = "12k"
|
|
else:
|
|
self.sample_rate = sample_rate
|
|
self.play_sample_rate = "32k"
|
|
|
|
await ctx.send(f"Ok {member}, I'm changing the sample rate to {self.sample_rate}")
|
|
|
|
# If the SDR is started, restart it with the updates
|
|
if self.sdr_started:
|
|
self.start_sdr()
|
|
else:
|
|
await ctx.send(f"{member}, {sample_rate} is not valid. please refer to the help page '@ help chbw'")
|
|
|
|
# Hidden admin commands
|
|
@self.command(name='saveprofile', hidden=True)
|
|
async def _saveprofile(ctx, profile_name: str, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.save_radio_config(profile_name)
|
|
await ctx.send(f"Ok {member}, I saved the current settings as {profile_name}")
|
|
|
|
@self.command(name='loadprofile', hidden=True)
|
|
async def _loadprofile(ctx, profile_name: str, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.load_radio_config(profile_name)
|
|
await ctx.send(f"Ok {member}, I loaded the settings saved as {profile_name}")
|
|
|
|
@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, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.start_sdr()
|
|
|
|
@self.command(name='stopsdr', hidden=True)
|
|
async def _stopsdr(ctx, member: discord.Member = None):
|
|
member = member or ctx.author.display_name
|
|
self.stop_sdr()
|
|
|
|
def load_opus(self):
|
|
# Check the system type and load the correct library
|
|
if self.system_os_type == 'Linux':
|
|
discord.opus.load_opus('./opus/libopus.so')
|
|
elif self.system_os_type == 'Windows':
|
|
discord.opus.load_opus('./opus/libopus.dll')
|
|
|
|
def check_device(self):
|
|
# Check to make sure the selected device is still available and has not changed it's index
|
|
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):
|
|
# Search the ./modules folder for any modules to load
|
|
# A valid module must be built as a 'cog', refer to the docs for more information
|
|
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):
|
|
# Reload a selected module for changes
|
|
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):
|
|
# Check and store the OS type of the system for later use
|
|
if os.name == 'nt':
|
|
self.system_os_type = 'Windows'
|
|
else:
|
|
self.system_os_type = 'Linux'
|
|
|
|
def start_sdr(self):
|
|
# Check to see if there is only one frequency
|
|
if type(self.freq) == str:
|
|
# Single freq sent
|
|
# Stop the SDR if it is running
|
|
self.stop_sdr()
|
|
|
|
# Start the radio
|
|
print(f"Starting freq: {self.freq}")
|
|
# Start the SDR receiver and pipe the output
|
|
self.sdr_process = Popen(["rtl_fm", "-M", str(self.mode), "-f", str(self.freq), "-g", str(self.gain),
|
|
"-l", str(self.squelch), "-s", str(self.sample_rate)],
|
|
stdout=PIPE)
|
|
# Use the piped output of the SDR receiver to generate audio
|
|
self.sdr_output_process = Popen(["play", "-t", "raw", "-r", str(self.play_sample_rate), "-es", "-b",
|
|
"16", "-c", "1", "-V1", "-"],
|
|
stdin=self.sdr_process.stdout)
|
|
# Set the started variable for later checks
|
|
self.sdr_started = True
|
|
|
|
def stop_sdr(self):
|
|
# Check to see if the SDR is running
|
|
if self.sdr_started:
|
|
# Wait for the running processes to close
|
|
while self.sdr_process.poll() is None and self.sdr_output_process.poll() is None:
|
|
self.sdr_process.terminate()
|
|
self.sdr_process.kill()
|
|
time.sleep(1)
|
|
self.sdr_output_process.terminate()
|
|
self.sdr_output_process.kill()
|
|
time.sleep(1)
|
|
|
|
self.sdr_started = False
|
|
|
|
def save_radio_config(self, profile_name):
|
|
config = configparser.SafeConfigParser()
|
|
|
|
if os.path.exists('./profiles.ini'):
|
|
config.read('./profiles.ini')
|
|
|
|
if not config.has_section(str(profile_name)):
|
|
config.add_section(str(profile_name))
|
|
|
|
config[str(profile_name)]['Frequency'] = self.freq
|
|
config[str(profile_name)]['Mode'] = self.mode
|
|
config[str(profile_name)]['Gain'] = str(self.gain)
|
|
config[str(profile_name)]['Squelch'] = str(self.squelch)
|
|
config[str(profile_name)]['Sample Rate'] = self.sample_rate
|
|
config[str(profile_name)]['Player Sample Rate'] = self.play_sample_rate
|
|
|
|
with open('./profiles.ini', 'w+') as config_file:
|
|
config.write(config_file)
|
|
|
|
|
|
|
|
def load_radio_config(self, profile_name):
|
|
config = configparser.ConfigParser()
|
|
config.read('./profiles.ini')
|
|
|
|
self.freq = config[str(profile_name)]['Frequency']
|
|
self.mode = config[str(profile_name)]['Mode']
|
|
self.gain = int(config[str(profile_name)]['Gain'])
|
|
self.squelch = int(config[str(profile_name)]['Squelch'])
|
|
self.sample_rate = config[str(profile_name)]['Sample Rate']
|
|
self.play_sample_rate = config[str(profile_name)]['Player Sample Rate'] |