From b6587447b9c19f49632fa4ffbd66886017f5754a Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 27 Apr 2025 02:06:12 -0400 Subject: [PATCH] Moved types to own file, added debug to drb api --- app/drb_cdb_api.py | 56 +++++++++----------------------------------- app/drb_cdb_types.py | 44 ++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 45 deletions(-) create mode 100644 app/drb_cdb_types.py diff --git a/app/drb_cdb_api.py b/app/drb_cdb_api.py index 21ffbab..bcd443f 100644 --- a/app/drb_cdb_api.py +++ b/app/drb_cdb_api.py @@ -3,51 +3,7 @@ import json import asyncio # Import asyncio for running the example usage from enum import Enum from typing import Dict, Any, List, Optional - -class DecodeMode(str, Enum): - P25 = "P25" - DMR = "DMR" - ANALOG = "NBFM" # Note: The API code uses "NBFM" for analog - -class TalkgroupTag: - """Represents a talkgroup tag.""" - def __init__(self, talkgroup: str, tagDec: int): - self.talkgroup = talkgroup - self.tagDec = tagDec - - # Add a method to convert to a dictionary, useful for sending as JSON - def to_dict(self) -> Dict[str, Any]: - return {"talkgroup": self.talkgroup, "tagDec": self.tagDec} - -class ConfigGenerator: - """Represents the configuration data structure for the API.""" - def __init__( - self, - type: DecodeMode, - systemName: str, - channels: List[str], - tags: Optional[List[TalkgroupTag]] = None, - whitelist: Optional[List[int]] = None - ): - self.type = type - self.systemName = systemName - self.channels = channels - self.tags = tags - self.whitelist = whitelist - - # Add a method to convert to a dictionary, useful for sending as JSON - def to_dict(self) -> Dict[str, Any]: - data = { - "type": self.type.value, # Use .value for Enum - "systemName": self.systemName, - "channels": self.channels, - } - if self.tags is not None: - # Convert list of TalkgroupTag objects to list of dictionaries - data["tags"] = [tag.to_dict() for tag in self.tags] - if self.whitelist is not None: - data["whitelist"] = self.whitelist - return data +from drb_cdb_types import ConfigGenerator class DRBCDBAPI: """ @@ -138,14 +94,17 @@ class DRBCDBAPI: async def start_op25(self): """Starts the OP25 process asynchronously.""" + print(f"Starting OP25") return await self._post("/op25/start") async def stop_op25(self): """Stops the OP25 process asynchronously.""" + print(f"Stopping OP25") return await self._post("/op25/stop") async def get_op25_status(self): """Gets the status of the OP25 process asynchronously.""" + print(f"Getting OP25 status") return await self._get("/op25/status") async def generate_op25_config(self, config_data: ConfigGenerator): @@ -156,12 +115,14 @@ class DRBCDBAPI: config_data: A ConfigGenerator object representing the configuration data. """ # Convert the ConfigGenerator object to a dictionary before sending as JSON + print(f"Generate OP25 config") return await self._post("/op25/generate-config", data=config_data.to_dict()) # --- Pulse Audio Endpoints --- async def get_pulse_status(self): """Gets the status of the Pulse Audio process asynchronously.""" + print(f"Checking Pulseaudio status") return await self._get("/pulse/status") # --- Bot Endpoints --- @@ -173,10 +134,12 @@ class DRBCDBAPI: Args: token: The Discord bot token. """ + print(f"Starting bot with token: '{token}'") return await self._post("/bot/start_bot", data={"token": token}) async def stop_bot(self): """Stops the Discord bot asynchronously.""" + print("Stopping bot") return await self._post("/bot/stop_bot") async def join_voice_channel(self, guild_id: int, channel_id: int): @@ -187,6 +150,7 @@ class DRBCDBAPI: guild_id: The ID of the guild. channel_id: The ID of the voice channel. """ + print(f"Joining voice channel: Guild ID: {guild_id}, Channel ID: {channel_id}") return await self._post("/bot/join_voice", data={"guild_id": guild_id, "channel_id": channel_id}) async def leave_voice_channel(self, guild_id: int): @@ -196,10 +160,12 @@ class DRBCDBAPI: Args: guild_id: The ID of the guild to leave the voice channel from. """ + print(f"Leaving voice channel on guild: {guild_id}") return await self._post("/bot/leave_voice", data={"guild_id": guild_id}) async def get_bot_status(self): """Gets the status of the Discord bot asynchronously.""" + print("Getting bot status") return await self._get("/bot/status") # Example Usage (assuming your FastAPI app is running on http://localhost:8000) diff --git a/app/drb_cdb_types.py b/app/drb_cdb_types.py new file mode 100644 index 0000000..bf8ca99 --- /dev/null +++ b/app/drb_cdb_types.py @@ -0,0 +1,44 @@ +class DecodeMode(str, Enum): + P25 = "P25" + DMR = "DMR" + ANALOG = "NBFM" # Note: The API code uses "NBFM" for analog + +class TalkgroupTag: + """Represents a talkgroup tag.""" + def __init__(self, talkgroup: str, tagDec: int): + self.talkgroup = talkgroup + self.tagDec = tagDec + + # Add a method to convert to a dictionary, useful for sending as JSON + def to_dict(self) -> Dict[str, Any]: + return {"talkgroup": self.talkgroup, "tagDec": self.tagDec} + +class ConfigGenerator: + """Represents the configuration data structure for the API.""" + def __init__( + self, + type: DecodeMode, + systemName: str, + channels: List[str], + tags: Optional[List[TalkgroupTag]] = None, + whitelist: Optional[List[int]] = None + ): + self.type = type + self.systemName = systemName + self.channels = channels + self.tags = tags + self.whitelist = whitelist + + # Add a method to convert to a dictionary, useful for sending as JSON + def to_dict(self) -> Dict[str, Any]: + data = { + "type": self.type.value, # Use .value for Enum + "systemName": self.systemName, + "channels": self.channels, + } + if self.tags is not None: + # Convert list of TalkgroupTag objects to list of dictionaries + data["tags"] = [tag.to_dict() for tag in self.tags] + if self.whitelist is not None: + data["whitelist"] = self.whitelist + return data