Industrial IoT Edge Platform

import asyncio
import json
from dataclasses import dataclass, asdict
from typing import Optional
from azure.iot.device import IoTHubModuleClient

@dataclass
class SensorReading:
    device_id: str
    timestamp: float
    value: float
    quality: str  # 'good', 'uncertain', 'bad'
    
class EdgeGateway:
    def __init__(self, connection_string: str):
        self.client = IoTHubModuleClient.create_from_connection_string(
            connection_string
        )
        self.local_buffer = []
        self.cloud_connected = False
        
    async def process_reading(self, reading: SensorReading):
        # Always store locally first
        self.local_buffer.append(asdict(reading))
        
        # Try to send to cloud
        if self.cloud_connected:
            try:
                message = json.dumps(asdict(reading))
                await self.client.send_message(message)
            except Exception:
                self.cloud_connected = False
                self._persist_to_disk(reading)
        else:
            # Store-and-forward pattern
            self._persist_to_disk(reading)
    
    def _persist_to_disk(self, reading: SensorReading):
        # SQLite for local resilience
        with self._get_db() as db:
            db.execute(
                "INSERT INTO pending VALUES (?, ?, ?, ?)",
                (reading.device_id, reading.timestamp, 
                 reading.value, reading.quality)
            )
            
    async def flush_buffer(self):
        # Send buffered messages when connection restored
        if not self.cloud_connected:
            return
            
        pending = self._get_pending_messages()
        for msg in pending:
            await self.client.send_message(json.dumps(msg))
            self._mark_sent(msg['timestamp'])