# TensorFlow Lite Micro on ESP32 (via C++) — conceptual example
# In practice: train model in Python, convert to .tflite, deploy to device

import tensorflow as tf
import numpy as np

# Step 1: Train anomaly detection model on a PC
def build_and_train_model(X_train):
    """Simple autoencoder — reconstructs normal patterns."""
    input_dim = X_train.shape[1]

    model = tf.keras.Sequential([
        tf.keras.layers.Dense(16, activation='relu', input_shape=(input_dim,)),
        tf.keras.layers.Dense(8,  activation='relu'),
        tf.keras.layers.Dense(4,  activation='relu'),  # Bottleneck
        tf.keras.layers.Dense(8,  activation='relu'),
        tf.keras.layers.Dense(16, activation='relu'),
        tf.keras.layers.Dense(input_dim, activation='sigmoid'),  # Reconstruct
    ])
    model.compile(optimizer='adam', loss='mse')
    model.fit(X_train, X_train, epochs=50, batch_size=32, verbose=0)
    return model

# Simulate 1000 normal sensor readings (temp, humidity, pressure normalized)
X_normal = np.random.normal(0, 1, (1000, 3)).astype(np.float32)
model = build_and_train_model(X_normal)

# Step 2: Convert to TensorFlow Lite
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]  # Quantize weights

# Further quantize to INT8 for ESP32 (eliminates float entirely)
converter.representative_dataset = lambda: [
    [X_normal[i:i+1]] for i in range(min(100, len(X_normal)))
]
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type  = tf.int8
converter.inference_output_type = tf.int8

tflite_model = converter.convert()

with open('anomaly_detector.tflite', 'wb') as f:
    f.write(tflite_model)

print(f"TFLite model: {len(tflite_model)} bytes "
      f"({'OK' if len(tflite_model) < 100000 else 'may not fit on MCU'})")
# A typical TinyML model fits in 10–100 KB

# Step 3: Load and run on device (Python simulation of what the C++ code does)
interpreter = tf.lite.Interpreter(model_path='anomaly_detector.tflite')
interpreter.allocate_tensors()
input_index  = interpreter.get_input_details()[0]['index']
output_index = interpreter.get_output_details()[0]['index']

def detect_anomaly(reading):
    """Run inference — would execute in < 5ms on an ESP32."""
    input_data = np.array([reading], dtype=np.float32).reshape(1, -1)
    interpreter.set_tensor(input_index, input_data)
    interpreter.invoke()
    reconstructed = interpreter.get_tensor(output_index)
    reconstruction_error = float(np.mean((input_data - reconstructed) ** 2))
    return reconstruction_error > 0.1, reconstruction_error

# Test: normal vs anomalous
normal_reading   = [0.1, 0.2, -0.1]  # Normal range
abnormal_reading = [3.5, -2.0, 4.1]  # Way out of normal range

is_anomaly, score = detect_anomaly(normal_reading)
print(f"Normal:   anomaly={is_anomaly}  score={score:.4f}")

is_anomaly, score = detect_anomaly(abnormal_reading)
print(f"Abnormal: anomaly={is_anomaly}  score={score:.4f}")

5G IoT capabilities:
  Connection density: 1,000,000 devices per km²
  Latency: < 1 ms (Ultra-Reliable Low Latency)
  Speed: Up to 10 Gbps
  Power: 10-year battery life in sleep mode (NB-IoT with 5G)

What this enables:
  Smart city sensors on every lamppost
  Industrial robots with sub-ms control loops
  Autonomous vehicles with real-time V2X communication
  Remote surgery with haptic feedback
  Dense indoor IoT without WiFi infrastructure

Python for 5G IoT:

# 5G network slicing concepts — network-aware IoT
class IoTConnection:
    SLICE_EMBB    = "enhanced_mobile_broadband"   # Video cameras
    SLICE_URLLC   = "ultra_reliable_low_latency"  # Industrial control
    SLICE_MMTC    = "massive_machine_type"         # Sensor nodes

    def __init__(self, device_type):
        self.slice = self._select_slice(device_type)

    def _select_slice(self, dtype):
        mapping = {
            "camera":    self.SLICE_EMBB,
            "robot":     self.SLICE_URLLC,
            "sensor":    self.SLICE_MMTC,
            "actuator":  self.SLICE_URLLC,
        }
        return mapping.get(dtype, self.SLICE_MMTC)

Before Matter:
  Philips Hue → Zigbee (works with some hubs)
  IKEA TRADFRI → Zigbee (works with some hubs)
  Amazon devices → Alexa API only
  Google Nest → Google Home only
  Apple HomeKit → Apple only

  Result: Multiple hubs, incompatible ecosystems

After Matter:
  Any Matter device works with ANY Matter controller
  (Amazon Alexa, Google Home, Apple HomeKit, Samsung SmartThings)

  All use IP (WiFi, Thread, Ethernet) as transport
  All use the same security model (X.509 + operational credentials)
  All discoverable via mDNS

