""" Database driver for OCPP WebSocket server - stores events in PostgreSQL database Provides persistent storage for OCPP events and charge point data """ import logging import asyncio from datetime import datetime from typing import Dict, Any, Optional, List logger = logging.getLogger(__name__) class DatabaseDriver: """PostgreSQL database driver for storing OCPP events and charge point data""" def __init__(self, connection_url: Optional[str] = None): """ Initialize PostgreSQL database driver Args: connection_url: PostgreSQL connection URL (optional, can be from config) """ self.connection_url = connection_url or self._get_database_url() self.connection = None self.is_connected = False logger.info(f"PostgreSQL driver initialized with URL: {self._mask_connection_url()}") def _get_database_url(self) -> str: """Get PostgreSQL database URL from config""" from ..config import get_config config = get_config() return config.get_database_url() or "postgresql://ocpp:password@localhost:5432/ocpp_db" def _mask_connection_url(self) -> str: """Mask sensitive parts of connection URL for logging""" if not self.connection_url: return "None" # Simple masking for passwords import re masked = re.sub(r'://([^:]+):([^@]+)@', r'://\1:***@', self.connection_url) return masked async def connect(self) -> bool: """Connect to PostgreSQL database""" try: if not self.connection_url.startswith("postgresql"): logger.error(f"Only PostgreSQL databases are supported. Got URL: {self._mask_connection_url()}") return False await self._connect_postgresql() await self._create_tables() self.is_connected = True logger.info("PostgreSQL database connection established successfully") return True except Exception as e: logger.error(f"Failed to connect to PostgreSQL database: {e}") self.is_connected = False return False async def _connect_postgresql(self): """Connect to PostgreSQL database""" try: import asyncpg self.connection = await asyncpg.connect(self.connection_url) logger.debug("Connected to PostgreSQL database") except ImportError: logger.error("asyncpg not installed. Install with: pip install asyncpg") raise async def _create_tables(self): """Create required PostgreSQL database tables""" try: await self._create_tables_postgresql() logger.debug("PostgreSQL database tables created/verified successfully") except Exception as e: logger.error(f"Failed to create PostgreSQL database tables: {e}") raise async def _create_tables_postgresql(self): """Create PostgreSQL tables""" await self.connection.execute(""" CREATE TABLE IF NOT EXISTS ocpp_events ( id SERIAL PRIMARY KEY, timestamp TIMESTAMP NOT NULL, charge_point_id VARCHAR(255) NOT NULL, event_type VARCHAR(100) NOT NULL, action VARCHAR(100), message_id VARCHAR(255), data JSONB, status VARCHAR(50) DEFAULT 'processed', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) await self.connection.execute(""" CREATE TABLE IF NOT EXISTS charge_points ( id VARCHAR(255) PRIMARY KEY, status VARCHAR(50) DEFAULT 'Available', last_seen TIMESTAMP, boot_notification JSONB, heartbeat_interval INTEGER DEFAULT 300, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) await self.connection.execute(""" CREATE TABLE IF NOT EXISTS transactions ( id SERIAL PRIMARY KEY, charge_point_id VARCHAR(255) NOT NULL, connector_id INTEGER NOT NULL, id_tag VARCHAR(255), start_timestamp TIMESTAMP, stop_timestamp TIMESTAMP, meter_start INTEGER, meter_stop INTEGER, status VARCHAR(50) DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) async def log_event(self, event_type: str, data: dict, source: str) -> dict: """Log an OCPP event to PostgreSQL database""" if not self.is_connected: if not await self.connect(): return {"status": "error", "error": "PostgreSQL database not connected"} try: timestamp = datetime.utcnow() action = data.get("action", event_type) message_id = data.get("message_id", data.get("uniqueId")) await self._log_event_postgresql(timestamp, source, event_type, action, message_id, data) logger.debug(f"[POSTGRESQL] Logged {event_type} event from {source}") return {"status": "logged", "database": "postgresql", "event_type": event_type} except Exception as e: logger.error(f"[POSTGRESQL] Failed to log event: {e}") return {"status": "error", "error": str(e)} async def _log_event_postgresql(self, timestamp, charge_point_id, event_type, action, message_id, data): """Log event to PostgreSQL""" await self.connection.execute( """INSERT INTO ocpp_events (timestamp, charge_point_id, event_type, action, message_id, data) VALUES ($1, $2, $3, $4, $5, $6)""", timestamp, charge_point_id, event_type, action, message_id, data ) async def send_message(self, charge_point_id: str, message: dict) -> dict: """Send/log a message - compatibility method""" action = message.get("action", "Unknown") data = message.get("data", message) return await self.log_event(action, data, charge_point_id) async def update_charge_point_status(self, charge_point_id: str, status: str, boot_notification: Optional[dict] = None) -> bool: """Update charge point status and information in PostgreSQL""" if not self.is_connected: if not await self.connect(): return False try: timestamp = datetime.utcnow() await self._update_charge_point_postgresql(charge_point_id, status, timestamp, boot_notification) logger.debug(f"Updated charge point {charge_point_id} status to {status}") return True except Exception as e: logger.error(f"Failed to update charge point status: {e}") return False async def _update_charge_point_postgresql(self, charge_point_id, status, timestamp, boot_notification): """Update charge point in PostgreSQL""" await self.connection.execute( """INSERT INTO charge_points (id, status, last_seen, boot_notification, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, last_seen = EXCLUDED.last_seen, boot_notification = EXCLUDED.boot_notification, updated_at = EXCLUDED.updated_at""", charge_point_id, status, timestamp, boot_notification, timestamp ) async def get_charge_points(self) -> List[Dict[str, Any]]: """Get all charge points from PostgreSQL database""" if not self.is_connected: if not await self.connect(): return [] try: return await self._get_charge_points_postgresql() except Exception as e: logger.error(f"Failed to get charge points: {e}") return [] async def _get_charge_points_postgresql(self): """Get charge points from PostgreSQL""" rows = await self.connection.fetch("SELECT * FROM charge_points ORDER BY last_seen DESC") return [dict(row) for row in rows] async def close(self): """Close PostgreSQL database connection""" if self.connection: try: await self.connection.close() self.is_connected = False logger.info("PostgreSQL database connection closed") except Exception as e: logger.error(f"Error closing PostgreSQL database connection: {e}") __all__ = ["DatabaseDriver"]