Rewrite start

services/mqtt_ingestion/config.py

@@ -0,0 +1,47 @@
+import os
+from dataclasses import dataclass
+import dotenv
+from typing import Optional
+
+dotenv.load_dotenv()
+
+@dataclass
+class RedisConfig:
+    host: str
+    port: int = 6379
+    db: int = 0
+    password: Optional[str] = None
+
+@dataclass
+class MQTTConfig:
+    broker: str
+    port: int = 1883
+    username: Optional[str] = None
+    password: Optional[str] = None
+    topic: str = "#"
+
+@dataclass
+class Payload:
+    device_id: str
+    sensor_type: str
+    value: float
+    timestamp: Optional[str] = None
+
+
+class Config:
+    def __init__(self):
+        self.redis = RedisConfig(
+            host=os.getenv('REDIS_HOST', 'localhost'),
+            port=int(os.getenv('REDIS_PORT', 6379)),
+            db=int(os.getenv('REDIS_DB', 0)),
+            password=os.getenv('REDIS_PASSWORD', None)
+        )
+        self.mqtt = MQTTConfig(
+            broker=os.getenv('MQTT_BROKER', 'localhost'),
+            port=int(os.getenv('MQTT_PORT', 1883)),
+            username=os.getenv('MQTT_USERNAME', None),
+            password=None,
+            topic="#"
+        )
+
+config = Config()

services/mqtt_ingestion/main.py

