Edge and Offline Handling¶

import asyncio
from azure.iot.device.aio import IoTHubModuleClient

class EdgeConnectivityManager:
    """Manage edge connectivity state."""

    def __init__(self):
        self.client = None
        self.is_connected = False
        self.offline_handler = OfflineDataHandler()

    async def connect(self):
        """Initialize IoT Edge module client."""
        self.client = IoTHubModuleClient.create_from_edge_environment()
        self.client.on_connection_state_change = self._on_connection_change
        await self.client.connect()

    def _on_connection_change(self, connected: bool):
        """Handle connectivity state changes."""
        self.is_connected = connected
        if connected:
            print("Connected - syncing pending data")
            asyncio.create_task(self._sync_offline_data())
        else:
            print("Disconnected - switching to offline mode")

    async def send_telemetry(self, data: dict):
        """Send telemetry with offline fallback."""
        if self.is_connected:
            try:
                await self.client.send_message(json.dumps(data))
            except Exception:
                self.offline_handler.store_event("telemetry", data)
        else:
            self.offline_handler.store_event("telemetry", data)

from hashlib import sha256

class IdempotentEventProcessor:
    """Ensure events are processed exactly once."""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl_seconds = 86400 * 7  # 7 days

    def generate_event_id(self, event: dict) -> str:
        """Generate deterministic event ID."""
        # Use business key fields for ID
        key_fields = sorted([
            event.get("device_id", ""),
            event.get("timestamp", ""),
            event.get("event_type", "")
        ])
        return sha256("|".join(key_fields).encode()).hexdigest()

    def should_process(self, event: dict) -> bool:
        """Check if event should be processed."""
        event_id = self.generate_event_id(event)

        # Try to set key with NX (only if not exists)
        result = self.redis.set(
            f"processed:{event_id}",
            "1",
            nx=True,
            ex=self.ttl_seconds
        )

        return result is not None

class ConflictResolver:
    """Resolve conflicts in offline data sync."""

    STRATEGIES = {
        "last_write_wins": lambda old, new: new,
        "first_write_wins": lambda old, new: old,
        "merge": lambda old, new: {**old, **new},
        "version_check": None  # Custom logic
    }

    def resolve(self, old_record: dict, new_record: dict, strategy: str = "last_write_wins"):
        """Resolve conflict between records."""
        if strategy == "version_check":
            return self._version_based_resolution(old_record, new_record)

        resolver = self.STRATEGIES.get(strategy, self.STRATEGIES["last_write_wins"])
        return resolver(old_record, new_record)

    def _version_based_resolution(self, old: dict, new: dict) -> dict:
        """Resolve based on version vectors."""
        old_version = old.get("_version", 0)
        new_version = new.get("_version", 0)

        if new_version > old_version:
            return new
        elif new_version < old_version:
            return old
        else:
            # Same version - use timestamp
            if new.get("_modified", "") > old.get("_modified", ""):
                return new
            return old

from collections import deque
import os

class HybridBuffer:
    """Memory buffer with disk overflow."""

    def __init__(self, memory_limit: int = 1000, disk_path: str = "/tmp/buffer"):
        self.memory_limit = memory_limit
        self.disk_path = disk_path
        self.memory_buffer = deque(maxlen=memory_limit)
        self.disk_overflow = False
        os.makedirs(disk_path, exist_ok=True)

    def append(self, item: dict):
        """Add item to buffer."""
        if len(self.memory_buffer) >= self.memory_limit:
            self._overflow_to_disk()

        self.memory_buffer.append(item)

    def _overflow_to_disk(self):
        """Move oldest items to disk."""
        overflow_count = len(self.memory_buffer) // 2
        overflow_items = [self.memory_buffer.popleft() for _ in range(overflow_count)]

        filename = f"{self.disk_path}/overflow_{datetime.utcnow().timestamp()}.json"
        with open(filename, 'w') as f:
            json.dump(overflow_items, f)

        self.disk_overflow = True

    def flush(self):
        """Get all buffered items."""
        items = list(self.memory_buffer)
        self.memory_buffer.clear()

        # Include disk overflow
        if self.disk_overflow:
            for filename in sorted(os.listdir(self.disk_path)):
                filepath = os.path.join(self.disk_path, filename)
                with open(filepath, 'r') as f:
                    items.extend(json.load(f))
                os.remove(filepath)
            self.disk_overflow = False

        return items