| 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
location: tuple
voltage_rating: float
current_load: float
capacity: float
status: str
last_updated: datetime
@dataclass
class PowerReading:
"""Real-time power measurement"""
meter_id: str
timestamp: datetime
voltage: float
current: float
power_factor: float
active_power: float
reactive_power: float
frequency: float
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
day_of_year = timestamp.timetuple().tm_yday
hour = timestamp.hour + timestamp.minute / 60.0
declination = 23.45 * math.sin(math.radians((360/365) * (day_of_year - 81)))
hour_angle = 15 * (hour - 12)
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))
)
if elevation <= 0:
return 0.0
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)"""
request = struct.pack(
'>BBHH',
slave_id,
0x03,
start_address,
count
)
response = self._send_request(request)
values = []
for i in range(count):
offset = 3 + (i * 2)
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