1에너지 부하 예측
제조 시설의 에너지 비용을 최적화하려면 먼저 정확한 부하 예측이 필요합니다. 시계열 예측 모델을 사용하여 시간대별, 요일별, 계절별 에너지 소비 패턴을 학습하고 미래 부하를 예측합니다. 이를 통해 피크 요금 시간대를 피해 생산 스케줄을 조정하거나 에너지 저장 시스템을 효율적으로 운영할 수 있습니다.
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime, timedelta
from sklearn.ensemble import GradientBoostingRegressor
@dataclass
class EnergyData:
"""에너지 데이터 포인트"""
timestamp: datetime
demand_kw: float # 전력 수요 (kW)
temperature: float # 외기 온도 (°C)
production_rate: float # 생산 가동률 (0-1)
is_holiday: bool # 휴일 여부
@dataclass
class EnergyTariff:
"""전력 요금 구조"""
peak_hours: List[int] # 피크 시간대
peak_rate: float # 피크 요금 (원/kWh)
off_peak_rate: float # 비피크 요금 (원/kWh)
demand_charge: float # 피크 수요 요금 (원/kW)
class EnergyDemandPredictor:
"""에너지 부하 예측 모델"""
def __init__(self, lookback_hours: int = 168):
self.lookback_hours = lookback_hours # 일주일
self.model = GradientBoostingRegressor(
n_estimators=100, max_depth=6, learning_rate=0.1
)
self.is_fitted = False
def _extract_features(self, data: List[EnergyData]) -> np.ndarray:
"""시계열 특성 추출"""
features = []
for d in data:
feat = [
d.timestamp.hour,
d.timestamp.weekday(),
d.timestamp.month,
np.sin(2 * np.pi * d.timestamp.hour / 24),
np.cos(2 * np.pi * d.timestamp.hour / 24),
d.temperature,
d.production_rate,
1 if d.is_holiday else 0
]
features.append(feat)
return np.array(features)
def fit(self, history: List[EnergyData]):
"""과거 데이터로 학습"""
X = self._extract_features(history[:-1])
y = np.array([d.demand_kw for d in history[1:]])
self.model.fit(X, y)
self.is_fitted = True
def predict(self, current: EnergyData, hours: int = 24) -> List[float]:
"""향후 n시간 수요 예측"""
predictions = []
feat = self._extract_features([current])[0]
for h in range(hours):
pred = self.model.predict([feat])[0]
predictions.append(pred)
# 시간 업데이트
next_time = current.timestamp + timedelta(hours=h+1)
feat[0] = next_time.hour
feat[1] = next_time.weekday()
return predictions2피크 수요 관리
전력 요금에서 피크 수요 요금(Demand Charge)은 전체 비용의 30-50%를 차지합니다. 월간 최대 수요(kW) 기준으로 부과되어 단 15분의 피크도 한 달 요금에 영향을 미칩니다. AI 기반 피크 관리는 실시간 부하 모니터링, 피크 발생 직전 비필수 장비 정지, ESS(Energy Storage System) 방전으로 피크를 억제합니다.
class PeakDemandOptimizer:
"""피크 수요 최적화"""
def __init__(self, peak_limit_kw: float,
ess_capacity_kwh: float, ess_max_power_kw: float):
self.peak_limit = peak_limit_kw
self.ess_capacity = ess_capacity_kwh
self.ess_max_power = ess_max_power_kw
self.ess_soc = 0.5 # State of Charge
self.controllable_loads = {} # {장비ID: (kW, 우선순위)}
def register_load(self, equip_id: str, kw: float, priority: int):
"""차단 가능 장비 등록 (높은 priority = 나중에 차단)"""
self.controllable_loads[equip_id] = (kw, priority)
def optimize(self, current_demand: float, tariff: EnergyTariff) -> Dict:
"""피크 억제 최적화"""
actions = {'ess_discharge_kw': 0, 'loads_to_shed': [], 'savings': 0}
excess = current_demand - self.peak_limit
if excess <= 0:
return actions # 피크 미만
# 1단계: ESS 방전
available_ess = min(self.ess_max_power, self.ess_soc * self.ess_capacity)
ess_discharge = min(excess, available_ess)
actions['ess_discharge_kw'] = ess_discharge
excess -= ess_discharge
# 2단계: 부하 차단 (우선순위 낮은 순)
if excess > 0:
sorted_loads = sorted(self.controllable_loads.items(),
key=lambda x: x[1][1])
for equip_id, (kw, _) in sorted_loads:
if excess <= 0: break
actions['loads_to_shed'].append(equip_id)
excess -= kw
# 절감액 계산
peak_reduction = current_demand - self.peak_limit - max(0, excess)
actions['savings'] = peak_reduction * tariff.demand_charge
return actions
class RealTimePeakController:
"""실시간 피크 제어"""
def __init__(self, optimizer: PeakDemandOptimizer,
predictor: EnergyDemandPredictor, tariff: EnergyTariff):
self.optimizer = optimizer
self.predictor = predictor
self.tariff = tariff
self.monthly_peak = 0
def control_cycle(self, current: EnergyData) -> Dict:
"""15분 주기 제어"""
forecast = self.predictor.predict(current, hours=4)
actions = self.optimizer.optimize(current.demand_kw, self.tariff)
effective = current.demand_kw - actions['ess_discharge_kw']
for eid in actions['loads_to_shed']:
effective -= self.optimizer.controllable_loads[eid][0]
self.monthly_peak = max(self.monthly_peak, effective)
return {'actions': actions, 'forecast': forecast,
'monthly_peak': self.monthly_peak}3설비별 에너지 효율 최적화
개별 설비의 에너지 효율을 최적화하면 전체 공장 에너지 소비를 15-30% 절감할 수 있습니다. 예를 들어 압축공기 시스템은 공장 전기 사용량의 20-30%를 차지하는데, 압력 설정 최적화, 누설 탐지, 부하 매칭 등을 통해 큰 절감 효과를 얻을 수 있습니다.
class CompressorOptimizer:
"""압축공기 시스템 최적화"""
def __init__(self, compressors: List[Dict]):
"""compressors: [{'id', 'capacity_cfm', 'power_kw', 'efficiency_curve'}]"""
self.compressors = compressors
self.min_pressure = 6.0 # bar
self.max_pressure = 7.0
def optimize_loading(self, required_cfm: float) -> Dict[str, float]:
"""압축기 부하 분배 (효율 높은 순)"""
sorted_comps = sorted(
self.compressors, key=lambda c: c['efficiency_curve'](0.8), reverse=True
)
allocation = {}
remaining = required_cfm
for comp in sorted_comps:
if remaining <= 0:
allocation[comp['id']] = 0
elif remaining >= comp['capacity_cfm']:
allocation[comp['id']] = 1.0
remaining -= comp['capacity_cfm']
else:
allocation[comp['id']] = remaining / comp['capacity_cfm']
remaining = 0
return allocation
def optimize_pressure(self, requirements: Dict[str, float]) -> float:
"""최적 공급 압력 (가장 높은 요구 + 마진)"""
optimal = max(requirements.values()) + 0.3
return np.clip(optimal, self.min_pressure, self.max_pressure)
def estimate_savings(self, current: float, optimal: float,
annual_kwh: float) -> float:
"""압력 최적화 절감 (1 bar = 7% 절감)"""
reduction = current - optimal
return annual_kwh * reduction * 0.074통합 에너지 관리 시스템
부하 예측, 피크 관리, 설비 최적화를 통합하는 EMS(Energy Management System)를 구축합니다. 실시간 모니터링, 자동 제어, 리포팅을 포함한 완전한 에너지 최적화 플랫폼으로 일반적으로 6-12개월 내에 투자 회수가 가능합니다.
class IntegratedEMS:
"""통합 에너지 관리 시스템"""
def __init__(self, predictor: EnergyDemandPredictor,
peak_controller: RealTimePeakController,
equipment_optimizers: Dict):
self.predictor = predictor
self.peak_controller = peak_controller
self.equipment_optimizers = equipment_optimizers
self.history = []
def run_cycle(self, current: EnergyData) -> Dict:
"""15분 주기 최적화"""
results = {'timestamp': current.timestamp}
# 피크 제어
results['peak'] = self.peak_controller.control_cycle(current)
# 설비별 최적화
for name, opt in self.equipment_optimizers.items():
results[name] = opt.optimize()
self.history.append(results)
return results
def report(self, days: int = 30) -> Dict:
"""절감 리포트"""
recent = self.history[-days*96:] # 15분 간격
total_savings = sum(
r.get('peak', {}).get('actions', {}).get('savings', 0)
for r in recent
)
return {
'period_days': days,
'total_savings_krw': total_savings,
'peak_demand_kw': self.peak_controller.monthly_peak
}
# 사용 예시
tariff = EnergyTariff([10,11,12,13,14,15,16,17], 150, 80, 8000)
optimizer = PeakDemandOptimizer(1000, 500, 200)
optimizer.register_load('HVAC-1', 50, 1)
optimizer.register_load('HVAC-2', 50, 1)
actions = optimizer.optimize(1100, tariff) # 피크 초과!
print(f"ESS 방전: {actions['ess_discharge_kw']:.0f} kW")
print(f"예상 절감: {actions['savings']:,.0f}원")ROI: 에너지 AI로 피크 15% 감소 시 연간 수억 원 절감이 가능합니다.