# python-matter-server (chip-repl) interaction example
import asyncio
from matter_server.client import MatterClient

async def control_matter_device():
    client = MatterClient("ws://localhost:5580/ws")
    await client.connect()

    # Commission a new device
    await client.commission_with_code("MT:ABC123DEF456")

    # Control a light (endpoint 1, OnOff cluster)
    await client.send_device_command(
        node_id=1,
        endpoint_id=1,
        cluster_id=6,    # OnOff cluster
        command_id=1,    # On command
        payload={}
    )
    print("Light turned on!")

asyncio.run(control_matter_device())

# digital_twin.py — simple digital twin for a CNC machine
from datetime import datetime
import json

class DigitalTwin:
    """
    Virtual model of a physical device.
    Synchronised with real-world sensor data via MQTT.
    Can simulate future states and run what-if scenarios.
    """
    def __init__(self, asset_id, asset_type):
        self.asset_id   = asset_id
        self.asset_type = asset_type
        self.state      = {}
        self.history    = []
        self.alerts     = []

    def update(self, sensor_data):
        """Synchronise twin with real-world reading."""
        self.state.update(sensor_data)
        self.state["last_sync"] = datetime.utcnow().isoformat()
        self.history.append({**sensor_data, "timestamp": self.state["last_sync"]})
        if len(self.history) > 1000:
            self.history.pop(0)
        self._evaluate_health()

    def _evaluate_health(self):
        """Compute health score from current state."""
        health = 100
        issues = []

        if self.state.get("temperature", 0) > 80:
            health -= 30
            issues.append("OVERHEATING")

        if self.state.get("vibration", 0) > 5.0:
            health -= 20
            issues.append("HIGH_VIBRATION")

        if self.state.get("bearing_temp", 0) > 70:
            health -= 25
            issues.append("BEARING_OVERHEAT")

        self.state["health_score"] = health
        self.state["health_status"] = (
            "HEALTHY" if health > 80 else
            "WARNING" if health > 50 else "CRITICAL"
        )
        if issues:
            self.alerts = issues

    def simulate_degradation(self, hours_ahead=168):
        """
        Project future state based on current degradation trend.
        Returns estimated time to failure.
        """
        if len(self.history) < 10:
            return "Insufficient history for prediction"

        # Linear trend in health score
        recent = [h.get("health_score", 100) for h in self.history[-10:]]
        trend  = (recent[-1] - recent[0]) / len(recent)

        if trend >= 0:
            return "Stable or improving — no failure predicted"

        steps_to_failure = -self.state["health_score"] / trend
        estimated_hours  = steps_to_failure  # Assuming 1 reading/hour

        return {
            "current_health":    self.state["health_score"],
            "trend_per_reading": round(trend, 3),
            "estimated_failure": f"~{estimated_hours:.0f} hours",
            "recommendation":    "Schedule maintenance" if estimated_hours < 72 else "Monitor closely"
        }

    def export(self):
        return json.dumps({
            "asset_id":   self.asset_id,
            "asset_type": self.asset_type,
            "state":      self.state,
            "alerts":     self.alerts,
        }, indent=2)

# Usage
twin = DigitalTwin("cnc-machine-07", "CNC Milling Machine")
import random
for i in range(20):
    twin.update({
        "temperature":   60 + i * 0.5 + random.uniform(-1, 1),
        "vibration":     1.5 + i * 0.1,
        "bearing_temp":  50 + i * 0.3,
        "health_score":  100 - i * 2,
        "spindle_speed": 3000
    })

print(twin.export())
print("\nDegradation prediction:", twin.simulate_degradation())

# Carbon footprint calculator for IoT deployment
def iot_carbon_footprint(
    num_devices,
    idle_power_w=0.5,          # Watts when idle
    active_power_w=0.2,         # Additional Watts when transmitting
    transmit_percent=0.01,      # 1% of time actively transmitting
    grid_carbon_gco2_kwh=450,   # gCO2/kWh (global average)
    device_lifetime_years=5
):
    """Estimate total carbon impact of an IoT deployment."""
    avg_power_w = (idle_power_w +
                   active_power_w * transmit_percent) * num_devices

    kwh_per_year    = avg_power_w * 8760 / 1000
    kgco2_per_year  = kwh_per_year * grid_carbon_gco2_kwh / 1000
    kgco2_lifetime  = kgco2_per_year * device_lifetime_years

    print(f"Fleet power:   {avg_power_w:.1f} W")
    print(f"Annual energy: {kwh_per_year:.1f} kWh")
    print(f"Annual CO₂:    {kgco2_per_year:.1f} kg CO₂")
    print(f"Lifetime CO₂:  {kgco2_lifetime:.1f} kg CO₂")

    # Offset recommendation
    trees_needed = kgco2_lifetime / 21  # 1 tree ≈ 21 kg CO₂/year
    print(f"Trees to offset (lifetime): {trees_needed:.0f}")

iot_carbon_footprint(10000, idle_power_w=0.01)  # ESP32 with deep sleep