discord-havoc/cogs/radio.py

import logging
import traceback
import asyncio
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

    async def radio_init(self) -> None:
        try:
            self.radio_state_loop.start()
        except Exception as e:
            logging.critical("Failed to start radio state loop: %s", str(e))
            traceback.print_exc()

    @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)
            #         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 not self.LAST_NP_TRACK or (np_track and not self.LAST_NP_TRACK == np_track):
                logging.critical("Setting: %s", 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))