logan/drb-core-server
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Moved types to their own file

Browse Source
This commit is contained in:
Logan Cusano
2025-05-24 14:56:56 -04:00
parent 588e794ae9
commit 1d5359d788
2 changed files with 110 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
app/internal/db_wrappers.py
View File

@@ -1,10 +1,10 @@
import os
import typing
import asyncio
from uuid import uuid4
from typing import Optional, List, Dict, Any
from enum import Enum
from internal.db_handler import MongoHandler
from internal.types import System
# Init vars
DB_NAME = os.getenv("DB_NAME", "default_db")
@@ -12,115 +12,6 @@ MONGO_URL = os.getenv("MONGO_URL", "mongodb://10.10.202.4:27017/")
SYSTEM_DB_COLLECTION_NAME = "radio_systems"
# --- Types
class DemodTypes(str, Enum):
P25 = "P25"
DMR = "DMR"
ANALOG = "NBFM"
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 System:
"""
A basic data model for a channel/system entry in a radio system.
"""
def __init__(self,
_id: str,
_type: DemodTypes,
name: str,
frequency_khz: List[int],
location: str,
avail_on_nodes: List[str],
description: Optional[str] = "",
tags: Optional[List[TalkgroupTag]] = None,
whitelist: Optional[List[int]] = None):
"""
Initializes a System object.
Args:
_id: A unique identifier for the entry (e.g., MongoDB ObjectId string).
_type: The demodulation type (P25, NBFM, etc.).
name: The name of the channel/system.
frequency_khz: The frequency in kilohertz.
location: The geographical location or coverage area.
avail_on_nodes: A list of node identifiers where this is available.
description: A brief description.
"""
self._id: str = _id
self.type: DemodTypes = _type
self.name: str = name
self.frequency_khz: List[int] = frequency_khz
self.location: str = location
self.avail_on_nodes: List[str] = avail_on_nodes
self.description: str = description or ""
self.tags: List[TalkgroupTag] = tags or None
self.whitelist: List[int] = whitelist or None
def __repr__(self) -> str:
"""
Provides a developer-friendly string representation of the object.
"""
# Use self.type.value for string representation of the enum
return (f"System(_id='{self._id}', type='{self.type.value}', name='{self.name}', "
f"frequency_khz={self.frequency_khz}, location='{self.location}', "
f"avail_on_nodes={self.avail_on_nodes}, description='{self.description}',"
f" tags='{self.tags}', whitelist='{self.whitelist}')")
def to_dict(self) -> Dict[str, Any]:
"""
Converts the System object to a dictionary suitable for MongoDB.
Converts the DemodTypes enum to its string value.
"""
return {
"_id": self._id,
"type": self.type.value, # Store the enum value (string)
"name": self.name,
"frequency_khz": self.frequency_khz,
"location": self.location,
"avail_on_nodes": self.avail_on_nodes,
"description": self.description,
"tags": self.tags,
"whitelist": self.whitelist,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "System":
"""
Creates a System object from a dictionary (e.g., from MongoDB).
Converts the 'type' string back to a DemodTypes enum member.
"""
# Ensure required keys exist and handle potential type mismatches if necessary
# Convert the type string back to the DemodTypes enum
system_type = DemodTypes(data.get("type")) if data.get("type") else None # Handle missing or invalid type
if system_type is None:
# Handle error: could raise an exception or return None/default
# For this example, let's raise an error if type is missing/invalid
raise ValueError(f"Invalid or missing 'type' in document data: {data}")
return cls(
_id=data.get("_id"),
_type=system_type,
name=data.get("name", ""), # Provide default empty string if name is missing
frequency_khz=data.get("frequency_khz", 0), # Provide default 0 if missing
location=data.get("location", ""),
avail_on_nodes=data.get("avail_on_nodes", []), # Provide default empty list
description=data.get("description", ""),
tags=data.get("tags", None),
whitelist=data.get("whitelist", None)
)
# --- System class ---
class SystemDbController():
def __init__(self):

109
app/internal/types.py Normal file
View File

@@ -0,0 +1,109 @@
from typing import Optional, List, Dict, Any
from enum import Enum
class DemodTypes(str, Enum):
P25 = "P25"
DMR = "DMR"
ANALOG = "NBFM"
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 System:
"""
A basic data model for a channel/system entry in a radio system.
"""
def __init__(self,
_id: str,
_type: DemodTypes,
name: str,
frequency_khz: List[str],
location: str,
avail_on_nodes: List[str],
description: Optional[str] = "",
tags: Optional[List[TalkgroupTag]] = None,
whitelist: Optional[List[int]] = None):
"""
Initializes a System object.
Args:
_id: A unique identifier for the entry (e.g., MongoDB ObjectId string).
_type: The demodulation type (P25, NBFM, etc.).
name: The name of the channel/system.
frequency_khz: The frequency in kilohertz.
location: The geographical location or coverage area.
avail_on_nodes: A list of node identifiers where this is available.
description: A brief description.
"""
self._id: str = _id
self.type: DemodTypes = _type
self.name: str = name
self.frequency_khz: List[int] = frequency_khz
self.location: str = location
self.avail_on_nodes: List[str] = avail_on_nodes
self.description: str = description or ""
self.tags: List[TalkgroupTag] = tags or None
self.whitelist: List[int] = whitelist or None
def __repr__(self) -> str:
"""
Provides a developer-friendly string representation of the object.
"""
# Use self.type.value for string representation of the enum
return (f"System(_id='{self._id}', type='{self.type.value}', name='{self.name}', "
f"frequency_khz={self.frequency_khz}, location='{self.location}', "
f"avail_on_nodes={self.avail_on_nodes}, description='{self.description}',"
f" tags='{self.tags}', whitelist='{self.whitelist}')")
def to_dict(self) -> Dict[str, Any]:
"""
Converts the System object to a dictionary suitable for MongoDB.
Converts the DemodTypes enum to its string value.
"""
return {
"_id": self._id,
"type": self.type.value, # Store the enum value (string)
"name": self.name,
"frequency_khz": self.frequency_khz,
"location": self.location,
"avail_on_nodes": self.avail_on_nodes,
"description": self.description,
"tags": self.tags,
"whitelist": self.whitelist,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "System":
"""
Creates a System object from a dictionary (e.g., from MongoDB).
Converts the 'type' string back to a DemodTypes enum member.
"""
# Ensure required keys exist and handle potential type mismatches if necessary
# Convert the type string back to the DemodTypes enum
system_type = DemodTypes(data.get("type")) if data.get("type") else None # Handle missing or invalid type
if system_type is None:
# Handle error: could raise an exception or return None/default
# For this example, let's raise an error if type is missing/invalid
raise ValueError(f"Invalid or missing 'type' in document data: {data}")
return cls(
_id=data.get("_id"),
_type=system_type,
name=data.get("name", ""), # Provide default empty string if name is missing
frequency_khz=data.get("frequency_khz", 0), # Provide default 0 if missing
location=data.get("location", ""),
avail_on_nodes=data.get("avail_on_nodes", []), # Provide default empty list
description=data.get("description", ""),
tags=data.get("tags", None),
whitelist=data.get("whitelist", None)
)