Implement presence change

app/internal/op25_config_utls.py

@@ -0,0 +1,70 @@
+import csv
+import json
+from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, AudioConfig, TerminalConfig, 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,9 @@ from internal.logger import create_logger
 
 LOGGER = create_logger(__name__)
 
-# Define FastAPI app
+def create_bot_router(bot_manager: DiscordBotManager): # Function to create router
     router = APIRouter()
 
-# Initialize Discord Bot Manager
-bot_manager = DiscordBotManager()
-
     # API Endpoints
     @router.post("/start_bot")
     async def start_bot(config: BotConfig):
@@ -56,3 +53,5 @@ async def get_status():
             "active_token": bot_manager.token
         }
         return status
+    
+    return router # Return the configured router

app/routers/op25_controller.py

@@ -3,18 +3,21 @@ import subprocess
 import os
 import signal
 import json
-import csv
 from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, AudioConfig, TerminalConfig, TalkgroupTag
 from internal.logger import create_logger
+from internal.bot_manager import DiscordBotManager
 from typing import List
+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 +108,21 @@ 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.
+    return router
 
-    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