diff --git a/app/internal/db_wrappers.py b/app/internal/db_wrappers.py index c89ae66..a773e5c 100644 --- a/app/internal/db_wrappers.py +++ b/app/internal/db_wrappers.py @@ -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): diff --git a/app/internal/types.py b/app/internal/types.py new file mode 100644 index 0000000..e90a3a2 --- /dev/null +++ b/app/internal/types.py @@ -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) + )