Merge pull request 'implement-bot-presence' (#6) from implement-bot-presence into master

Reviewed-on: #6

.gitignore

@@ -3,3 +3,4 @@ bot-poc.py
 configs*
 .env
 *.log*
+.venv

app/internal/bot_manager.py

@@ -165,9 +165,15 @@ class DiscordBotManager:
                 LOGGER.info("Loaded OPUS library for armv7l")
                 return "armv7l"
 
-    async def set_presence(self, presence: str):
-        """ Set the presense (activity) of the bot """
+    async def set_presence(self, system_name: str):
+        """ Set the presence (activity) of the bot """
+        if not self.bot:
+            LOGGER.warning("Bot is not running, cannot set presence.")
+            return
+
         try:
-            await self.bot.change_presence(activity=Activity(type=ActivityType.listening, name=presence))
+            activity = Activity(type=ActivityType.listening, name=system_name)
+            await self.bot.change_presence(activity=activity)
+            LOGGER.info(f"Bot presence set to 'Listening to {system_name}'")
         except Exception as pe:
             LOGGER.error(f"Unable to set presence: '{pe}'")

app/internal/op25_config_utls.py

@@ -0,0 +1,70 @@
+import csv
+import json
+from models import TalkgroupTag
+from typing import List
+from internal.logger import create_logger
+
+LOGGER = create_logger(__name__)
+
+def save_talkgroup_tags(talkgroup_tags: List[TalkgroupTag]) -> None:
+    """
+    Writes a list of tags to the tags file.
+
+    Args:
+        talkgroup_tags (List[TalkgroupTag]): The list of TalkgroupTag instances.
+    """
+    with open("/configs/active.cfg.tags.tsv", 'w', newline='', encoding='utf-8') as file:
+        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
+        # Write rows
+        for tag in talkgroup_tags:
+            writer.writerow([tag.talkgroup, tag.tagDec])
+
+def save_whitelist(talkgroup_tags: List[int]) -> None:
+    """
+    Writes a list of talkgroups to the whitelists file.
+
+    Args:
+        talkgroup_tags (List[int]): The list of decimals to whitelist.
+    """
+    with open("/configs/active.cfg.whitelist.tsv", 'w', newline='', encoding='utf-8') as file:
+        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
+        # Write rows
+        for tag in talkgroup_tags:
+            writer.writerow([tag])
+
+def del_none_in_dict(d):
+    """
+    Delete keys with the value ``None`` in a dictionary, recursively.
+
+    This alters the input so you may wish to ``copy`` the dict first.
+    """
+    for key, value in list(d.items()):
+        LOGGER.info(f"Key: '{key}'\nValue: '{value}'")
+        if value is None:
+            del d[key]
+        elif isinstance(value, dict):
+            del_none_in_dict(value)
+        elif isinstance(value, list):
+            for iterative_value in value:
+                del_none_in_dict(iterative_value)
+    return d  # For convenience
+
+def get_current_system_from_config() -> str:
+    # Get the current config
+    with open('/configs/active.cfg.json', 'r') as f:
+        json_data = f.read()
+        if isinstance(json_data, str):
+            try:
+                data = json.loads(json_data)
+            except json.JSONDecodeError:
+                return None
+        elif isinstance(json_data, dict):
+            data = json_data
+        else:
+            return None
+
+        if "channels" in data and isinstance(data["channels"], list) and len(data["channels"]) > 0:
+            first_channel = data["channels"][0]
+            if "name" in first_channel:
+                return first_channel["name"]
+    return None

app/main.py

@@ -3,6 +3,7 @@ import routers.op25_controller as op25_controller
 import routers.pulse as pulse
 import routers.bot as bot
 from internal.logger import create_logger
+from internal.bot_manager import DiscordBotManager
 
 # Initialize logging
 LOGGER = create_logger(__name__)
@@ -10,6 +11,9 @@ LOGGER = create_logger(__name__)
 # Define FastAPI app
 app = FastAPI()
 
-app.include_router(op25_controller.router, prefix="/op25")
+# Initialize Discord Bot Manager
+bot_manager_instance = DiscordBotManager()
+
+app.include_router(op25_controller.create_op25_router(bot_manager=bot_manager_instance), prefix="/op25")
 app.include_router(pulse.router, prefix="/pulse")
-app.include_router(bot.router, prefix="/bot")
+app.include_router(bot.create_bot_router(bot_manager=bot_manager_instance), prefix="/bot")

app/routers/bot.py

@@ -5,12 +5,10 @@ from internal.logger import create_logger
 
 LOGGER = create_logger(__name__)
 
-# Define FastAPI app
+# Function to create router
+def create_bot_router(bot_manager: DiscordBotManager):
     router = APIRouter()
 
-# Initialize Discord Bot Manager
-bot_manager = DiscordBotManager()
-
     # API Endpoints
     @router.post("/start_bot")
     async def start_bot(config: BotConfig):
@@ -56,3 +54,5 @@ async def get_status():
             "active_token": bot_manager.token
         }
         return status
