energy-expert

name	energy-expert
version	1.0.0
description	Expert-level energy systems, smart grids, renewable energy, power management, and energy analytics
category	domains
tags	["energy","smart-grid","renewable","power","utilities","scada"]
allowed-tools	["Read","Write","Edit"]

Energy Expert

Expert guidance for energy systems, smart grid technology, renewable energy integration, power management, and energy sector software development.

Core Concepts

Energy Systems

• Smart grid infrastructure
• Renewable energy systems (solar, wind, hydro)
• Power generation and distribution
• Energy storage systems (batteries, pumped hydro)
• Demand response management
• Energy trading and markets
• Grid stability and load balancing

Smart Grid Technology

• Advanced Metering Infrastructure (AMI)
• Supervisory Control and Data Acquisition (SCADA)
• Distribution Management Systems (DMS)
• Energy Management Systems (EMS)
• Outage Management Systems (OMS)
• Geographic Information Systems (GIS)
• Real-time monitoring and control

Standards and Protocols

• IEC 61850 (power utility automation)
• Modbus (industrial protocol)
• DNP3 (Distributed Network Protocol)
• IEEE 2030 (smart grid interoperability)
• OpenADR (automated demand response)
• CIM (Common Information Model)
• MQTT for IoT devices

Smart Grid Monitoring System

from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
import numpy as np

@dataclass
class GridNode:
    """Represents a node in the power grid"""
    node_id: str
    node_type: str  # 'substation', 'transformer', 'meter'
    location: tuple  # (latitude, longitude)
    voltage_rating: float  # kV
    current_load: float  # MW
    capacity: float  # MW
    status: str  # 'online', 'offline', 'maintenance'
    last_updated: datetime

@dataclass
class PowerReading:
    """Real-time power measurement"""
    meter_id: str
    timestamp: datetime
    voltage: float  # Volts
    current: float  # Amperes
    power_factor: float
    active_power: float  # kW
    reactive_power: float  # kVAR
    frequency: float  # Hz

class SmartGridMonitor:
    """Smart grid monitoring and control system"""

    def __init__():
        .nodes = {}
        .alert_thresholds = {
            : ,  
            : ,  
            :   
        }

     () -> :
        
        alerts = []

        
        nominal_voltage =   
        voltage_deviation = (reading.voltage - nominal_voltage) / nominal_voltage

         voltage_deviation > .alert_thresholds[]:
            alerts.append({
                : ,
                : ,
                : voltage_deviation,
                : 
            })

        
        nominal_frequency =   
        freq_deviation = (reading.frequency - nominal_frequency)

         freq_deviation > .alert_thresholds[]:
            alerts.append({
                : ,
                : ,
                : freq_deviation,
                : 
            })

        
         reading.power_factor < :
            alerts.append({
                : ,
                : ,
                : reading.power_factor,
                : 
            })

         {
            : reading.meter_id,
            : reading.timestamp,
            : {
                : reading.voltage,
                : reading.current,
                : reading.active_power,
                : reading.power_factor
            },
            : alerts
        }

     () -> :
        
        node = .nodes.get(node_id)
          node:
             {: }

        load_percentage = (node.current_load / node.capacity) * 
        available_capacity = node.capacity - node.current_load

        status = 
         load_percentage > :
            status = 
         load_percentage > :
            status = 

         {
            : node_id,
            : node.current_load,
            : node.capacity,
            : load_percentage,
            : available_capacity,
            : status
        }

     () -> np.ndarray:
        
        
        
        window_size =   

         (historical_data) < window_size:
             np.array([np.mean(historical_data)] * hours_ahead)

        recent_data = np.array(historical_data[-window_size:])

        
        hourly_pattern = np.zeros()
         i  ():
            hourly_indices = ((i, (recent_data), ))
            hourly_pattern[i] = np.mean(recent_data[hourly_indices])

        
        predictions = []
         hour  (hours_ahead):
            hour_of_day = hour % 
            predictions.append(hourly_pattern[hour_of_day])

         np.array(predictions)

Renewable Energy Integration

from datetime import datetime, timedelta
import math

