import discord from discord import app_commands from discord.ext import commands from typing import Optional from app.internal.c2_client import c2 from app.internal.logger import logger class RadioCommands(commands.Cog): def __init__(self, bot: commands.Bot): self.bot = bot # ------------------------------------------------------------------ # Autocomplete helpers # ------------------------------------------------------------------ async def system_autocomplete( self, interaction: discord.Interaction, current: str ) -> list[app_commands.Choice[str]]: systems = await c2.get_systems() return [ app_commands.Choice(name=s["name"], value=s["system_id"]) for s in systems if current.lower() in s["name"].lower() ][:25] async def token_autocomplete( self, interaction: discord.Interaction, current: str ) -> list[app_commands.Choice[str]]: tokens = await c2.get_tokens() return [ app_commands.Choice(name=f"{t['name']} {'(in use)' if t.get('in_use') else '(free)'}", value=t["token_id"]) for t in tokens if current.lower() in t["name"].lower() ][:25] # ------------------------------------------------------------------ # /join # ------------------------------------------------------------------ @app_commands.command(name="join", description="Stream a radio system to your voice channel.") @app_commands.describe( system="The radio system to listen to.", token="Optionally pick a specific bot token from the pool.", ) @app_commands.autocomplete(system=system_autocomplete, token=token_autocomplete) async def join(self, interaction: discord.Interaction, system: str, token: Optional[str] = None): await interaction.response.defer(ephemeral=True) if not interaction.user.voice or not interaction.user.voice.channel: await interaction.followup.send("You need to be in a voice channel first.") return channel = interaction.user.voice.channel guild_id = str(interaction.guild_id) channel_id = str(channel.id) node = await c2.find_node_for_system(system) if not node: await interaction.followup.send( "No online node is assigned to that system. Check `/status` for availability." ) return cmd: dict = { "action": "discord_join", "guild_id": guild_id, "channel_id": channel_id, } if token: cmd["preferred_token_id"] = token ok = await c2.send_command(node["node_id"], cmd) if ok: systems = await c2.get_systems() system_name = next((s["name"] for s in systems if s["system_id"] == system), system) await interaction.followup.send( f"Streaming **{system_name}** from node `{node['node_id']}` to {channel.mention}." ) else: await interaction.followup.send("Failed to contact the node. It may be offline.") # ------------------------------------------------------------------ # /leave # ------------------------------------------------------------------ @app_commands.command(name="leave", description="Stop streaming radio in this server.") async def leave(self, interaction: discord.Interaction): await interaction.response.defer(ephemeral=True) nodes = await c2.get_nodes() streaming_nodes = [ n for n in nodes if n.get("status") in ("online", "recording") ] if not streaming_nodes: await interaction.followup.send("No nodes appear to be active right now.") return for node in streaming_nodes: await c2.send_command(node["node_id"], {"action": "discord_leave"}) await interaction.followup.send("Disconnected.") # ------------------------------------------------------------------ # /status # ------------------------------------------------------------------ @app_commands.command(name="status", description="Show all node and system status.") async def status(self, interaction: discord.Interaction): await interaction.response.defer() nodes = await c2.get_nodes() systems = await c2.get_systems() system_map = {s["system_id"]: s["name"] for s in systems} status_emoji = { "online": "🟢", "recording": "🔴", "offline": "⚫", "unconfigured": "🟡", } embed = discord.Embed(title="DRB Node Status", color=0x2b2d31) if not nodes: embed.description = "No nodes registered." else: for node in sorted(nodes, key=lambda n: n.get("name", "")): s = node.get("status", "offline") emoji = status_emoji.get(s, "⚪") system_name = system_map.get(node.get("assigned_system_id", ""), "Unassigned") embed.add_field( name=f"{emoji} {node.get('name', node['node_id'])}", value=f"`{s}` — {system_name}", inline=True, ) await interaction.followup.send(embed=embed) # ------------------------------------------------------------------ # /help # ------------------------------------------------------------------ @app_commands.command(name="help", description="How to use this bot.") async def help(self, interaction: discord.Interaction): embed = discord.Embed( title="DRB Radio Bot — Help", color=0x5865f2, ) embed.add_field( name="/join `system`", value=( "Join your current voice channel and start streaming a radio system.\n" "Use the `token` option to pick a specific bot from the pool (optional).\n" "You must be in a voice channel first." ), inline=False, ) embed.add_field( name="/leave", value="Stop streaming and disconnect all active radio bots in this server.", inline=False, ) embed.add_field( name="/status", value="Show all registered edge nodes and which systems they're monitoring.", inline=False, ) embed.add_field( name="Radio bot direct commands", value=( "Once a radio bot is in your voice channel, you can control it by **mentioning it** in any text channel:\n" "- **@botname joinme** — Move the bot to your current voice channel\n" "- **@botname leave** — Disconnect the bot from voice\n\n" "The radio bot's green speaking ring lights up only when radio is actively transmitting." ), inline=False, ) embed.set_footer(text="Audio is streamed live from SDR edge nodes via Icecast.") await interaction.response.send_message(embed=embed) async def setup(bot: commands.Bot): await bot.add_cog(RadioCommands(bot))