148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
import logging
|
|
import traceback
|
|
from typing import Optional
|
|
from discord.ext import bridge, commands, tasks
|
|
from util.radio_util import get_now_playing, skip
|
|
import discord
|
|
from disc_havoc import Havoc
|
|
|
|
|
|
class Radio(commands.Cog):
|
|
"""Radio Cog for Havoc"""
|
|
|
|
def __init__(self, bot: Havoc) -> None:
|
|
self.bot: Havoc = bot
|
|
self.channels: dict[str, tuple] = {
|
|
"sfm": (
|
|
1145182936002482196,
|
|
1221615558492029050,
|
|
), # Tuple: Guild Id, Chan Id
|
|
}
|
|
self.STREAM_URL: str = "https://stream.codey.lol/sfm.ogg"
|
|
self.LAST_NP_TRACK: Optional[str] = None
|
|
try:
|
|
self.radio_state_loop.cancel()
|
|
except Exception as e:
|
|
logging.debug("Failed to cancel radio_state_loop: %s", str(e))
|
|
|
|
@commands.Cog.listener()
|
|
async def on_ready(self) -> None:
|
|
"""Run on Bot Ready"""
|
|
await self.radio_init()
|
|
|
|
def is_radio_chan(): # type: ignore
|
|
"""Check if channel is radio chan"""
|
|
|
|
def predicate(ctx):
|
|
try:
|
|
return ctx.channel.id == 1221615558492029050
|
|
except Exception as e:
|
|
logging.debug("Exception: %s", str(e))
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
return commands.check(predicate)
|
|
|
|
@bridge.bridge_command()
|
|
@commands.is_owner()
|
|
async def reinitradio(self, ctx) -> None:
|
|
"""
|
|
Reinitialize serious.FM
|
|
"""
|
|
loop: discord.asyncio.AbstractEventLoop = self.bot.loop
|
|
loop.create_task(self.radio_init())
|
|
await ctx.respond("Done!", ephemeral=True)
|
|
|
|
async def radio_init(self) -> None:
|
|
"""Init Radio"""
|
|
try:
|
|
(radio_guild, radio_chan) = self.channels["sfm"]
|
|
guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild)
|
|
if not guild:
|
|
return
|
|
channel = guild.get_channel(radio_chan)
|
|
if not isinstance(channel, discord.VoiceChannel):
|
|
return
|
|
if not self.bot.voice_clients:
|
|
await channel.connect()
|
|
try:
|
|
try:
|
|
self.radio_state_loop.cancel()
|
|
except Exception as e:
|
|
logging.debug("Failed to cancel radio_state_loop: %s", str(e))
|
|
self.radio_state_loop.start()
|
|
logging.info("radio_state_loop task started!")
|
|
except Exception as e:
|
|
logging.critical("Could not start task... Exception: %s", str(e))
|
|
traceback.print_exc()
|
|
except Exception as e:
|
|
logging.debug("Exception: %s", str(e))
|
|
traceback.print_exc()
|
|
return
|
|
|
|
@tasks.loop(seconds=5.0)
|
|
async def radio_state_loop(self) -> None:
|
|
"""Radio State Loop"""
|
|
try:
|
|
(radio_guild, radio_chan) = self.channels["sfm"]
|
|
try:
|
|
vc: discord.VoiceProtocol = self.bot.voice_clients[-1]
|
|
except Exception as e:
|
|
logging.debug(
|
|
"No voice client, establishing new VC connection... (Exception: %s)",
|
|
str(e),
|
|
)
|
|
guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild)
|
|
if not guild:
|
|
return
|
|
channel = guild.get_channel(radio_chan)
|
|
if not isinstance(channel, discord.VoiceChannel):
|
|
return
|
|
await channel.connect()
|
|
vc = self.bot.voice_clients[-1]
|
|
|
|
if not vc.is_playing() or vc.is_paused(): # type: ignore
|
|
"""
|
|
Mypy does not seem aware of the is_playing, play, and is_paused methods,
|
|
but they exist.
|
|
"""
|
|
logging.info("Detected VC not playing... playing!")
|
|
source: discord.FFmpegAudio = discord.FFmpegOpusAudio(
|
|
self.STREAM_URL, before_options="-timeout 3000000"
|
|
)
|
|
vc.play( # type: ignore
|
|
source,
|
|
after=lambda e: logging.info("Error: %s", e) if e else None,
|
|
)
|
|
# Get Now Playing
|
|
np_track: Optional[str] = await get_now_playing()
|
|
if np_track and not self.LAST_NP_TRACK == np_track:
|
|
self.LAST_NP_TRACK = np_track
|
|
await self.bot.change_presence(
|
|
activity=discord.Activity(
|
|
type=discord.ActivityType.listening, name=np_track
|
|
)
|
|
)
|
|
except Exception as e:
|
|
logging.debug("Exception: %s", str(e))
|
|
traceback.print_exc()
|
|
|
|
@bridge.bridge_command()
|
|
@commands.is_owner()
|
|
async def skip(self, ctx) -> None:
|
|
"""
|
|
Skip - Convenience Command
|
|
"""
|
|
|
|
await skip()
|
|
return await ctx.respond("OK", ephemeral=True)
|
|
|
|
def cog_unload(self) -> None:
|
|
"""Run on Cog Unload"""
|
|
self.radio_state_loop.cancel()
|
|
|
|
|
|
def setup(bot) -> None:
|
|
"""Run on Cog Load"""
|
|
bot.add_cog(Radio(bot))
|