class RenewableEnergyManager:
    """Manage renewable energy sources in the grid"""

    def __init__(self):
        self.solar_farms = {}
        self.wind_farms = {}
        self.energy_storage = {}

    def calculate_solar_output(self,
                              capacity_kw: float,
                              location: tuple,
                              timestamp: datetime,
                              cloud_cover: float = 0.0) -> float:
        """Calculate solar panel output based on conditions"""
        lat, lon = location

        # Calculate solar angle (simplified)
        day_of_year = timestamp.timetuple().tm_yday
        hour = timestamp.hour + timestamp.minute / 60.0

        # Solar declination
        declination = 23.45 * math.sin(math.radians((360/365) * (day_of_year - 81)))

        # Hour angle
        hour_angle = 15 * (hour - 12)

        # Solar elevation angle
        elevation = math.asin(
            math.sin(math.radians(lat)) * math.sin(math.radians(declination)) +
            math.cos(math.radians(lat)) * math.cos(math.radians(declination)) *
            math.cos(math.radians(hour_angle))
        )

        # Base output (0-1 scale)
        if elevation <= 0:
            return 0.0  # Night time

        base_output = math.sin(elevation)

        
        cloud_factor =  - (cloud_cover * )

        
        output_kw = capacity_kw * base_output * cloud_factor

         (, output_kw)

     () -> :
        

        
         wind_speed_ms < cut_in_speed:
             

        
         wind_speed_ms > cut_out_speed:
             

        
         wind_speed_ms < rated_speed:
            power_coefficient = ((wind_speed_ms - cut_in_speed) /
                               (rated_speed - cut_in_speed)) ** 
             capacity_kw * power_coefficient

        
         capacity_kw

     () -> :
        

        surplus = renewable_output - current_demand

        action = 
        amount = 

        
         surplus >   storage_level < storage_capacity:
            charge_amount = (surplus, storage_capacity - storage_level)
            action = 
            amount = charge_amount

        
         surplus <   storage_level > :
            discharge_amount = ((surplus), storage_level)

            
             grid_price > :  
                action = 
                amount = discharge_amount

        new_storage_level = storage_level
         action == :
            new_storage_level = storage_level + amount
         action == :
            new_storage_level = storage_level - amount

         {
            : action,
            : amount,
            : new_storage_level,
            : (new_storage_level / storage_capacity) * 
        }

SCADA Integration

import struct
from typing import Dict, Any

class ModbusClient:
    """Modbus protocol client for SCADA systems"""

    def __init__(self, host: str, port: int = 502):
        self.host = host
        self.port = port
        self.connected = False

    def read_holding_registers(self,
                               slave_id: int,
                               start_address: int,
                               count: int) -> List[int]:
        """Read holding registers (function code 0x03)"""
        # Build Modbus request
        request = struct.pack(
            '>BBHH',
            slave_id,
            0x03,  # Function code
            start_address,
            count
        )

        # Send request and receive response
        # In production, use pymodbus library
        response = self._send_request(request)

        # Parse response
        values = []
        for i in range(count):
            offset = 3 + (i * 2)  # Skip header
            value = struct.unpack('>H', response[offset:offset+2])[0]
            values.append(value)

         values

     () -> :
        
        request = struct.pack(
            ,
            slave_id,
            ,  
            address,
            value
        )

        response = ._send_request(request)
         response   

     () -> :
        
        
        

 :
    

     ():
        .devices = {}
        .alarm_conditions = []

     () -> :
        
        modbus = ModbusClient()

        :
            
            voltages = modbus.read_holding_registers(, , )

            
            currents = modbus.read_holding_registers(, , )

            
            breaker_status = modbus.read_holding_registers(, , )

            
            total_power = (
                v * c  v, c  (voltages, currents)
            ) /   

             {
                : substation_id,
                : voltages,
                : currents,
                : total_power,
                : {
                    :   status  
                     i, status  (breaker_status)
                },
                : 
            }

         Exception  e:
             {
                : substation_id,
                : ,
                : (e)
            }

     () -> :
        
        modbus = ModbusClient()

        value =   action ==   
        register =  + breaker_id - 

        success = modbus.write_single_register(, register, value)

         success:
            ._log_control_action(substation_id, breaker_id, action)

         success

     ():
        
        timestamp = datetime.now().isoformat()
        ()

Energy Trading and Markets

from decimal import Decimal
from datetime import datetime, timedelta

