smart-grid-telemetry-iot

Smart Grid & Energy Telemetry IoT Skill Guide

This skill governs standard communication protocols, time-series telemetry parsing, and electrical metrics calculation for modern smart grids and renewable microgrid systems.

1. Smart Grid Telemetry Architecture

[ Substation Sensors / Inverters / BESS ]
  |-- Modbus TCP (Port 502) / IEC 61850 / DNP3
  |-- Electric Metrics: Voltage (V), Current (I), Active Power (kW), Frequency (Hz)
                 |
                 v
[ Edge IoT Gateway (Pymodbus / Edge Daemon) ]
  |-- Register Polling (Holding Registers: IEEE 754 32-bit Float)
  |-- Local Buffer & Network Fault Tolerance
                 |
                 v (MQTT / TLS / Kafka)
[ Real-Time Microgrid Energy Management System (EMS) ]
  |-- State-of-Charge (SoC) Coulomb Counting
  |-- Power Factor & Peak Shaving Automation
  |-- Time-Series Storage (TimescaleDB / InfluxDB)

2. Production Code Implementations

A. Modbus TCP Energy Meter Ingestion (Python / pymodbus)

from pymodbus.client import ModbusTcpClient
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder
from pydantic import BaseModel
import time


class MeterTelemetry(BaseModel):
    timestamp_epoch: float
    voltage_l1_volts: float
    current_l1_amps: float
    active_power_kw: float
    power_factor: float
    frequency_hz: float


class SmartMeterReader:
    def __init__(self, host: str, port: int = 502, unit_id: int = 1):
        self.client = ModbusTcpClient(host=host, port=port)
        self.unit_id = unit_id

    def connect(self) -> bool:
        return self.client.connect()

    def read_metrics(self) -> MeterTelemetry:
        # Read 10 consecutive holding registers starting at 0x1000
        result = self.client.read_holding_registers(address=0x1000, count=10, slave=self.unit_id)
        if result.isError():
            raise IOError(f"Modbus read error: {result}")

        # Decode IEEE 754 32-bit floating point numbers (Big-Endian word order)
        decoder = BinaryPayloadDecoder.fromRegisters(
            result.registers,
            byteorder=Endian.BIG,
            wordorder=Endian.BIG,
        )

        voltage = decoder.decode_32bit_float()
        current = decoder.decode_32bit_float()
        active_power = decoder.decode_32bit_float()
        power_factor = decoder.decode_32bit_float()
        frequency = decoder.decode_32bit_float()

        return MeterTelemetry(
            timestamp_epoch=time.time(),
            voltage_l1_volts=round(voltage, 2),
            current_l1_amps=round(current, 2),
            active_power_kw=round(active_power, 2),
            power_factor=round(power_factor, 3),
            frequency_hz=round(frequency, 2),
        )

    def close(self):
        self.client.close()