Хімічний завод втрачає до 50 млн грн на рік на штрафах за перевищення ГДВ. Традиційний CEMS фіксує порушення постфактум — після лабораторного аналізу проб, коли штраф уже нараховано. AI-шар у Continuous Emission Monitoring System (CEMS) змінює правила гри: детекція аномалій у реальному часі, предиктивна модель замінює датчики, що вийшли з ладу, а автоматична звітність виключає людський фактор. Ми впроваджуємо такі рішення на хімічних виробництвах — від інтеграції з існуючими аналізаторами до побудови повного compliance-модуля під вимоги України (ГДВ) та ЄС (BAT AEL).
Ось реальний кейс. На заводі з виробництва аміаку часті хибні спрацьовування сигналізації по NOx призводили до зупинок. Виявилося, аналізатор завищував показання через перегрів оптичної комірки. Ми розробили soft sensor на основі Gradient Boosting, який передбачав коректну концентрацію за температурою реактора та витратою повітря. Завод скоротив простої на 40% і уникнув штрафу в 12 млн грн за уявне перевищення.
Архітектура CEMS з AI
Джерела даних:
cems_architecture = { 'analyzers': { 'NOx': 'хемілюмінесцентний, діапазон 0-1000 ppm', 'SO2': 'ультрафіолетовий флуоресцентний, 0-2000 ppm', 'CO': 'недисперсійний інфрачервоний (NDIR)', 'CO2': 'NDIR, 0-20% об'ємних', 'HCl': 'NDIR для хлорорганічних виробництв', 'PM2.5/PM10': 'оптичний розсіювач + бета-абсорбція', 'VOC': 'ПІД (Photoionization Detector)' }, 'flow_meter': { 'type': 'ультразвуковий або термічний', 'use': 'розрахунок масового потоку викидів (г/с, т/рік)' }, 'scada_process': { 'parameters': 'температура, тиск, витрата сировини, режими реактора', 'use': 'кореляція викидів з технологічними параметрами' } } Моніторинг відповідності нормативам
Розрахунок нормативних показників:
import pandas as pd import numpy as np def check_regulatory_compliance(emissions_data: pd.DataFrame, permits: dict, regulation: str = 'UA_ND') -> dict: """ Україна: нормативи ГДВ (Гранично Допустимі Викиди) по кожному джерелу. ЄС: EU IED (Industrial Emissions Directive) — BAT Associated Emission Levels. """ violations = [] for pollutant, permit_value in permits.items(): if pollutant not in emissions_data.columns: continue # Миттєве перевищення instantaneous = emissions_data[pollutant].iloc[-1] if instantaneous > permit_value: violations.append({ 'pollutant': pollutant, 'type': 'instantaneous_exceedance', 'current_value': round(instantaneous, 3), 'permit': permit_value, 'exceedance_factor': round(instantaneous / permit_value, 2) }) # Середньодобове (НОРМАТИВНА ВИМОГА: не перевищувати ГДВ більше 3 днів на рік) daily_avg = emissions_data[pollutant].resample('D').mean() if len(daily_avg) > 0: days_exceeded = (daily_avg > permit_value).sum() if days_exceeded > 0: violations.append({ 'pollutant': pollutant, 'type': 'daily_average_exceeded', 'days_exceeded': int(days_exceeded), 'avg_exceedance': round(daily_avg[daily_avg > permit_value].mean(), 3) }) # Річний сумарний викид if regulation == 'UA_ND' and 'annual_limit_tonnes' in permits: annual_actual = emissions_data[pollutant].sum() * 3600 * 1e-6 # г/с → т/рік (спрощення) annual_limit = permits['annual_limit_tonnes'].get(pollutant, float('inf')) if annual_actual > annual_limit * 0.9: violations.append({ 'pollutant': pollutant, 'type': 'annual_limit_approaching', 'current_tonnes': round(annual_actual, 2), 'annual_limit': annual_limit, 'utilization_pct': round(annual_actual / annual_limit * 100, 1) }) return { 'compliance': len(violations) == 0, 'violations': violations, 'regulatory_status': 'compliant' if not violations else 'violation' } Як ми виявляємо аномальні викиди?
Детектор на основі EWMA та Z-score аналізує кожне нове вимірювання. Якщо відхилення перевищує 4 сигми — формується alert. При перевищенні 6 сигм — emergency shutdown check. AI-детектор виявляє аномалії в 10 разів швидше традиційного підходу (секунди замість годин на лабораторний аналіз). Це мінімізує ризик штрафів до 50% та запобігає екологічній шкоді.
class EmissionSpikeDetector: def __init__(self, pollutants: list, ewma_alpha: float = 0.1): self.baselines = {p: {'mean': None, 'std': None} for p in pollutants} self.alpha = ewma_alpha self.history = {p: [] for p in pollutants} def update_and_detect(self, timestamp, readings: dict) -> dict: alerts = [] for pollutant, value in readings.items(): if pollutant not in self.baselines: continue self.history[pollutant].append(value) if len(self.history[pollutant]) < 30: # Накопичуємо baseline if len(self.history[pollutant]) == 30: self.baselines[pollutant]['mean'] = np.mean(self.history[pollutant]) self.baselines[pollutant]['std'] = np.std(self.history[pollutant]) continue mean = self.baselines[pollutant]['mean'] std = self.baselines[pollutant]['std'] # Z-score z = (value - mean) / (std + 1e-9) # EWMA оновлення (повільно, щоб не підлаштуватися під аварію) if abs(z) < 2: # оновлюємо baseline лише в нормальному режимі self.baselines[pollutant]['mean'] = ( self.alpha * value + (1 - self.alpha) * mean ) # Швидкість наростання (ROC) if len(self.history[pollutant]) >= 5: rate_of_change = (value - self.history[pollutant][-5]) / 4 # за 4 інтервали if abs(z) > 4 or rate_of_change > std * 3: alerts.append({ 'pollutant': pollutant, 'value': value, 'z_score': round(z, 1), 'rate_of_change': round(rate_of_change, 3), 'severity': 'emergency' if abs(z) > 6 else 'alert', 'action': 'emergency_shutdown_check' if abs(z) > 6 else 'investigate_source' }) return {'timestamp': str(timestamp), 'alerts': alerts, 'healthy': len(alerts) == 0} Чому предиктивна модель викидів знижує штрафи?
Soft sensor дозволяє прогнозувати концентрації за технологічними параметрами: температура реактора, витрата сировини, завантаження. При відмові газоаналізатора модель заміщує його дані — це гарантує безперервність моніторингу та виключає «сліпі зони». Точність моделі на основі Gradient Boosting — ±5% для NOx та SO2. Для вибору найкращої моделі ми порівнюємо Gradient Boosting, LSTM та Random Forest:
| Модель | Точність (MAPE) | Час навчання | Інтерпретованість |
|---|---|---|---|
| Gradient Boosting | 5% | 2 хвилини | Висока |
| LSTM | 4% | 30 хвилин | Низька |
| Random Forest | 6% | 1 хвилина | Середня |
Gradient Boosting дає найкращий баланс точності та швидкості.
from sklearn.ensemble import GradientBoostingRegressor def train_emission_prediction_model(process_data: pd.DataFrame, emission_data: pd.DataFrame, pollutant: str) -> GradientBoostingRegressor: """ Передбачаємо викид за технологічними параметрами. Використання: 1) моніторинг при відмові аналізатора, 2) оптимізація режиму. """ process_features = [ 'reactor_temperature', 'feed_flow_rate', 'pressure', 'oxygen_content', 'fuel_type_encoded', 'load_pct', 'catalyst_activity' ] combined = process_data.merge(emission_data[['timestamp', pollutant]], on='timestamp', how='inner') combined = combined.dropna(subset=process_features + [pollutant]) model = GradientBoostingRegressor( n_estimators=200, max_depth=5, learning_rate=0.05 ) model.fit(combined[process_features], combined[pollutant]) return model # використовувати: model.predict([[T, F, P, O2, fuel, load, cat]]) Інтеграція та звітність
Система передає дані безпосередньо до CEMS державної системи моніторингу та формує звіти для Державної екологічної інспекції згідно з відповідними наказами. Підтримується інтеграція з EU IED та SAP EHS для управління інцидентами. Зберігання даних — 5 років.
Детальніше про налаштування EWMA-детектора
Параметр alpha (зазвичай 0.1) визначає швидкість адаптації baseline. Менше alpha — стабільніший baseline, але повільна реакція на дрейф датчика. Ми рекомендуємо alpha=0.05 для критичних показників.Що входить у роботу
- Інтеграція з існуючими газоаналізаторами (NOx, SO2, CO, CO2, HCl, VOC, PM).
- Налаштування дашборду реального часу (Grafana + InfluxDB).
- Детектор аномальних викидів з повідомленнями (Telegram, email).
- Предиктивна модель викидів (Gradient Boosting) для soft sensor.
- Модуль compliance-звітності (ГДВ, BAT AEL).
- Навчання персоналу (2 дні).
- Технічна підтримка 24/7.
Порівняння: традиційний CEMS vs CEMS з AI
| Параметр | Традиційний CEMS | CEMS з AI |
|---|---|---|
| Виявлення перевищень | Постфактум (години) | Real-time (секунди) |
| Soft sensor | Відсутній | Прогноз при відмові датчика, точність ±5% |
| Оптимізація режиму | Немає | Рекомендації щодо зниження емісії на 15-30% |
| Звітність | Ручна | Автоматична, під вимоги регуляторів |
| Вартість володіння | Висока через штрафи та простої | Нижча за рахунок предиктивного обслуговування |
Процес впровадження
- Аудит — обстеження існуючих аналізаторів, SCADA, нормативної бази (1-2 дні).
- Проектування — архітектура збору даних, вибір AI-моделей (3-5 днів).
- Інтеграція — підключення до CEMS, налаштування потокової передачі даних (5-7 днів).
- Розробка AI-модулів — детектор спайків, предиктивна модель, дашборд (7-10 днів).
- Тестування — валідація на історичних даних, навантажувальне тестування (3-5 днів).
- Деплой та навчання — запуск у продакшн, навчання операторів (2 дні).
Терміни та вартість
Базовий функціонал (CEMS + compliance + spike detector + дашборд) — від 3 до 4 тижнів. Повний цикл з предиктивною моделлю, оптимізацією режиму та інтеграцією з SAP EHS — від 2 до 3 місяців. Вартість розраховується індивідуально після аудиту. Замовте аудит вашої системи моніторингу викидів — ми оцінимо потенціал зниження штрафів та оптимізації. Отримайте консультацію щодо впровадження AI-CEMS на вашому підприємстві. Досвід впровадження на 20+ виробництвах — гарантуємо сертифіковане рішення.