Da MQTT a InfluxDB: Piattaforma IoT Energetica Real-Time

# Schema topic namespace per impianto energetico
# Struttura: {tipo_impianto}/{plant_id}/{subsystem}/{device_id}/{metric_type}

# === FOTOVOLTAICO ===
# Inverter string-level metrics
plant/PV001/inverter/INV001/metrics
plant/PV001/inverter/INV001/alarms
plant/PV001/inverter/INV001/status

# String-level measurements
plant/PV001/string/STR001/metrics
plant/PV001/string/STR001/metrics/voltage
plant/PV001/string/STR001/metrics/current

# Meteorological sensors
plant/PV001/weather/WS001/metrics

# Plant-level aggregates (calcolati dal gateway)
plant/PV001/totals/power_ac
plant/PV001/totals/energy_today

# === EOLICO ===
wind/WF001/turbine/T001/nacelle/metrics
wind/WF001/turbine/T001/blade/metrics
wind/WF001/turbine/T001/gearbox/temperature
wind/WF001/substation/SS001/metrics

# === STAZIONI DI RICARICA EV ===
ev/SITE001/charger/CHRG001/connector/1/metrics
ev/SITE001/charger/CHRG001/session/current
ev/SITE001/charger/CHRG001/status
ev/SITE001/totals/energy_dispensed

# === SISTEMA ===
# Heartbeat e connectivity
plant/PV001/gateway/GW001/heartbeat    # ogni 60s
plant/PV001/gateway/GW001/status       # retain=true
plant/PV001/gateway/GW001/version      # retain=true

# === WILDCARD SUBSCRIPTIONS ===
# Telegraf sottoscrive tutti i metrics:
plant/+/inverter/+/metrics     # tutti gli inverter di tutti gli impianti
plant/PV001/+/+/metrics        # tutti i metrics dell'impianto PV001
plant/#                        # tutto il plant PV001 (usare con cautela)

# /etc/mosquitto/mosquitto.conf - Produzione con TLS e ACL

# Network listeners
listener 1883 localhost
listener 8883
  protocol mqtt
  cafile /etc/mosquitto/certs/ca.crt
  certfile /etc/mosquitto/certs/server.crt
  keyfile /etc/mosquitto/certs/server.key
  require_certificate true    # mutual TLS
  use_identity_as_username true

# WebSocket listener per dashboard web
listener 9001
  protocol websockets
  cafile /etc/mosquitto/certs/ca.crt
  certfile /etc/mosquitto/certs/server.crt
  keyfile /etc/mosquitto/certs/server.key

# Autenticazione
allow_anonymous false
password_file /etc/mosquitto/passwd
acl_file /etc/mosquitto/acl

# Persistenza
persistence true
persistence_location /var/lib/mosquitto/
autosave_interval 60

# Logging
log_dest file /var/log/mosquitto/mosquitto.log
log_type error
log_type warning
log_type notice
log_type information
log_timestamp true
log_timestamp_format %Y-%m-%dT%H:%M:%S

# Performance tuning
max_queued_messages 10000
max_packet_size 1048576  # 1MB
message_size_limit 262144  # 256KB per messaggio
max_connections 10000
max_keepalive 300

# /etc/mosquitto/acl - Access Control List per ruoli

# Telegraf: legge tutto
user telegraf_reader
topic read plant/#
topic read wind/#
topic read ev/#

# Gateway PV001: scrive solo per il suo impianto
user gateway_PV001
topic write plant/PV001/#
topic read plant/PV001/commands/#

# Gateway EV SITE001: scrive solo per il suo sito
user gateway_SITE001
topic write ev/SITE001/#

# Dashboard/SCADA: solo lettura
user dashboard_user
topic read plant/#
topic read wind/#
topic read ev/#

# Admin: tutto (solo per maintenance)
user mqtt_admin
topic #

#!/usr/bin/env python3
"""
gateway_modbus_mqtt.py
Gateway edge per inverter fotovoltaici Modbus TCP → MQTT
Compatibile con registro SunSpec (standard de facto per inverter PV)
"""

import asyncio
import json
import logging
import time
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional
import aiofiles
from pymodbus.client import AsyncModbusTcpClient
from pymodbus.exceptions import ModbusException
import aiomqtt

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('gateway')


@dataclass
class InverterMetrics:
    """Metriche SunSpec inverter (Model 103 - Three Phase Inverter)"""
    timestamp: str
    plant_id: str
    inverter_id: str
    # AC output
    ac_power_w: float          # Potenza AC totale (W)
    ac_voltage_l1_v: float     # Tensione fase L1 (V)
    ac_voltage_l2_v: float     # Tensione fase L2 (V)
    ac_voltage_l3_v: float     # Tensione fase L3 (V)
    ac_current_a: float        # Corrente totale AC (A)
    ac_frequency_hz: float     # Frequenza rete (Hz)
    # DC input
    dc_voltage_v: float        # Tensione DC ingresso (V)
    dc_current_a: float        # Corrente DC ingresso (A)
    dc_power_w: float          # Potenza DC ingresso (W)
    # Energy
    energy_wh_total: float     # Energia prodotta totale (Wh)
    energy_wh_today: float     # Energia prodotta oggi (Wh)
    # Thermal
    temp_cabinet_c: float      # Temperatura cabinet inverter (C)
    temp_heatsink_c: float     # Temperatura heatsink (C)
    # Status
    operating_state: int       # 1=Off, 2=Sleeping, 3=Starting, 4=MPPT, 5=Throttled
    error_code: int            # Codice errore (0=no error)
    # Efficiency
    efficiency_pct: float      # Efficienza conversione (%)