+
+    return router

app/routers/op25_controller.py

@@ -3,18 +3,20 @@ import subprocess
 import os
 import signal
 import json
-import csv
-from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, AudioConfig, TerminalConfig, TalkgroupTag
+from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, AudioConfig, TerminalConfig
 from internal.logger import create_logger
-from typing import List
+from internal.bot_manager import DiscordBotManager
+from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
 
-router = APIRouter()
 LOGGER = create_logger(__name__)
 
 op25_process = None
 OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
 OP25_SCRIPT = "run_multi-rx_service.sh"
 
+def create_op25_router(bot_manager: DiscordBotManager):
+    router = APIRouter()
+
     @router.post("/start")
     async def start_op25():
         global op25_process
@@ -105,49 +107,20 @@ async def generate_config(generator: ConfigGenerator):
             with open('/configs/active.cfg.json', 'w') as f:
                 json.dump(del_none_in_dict(config_dict), f, indent=2)
 
+            # Set the presence of the bot (if it's online)
+            await bot_manager.set_presence(generator.systemName)
+
             return {"message": "Config exported to '/configs/active.cfg.json'"}
         except Exception as e:
             raise HTTPException(status_code=500, detail=str(e))
 
-def save_talkgroup_tags(talkgroup_tags: List[TalkgroupTag]) -> None:
-    """
-    Writes a list of tags to the tags file.
+    @router.post("/update-prensece")
+    async def update_prensece():
+        current_system = get_current_system_from_config()
+        if not current_system:
+            raise HTTPException(status_code=500, detail="Unable to get current system.")
 
-    Args:
-        talkgroup_tags (List[TalkgroupTag]): The list of TalkgroupTag instances.
-    """
-    with open("/configs/active.cfg.tags.tsv", 'w', newline='', encoding='utf-8') as file:
-        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
-        # Write rows
-        for tag in talkgroup_tags:
-            writer.writerow([tag.talkgroup, tag.tagDec])
+        await bot_manager.set_presence(current_system)
+        return current_system
 
-def save_whitelist(talkgroup_tags: List[int]) -> None:
-    """
-    Writes a list of talkgroups to the whitelists file.
-
-    Args:
-        talkgroup_tags (List[int]): The list of decimals to whitelist.
-    """
-    with open("/configs/active.cfg.whitelist.tsv", 'w', newline='', encoding='utf-8') as file:
-        writer = csv.writer(file, delimiter='\t', lineterminator='\n')
-        # Write rows
-        for tag in talkgroup_tags:
-            writer.writerow([tag])
-
-def del_none_in_dict(d):
-    """
-    Delete keys with the value ``None`` in a dictionary, recursively.
-
-    This alters the input so you may wish to ``copy`` the dict first.
-    """
-    for key, value in list(d.items()):
-        LOGGER.info(f"Key: '{key}'\nValue: '{value}'")
-        if value is None:
-            del d[key]
-        elif isinstance(value, dict):
-            del_none_in_dict(value)
-        elif isinstance(value, list):
-            for iterative_value in value:
-                del_none_in_dict(iterative_value)
-    return d  # For convenience
+    return router