Tracking High-Frequency Instrument Usage with IoT Sensors

import os
import sys
import json
import time
import hashlib
import sqlite3
import logging
import requests
import paho.mqtt.client as mqtt
from datetime import datetime, timezone
from typing import Dict, Any

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
MQTT_BROKER = os.getenv("MQTT_BROKER", "localhost")
MQTT_PORT = int(os.getenv("MQTT_PORT", 1883))
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "lab/instruments/+/usage/telemetry")
REST_ENDPOINT = os.getenv("REST_ENDPOINT", "https://hub.university.edu/api/v1/usage/ingest")
FALLBACK_DB = os.getenv("FALLBACK_DB", "/var/lib/lab_iot/fallback_usage.db")
AUDIT_LOG = os.getenv("AUDIT_LOG", "/var/log/lab_iot/audit_verification.log")
MAX_RETRIES = int(os.getenv("MAX_RETRIES", "5"))
BASE_BACKOFF = float(os.getenv("BASE_BACKOFF", "1.5"))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler(AUDIT_LOG, mode="a")
    ]
)
logger = logging.getLogger("iot_ingestion_worker")

# ---------------------------------------------------------------------------
# Persistence Layer
# ---------------------------------------------------------------------------
def init_fallback_db(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS usage_buffer (
            payload_hash TEXT PRIMARY KEY,
            timestamp TEXT NOT NULL,
            instrument_id TEXT NOT NULL,
            raw_json TEXT NOT NULL,
            retry_count INTEGER DEFAULT 0,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn

def buffer_payload(conn: sqlite3.Connection, payload_hash: str, data: Dict[str, Any]) -> None:
    """Idempotent insert. Ignores duplicates based on payload_hash."""
    conn.execute(
        "INSERT OR IGNORE INTO usage_buffer (payload_hash, timestamp, instrument_id, raw_json) VALUES (?, ?, ?, ?)",
        (payload_hash, data.get("timestamp", datetime.now(timezone.utc).isoformat()), data.get("instrument_id", "unknown"), json.dumps(data))
    )
    conn.commit()

def fetch_unsent(conn: sqlite3.Connection, limit: int = 50) -> list[Dict[str, Any]]:
    cursor = conn.execute(
        "SELECT payload_hash, raw_json, retry_count FROM usage_buffer ORDER BY created_at ASC LIMIT ?", (limit,)
    )
    return [{"hash": row[0], "json": json.loads(row[1]), "retries": row[2]} for row in cursor.fetchall()]

def purge_sent(conn: sqlite3.Connection, hashes: list[str]) -> None:
    conn.executemany("DELETE FROM usage_buffer WHERE payload_hash = ?", [(h,) for h in hashes])
    conn.commit()

# ---------------------------------------------------------------------------
# Cryptographic & Validation Layer
# ---------------------------------------------------------------------------
def compute_audit_hash(payload: Dict[str, Any]) -> str:
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

def validate_payload(payload: Dict[str, Any]) -> bool:
    required = {"instrument_id", "timestamp", "state", "metrics"}
    return required.issubset(payload.keys()) and isinstance(payload.get("metrics"), dict)

# ---------------------------------------------------------------------------
# Network & Retry Logic
# ---------------------------------------------------------------------------
def post_to_rest(session: requests.Session, payload: Dict[str, Any], payload_hash: str) -> bool:
    """Send to institutional API with exponential backoff and idempotency key."""
    headers = {
        "Content-Type": "application/json",
        "X-Idempotency-Key": payload_hash,
        "User-Agent": "LabIoT-IngestionWorker/1.0"
    }
    for attempt in range(MAX_RETRIES):
        try:
            resp = session.post(REST_ENDPOINT, json=payload, headers=headers, timeout=5.0)
            if resp.status_code == 200:
                return True
            if 400 <= resp.status_code < 500:
                logger.error(f"Client error {resp.status_code}: {resp.text}")
                return False
            logger.warning(f"Server error {resp.status_code} on attempt {attempt+1}")
        except requests.exceptions.RequestException as e:
            logger.warning(f"Network failure on attempt {attempt+1}: {e}")
        
        backoff = min(BASE_BACKOFF * (2 ** attempt), 30.0)
        time.sleep(backoff)
    return False

# ---------------------------------------------------------------------------
# MQTT Callbacks & Main Loop
# ---------------------------------------------------------------------------
class IngestionWorker:
    def __init__(self):
        self.db = init_fallback_db(FALLBACK_DB)
        self.session = requests.Session()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_disconnect = self._on_disconnect

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            logger.info("Connected to MQTT broker. Subscribing to %s", MQTT_TOPIC)
            client.subscribe(MQTT_TOPIC, qos=1)
        else:
            logger.error("Failed to connect to MQTT broker: %s", reason_code)

    def _on_message(self, client, userdata, msg):
        try:
            payload = json.loads(msg.payload.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError) as e:
            logger.error("Malformed payload received: %s", e)
            return

        if not validate_payload(payload):
            logger.warning("Schema validation failed for %s", payload.get("instrument_id"))
            return

        p_hash = compute_audit_hash(payload)
        success = post_to_rest(self.session, payload, p_hash)
        
        if success:
            logger.info("Ingested %s (hash: %s)", payload["instrument_id"], p_hash[:12])
        else:
            logger.warning("Primary ingest failed. Buffering %s", p_hash[:12])
            buffer_payload(self.db, p_hash, payload)

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        logger.info("Disconnected from MQTT broker. Reason: %s", reason_code)

    def flush_buffer(self):
        """Reconcile local buffer with REST endpoint."""
        unsent = fetch_unsent(self.db)
        if not unsent:
            return
        logger.info("Flushing %d buffered records...", len(unsent))
        flushed_hashes = []
        for item in unsent:
            if post_to_rest(self.session, item["json"], item["hash"]):
                flushed_hashes.append(item["hash"])
        purge_sent(self.db, flushed_hashes)

    def run(self):
        self.client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
        self.client.loop_start()
        try:
            while True:
                self.flush_buffer()
                time.sleep(10)
        except KeyboardInterrupt:
            logger.info("Shutdown requested. Flushing final buffer...")
            self.flush_buffer()
        finally:
            self.client.loop_stop()
            self.db.close()
            logger.info("Worker terminated gracefully.")

if __name__ == "__main__":
    worker = IngestionWorker()
    worker.run()