class ModbusMqttGateway:
    """
    Gateway che legge metriche Modbus da inverter SunSpec
    e le pubblica su MQTT con buffering locale per disconnessioni WAN
    """

    def __init__(self, config: dict):
        self.plant_id = config['plant_id']
        self.inverters = config['inverters']  # lista di {id, host, port, unit_id}
        self.mqtt_host = config['mqtt']['host']
        self.mqtt_port = config['mqtt']['port']
        self.mqtt_user = config['mqtt']['username']
        self.mqtt_password = config['mqtt']['password']
        self.poll_interval = config.get('poll_interval_sec', 5)
        self.buffer_file = config.get('buffer_file', '/var/lib/gateway/buffer.jsonl')
        self.clients: dict[str, AsyncModbusTcpClient] = {}
        self._buffer: list[dict] = []
        self._mqtt_connected = False

    async def connect_inverter(self, inverter_cfg: dict) -> AsyncModbusTcpClient:
        """Crea e connette client Modbus TCP"""
        client = AsyncModbusTcpClient(
            host=inverter_cfg['host'],
            port=inverter_cfg.get('port', 502),
            timeout=5,
            retries=3,
            retry_on_empty=True,
        )
        await client.connect()
        if not client.connected:
            raise ConnectionError(
                f"Cannot connect to inverter {inverter_cfg['id']} "
                f"at {inverter_cfg['host']}:{inverter_cfg.get('port', 502)}"
            )
        logger.info(f"Connected to inverter {inverter_cfg['id']}")
        return client

    async def read_sunspec_model103(
        self,
        client: AsyncModbusTcpClient,
        unit_id: int,
        plant_id: str,
        inverter_id: str
    ) -> Optional[InverterMetrics]:
        """
        Legge registri SunSpec Model 103 (Three Phase Inverter)
        SunSpec base address: 40001 (Modbus address 40000)
        Model 103 start: 40070 (tipico, dipende da inverter)
        """
        try:
            # Leggi blocco principale Model 103: 40069 - 40094 (26 registri)
            result = await client.read_holding_registers(
                address=40069,
                count=26,
                slave=unit_id
            )
            if result.isError():
                logger.error(f"Modbus error reading INV {inverter_id}: {result}")
                return None

            regs = result.registers
            # Scala con SF (scale factors) SunSpec
            ac_current_sf = self._sf(regs[10])
            ac_voltage_sf = self._sf(regs[14])
            ac_power_sf = self._sf(regs[17])
            ac_freq_sf = self._sf(regs[19])
            dc_current_sf = self._sf(regs[21])
            dc_voltage_sf = self._sf(regs[23])
            dc_power_sf = self._sf(regs[25])

            # Leggi energia prodotta (registri 40093-40096)
            energy_result = await client.read_holding_registers(
                address=40093, count=4, slave=unit_id
            )
            energy_regs = energy_result.registers
            energy_total_wh = ((energy_regs[0] << 16) | energy_regs[1]) * 1.0

            # Leggi temperatura (registri 40103-40108)
            temp_result = await client.read_holding_registers(
                address=40103, count=3, slave=unit_id
            )
            temp_regs = temp_result.registers

            ac_power_raw = self._signed(regs[16])
            dc_power_w = self._signed(regs[24]) * (10 ** dc_power_sf)
            efficiency = (ac_power_raw / dc_power_w * 100) if dc_power_w > 0 else 0.0

            return InverterMetrics(
                timestamp=datetime.now(timezone.utc).isoformat(),
                plant_id=plant_id,
                inverter_id=inverter_id,
                ac_power_w=ac_power_raw * (10 ** ac_power_sf),
                ac_voltage_l1_v=regs[11] * (10 ** ac_voltage_sf),
                ac_voltage_l2_v=regs[12] * (10 ** ac_voltage_sf),
                ac_voltage_l3_v=regs[13] * (10 ** ac_voltage_sf),
                ac_current_a=(regs[7] + regs[8] + regs[9]) * (10 ** ac_current_sf),
                ac_frequency_hz=regs[18] * (10 ** ac_freq_sf),
                dc_voltage_v=self._signed(regs[22]) * (10 ** dc_voltage_sf),
                dc_current_a=self._signed(regs[20]) * (10 ** dc_current_sf),
                dc_power_w=dc_power_w,
                energy_wh_total=energy_total_wh,
                energy_wh_today=0.0,  # da registro specifico vendor
                temp_cabinet_c=self._signed(temp_regs[0]) * 0.01,
                temp_heatsink_c=self._signed(temp_regs[1]) * 0.01,
                operating_state=regs[0],
                error_code=regs[1],
                efficiency_pct=round(efficiency, 2),
            )
        except ModbusException as e:
            logger.error(f"Modbus exception for {inverter_id}: {e}")
            return None

    def _signed(self, val: int) -> int:
        """Converte uint16 in int16 (signed)"""
        return val if val < 32768 else val - 65536

    def _sf(self, val: int) -> int:
        """Scale factor SunSpec: int16 nel range -10..10"""
        s = self._signed(val)
        return s if -10 <= s <= 10 else 0

    async def publish_metrics(
        self,
        mqtt_client: aiomqtt.Client,
        metrics: InverterMetrics
    ) -> None:
        """Pubblica metriche su MQTT con QoS 1"""
        topic = f"plant/{metrics.plant_id}/inverter/{metrics.inverter_id}/metrics"
        payload = json.dumps(asdict(metrics), default=str)
        await mqtt_client.publish(
            topic=topic,
            payload=payload.encode(),
            qos=1,
            retain=False,
        )
        # Pubblica anche stato con retain
        status = {
            "timestamp": metrics.timestamp,
            "operating_state": metrics.operating_state,
            "error_code": metrics.error_code,
            "online": True,
        }
        await mqtt_client.publish(
            topic=f"plant/{metrics.plant_id}/inverter/{metrics.inverter_id}/status",
            payload=json.dumps(status).encode(),
            qos=1,
            retain=True,  # retain per status
        )

    async def run(self):
        """Loop principale del gateway"""
        # Carica buffer da disco se esiste
        await self._load_buffer()

        # Will message: pubblica offline status se il gateway crolla
        will = aiomqtt.Will(
            topic=f"plant/{self.plant_id}/gateway/GW001/status",
            payload=json.dumps({"online": False, "reason": "unexpected_disconnect"}).encode(),
            qos=1,
            retain=True,
        )

        async with aiomqtt.Client(
            hostname=self.mqtt_host,
            port=self.mqtt_port,
            username=self.mqtt_user,
            password=self.mqtt_password,
            will=will,
            keepalive=60,
            tls_params=aiomqtt.TLSParameters(
                ca_certs='/etc/gateway/certs/ca.crt',
                certfile='/etc/gateway/certs/gateway.crt',
                keyfile='/etc/gateway/certs/gateway.key',
            ),
        ) as mqtt:
            self._mqtt_connected = True
            logger.info("MQTT connected")

            # Pubblica online status con retain
            await mqtt.publish(
                topic=f"plant/{self.plant_id}/gateway/GW001/status",
                payload=json.dumps({"online": True}).encode(),
                qos=1, retain=True,
            )

            # Svuota buffer offline
            await self._flush_buffer(mqtt)

            # Connetti a tutti gli inverter
            for inv_cfg in self.inverters:
                try:
                    self.clients[inv_cfg['id']] = await self.connect_inverter(inv_cfg)
                except ConnectionError as e:
                    logger.error(f"Failed to connect: {e}")

            # Loop principale di polling
            while True:
                poll_start = time.monotonic()
                tasks = [
                    self._poll_and_publish(inv_cfg, mqtt)
                    for inv_cfg in self.inverters
                ]
                await asyncio.gather(*tasks, return_exceptions=True)

                elapsed = time.monotonic() - poll_start
                sleep_time = max(0, self.poll_interval - elapsed)
                await asyncio.sleep(sleep_time)

    async def _poll_and_publish(self, inv_cfg: dict, mqtt: aiomqtt.Client):
        """Polling singolo inverter e pubblicazione"""
        inv_id = inv_cfg['id']
        client = self.clients.get(inv_id)
        if client is None or not client.connected:
            try:
                client = await self.connect_inverter(inv_cfg)
                self.clients[inv_id] = client
            except ConnectionError:
                return

        metrics = await self.read_sunspec_model103(
            client, inv_cfg.get('unit_id', 1),
            self.plant_id, inv_id
        )
        if metrics:
            try:
                await self.publish_metrics(mqtt, metrics)
            except Exception as e:
                logger.warning(f"MQTT publish failed, buffering: {e}")
                self._buffer.append(asdict(metrics))
                await self._save_buffer()

    async def _load_buffer(self):
        """Carica buffer dal disco"""
        try:
            async with aiofiles.open(self.buffer_file) as f:
                async for line in f:
                    self._buffer.append(json.loads(line))
            logger.info(f"Loaded {len(self._buffer)} buffered messages")
        except FileNotFoundError:
            pass

    async def _save_buffer(self):
        """Salva buffer su disco"""
        async with aiofiles.open(self.buffer_file, 'w') as f:
            for item in self._buffer:
                await f.write(json.dumps(item) + '\n')

    async def _flush_buffer(self, mqtt: aiomqtt.Client):
        """Pubblica messaggi bufferizzati"""
        if not self._buffer:
            return
        logger.info(f"Flushing {len(self._buffer)} buffered messages")
        published = []
        for item in self._buffer:
            try:
                metrics = InverterMetrics(**item)
                await self.publish_metrics(mqtt, metrics)
                published.append(item)
            except Exception as e:
                logger.error(f"Failed to flush buffered message: {e}")
                break
        self._buffer = [i for i in self._buffer if i not in published]
        await self._save_buffer()


