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__(self):
self.nodes = {}
self.alert_thresholds = {
'voltage_deviation': 0.05, # 5% deviation
'overload': 0.95, # 95% capacity
'frequency_deviation': 0.5 # Hz
}
def process_meter_reading(self, reading: PowerReading) -> dict:
"""Process AMI meter reading"""
alerts = []
# Voltage quality check
nominal_voltage = 240.0 # Volts
voltage_deviation = abs(reading.voltage - nominal_voltage) / nominal_voltage
if voltage_deviation > self.alert_thresholds['voltage_deviation']:
alerts.append({
'type': 'voltage_deviation',
'severity': 'warning',
'value': voltage_deviation,
'message': f'Voltage deviation: {voltage_deviation:.2%}'
})
# Frequency check
nominal_frequency = 60.0 # Hz (US) or 50.0 (Europe)
freq_deviation = abs(reading.frequency - nominal_frequency)
if freq_deviation > self.alert_thresholds['frequency_deviation']:
alerts.append({
'type': 'frequency_deviation',
'severity': 'critical',
'value': freq_deviation,
'message': f'Frequency deviation: {freq_deviation:.2f} Hz'
})
# Power factor check
if reading.power_factor < 0.9:
alerts.append({
'type': 'poor_power_factor',
'severity': 'info',
'value': reading.power_factor,
'message': f'Low power factor: {reading.power_factor:.2f}'
})
return {
'meter_id': reading.meter_id,
'timestamp': reading.timestamp,
'metrics': {
'voltage': reading.voltage,
'current': reading.current,
'power': reading.active_power,
'power_factor': reading.power_factor
},
'alerts': alerts
}
def calculate_grid_load(self, node_id: str) -> dict:
"""Calculate load metrics for grid node"""
node = self.nodes.get(node_id)
if not node:
return {'error': 'Node not found'}
load_percentage = (node.current_load / node.capacity) * 100
available_capacity = node.capacity - node.current_load
status = 'normal'
if load_percentage > 95:
status = 'critical'
elif load_percentage > 80:
status = 'warning'
return {
'node_id': node_id,
'current_load_mw': node.current_load,
'capacity_mw': node.capacity,
'load_percentage': load_percentage,
'available_capacity_mw': available_capacity,
'status': status
}
def predict_demand(self, historical_data: List[float], hours_ahead: int = 24) -> np.ndarray:
"""Predict energy demand using time series analysis"""
# Simple moving average prediction
# In production, use LSTM or ARIMA models
window_size = 168 # 1 week of hourly data
if len(historical_data) < window_size:
return np.array([np.mean(historical_data)] * hours_ahead)
recent_data = np.array(historical_data[-window_size:])
# Calculate seasonal pattern (24-hour cycle)
hourly_pattern = np.zeros(24)
for i in range(24):
hourly_indices = list(range(i, len(recent_data), 24))
hourly_pattern[i] = np.mean(recent_data[hourly_indices])
# Generate predictions
predictions = []
for hour in range(hours_ahead):
hour_of_day = hour % 24
predictions.append(hourly_pattern[hour_of_day])
return 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)
# 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
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/