@@ -0,0 +1,119 @@
+import logging
+import signal
+import sys
+from config import config
+from mqtt_client import MQTTClient
+from redis_writer import RedisWriter
+
+# Setup logging
+logging.basicConfig(
+    level=getattr(logging,'INFO'),
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+class MQTTIngestionService:
+    def __init__(self):
+        self.running = False
+        self.redis_writer = None
+        self.mqtt_client = None
+        
+        # Setup signal handlers for graceful shutdown
+        signal.signal(signal.SIGTERM, self._signal_handler)
+        signal.signal(signal.SIGINT, self._signal_handler)
+    
+    def _signal_handler(self, signum, frame):
+        """Handle shutdown signals"""
+        logger.info(f"Received signal {signum}, shutting down...")
+        self.stop()
+    
+    def _handle_sensor_data(self, device_id: str, sensor_type: str, value: float):
+        """
+        This function is called by MQTT client when a message arrives.
+        It just passes the data to Redis writer.
+        """
+        success = self.redis_writer.write_sensor_data(device_id, sensor_type, value)
+        if success:
+            logger.info(f"Processed {device_id}/{sensor_type}: {value}")
+        else:
+            logger.error(f"Failed to process {device_id}/{sensor_type}: {value}")
+    
+    def start(self):
+        """Start the service"""
+        logger.info("Starting MQTT Ingestion Service...")
+        
+        try:
+            # Initialize Redis writer
+            self.redis_writer = RedisWriter()
+            
+            # Initialize MQTT client with our message handler
+            self.mqtt_client = MQTTClient(self._handle_sensor_data)
+            
+            # Connect to MQTT
+            if not self.mqtt_client.connect():
+                logger.error("Failed to connect to MQTT, exiting")
+                return False
+            
+            self.running = True
+            logger.info("Service started successfully")
+            
+            # Start MQTT loop (this blocks)
+            self.mqtt_client.start_loop()
+            
+        except Exception as e:
+            logger.error(f"Service startup failed: {e}")
+            return False
+        
+        return True
+    
+    def stop(self):
+        """Stop the service gracefully"""
+        if not self.running:
+            return
+            
+        logger.info("Stopping service...")
+        self.running = False
+        
+        # Stop MQTT client
+        if self.mqtt_client:
+            self.mqtt_client.stop()
+        
+        # Close Redis connection
+        if self.redis_writer:
+            self.redis_writer.close()
+        
+        logger.info("Service stopped")
+    
+    def health_check(self) -> bool:
+        """Check if service is healthy"""
+        if not self.running:
+            return False
+            
+        # Check Redis connection
+        if not self.redis_writer or not self.redis_writer.health_check():
+            return False
+            
+        return True
+
+def main():
+    """Entry point"""
+    service = MQTTIngestionService(
+        redis_config=config.redis,
+        mqtt_config=config.mqtt
+    )
+    
+    try:
+        # Start the service (blocks until shutdown)
+        success = service.start()
+        if not success:
+            sys.exit(1)
+    except KeyboardInterrupt:
+        logger.info("Received keyboard interrupt")
+    except Exception as e:
+        logger.error(f"Unexpected error: {e}")
+        sys.exit(1)
+    finally:
+        service.stop()
+
+if __name__ == "__main__":
+    main()

services/mqtt_ingestion/mqtt_client.py

@@ -0,0 +1,88 @@
+import logging
+import paho.mqtt.client as mqtt
+from typing import Callable
+from config import config
+
+logger = logging.getLogger(__name__)
+
+class MQTTClient:
+    def __init__(self, message_handler: Callable[[str, str, float], None]):
+        """
+        Args:
+            message_handler: Function that takes (device_id, sensor_type, value)
+        """
+        self.message_handler = message_handler
+        self.client = mqtt.Client()
+        self._setup_callbacks()
+        
+    def _setup_callbacks(self):
+        self.client.on_connect = self._on_connect
+        self.client.on_message = self._on_message
+        self.client.on_disconnect = self._on_disconnect
+        
+        # Set credentials if provided
+        if config.mqtt.username:
+            self.client.username_pw_set(
+                config.mqtt.username, 
+                config.mqtt.password
+            )
+    
+    def _on_connect(self, client, userdata, flags, rc):
+        if rc == 0:
+            logger.info(f"Connected to MQTT broker {config.mqtt.broker}")
+            # Subscribe to all device topics: devices/+/+
+            client.subscribe(config.mqtt.topic_pattern)
+            logger.info(f"Subscribed to {config.mqtt.topic_pattern}")
+        else:
+            logger.error(f"Failed to connect to MQTT broker, code: {rc}")
+    
+    def _on_message(self, client, userdata, msg):
+        try:
+            # Parse topic: devices/{device_id}/{sensor_type}
+            topic_parts = msg.topic.split('/')
+            if len(topic_parts) != 3 or topic_parts[0] != 'devices':
+                logger.warning(f"Invalid topic format: {msg.topic}")
+                return
+                
+            device_id = topic_parts[1]
+            sensor_type = topic_parts[2]
+            
+            # Parse payload as float
+            try:
+                value = float(msg.payload.decode())
+            except ValueError:
+                logger.error(f"Invalid payload for {msg.topic}: {msg.payload}")
+                return
+            
+            # Call the handler (this will be our Redis writer)
+            self.message_handler(device_id, sensor_type, value)
+            
+        except Exception as e:
+            logger.error(f"Error processing MQTT message: {e}")
+    
+    def _on_disconnect(self, client, userdata, rc):
+        if rc != 0:
+            logger.warning("Unexpected MQTT disconnection")
+        else:
+            logger.info("MQTT client disconnected")
+    
+    def connect(self):
+        """Connect to MQTT broker"""
+        try:
+            self.client.connect(
+                config.mqtt.broker, 
+                config.mqtt.port, 
+                config.mqtt.keepalive
+            )
+            return True
+        except Exception as e:
+            logger.error(f"Failed to connect to MQTT: {e}")
+            return False
+    
+    def start_loop(self):
+        """Start the MQTT loop (blocking)"""
+        self.client.loop_forever()
+    
+    def stop(self):
+        """Stop the MQTT client"""
+        self.client.disconnect()

services/mqtt_ingestion/redis_writer.py

@@ -0,0 +1,37 @@
+import redis
+import logging
+from datetime import datetime
+from typing import Optional
+from config import Payload
+
+class RedisWriter:
+    def __init__(self, host: str, port: int, db: int, password: Optional[str] = None):
+        self.logger = logging.getLogger(__name__)
+        self.redis_client = redis.StrictRedis(host=host, port=port, db=db, password=password)
+        try:
+            self.redis_client.ping()
+            self.logger.info("Connected to Redis server successfully.")
+        except redis.ConnectionError as e:
+            self.logger.error(f"Failed to connect to Redis server: {e}")
+            raise
+
+    def write_message(self, topic: str, payload: Payload):
+        """
+        Write a message to a Redis stream with the topic and payload.
+        - Stream: mqtt_stream: {device_id}:{sensor_type}
+        """
+        device_id = payload.device_id
+        sensor_type = payload.sensor_type
+        timestamp = datetime.utcnow().isoformat()
+        stream_key= f"mqtt_stream:{device_id}:{sensor_type}"
+        stream_data = {
+            "value": str(payload),
+            "source": "mqtt",
+            "timestamp": timestamp
+        }
+        try:
+            message_id = self.redis_client.xadd(stream_key, stream_data,maxlen=1000)
+            self.logger.info(f"Message written to Redis: {stream_data}")
+            return message_id
+        except redis.RedisError as e:
+            self.logger.error(f"Failed to write message to Redis: {e}")

services/mqtt_ingestion/requirements.txt

@@ -0,0 +1,2 @@
+redis
+paho-mqtt