if __name__ == '__main__':
    config = {
        'plant_id': 'PV001',
        'poll_interval_sec': 5,
        'mqtt': {
            'host': '10.0.1.10',
            'port': 8883,
            'username': 'gateway_PV001',
            'password': 'secret',
        },
        'inverters': [
            {'id': 'INV001', 'host': '192.168.1.101', 'port': 502, 'unit_id': 1},
            {'id': 'INV002', 'host': '192.168.1.102', 'port': 502, 'unit_id': 1},
            # ... fino a INV200 per impianto da 10 MW
        ],
        'buffer_file': '/var/lib/gateway/buffer.jsonl',
    }
    asyncio.run(ModbusMqttGateway(config).run())

# telegraf.conf - Configurazione completa per piattaforma IoT energetica

# ============================================================
# AGENT SETTINGS
# ============================================================
[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 10000
  metric_buffer_limit = 100000
  collection_jitter = "1s"
  flush_interval = "10s"
  flush_jitter = "1s"
  precision = "1ns"
  debug = false
  quiet = false
  logtarget = "file"
  logfile = "/var/log/telegraf/telegraf.log"
  logfile_rotation_interval = "24h"
  logfile_rotation_max_size = "50MB"
  logfile_rotation_max_archives = 5

# ============================================================
# OUTPUT: InfluxDB 2.x / 3.x
# ============================================================
[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "$INFLUXDB_TOKEN"
  organization = "energy-corp"
  bucket = "plant_metrics_raw"
  timeout = "5s"

  # Tag per routing - usato per bucket separati per plant
  [outputs.influxdb_v2.tagpass]
    data_type = ["inverter_metrics", "weather_metrics", "string_metrics"]

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "$INFLUXDB_TOKEN"
  organization = "energy-corp"
  bucket = "ev_metrics_raw"
  timeout = "5s"

  [outputs.influxdb_v2.tagpass]
    data_type = ["ev_metrics"]

# Output di backup su file locale (per audit e recovery)
[[outputs.file]]
  files = ["/var/log/telegraf/metrics.jsonl"]
  rotation_interval = "1h"
  rotation_max_size = "500MB"
  data_format = "json"
  [outputs.file.tagpass]
    error_code = ["*"]  # solo metriche con errori

# ============================================================
# INPUT: MQTT Consumer - Inverter Fotovoltaici
# ============================================================
[[inputs.mqtt_consumer]]
  name_override = "pv_inverter"
  servers = ["ssl://mosquitto:8883"]
  topics = ["plant/+/inverter/+/metrics"]
  qos = 1
  connection_timeout = "30s"
  max_undelivered_messages = 5000

  # Autenticazione mTLS
  tls_ca = "/etc/telegraf/certs/ca.crt"
  tls_cert = "/etc/telegraf/certs/telegraf.crt"
  tls_key = "/etc/telegraf/certs/telegraf.key"

  username = "telegraf_reader"
  password = "$MQTT_PASSWORD"

  client_id = "telegraf-energy-01"
  persistent_session = true  # QoS 1+ richiede sessione persistente

  # Parsing topic come tag
  # topic: plant/PV001/inverter/INV001/metrics
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "plant/+/inverter/+/metrics"
    measurement = "_/plant_id/_/inverter_id/_"
    tags = "_/plant_id/_/inverter_id/_"
    # Estrae plant_id=PV001, inverter_id=INV001 come tag

  # Formato payload
  data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_power_w"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_voltage_l1_v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_voltage_l2_v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_voltage_l3_v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_current_a"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "ac_frequency_hz"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "dc_voltage_v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "dc_current_a"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "dc_power_w"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "energy_wh_total"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temp_cabinet_c"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temp_heatsink_c"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "operating_state"
      type = "int"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "error_code"
      type = "int"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "efficiency_pct"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "plant_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "inverter_id"
    [[inputs.mqtt_consumer.json_v2.timestamp]]
      path = "timestamp"
      format = "2006-01-02T15:04:05.999999999Z07:00"  # RFC3339Nano

# INPUT: MQTT Consumer - Stazioni Meteo
[[inputs.mqtt_consumer]]
  name_override = "pv_weather"
  servers = ["ssl://mosquitto:8883"]
  topics = ["plant/+/weather/+/metrics"]
  qos = 0  # QoS 0 per dati meteo (perdita occasionale accettabile)

  tls_ca = "/etc/telegraf/certs/ca.crt"
  tls_cert = "/etc/telegraf/certs/telegraf.crt"
  tls_key = "/etc/telegraf/certs/telegraf.key"
  username = "telegraf_reader"
  password = "$MQTT_PASSWORD"
  client_id = "telegraf-weather-01"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "plant/+/weather/+/metrics"
    tags = "_/plant_id/_/station_id/_"

  data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "irradiance_wm2"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temperature_c"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "wind_speed_ms"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity_pct"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.timestamp]]
      path = "timestamp"
      format = "2006-01-02T15:04:05Z07:00"

