Moved types to own file, added debug to drb api

This commit is contained in:
Logan Cusano
2025-04-27 02:06:12 -04:00
parent e414a71013
commit b6587447b9
2 changed files with 55 additions and 45 deletions

View File

@@ -3,51 +3,7 @@ import json
import asyncio # Import asyncio for running the example usage import asyncio # Import asyncio for running the example usage
from enum import Enum from enum import Enum
from typing import Dict, Any, List, Optional from typing import Dict, Any, List, Optional
from drb_cdb_types import ConfigGenerator
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
class DRBCDBAPI: class DRBCDBAPI:
""" """
@@ -138,14 +94,17 @@ class DRBCDBAPI:
async def start_op25(self): async def start_op25(self):
"""Starts the OP25 process asynchronously.""" """Starts the OP25 process asynchronously."""
print(f"Starting OP25")
return await self._post("/op25/start") return await self._post("/op25/start")
async def stop_op25(self): async def stop_op25(self):
"""Stops the OP25 process asynchronously.""" """Stops the OP25 process asynchronously."""
print(f"Stopping OP25")
return await self._post("/op25/stop") return await self._post("/op25/stop")
async def get_op25_status(self): async def get_op25_status(self):
"""Gets the status of the OP25 process asynchronously.""" """Gets the status of the OP25 process asynchronously."""
print(f"Getting OP25 status")
return await self._get("/op25/status") return await self._get("/op25/status")
async def generate_op25_config(self, config_data: ConfigGenerator): async def generate_op25_config(self, config_data: ConfigGenerator):
@@ -156,12 +115,14 @@ class DRBCDBAPI:
config_data: A ConfigGenerator object representing the configuration data. config_data: A ConfigGenerator object representing the configuration data.
""" """
# Convert the ConfigGenerator object to a dictionary before sending as JSON # 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()) return await self._post("/op25/generate-config", data=config_data.to_dict())
# --- Pulse Audio Endpoints --- # --- Pulse Audio Endpoints ---
async def get_pulse_status(self): async def get_pulse_status(self):
"""Gets the status of the Pulse Audio process asynchronously.""" """Gets the status of the Pulse Audio process asynchronously."""
print(f"Checking Pulseaudio status")
return await self._get("/pulse/status") return await self._get("/pulse/status")
# --- Bot Endpoints --- # --- Bot Endpoints ---
@@ -173,10 +134,12 @@ class DRBCDBAPI:
Args: Args:
token: The Discord bot token. token: The Discord bot token.
""" """
print(f"Starting bot with token: '{token}'")
return await self._post("/bot/start_bot", data={"token": token}) return await self._post("/bot/start_bot", data={"token": token})
async def stop_bot(self): async def stop_bot(self):
"""Stops the Discord bot asynchronously.""" """Stops the Discord bot asynchronously."""
print("Stopping bot")
return await self._post("/bot/stop_bot") return await self._post("/bot/stop_bot")
async def join_voice_channel(self, guild_id: int, channel_id: int): 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. guild_id: The ID of the guild.
channel_id: The ID of the voice channel. 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}) 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): async def leave_voice_channel(self, guild_id: int):
@@ -196,10 +160,12 @@ class DRBCDBAPI:
Args: Args:
guild_id: The ID of the guild to leave the voice channel from. 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}) return await self._post("/bot/leave_voice", data={"guild_id": guild_id})
async def get_bot_status(self): async def get_bot_status(self):
"""Gets the status of the Discord bot asynchronously.""" """Gets the status of the Discord bot asynchronously."""
print("Getting bot status")
return await self._get("/bot/status") return await self._get("/bot/status")
# Example Usage (assuming your FastAPI app is running on http://localhost:8000) # Example Usage (assuming your FastAPI app is running on http://localhost:8000)

44
app/drb_cdb_types.py Normal file
View File

@@ -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