demand response

microservices/demand-response-service/main.py

@@ -23,22 +23,130 @@ from demand_response_service import DemandResponseService
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+# Background task functions
+async def event_scheduler_task():
+    """Background task for checking and executing scheduled events"""
+    logger.info("Starting event scheduler task")
+
+    while True:
+        try:
+            db = await get_database()
+            redis = await get_redis()
+            service = DemandResponseService(db, redis)
+
+            # Check for events that need to be executed
+            await service.check_scheduled_events()
+
+            # Sleep for 60 seconds between checks
+            await asyncio.sleep(60)
+
+        except asyncio.CancelledError:
+            logger.info("Event scheduler task cancelled")
+            raise
+        except Exception as e:
+            logger.error(f"Error in event scheduler task: {e}")
+            await asyncio.sleep(120)  # Wait longer on error
+
+
+async def auto_response_task():
+    """Background task for automatic demand response"""
+    logger.info("Starting auto-response task")
+
+    while True:
+        try:
+            db = await get_database()
+            redis = await get_redis()
+            service = DemandResponseService(db, redis)
+
+            # Check for auto-response opportunities
+            await service.process_auto_responses()
+
+            # Sleep for 30 seconds between checks
+            await asyncio.sleep(30)
+
+        except asyncio.CancelledError:
+            logger.info("Auto-response task cancelled")
+            raise
+        except Exception as e:
+            logger.error(f"Error in auto-response task: {e}")
+            await asyncio.sleep(90)  # Wait longer on error
+
+
+async def energy_data_subscriber_task():
+    """Subscribe to energy_data Redis channel for device power updates"""
+    logger.info("Starting energy data subscriber task")
+
+    try:
+        redis = await get_redis()
+        db = await get_database()
+        service = DemandResponseService(db, redis)
+
+        pubsub = redis.pubsub()
+        await pubsub.subscribe("energy_data")
+
+        logger.info("Subscribed to energy_data channel")
+
+        while True:
+            try:
+                message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0)
+                if message and message.get('type') == 'message':
+                    import json
+                    data = json.loads(message['data'])
+
+                    # Format: {"sensorId": "sensor_1", "timestamp": 123, "value": 3.5, "unit": "kWh"}
+                    sensor_id = data.get("sensorId")
+                    power_kw = data.get("value", 0.0)
+
+                    # Update service cache
+                    service.update_device_power_cache(sensor_id, power_kw)
+
+            except json.JSONDecodeError as e:
+                logger.warning(f"Invalid JSON in energy_data message: {e}")
+            except Exception as e:
+                logger.error(f"Error processing energy data message: {e}")
+                await asyncio.sleep(5)
+
+    except asyncio.CancelledError:
+        logger.info("Energy data subscriber task cancelled")
+        raise
+    except Exception as e:
+        logger.error(f"Energy data subscriber task failed: {e}")
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Application lifespan manager"""
     logger.info("Demand Response Service starting up...")
     await connect_to_mongo()
     await connect_to_redis()
-    
+
+    # Create global service instance for shutdown cleanup
+    db = await get_database()
+    redis = await get_redis()
+    app.state.dr_service = DemandResponseService(db, redis)
+
     # Start background tasks
     asyncio.create_task(event_scheduler_task())
     asyncio.create_task(auto_response_task())
-    
+    asyncio.create_task(energy_data_subscriber_task())
+
     logger.info("Demand Response Service startup complete")
-    
+
     yield
-    
+
     logger.info("Demand Response Service shutting down...")
+
+    # Cancel all active DR events gracefully
+    if hasattr(app.state, 'dr_service'):
+        active_event_ids = list(app.state.dr_service.active_events.keys())
+        if active_event_ids:
+            logger.info(f"Cancelling {len(active_event_ids)} active events...")
+            for event_id in active_event_ids:
+                try:
+                    await app.state.dr_service.cancel_event(event_id)
+                except Exception as e:
+                    logger.error(f"Error cancelling event {event_id}: {e}")
+
     await close_mongo_connection()
     logger.info("Demand Response Service shutdown complete")