# INPUT: MQTT Consumer - Stazioni EV
[[inputs.mqtt_consumer]]
  name_override = "ev_charger"
  servers = ["ssl://mosquitto:8883"]
  topics = ["ev/+/charger/+/connector/+/metrics"]
  qos = 1
  tls_ca = "/etc/telegraf/certs/ca.crt"
  tls_cert = "/etc/telegraf/certs/telegraf.crt"
  tls_key = "/etc/telegraf/certs/telegraf.key"
  username = "telegraf_reader"
  password = "$MQTT_PASSWORD"
  client_id = "telegraf-ev-01"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "ev/+/charger/+/connector/+/metrics"
    tags = "_/site_id/_/charger_id/_/connector_id/_"

  data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "power_kw"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "energy_kwh"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "current_a"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "voltage_v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "site_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "charger_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "connector_id"

# ============================================================
# PROCESSORS: Trasformazione dati
# ============================================================

# Aggiungi tag data_type per routing output
[[processors.converter]]
  namepass = ["pv_inverter"]
  [processors.converter.tags]
    string = ["data_type"]

[[processors.override]]
  namepass = ["pv_inverter"]
  [processors.override.tags]
    data_type = "inverter_metrics"

[[processors.override]]
  namepass = ["ev_charger"]
  [processors.override.tags]
    data_type = "ev_metrics"

# Filtra metriche con valori anomali (sanity check)
[[processors.dedup]]
  dedup_interval = "1s"  # evita duplicati QoS 1

[[processors.enum]]
  namepass = ["pv_inverter"]
  [[processors.enum.mapping]]
    field = "operating_state"
    dest = "operating_state_str"
    [processors.enum.mapping.value_mappings]
      1 = "off"
      2 = "sleeping"
      3 = "starting"
      4 = "mppt"
      5 = "throttled"
      6 = "shutting_down"
      7 = "fault"
      8 = "standby"

# ============================================================
# MONITORING: Telegraf self-monitoring
# ============================================================
[[inputs.internal]]
  collect_memstats = true

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "$INFLUXDB_TOKEN"
  organization = "energy-corp"
  bucket = "telegraf_monitoring"
  [outputs.influxdb_v2.namepass]
    internal = ["*"]

#!/bin/bash
# setup_influxdb.sh - Configurazione iniziale InfluxDB per piattaforma energetica

INFLUX_HOST="http://localhost:8086"
INFLUX_TOKEN="my-super-secret-admin-token"
INFLUX_ORG="energy-corp"

# Setup iniziale (solo prima volta)
influx setup \
  --host "$INFLUX_HOST" \
  --username admin \
  --password "SecurePassword123!" \
  --org "$INFLUX_ORG" \
  --bucket plant_metrics_raw \
  --retention 720h \  # 30 giorni dati grezzi
  --token "$INFLUX_TOKEN" \
  --force

# Bucket dati grezzi PV (alta risoluzione: 30 giorni)
influx bucket create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --name "plant_metrics_raw" \
  --retention "720h" \
  --shard-group-duration "24h"  # shard giornalieri per query efficienti

# Bucket downsampled 1 ora (retention 1 anno)
influx bucket create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --name "plant_metrics_1h" \
  --retention "8760h" \
  --shard-group-duration "168h"  # shard settimanali

# Bucket downsampled 1 giorno (retention 5 anni)
influx bucket create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --name "plant_metrics_1d" \
  --retention "43800h" \  # ~5 anni
  --shard-group-duration "720h"  # shard mensili

# Bucket per alerting e eventi
influx bucket create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --name "energy_alerts" \
  --retention "8760h"  # 1 anno per audit

# Bucket per stazioni EV
influx bucket create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --name "ev_metrics_raw" \
  --retention "720h"

# Token con permessi limitati per Telegraf (write only)
influx auth create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --description "Telegraf write-only token" \
  --write-bucket plant_metrics_raw \
  --write-bucket ev_metrics_raw \
  --write-bucket telegraf_monitoring

# Token per Grafana (read only)
influx auth create \
  --host "$INFLUX_HOST" \
  --token "$INFLUX_TOKEN" \
  --org "$INFLUX_ORG" \
  --description "Grafana read-only token" \
  --read-bucket plant_metrics_raw \
  --read-bucket plant_metrics_1h \
  --read-bucket plant_metrics_1d \
  --read-bucket ev_metrics_raw \
  --read-bucket energy_alerts

echo "InfluxDB setup completato"

// task_downsample_1h.flux
// Task: Downsampling dati inverter da 5s a 1h
// Schedulato ogni ora, processa ultima ora

