Imported from personamanagmentlayer/pcl (stdlib/domains/energy-expert/SKILL.md). Install upstream with npx skills add personamanagmentlayer/pcl --skill energy-expert. Copyright stays with the author.

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

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)

        # Apply cloud cover factor
        cloud_factor = 1.0 - (cloud_cover * 0.75)

        # Calculate actual output
        output_kw = capacity_kw * base_output * cloud_factor

        return max(0.0, output_kw)

    def calculate_wind_output(self,
                            capacity_kw: float,
                            wind_speed_ms: float,
                            cut_in_speed: float = 3.0,
                            rated_speed: float = 12.0,
                            cut_out_speed: float = 25.0) -> float:
        """Calculate wind turbine output based on wind speed"""

        # Below cut-in speed
        if wind_speed_ms < cut_in_speed:
            return 0.0

        # Above cut-out speed (safety shutdown)
        if wind_speed_ms > cut_out_speed:
            return 0.0

        # Between cut-in and rated speed (cubic relationship)
        if wind_speed_ms < rated_speed:
            power_coefficient = ((wind_speed_ms - cut_in_speed) /
                               (rated_speed - cut_in_speed)) ** 3
            return capacity_kw * power_coefficient

        # At or above rated speed
        return capacity_kw

    def optimize_energy_storage(self,
                               current_demand: float,
                               renewable_output: float,
                               storage_capacity: float,
                               storage_level: float,
                               grid_price: float) -> dict:
        """Optimize battery storage charge/discharge"""

        surplus = renewable_output - current_demand

        action = 'hold'
        amount = 0.0

        # Surplus energy - charge battery
        if surplus > 0 and storage_level < storage_capacity:
            charge_amount = min(surplus, storage_capacity - storage_level)
            action = 'charge'
            amount = charge_amount

        # Deficit and high price - discharge battery
        elif surplus < 0 and storage_level > 0:
            discharge_amount = min(abs(surplus), storage_level)

            # Only discharge if grid price is high
            if grid_price > 0.15:  # $0.15/kWh threshold
                action = 'discharge'
                amount = discharge_amount

        new_storage_level = storage_level
        if action == 'charge':
            new_storage_level = storage_level + amount
        elif action == 'discharge':
            new_storage_level = storage_level - amount

        return {
            'action': action,
            'amount_kwh': amount,
            'storage_level_kwh': new_storage_level,
            'storage_percentage': (new_storage_level / storage_capacity) * 100
        }

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)

        return values

    def write_single_register(self,
                            slave_id: int,
                            address: int,
                            value: int) -> bool:
        """Write single register (function code 0x06)"""
        request = struct.pack(
            '>BBHH',
            slave_id,
            0x06,  # Function code
            address,
            value
        )

        response = self._send_request(request)
        return response is not None

    def _send_request(self, request: bytes) -> bytes:
        """Send Modbus request and receive response"""
        # Implement actual TCP/RTU communication
        pass

class SCADASystem:
    """SCADA system for power grid control"""

    def __init__(self):
        self.devices = {}
        self.alarm_conditions = []

    def monitor_substation(self, substation_id: str) -> dict:
        """Monitor substation parameters via SCADA"""
        modbus = ModbusClient(f'substation-{substation_id}.local')

        try:
            # Read voltage (registers 0-2 for 3-phase)
            voltages = modbus.read_holding_registers(1, 0, 3)

            # Read current (registers 3-5)
            currents = modbus.read_holding_registers(1, 3, 3)

            # Read breaker status (registers 10-15)
            breaker_status = modbus.read_holding_registers(1, 10, 6)

            # Calculate power
            total_power = sum(
                v * c for v, c in zip(voltages, currents)
            ) / 1000.0  # Convert to kW

            return {
                'substation_id': substation_id,
                'voltages_v': voltages,
                'currents_a': currents,
                'power_kw': total_power,
                'breakers': {
                    f'breaker_{i+1}': 'closed' if status else 'open'
                    for i, status in enumerate(breaker_status)
                },
                'status': 'online'
            }

        except Exception as e:
            return {
                'substation_id': substation_id,
                'status': 'error',
                'error': str(e)
            }

    def control_breaker(self,
                       substation_id: str,
                       breaker_id: int,
                       action: str) -> bool:
        """Control circuit breaker (open/close)"""
        modbus = ModbusClient(f'substation-{substation_id}.local')

        value = 1 if action == 'close' else 0
        register = 10 + breaker_id - 1

        success = modbus.write_single_register(1, register, value)

        if success:
            self._log_control_action(substation_id, breaker_id, action)

        return success

    def _log_control_action(self, substation_id: str, breaker_id: int, action: str):
        """Log control actions for audit trail"""
        timestamp = datetime.now().isoformat()
        print(f"[{timestamp}] Breaker control: {substation_id}/breaker_{breaker_id} -> {action}")

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': self._generate_id(),
            'participant_id': participant_id,
            'type': 'sell',
            'quantity_mwh': quantity_mwh,
            'price_per_mwh': price_per_mwh,
            'delivery_hour': delivery_hour,
            'timestamp': datetime.now(),
            'status': 'pending'
        }

        self.offers.append(offer)
        return offer['offer_id']

    def clear_market(self, delivery_hour: datetime) -> dict:
        """Clear energy market using merit order"""
        # Filter bids and offers for delivery hour
        hour_bids = [b for b in self.bids
                    if b['delivery_hour'] == delivery_hour and b['status'] == 'pending']
        hour_offers = [o for o in self.offers
                      if o['delivery_hour'] == delivery_hour and o['status'] == 'pending']

        # Sort bids (descending price) and offers (ascending price)
        sorted_bids = sorted(hour_bids, key=lambda x: x['price_per_mwh'], reverse=True)
        sorted_offers = sorted(hour_offers, key=lambda x: x['price_per_mwh'])

        # Match bids and offers
        matches = []
        total_cleared_volume = Decimal('0')
        clearing_price = Decimal('0')

        bid_idx = 0
        offer_idx = 0

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

            # Check if bid price >= offer price
            if bid['price_per_mwh'] >= offer['price_per_mwh']:
                # Match found
                volume = min(bid['quantity_mwh'], offer['quantity_mwh'])
                clearing_price = (bid['price_per_mwh'] + offer['price_per_mwh']) / 2

                matches.append({
                    'bid_id': bid['bid_id'],
                    'offer_id': offer['offer_id'],
                    'volume_mwh': volume,
                    'price_per_mwh': clearing_price
                })

                total_cleared_volume += volume

                # Update quantities
                bid['quantity_mwh'] -= volume
                offer['quantity_mwh'] -= volume

                if bid['quantity_mwh'] == 0:
                    bid_idx += 1
                if offer['quantity_mwh'] == 0:
                    offer_idx += 1
            else:
                break

        return {
            'delivery_hour': delivery_hour,
            'clearing_price': clearing_price,
            'total_volume_mwh': total_cleared_volume,
            'matches': matches
        }

    def _generate_id(self) -> str:
        """Generate unique transaction ID"""
        import uuid
        return str(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

Reference Documentation

Detailed material lives alongside this skill and is read on demand:

• Smart Grid Monitoring System

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/