class EnergyTradingSystem:
    """Energy trading and market operations"""

    def __init__(self):
        self.bids = []
        self.offers = []
        self.market_prices = {}

    def submit_bid(self,
                   participant_id: str,
                   quantity_mwh: Decimal,
                   price_per_mwh: Decimal,
                   delivery_hour: datetime) -> str:
        """Submit bid to purchase energy"""
        bid = {
            'bid_id': self._generate_id(),
            'participant_id': participant_id,
            'type': 'buy',
            'quantity_mwh': quantity_mwh,
            'price_per_mwh': price_per_mwh,
            'delivery_hour': delivery_hour,
            'timestamp': datetime.now(),
            'status': 'pending'
        }

        self.bids.append(bid)
        return bid['bid_id']

    def submit_offer(self,
                    participant_id: str,
                    quantity_mwh: Decimal,
                    price_per_mwh: Decimal,
                    delivery_hour: datetime) -> str:
        """Submit offer to sell energy"""
        offer = {
            'offer_id': ._generate_id(),
            : participant_id,
            : ,
            : quantity_mwh,
            : price_per_mwh,
            : delivery_hour,
            : datetime.now(),
            : 
        }

        .offers.append(offer)
         offer[]

     () -> :
        
        
        hour_bids = [b  b  .bids
                     b[] == delivery_hour  b[] == ]
        hour_offers = [o  o  .offers
                       o[] == delivery_hour  o[] == ]

        
        sorted_bids = (hour_bids, key= x: x[], reverse=)
        sorted_offers = (hour_offers, key= x: x[])

        
        matches = []
        total_cleared_volume = Decimal()
        clearing_price = Decimal()

        bid_idx = 
        offer_idx = 

         bid_idx < (sorted_bids)  offer_idx < (sorted_offers):
            bid = sorted_bids[bid_idx]
            offer = sorted_offers[offer_idx]

            
             bid[] >= offer[]:
                
                volume = (bid[], offer[])
                clearing_price = (bid[] + offer[]) / 

                matches.append({
                    : bid[],
                    : offer[],
                    : volume,
                    : clearing_price
                })

                total_cleared_volume += volume

                
                bid[] -= volume
                offer[] -= volume

                 bid[] == :
                    bid_idx += 
                 offer[] == :
                    offer_idx += 
            :
                

         {
            : delivery_hour,
            : clearing_price,
            : total_cleared_volume,
            : matches
        }

     () -> :
        
         uuid
         (uuid.uuid4())

Best Practices

Smart Grid Operations

• Implement real-time monitoring with sub-second latency
• Use redundant communication paths for critical systems
• Deploy edge computing for local decision-making
• Maintain comprehensive audit logs for all control actions
• Implement cybersecurity measures (IEC 62351)
• Use time synchronization (IEEE 1588 PTP)

Renewable Energy Integration

• Forecast renewable generation using ML models
• Implement dynamic curtailment strategies
• Use energy storage for grid stabilization
• Support virtual power plants (VPP)
• Enable peer-to-peer energy trading
• Monitor power quality metrics

Data Management

• Use time-series databases (InfluxDB, TimescaleDB)
• Implement data compression for long-term storage
• Archive historical data with proper retention policies
• Ensure data integrity and traceability
• Support real-time analytics and visualization
• Implement anomaly detection algorithms

System Design

• Design for 99.999% availability
• Implement graceful degradation
• Use microservices architecture
• Support multi-region deployments
• Enable automatic failover
• Implement load balancing

Anti-Patterns

❌ Single point of failure in critical systems ❌ No backup power for control systems ❌ Ignoring cybersecurity requirements ❌ Insufficient data validation ❌ No disaster recovery plan ❌ Inadequate alarm management (alarm floods) ❌ Poor time synchronization ❌ No testing of protection schemes

Resources

• IEC 61850 Standard: https://www.iec.ch/
• IEEE Smart Grid: https://smartgrid.ieee.org/
• OpenADR Alliance: https://www.openadr.org/
• Modbus Protocol: https://modbus.org/
• DNP3 Protocol: https://www.dnp.org/
• NIST Smart Grid Framework: https://www.nist.gov/smartgrid
• GridWise Architecture Council: https://www.gridwiseac.org/