option task = {
  name: "downsample_pv_inverter_1h",
  every: 1h,
  offset: 5m,  // aspetta 5 min per assicurarsi che tutti i dati siano arrivati
}

from(bucket: "plant_metrics_raw")
  |> range(start: -task.every)
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) =>
      r._field == "ac_power_w" or
      r._field == "dc_power_w" or
      r._field == "energy_wh_total" or
      r._field == "ac_voltage_l1_v" or
      r._field == "temp_cabinet_c" or
      r._field == "efficiency_pct"
  )
  |> aggregateWindow(
      every: 1h,
      fn: (tables=<-, column) => tables
        |> mean(column: column),  // media oraria
      createEmpty: false
  )
  // Aggiungi statistiche aggiuntive per power
  |> map(fn: (r) => ({r with _measurement: "pv_inverter_1h"}))
  |> to(bucket: "plant_metrics_1h", org: "energy-corp")

// task_plant_kpi.flux
// Task: Calcolo KPI giornalieri per impianto
// Produccion, efficienza, performance ratio

option task = {
  name: "calculate_plant_daily_kpi",
  every: 15m,
}

// Potenza attiva totale per impianto (somma tutti inverter)
totalPower = from(bucket: "plant_metrics_raw")
  |> range(start: -15m)
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) => r._field == "ac_power_w")
  |> aggregateWindow(every: 15m, fn: sum, createEmpty: false)
  |> group(columns: ["plant_id", "_time"])
  |> sum()
  |> map(fn: (r) => ({r with
      _measurement: "plant_totals",
      _field: "total_ac_power_w",
  }))

// Energia prodotta totale (max energy_wh_total - min energy_wh_total per periodo)
energyDelta = from(bucket: "plant_metrics_raw")
  |> range(start: -15m)
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) => r._field == "energy_wh_total")
  |> difference()
  |> group(columns: ["plant_id", "_time"])
  |> sum()
  |> map(fn: (r) => ({r with
      _measurement: "plant_totals",
      _field: "energy_delta_wh",
  }))

// Temperatura media inverter
avgTemp = from(bucket: "plant_metrics_raw")
  |> range(start: -15m)
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) => r._field == "temp_cabinet_c")
  |> aggregateWindow(every: 15m, fn: mean, createEmpty: false)
  |> group(columns: ["plant_id", "_time"])
  |> mean()
  |> map(fn: (r) => ({r with
      _measurement: "plant_totals",
      _field: "avg_inverter_temp_c",
  }))

union(tables: [totalPower, energyDelta, avgTemp])
  |> to(bucket: "plant_metrics_1h", org: "energy-corp")

// task_thermal_alert.flux
// Task: Rilevamento surriscaldamento inverter
// Alert se temperatura > 75°C per più di 5 minuti

option task = {
  name: "thermal_alert_inverter",
  every: 1m,
}

threshold = 75.0

from(bucket: "plant_metrics_raw")
  |> range(start: -6m)
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) => r._field == "temp_cabinet_c")
  |> filter(fn: (r) => r._value > threshold)
  // Conta quanti valori sopra soglia negli ultimi 6 min (= almeno 5 min)
  |> aggregateWindow(every: 6m, fn: count, createEmpty: false)
  |> filter(fn: (r) => r._value >= 60)  // 60 campioni * 5sec = 5 min
  |> map(fn: (r) => ({r with
      _measurement: "energy_alert",
      _field: "alert_type",
      _value: "thermal_overtemperature",
      severity: "high",
      description: "Inverter temperature exceeds " + string(v: threshold) + "°C for > 5 minutes",
  }))
  |> to(bucket: "energy_alerts", org: "energy-corp")

// query_production_analysis.flux
// Analisi produzione giornaliera per impianto PV001
// Confronto con irraggiamento (Performance Ratio)

import "math"

// Produzione AC ogni 5 minuti
production = from(bucket: "plant_metrics_raw")
  |> range(start: today(), stop: now())
  |> filter(fn: (r) => r._measurement == "pv_inverter")
  |> filter(fn: (r) => r.plant_id == "PV001")
  |> filter(fn: (r) => r._field == "ac_power_w")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: true)
  |> group(columns: ["plant_id"])
  |> sum()  // somma tutti gli inverter per potenza totale impianto

// Irraggiamento dalla stazione meteo
irradiance = from(bucket: "plant_metrics_raw")
  |> range(start: today(), stop: now())
  |> filter(fn: (r) => r._measurement == "pv_weather")
  |> filter(fn: (r) => r.plant_id == "PV001")
  |> filter(fn: (r) => r._field == "irradiance_wm2")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: true)

// Join per calcolo Performance Ratio
// PR = (energia prodotta / (irraggiamento * superficie)) * 100
join(
  tables: {production: production, irradiance: irradiance},
  on: ["_time", "plant_id"]
)
// PV001: 10 MW peak, superficie pannelli ~50.000 m2
|> map(fn: (r) => ({r with
    performance_ratio: if r.irradiance_irradiance_wm2 > 50.0 then
        (r.production__value / (r.irradiance_irradiance_wm2 * 50000.0)) * 100.0
    else
        0.0,
}))
|> keep(columns: ["_time", "production__value", "irradiance_irradiance_wm2", "performance_ratio"])

# docker-compose.yml
# Stack IoT Energetico: Mosquitto + Telegraf + InfluxDB + Grafana
# Produzione-ready con TLS, health checks e persistent volumes

version: '3.9'

networks:
  iot_network:
    driver: bridge
    ipam:
      config:
        - subnet: 172.20.0.0/24
  # Network separata per accesso esterno (solo Grafana e Mosquitto esposti)
  external_network:
    driver: bridge

volumes:
  mosquitto_data:
  mosquitto_logs:
  influxdb_data:
  influxdb_config:
  grafana_data:
  grafana_provisioning:
  telegraf_buffer:

services:
  # ============================================================
  # MOSQUITTO MQTT BROKER
  # ============================================================
  mosquitto:
    image: eclipse-mosquitto:2.0.20
    container_name: mosquitto
    restart: unless-stopped
    networks:
      - iot_network
      - external_network
    ports:
      - "1883:1883"   # MQTT plain (solo per sviluppo/testing locale)
      - "8883:8883"   # MQTT over TLS
      - "9001:9001"   # WebSocket over TLS
    volumes:
      - ./config/mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
      - ./config/mosquitto/passwd:/mosquitto/config/passwd:ro
      - ./config/mosquitto/acl:/mosquitto/config/acl:ro
      - ./certs:/etc/mosquitto/certs:ro
      - mosquitto_data:/mosquitto/data
      - mosquitto_logs:/mosquitto/log
    healthcheck:
      test: ["CMD", "mosquitto_pub", "-h", "localhost", "-p", "1883",
             "-t", "health", "-m", "ping", "--quiet"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s
    deploy:
      resources:
        limits:
          memory: 512m
          cpus: '1.0'

  # ============================================================
  # INFLUXDB 2.7 (stable production)
  # ============================================================
  influxdb:
    image: influxdb:2.7-alpine
    container_name: influxdb
    restart: unless-stopped
    networks:
      - iot_network
    ports:
      - "8086:8086"   # solo su rete interna, proxy via Nginx se necessario
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: {{ "$INFLUXDB_ADMIN_PASSWORD" }}
      DOCKER_INFLUXDB_INIT_ORG: energy-corp
      DOCKER_INFLUXDB_INIT_BUCKET: plant_metrics_raw
      DOCKER_INFLUXDB_INIT_RETENTION: 720h
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: {{ "$INFLUXDB_TOKEN" }}
      # Performance tuning
      INFLUXD_STORAGE_CACHE_MAX_MEMORY_SIZE: 1073741824  # 1GB cache
      INFLUXD_STORAGE_COMPACT_THROUGHPUT_BURST: 50331648  # 48MB/s compaction
      INFLUXD_HTTP_MAX_BODY_SIZE: 104857600  # 100MB max body
      INFLUXD_QUERY_MEMORY_BYTES: 2147483648  # 2GB query memory limit
    volumes:
      - influxdb_data:/var/lib/influxdb2
      - influxdb_config:/etc/influxdb2
    healthcheck:
      test: ["CMD", "influx", "ping"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 60s
    deploy:
      resources:
        limits:
          memory: 4g
          cpus: '2.0'

  # ============================================================
  # TELEGRAF
  # ============================================================
  telegraf:
    image: telegraf:1.34-alpine
    container_name: telegraf
    restart: unless-stopped
    networks:
      - iot_network
      - external_network  # per raggiungere MQTT su rete esterna
    depends_on:
      influxdb:
        condition: service_healthy
      mosquitto:
        condition: service_healthy
    environment:
      INFLUXDB_TOKEN: {{ "$INFLUXDB_TOKEN" }}
      MQTT_PASSWORD: {{ "$MQTT_PASSWORD" }}
      HOSTNAME: telegraf-energy-01
    volumes:
      - ./config/telegraf/telegraf.conf:/etc/telegraf/telegraf.conf:ro
      - ./certs:/etc/telegraf/certs:ro
      - telegraf_buffer:/var/lib/telegraf
      - /var/log/telegraf:/var/log/telegraf
    healthcheck:
      test: ["CMD", "telegraf", "--test", "--config", "/etc/telegraf/telegraf.conf"]
      interval: 60s
      timeout: 30s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 1g
          cpus: '1.0'

  # ============================================================
  # GRAFANA
  # ============================================================
  grafana:
    image: grafana/grafana-oss:11.4.0
    container_name: grafana
    restart: unless-stopped
    networks:
      - iot_network
      - external_network
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_USER: admin
      GF_SECURITY_ADMIN_PASSWORD: {{ "$GRAFANA_ADMIN_PASSWORD" }}
      GF_SECURITY_SECRET_KEY: {{ "$GRAFANA_SECRET_KEY" }}
      GF_SERVER_ROOT_URL: "https://grafana.energy-corp.it"
      GF_SERVER_DOMAIN: "grafana.energy-corp.it"
      # Analytics disabilitate per privacy
      GF_ANALYTICS_REPORTING_ENABLED: "false"
      GF_ANALYTICS_CHECK_FOR_UPDATES: "false"
      # SMTP per alerting
      GF_SMTP_ENABLED: "true"
      GF_SMTP_HOST: {{ "$SMTP_HOST" }}
      GF_SMTP_USER: {{ "$SMTP_USER" }}
      GF_SMTP_PASSWORD: {{ "$SMTP_PASSWORD" }}
      GF_SMTP_FROM_ADDRESS: "alerts@energy-corp.it"
      # InfluxDB datasource
      GF_DATASOURCES_DEFAULT: InfluxDB
      INFLUXDB_TOKEN_GRAFANA: {{ "$INFLUXDB_TOKEN_GRAFANA" }}
    volumes:
      - grafana_data:/var/lib/grafana
      - ./config/grafana/provisioning:/etc/grafana/provisioning:ro
      - ./config/grafana/dashboards:/var/lib/grafana/dashboards:ro
    depends_on:
      influxdb:
        condition: service_healthy
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:3000/api/health || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 512m
          cpus: '0.5'

  # ============================================================
  # NGINX REVERSE PROXY (opzionale ma raccomandato)
  # ============================================================
  nginx:
    image: nginx:1.27-alpine
    container_name: nginx
    restart: unless-stopped
    networks:
      - external_network
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./config/nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certs:/etc/nginx/certs:ro
      - /var/log/nginx:/var/log/nginx
    depends_on:
      - grafana
      - mosquitto
    healthcheck:
      test: ["CMD", "nginx", "-t"]
      interval: 30s
      timeout: 10s
      retries: 3

# config/grafana/provisioning/datasources/influxdb.yml
apiVersion: 1
datasources:
  - name: InfluxDB-Raw
    type: influxdb
    access: proxy
    url: http://influxdb:8086
    jsonData:
      version: Flux
      organization: energy-corp
      defaultBucket: plant_metrics_raw
      tlsSkipVerify: false
    secureJsonData:
      token: $INFLUXDB_TOKEN_GRAFANA

  - name: InfluxDB-1h
    type: influxdb
    access: proxy
    url: http://influxdb:8086
    jsonData:
      version: Flux
      organization: energy-corp
      defaultBucket: plant_metrics_1h
    secureJsonData:
      token: $INFLUXDB_TOKEN_GRAFANA

  - name: InfluxDB-Alerts
    type: influxdb
    access: proxy
    url: http://influxdb:8086
    jsonData:
      version: Flux
      organization: energy-corp
      defaultBucket: energy_alerts
    secureJsonData:
      token: $INFLUXDB_TOKEN_GRAFANA

# config/grafana/provisioning/alerting/rules.yml
apiVersion: 1
groups:
  - name: "PV Plant Alerts"
    folder: "Energy Alerts"
    interval: 1m
    rules:
      # Alert: Produzione impianto drasticamente ridotta
      - uid: pv001-production-drop
        title: "PV001 - Produzione ridotta >30%"
        condition: C
        data:
          # Query A: Produzione attuale (15 min rolling mean)
          - refId: A
            datasourceUid: influxdb-raw-uid
            model:
              query: |
                from(bucket: "plant_metrics_raw")
                  |> range(start: -15m)
                  |> filter(fn: (r) => r._measurement == "pv_inverter")
                  |> filter(fn: (r) => r.plant_id == "PV001")
                  |> filter(fn: (r) => r._field == "ac_power_w")
                  |> mean()
                  |> sum()
          # Query B: Irraggiamento attuale (per normalizzare)
          - refId: B
            datasourceUid: influxdb-raw-uid
            model:
              query: |
                from(bucket: "plant_metrics_raw")
                  |> range(start: -15m)
                  |> filter(fn: (r) => r._measurement == "pv_weather")
                  |> filter(fn: (r) => r.plant_id == "PV001")
                  |> filter(fn: (r) => r._field == "irradiance_wm2")
                  |> mean()
          # Condition C: se irraggiamento > 200 W/m2 (giorno chiaro)
          # e produzione normalizzata < 70% del teorico
          - refId: C
            datasourceUid: __expr__
            model:
              type: math
              expression: "$A / ($B * 50000) < 0.70 and $B > 200"
        for: 10m  # alert solo se condizione persiste 10 min
        labels:
          severity: "warning"
          plant_id: "PV001"
          category: "production"
        annotations:
          summary: "PV001: produzione sotto il 70% del teorico"
          description: |
            Produzione attuale: {{ $A }} W
            Irraggiamento: {{ $B }} W/m2
            Performance attesa: > 70%
          runbook_url: "https://wiki.energy-corp.it/runbook/pv-production-drop"

      # Alert: Inverter in fault
      - uid: pv001-inverter-fault
        title: "PV001 - Inverter Fault Rilevato"
        condition: C
        data:
          - refId: A
            datasourceUid: influxdb-raw-uid
            model:
              query: |
                from(bucket: "plant_metrics_raw")
                  |> range(start: -2m)
                  |> filter(fn: (r) => r._measurement == "pv_inverter")
                  |> filter(fn: (r) => r._field == "operating_state")
                  |> filter(fn: (r) => r._value == 7)  // stato fault
                  |> count()
          - refId: C
            datasourceUid: __expr__
            model:
              type: threshold
              expression: "$A"
              conditions:
                - evaluator:
                    params: [0]
                    type: gt
        for: 1m
        labels:
          severity: "critical"
          plant_id: "PV001"
        annotations:
          summary: "Inverter in stato FAULT rilevato"

      # Alert: Temperatura critica
      - uid: pv001-thermal-critical
        title: "PV001 - Temperatura Inverter Critica"
        condition: C
        data:
          - refId: A
            datasourceUid: influxdb-raw-uid
            model:
              query: |
                from(bucket: "plant_metrics_raw")
                  |> range(start: -5m)
                  |> filter(fn: (r) => r._measurement == "pv_inverter")
                  |> filter(fn: (r) => r._field == "temp_cabinet_c")
                  |> max()
          - refId: C
            datasourceUid: __expr__
            model:
              type: threshold
              expression: "$A"
              conditions:
                - evaluator:
                    params: [85]
                    type: gt
        for: 5m
        labels:
          severity: "critical"
          category: "thermal"

# Contact points
contactPoints:
  - name: "PagerDuty Critical"
    receivers:
      - uid: pagerduty-critical
        type: pagerduty
        settings:
          integrationKey: $PAGERDUTY_INTEGRATION_KEY
          severity: critical
          class: "EnergyAlerts"
          component: "PV Plant"

  - name: "Slack Operations"
    receivers:
      - uid: slack-ops
        type: slack
        settings:
          url: $SLACK_WEBHOOK_URL
          channel: "#energy-ops-alerts"
          username: "Grafana Alert Bot"
          iconEmoji: ":zap:"
          title: "{{ template \"slack.title\" . }}"
          text: "{{ template \"slack.message\" . }}"

# Routing
policies:
  - receiver: "Slack Operations"
    group_by: ["plant_id", "severity"]
    group_wait: 30s
    group_interval: 5m
    repeat_interval: 4h
    routes:
      - receiver: "PagerDuty Critical"
        matchers:
          - "severity = critical"
        continue: true  # invia anche a Slack

#!/bin/bash
# generate_certs.sh - Genera CA e certificati per mutual TLS

set -euo pipefail
CERTS_DIR="./certs"
mkdir -p "$CERTS_DIR"

# 1. CA (Certificate Authority) - Root of trust
openssl genrsa -out "$CERTS_DIR/ca.key" 4096
openssl req -x509 -new -nodes \
    -key "$CERTS_DIR/ca.key" \
    -sha256 -days 3650 \
    -out "$CERTS_DIR/ca.crt" \
    -subj "/C=IT/ST=Puglia/L=Bari/O=EnergyCorp/CN=Energy IoT CA"

# 2. Server certificate (Mosquitto)
openssl genrsa -out "$CERTS_DIR/server.key" 2048
openssl req -new \
    -key "$CERTS_DIR/server.key" \
    -out "$CERTS_DIR/server.csr" \
    -subj "/C=IT/O=EnergyCorp/CN=mosquitto"

# SAN per tutti i possibili hostname
cat > "$CERTS_DIR/server-ext.cnf" <<EOF
[req]
req_extensions = v3_req
[v3_req]
subjectAltName = @alt_names
[alt_names]
DNS.1 = mosquitto
DNS.2 = mqtt.energy-corp.it
DNS.3 = localhost
IP.1 = 127.0.0.1
IP.2 = 172.20.0.10
EOF

openssl x509 -req \
    -in "$CERTS_DIR/server.csr" \
    -CA "$CERTS_DIR/ca.crt" \
    -CAkey "$CERTS_DIR/ca.key" \
    -CAcreateserial \
    -out "$CERTS_DIR/server.crt" \
    -days 825 \
    -sha256 \
    -extfile "$CERTS_DIR/server-ext.cnf" \
    -extensions v3_req

# 3. Client certificate per Telegraf
openssl genrsa -out "$CERTS_DIR/telegraf.key" 2048
openssl req -new \
    -key "$CERTS_DIR/telegraf.key" \
    -out "$CERTS_DIR/telegraf.csr" \
    -subj "/C=IT/O=EnergyCorp/CN=telegraf"
openssl x509 -req \
    -in "$CERTS_DIR/telegraf.csr" \
    -CA "$CERTS_DIR/ca.crt" \
    -CAkey "$CERTS_DIR/ca.key" \
    -CAcreateserial \
    -out "$CERTS_DIR/telegraf.crt" \
    -days 825 -sha256

# 4. Client certificate per gateway PV001
openssl genrsa -out "$CERTS_DIR/gateway_PV001.key" 2048
openssl req -new \
    -key "$CERTS_DIR/gateway_PV001.key" \
    -out "$CERTS_DIR/gateway_PV001.csr" \
    -subj "/C=IT/O=EnergyCorp/CN=gateway_PV001"
openssl x509 -req \
    -in "$CERTS_DIR/gateway_PV001.csr" \
    -CA "$CERTS_DIR/ca.crt" \
    -CAkey "$CERTS_DIR/ca.key" \
    -CAcreateserial \
    -out "$CERTS_DIR/gateway_PV001.crt" \
    -days 365 -sha256  # 1 anno per gateway (rotazione annuale)

echo "Certificati generati in $CERTS_DIR/"
echo "CA: ca.crt"
echo "Server (Mosquitto): server.crt + server.key"
echo "Client Telegraf: telegraf.crt + telegraf.key"
echo "Client Gateway PV001: gateway_PV001.crt + gateway_PV001.key"

#!/usr/bin/env python3
"""
mqtt_benchmark.py
Benchmark del broker MQTT per simulare 200 inverter
Verifica throughput, latenza e affidabilità
"""

import asyncio
import json
import time
import random
import statistics
from datetime import datetime, timezone
import aiomqtt


async def simulate_inverter(
    plant_id: str,
    inverter_id: str,
    broker_host: str,
    broker_port: int,
    duration_sec: int,
    poll_interval: float = 5.0
) -> dict:
    """Simula un inverter che pubblica metriche su MQTT"""
    messages_sent = 0
    latencies = []
    errors = 0
    start_time = time.monotonic()

    async with aiomqtt.Client(
        hostname=broker_host,
        port=broker_port,
        client_id=f"bench_{plant_id}_{inverter_id}",
        keepalive=30,
    ) as client:
        while time.monotonic() - start_time < duration_sec:
            # Simula metriche realistiche
            base_power = 50000 * random.uniform(0.3, 1.0)  # 0-50 kW
            payload = {
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "plant_id": plant_id,
                "inverter_id": inverter_id,
                "ac_power_w": base_power,
                "ac_voltage_l1_v": random.uniform(225, 235),
                "dc_voltage_v": random.uniform(550, 650),
                "dc_current_a": base_power / random.uniform(550, 650),
                "temp_cabinet_c": random.uniform(35, 65),
                "efficiency_pct": random.uniform(94, 98.5),
                "operating_state": 4,  # MPPT
                "error_code": 0,
                "energy_wh_total": messages_sent * base_power * poll_interval / 3600,
            }
            topic = f"plant/{plant_id}/inverter/{inverter_id}/metrics"
            t_send = time.monotonic()
            try:
                await client.publish(
                    topic=topic,
                    payload=json.dumps(payload).encode(),
                    qos=1,
                )
                latencies.append((time.monotonic() - t_send) * 1000)  # ms
                messages_sent += 1
            except Exception:
                errors += 1
            await asyncio.sleep(poll_interval)

    return {
        "inverter_id": inverter_id,
        "messages_sent": messages_sent,
        "errors": errors,
        "avg_latency_ms": statistics.mean(latencies) if latencies else 0,
        "p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0,
    }


async def run_benchmark(
    num_inverters: int = 200,
    duration_sec: int = 60,
    broker_host: str = "localhost",
    broker_port: int = 1883,
) -> None:
    """Esegui benchmark con N inverter simulati"""
    print(f"Avvio benchmark: {num_inverters} inverter per {duration_sec}s")
    print(f"Broker: {broker_host}:{broker_port}")
    print("-" * 60)

    start = time.monotonic()
    tasks = [
        simulate_inverter(
            plant_id="BENCH001",
            inverter_id=f"INV{i:03d}",
            broker_host=broker_host,
            broker_port=broker_port,
            duration_sec=duration_sec,
            poll_interval=5.0,
        )
        for i in range(1, num_inverters + 1)
    ]
    results = await asyncio.gather(*tasks)
    elapsed = time.monotonic() - start

    # Aggregazione risultati
    total_messages = sum(r["messages_sent"] for r in results)
    total_errors = sum(r["errors"] for r in results)
    all_latencies_avg = [r["avg_latency_ms"] for r in results]
    all_latencies_p99 = [r["p99_latency_ms"] for r in results]

    print(f"Risultati benchmark:")
    print(f"  Durata: {elapsed:.1f}s")
    print(f"  Messaggi inviati: {total_messages:,}")
    print(f"  Errori: {total_errors} ({total_errors/total_messages*100:.2f}%)")
    print(f"  Throughput: {total_messages/elapsed:.1f} msg/sec")
    print(f"  Latenza media: {statistics.mean(all_latencies_avg):.2f} ms")
    print(f"  Latenza P99: {statistics.mean(all_latencies_p99):.2f} ms")
    print(f"  Inverter senza errori: {sum(1 for r in results if r['errors'] == 0)}/{num_inverters}")


if __name__ == "__main__":
    asyncio.run(run_benchmark(
        num_inverters=200,
        duration_sec=120,
        broker_host="localhost",
        broker_port=1883,
    ))