学習目標
この章を読むことで、以下を習得できます：
- ✅ ストリーミングデータの処理とバッファリング技術を理解する
- ✅ リアルタイム統計モニタリング（移動窓）を実装できる
- ✅ オンライン機械学習モデル（増分学習）を構築できる
- ✅ Plotly Dashによるリアルタイムダッシュボードを作成できる
- ✅ データベースストリーミング（InfluxDB、TimescaleDB）を活用できる
- ✅ アラートシステムと異常検知の自動化を実装できる
- ✅ 本番環境での監視システムを設計・最適化できる
5.1 リアルタイム処理の基礎
リアルタイムデータ処理の課題
化学プロセスのリアルタイム監視では、以下の技術的課題に対応する必要があります：
| 課題 | 説明 | 解決策 |
|---|---|---|
| 低レイテンシ | ミリ秒〜秒単位での応答 | 効率的なアルゴリズム、バッファリング |
| 高スループット | 多数のセンサーからのデータ | 並列処理、分散システム |
| データ品質 | 欠損値、ノイズ、外れ値 | リアルタイムフィルタリング |
| メモリ制約 | 無限のデータストリーム | 移動窓、メモリ効率的なデータ構造 |
| モデル適応 | プロセス条件の変化 | オンライン学習、適応アルゴリズム |
5.2 ストリーミングデータ処理
コード例1: pandasによるストリーミングデータ処理の基礎
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from collections import deque
import time
class StreamingDataProcessor:
"""
リアルタイムデータストリーム処理クラス
Parameters:
-----------
window_size : int
移動窓のサイズ（データポイント数）
sampling_rate : float
サンプリングレート [Hz]
"""
def __init__(self, window_size=100, sampling_rate=1.0):
self.window_size = window_size
self.sampling_rate = sampling_rate
# データバッファ（高速なdequeを使用）
self.buffer = deque(maxlen=window_size)
# 統計量の履歴
self.timestamps = []
self.means = []
self.stds = []
self.mins = []
self.maxs = []
def add_data_point(self, value, timestamp=None):
"""
新しいデータポイントを追加
Parameters:
-----------
value : float
データ値
timestamp : float
タイムスタンプ（Noneの場合は現在時刻）
"""
if timestamp is None:
timestamp = time.time()
self.buffer.append(value)
# 統計量の計算（窓内のデータ）
if len(self.buffer) > 0:
data_array = np.array(self.buffer)
self.timestamps.append(timestamp)
self.means.append(np.mean(data_array))
self.stds.append(np.std(data_array))
self.mins.append(np.min(data_array))
self.maxs.append(np.max(data_array))
def get_statistics(self):
"""
現在の窓での統計量を取得
Returns:
--------
stats : dict
統計量の辞書
"""
if len(self.buffer) == 0:
return {}
data_array = np.array(self.buffer)
return {
'mean': np.mean(data_array),
'std': np.std(data_array),
'min': np.min(data_array),
'max': np.max(data_array),
'median': np.median(data_array),
'count': len(data_array)
}
def detect_anomaly(self, threshold_std=3.0):
"""
異常検知（移動平均 ± threshold_std × 標準偏差）
Parameters:
-----------
threshold_std : float
標準偏差の倍数（閾値）
Returns:
--------
is_anomaly : bool
異常かどうか
"""
if len(self.buffer) < 2:
return False
data_array = np.array(self.buffer)
mean = np.mean(data_array)
std = np.std(data_array)
latest_value = data_array[-1]
# 3シグマルール
is_anomaly = abs(latest_value - mean) > threshold_std * std
return is_anomaly
# シミュレーション：リアルタイムデータストリームの処理
np.random.seed(42)
# プロセッサの初期化
processor = StreamingDataProcessor(window_size=50, sampling_rate=10.0)
# データ生成とストリーミング処理
n_points = 300
anomaly_indices = [100, 150, 250] # 異常を注入する位置
for i in range(n_points):
# 通常データ + 異常
if i in anomaly_indices:
value = 180 + np.random.randn() * 15 # 異常に大きな変動
else:
value = 180 + 2 * np.sin(0.1 * i) + np.random.randn() * 1
# データポイントの追加
processor.add_data_point(value, timestamp=i)
# リアルタイム異常検知
if processor.detect_anomaly(threshold_std=3.0) and i > 50:
print(f"Anomaly detected at time {i}: value = {value:.2f}")
# 可視化
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
# 生データと移動平均
time_axis = processor.timestamps
axes[0].plot(time_axis, processor.means, color='#11998e', linewidth=2, label='Moving Mean')
axes[0].fill_between(time_axis,
np.array(processor.means) - np.array(processor.stds),
np.array(processor.means) + np.array(processor.stds),
color='#11998e', alpha=0.2, label='±1σ')
axes[0].plot(time_axis, processor.mins, color='red', linewidth=1, alpha=0.5, label='Min/Max')
axes[0].plot(time_axis, processor.maxs, color='red', linewidth=1, alpha=0.5)
# 異常位置をマーク
for idx in anomaly_indices:
if idx < len(time_axis):
axes[0].axvline(x=time_axis[idx], color='red', linestyle='--',
linewidth=2, alpha=0.7)
axes[0].set_xlabel('Time [s]')
axes[0].set_ylabel('Temperature [°C]')
axes[0].set_title('Real-time Streaming Data Processing (Moving Window)', fontweight='bold')
axes[0].legend()
axes[0].grid(alpha=0.3)
# 移動標準偏差（変動性の監視）
axes[1].plot(time_axis, processor.stds, color='orange', linewidth=2)
axes[1].axhline(y=np.mean(processor.stds) + 2*np.std(processor.stds),
color='red', linestyle='--', linewidth=2, label='Threshold (Mean + 2σ)')
axes[1].set_xlabel('Time [s]')
axes[1].set_ylabel('Moving Std Dev [°C]')
axes[1].set_title('Process Variability Monitoring', fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
# 最終統計
final_stats = processor.get_statistics()
print("\nFinal Window Statistics:")
print("=" * 50)
for key, value in final_stats.items():
print(f"{key:10s}: {value:10.4f}")解説: dequeを使った移動窓は、固定サイズのバッファを効率的に管理し、O(1)の時間複雑度で新しいデータを追加できます。リアルタイム異常検知は3シグマルールに基づき、統計的な外れ値を検出します。
コード例2: リアルタイム統計モニタリング（EWMA）
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
class EWMAMonitor:
"""
指数移動平均（EWMA）によるリアルタイムモニタリング
Parameters:
-----------
alpha : float
平滑化パラメータ（0 < alpha <= 1）
小さいほど過去のデータに重みを置く
"""
def __init__(self, alpha=0.1):
self.alpha = alpha
self.ewma = None
self.ewmvar = None # EWMA variance
# 履歴
self.values = []
self.ewma_values = []
self.ewmvar_values = []
self.control_limits_upper = []
self.control_limits_lower = []
def update(self, new_value):
"""
新しいデータポイントでEWMAを更新
Parameters:
-----------
new_value : float
新しいデータ値
"""
# 初期化
if self.ewma is None:
self.ewma = new_value
self.ewmvar = 0
else:
# EWMA更新
self.ewma = self.alpha * new_value + (1 - self.alpha) * self.ewma
# EWMA分散更新
deviation = new_value - self.ewma
self.ewmvar = self.alpha * (deviation ** 2) + (1 - self.alpha) * self.ewmvar
# 管理限界（UCL/LCL: Upper/Lower Control Limit）
ewm_std = np.sqrt(self.ewmvar)
ucl = self.ewma + 3 * ewm_std
lcl = self.ewma - 3 * ewm_std
# 履歴に保存
self.values.append(new_value)
self.ewma_values.append(self.ewma)
self.ewmvar_values.append(self.ewmvar)
self.control_limits_upper.append(ucl)
self.control_limits_lower.append(lcl)
def is_out_of_control(self, threshold=3.0):
"""
管理限界を超えているか判定
Returns:
--------
bool : 管理限界外かどうか
"""
if len(self.values) < 2:
return False
latest_value = self.values[-1]
ucl = self.control_limits_upper[-1]
lcl = self.control_limits_lower[-1]
return latest_value > ucl or latest_value < lcl
# シミュレーション
np.random.seed(42)
n_points = 300
# プロセスシフト（100時点、200時点で平均がシフト）
data = []
for i in range(n_points):
if i < 100:
value = 180 + np.random.randn() * 2
elif i < 200:
value = 185 + np.random.randn() * 2 # 平均シフト（+5°C）
else:
value = 180 + np.random.randn() * 3 # 変動増加
data.append(value)
# EWMAモニタリング
monitor = EWMAMonitor(alpha=0.2)
for i, value in enumerate(data):
monitor.update(value)
if monitor.is_out_of_control():
print(f"Out of control at time {i}: value = {value:.2f}")
# 可視化
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(len(data))
# EWMA管理図
axes[0].plot(time_axis, monitor.values, color='gray', alpha=0.5,
linewidth=1, label='Raw Data')
axes[0].plot(time_axis, monitor.ewma_values, color='#11998e',
linewidth=2.5, label='EWMA')
axes[0].plot(time_axis, monitor.control_limits_upper, color='red',
linestyle='--', linewidth=2, label='UCL/LCL')
axes[0].plot(time_axis, monitor.control_limits_lower, color='red',
linestyle='--', linewidth=2)
axes[0].fill_between(time_axis, monitor.control_limits_lower,
monitor.control_limits_upper,
color='green', alpha=0.1, label='Control Region')
# プロセスシフト位置をマーク
axes[0].axvline(x=100, color='blue', linestyle=':', linewidth=2, alpha=0.7)
axes[0].axvline(x=200, color='blue', linestyle=':', linewidth=2, alpha=0.7)
axes[0].set_xlabel('Time [samples]')
axes[0].set_ylabel('Temperature [°C]')
axes[0].set_title('EWMA Control Chart (α=0.2)', fontweight='bold')
axes[0].legend()
axes[0].grid(alpha=0.3)
# EWMA分散（変動性監視）
axes[1].plot(time_axis, np.sqrt(monitor.ewmvar_values), color='orange', linewidth=2)
axes[1].set_xlabel('Time [samples]')
axes[1].set_ylabel('EWMA Std Dev [°C]')
axes[1].set_title('Process Variability (EWMA Variance)', fontweight='bold')
axes[1].axvline(x=200, color='blue', linestyle=':', linewidth=2, alpha=0.7,
label='Variability Increase')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
print(f"\nEWMA Monitor Summary:")
print(f" Total samples: {len(data)}")
print(f" Final EWMA: {monitor.ewma:.2f}")
print(f" Final EWMA Std: {np.sqrt(monitor.ewmvar):.2f}")解説: EWMA（Exponentially Weighted Moving Average）は、小さなプロセスシフトを迅速に検出できるリアルタイム監視手法です。単純移動平均よりも最近のデータに重みを置き、レイテンシが低いため、化学プロセスの制御に適しています。
5.3 オンライン機械学習
コード例3: オンライン線形回帰（増分学習）
import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import SGDRegressor
from sklearn.preprocessing import StandardScaler
class OnlineLinearRegressor:
"""
オンライン学習による線形回帰モデル
Parameters:
-----------
learning_rate : str
学習率スケジュール ('constant', 'optimal', 'adaptive')
"""
def __init__(self, learning_rate='constant', eta0=0.01):
self.model = SGDRegressor(
learning_rate=learning_rate,
eta0=eta0,
random_state=42,
warm_start=True # 増分学習を有効化
)
self.scaler_X = StandardScaler()
self.scaler_y = StandardScaler()
self.is_fitted = False
# 履歴
self.predictions = []
self.actuals = []
self.errors = []
def partial_fit(self, X, y):
"""
新しいデータで部分的にフィット（増分学習）
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
特徴量
y : array-like, shape (n_samples,)
目的変数
"""
X = np.atleast_2d(X)
y = np.atleast_1d(y)
# スケーリング
if not self.is_fitted:
self.scaler_X.fit(X)
self.scaler_y.fit(y.reshape(-1, 1))
self.is_fitted = True
X_scaled = self.scaler_X.transform(X)
y_scaled = self.scaler_y.transform(y.reshape(-1, 1)).ravel()
# 増分学習
self.model.partial_fit(X_scaled, y_scaled)
def predict(self, X):
"""
予測
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
特徴量
Returns:
--------
y_pred : array-like
予測値
"""
if not self.is_fitted:
return np.zeros(X.shape[0])
X = np.atleast_2d(X)
X_scaled = self.scaler_X.transform(X)
y_scaled_pred = self.model.predict(X_scaled)
y_pred = self.scaler_y.inverse_transform(y_scaled_pred.reshape(-1, 1)).ravel()
return y_pred
def update_and_predict(self, X, y):
"""
予測してから学習（オンライン学習の典型パターン）
Parameters:
-----------
X : array-like
特徴量
y : float
実際の値
Returns:
--------
y_pred : float
予測値
"""
# 予測
y_pred = self.predict(X)
# 学習
self.partial_fit(X, y)
# 履歴更新
self.predictions.append(y_pred[0] if len(y_pred) > 0 else 0)
self.actuals.append(y)
self.errors.append(y - (y_pred[0] if len(y_pred) > 0 else 0))
return y_pred[0] if len(y_pred) > 0 else 0
# シミュレーション：オンライン回帰
np.random.seed(42)
n_samples = 500
# 時間変化する線形関係（概念ドリフト）
X_stream = []
y_stream = []
for i in range(n_samples):
# 特徴量
x1 = np.random.randn()
x2 = np.random.randn()
X_stream.append([x1, x2])
# 真の関係（時間とともに変化）
if i < 200:
# Phase 1: y = 3*x1 + 2*x2
y_true = 3*x1 + 2*x2
elif i < 400:
# Phase 2: y = 2*x1 + 4*x2 (関係性が変化)
y_true = 2*x1 + 4*x2
else:
# Phase 3: y = 1*x1 + 5*x2 (さらに変化)
y_true = 1*x1 + 5*x2
y_stream.append(y_true + np.random.randn() * 0.5) # ノイズ追加
X_stream = np.array(X_stream)
y_stream = np.array(y_stream)
# オンライン学習
online_model = OnlineLinearRegressor(learning_rate='constant', eta0=0.01)
# 初期学習（最初の10サンプル）
online_model.partial_fit(X_stream[:10], y_stream[:10])
# ストリーミング学習
for i in range(10, n_samples):
X_current = X_stream[i:i+1]
y_current = y_stream[i]
online_model.update_and_predict(X_current, y_current)
# 可視化
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(10, n_samples)
# 予測 vs 実際
axes[0].plot(time_axis, online_model.actuals, color='#11998e',
linewidth=1.5, label='Actual', alpha=0.7)
axes[0].plot(time_axis, online_model.predictions, color='orange',
linewidth=2, label='Predicted (Online)', linestyle='--')
axes[0].axvline(x=200, color='red', linestyle=':', linewidth=2, alpha=0.7,
label='Concept Drift')
axes[0].axvline(x=400, color='red', linestyle=':', linewidth=2, alpha=0.7)
axes[0].set_xlabel('Time [samples]')
axes[0].set_ylabel('Target Value')
axes[0].set_title('Online Learning with Concept Drift', fontweight='bold')
axes[0].legend()
axes[0].grid(alpha=0.3)
# 予測誤差（オンライン学習の適応性を評価）
axes[1].plot(time_axis, online_model.errors, color='purple', linewidth=1.5)
axes[1].axhline(y=0, color='red', linestyle='--', linewidth=1)
axes[1].fill_between(time_axis, -1, 1, color='green', alpha=0.1, label='Acceptable Error')
axes[1].axvline(x=200, color='red', linestyle=':', linewidth=2, alpha=0.7)
axes[1].axvline(x=400, color='red', linestyle=':', linewidth=2, alpha=0.7)
axes[1].set_xlabel('Time [samples]')
axes[1].set_ylabel('Prediction Error')
axes[1].set_title('Prediction Error Over Time', fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
# 各フェーズでのRMSE
phase1_rmse = np.sqrt(np.mean(np.array(online_model.errors[:190])**2))
phase2_rmse = np.sqrt(np.mean(np.array(online_model.errors[190:390])**2))
phase3_rmse = np.sqrt(np.mean(np.array(online_model.errors[390:])**2))
print(f"\nOnline Learning Performance:")
print(f" Phase 1 RMSE: {phase1_rmse:.4f}")
print(f" Phase 2 RMSE: {phase2_rmse:.4f}")
print(f" Phase 3 RMSE: {phase3_rmse:.4f}")解説: オンライン学習（増分学習）は、データが到着するたびにモデルを更新し、概念ドリフト（データ分布の変化）に適応します。バッチ学習と異なり、全データを保持する必要がなく、メモリ効率的です。
コード例4: オンラインPCAによる適応的異常検知
import numpy as np
import matplotlib.pyplot as plt
from sklearn.decomposition import IncrementalPCA
class OnlineAnomalyDetector:
"""
オンラインPCAによる適応的異常検知
Parameters:
-----------
n_components : int
主成分数
threshold : float
異常判定の閾値（SPEの標準偏差の倍数）
"""
def __init__(self, n_components=2, threshold=3.0):
self.n_components = n_components
self.threshold = threshold
self.ipca = IncrementalPCA(n_components=n_components)
self.is_fitted = False
self.spe_history = []
self.anomalies = []
def fit(self, X_initial):
"""
初期データでPCAをフィット
Parameters:
-----------
X_initial : array-like, shape (n_samples, n_features)
初期学習データ
"""
self.ipca.partial_fit(X_initial)
self.is_fitted = True
def detect(self, X):
"""
異常検知
Parameters:
-----------
X : array-like, shape (n_features,) or (n_samples, n_features)
監視対象データ
Returns:
--------
is_anomaly : bool or array
異常かどうか
spe : float or array
SPE (Squared Prediction Error)
"""
if not self.is_fitted:
return False, 0.0
X = np.atleast_2d(X)
# 主成分空間への投影と再構成
X_transformed = self.ipca.transform(X)
X_reconstructed = self.ipca.inverse_transform(X_transformed)
# SPE（二乗予測誤差）計算
spe = np.sum((X - X_reconstructed) ** 2, axis=1)
# 閾値判定
if len(self.spe_history) > 30:
spe_mean = np.mean(self.spe_history)
spe_std = np.std(self.spe_history)
threshold_value = spe_mean + self.threshold * spe_std
else:
threshold_value = np.inf # 初期段階は閾値なし
is_anomaly = spe > threshold_value
# 履歴更新
self.spe_history.extend(spe.tolist())
return is_anomaly[0] if len(is_anomaly) == 1 else is_anomaly, spe[0] if len(spe) == 1 else spe
# シミュレーション：多変量プロセスデータ
np.random.seed(42)
n_samples = 500
n_features = 10
# 正常データ生成（相関のある多変量データ）
mean_normal = np.zeros(n_features)
cov_normal = np.eye(n_features) + 0.3 * np.random.randn(n_features, n_features)
cov_normal = cov_normal @ cov_normal.T # 正定値対称行列
X_normal = np.random.multivariate_normal(mean_normal, cov_normal, size=400)
# 異常データの注入（大きな偏差）
anomaly_indices = [100, 150, 250, 350, 450]
X_stream = []
anomaly_labels = []
for i in range(n_samples):
if i < 400:
X_stream.append(X_normal[i])
anomaly_labels.append(0)
else:
# 末尾100サンプルは異常を注入
if i in anomaly_indices:
X_anomaly = X_normal[i-400] + np.random.randn(n_features) * 5
else:
X_anomaly = X_normal[i-400] + np.random.randn(n_features) * 0.5
X_stream.append(X_anomaly)
anomaly_labels.append(1 if i in anomaly_indices else 0)
X_stream = np.array(X_stream)
anomaly_labels = np.array(anomaly_labels)
# オンラインPCA異常検知
detector = OnlineAnomalyDetector(n_components=3, threshold=3.0)
detector.fit(X_stream[:50]) # 初期学習
detected_anomalies = []
spe_values = []
for i in range(50, n_samples):
X_current = X_stream[i:i+1]
is_anomaly, spe = detector.detect(X_current)
detected_anomalies.append(is_anomaly)
spe_values.append(spe)
# 定期的にモデル更新
if i % 10 == 0:
detector.ipca.partial_fit(X_stream[i-10:i])
# 可視化
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(50, n_samples)
# SPE値と閾値
axes[0].plot(time_axis, spe_values, color='#11998e', linewidth=1.5, label='SPE')
# 動的閾値
if len(detector.spe_history) > 30:
spe_mean = np.mean(detector.spe_history)
spe_std = np.std(detector.spe_history)
threshold_line = spe_mean + 3 * spe_std
axes[0].axhline(y=threshold_line, color='red', linestyle='--',
linewidth=2, label=f'Threshold (μ+3σ)')
# 真の異常位置をマーク
for idx in anomaly_indices:
if idx >= 50:
axes[0].axvline(x=idx, color='red', linestyle=':', linewidth=2, alpha=0.7)
axes[0].set_xlabel('Time [samples]')
axes[0].set_ylabel('SPE (Squared Prediction Error)')
axes[0].set_title('Online PCA Anomaly Detection', fontweight='bold')
axes[0].legend()
axes[0].set_yscale('log')
axes[0].grid(alpha=0.3)
# 検知結果の可視化
axes[1].scatter(time_axis[np.array(detected_anomalies)],
np.ones(np.sum(detected_anomalies)),
color='red', s=100, marker='x', label='Detected Anomalies', zorder=5)
axes[1].scatter(time_axis[~np.array(detected_anomalies)],
np.zeros(np.sum(~np.array(detected_anomalies))),
color='green', s=50, marker='o', alpha=0.3, label='Normal')
# 真の異常
true_anomaly_times = [idx for idx in anomaly_indices if idx >= 50]
axes[1].scatter(true_anomaly_times, [1.5]*len(true_anomaly_times),
color='blue', s=150, marker='*', label='True Anomalies', zorder=5)
axes[1].set_xlabel('Time [samples]')
axes[1].set_yticks([0, 1, 1.5])
axes[1].set_yticklabels(['Normal', 'Detected', 'True'])
axes[1].set_title('Anomaly Detection Results', fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
# 検知精度の評価
true_labels = anomaly_labels[50:]
detected_labels = np.array(detected_anomalies).astype(int)
from sklearn.metrics import precision_score, recall_score, f1_score
precision = precision_score(true_labels, detected_labels, zero_division=0)
recall = recall_score(true_labels, detected_labels, zero_division=0)
f1 = f1_score(true_labels, detected_labels, zero_division=0)
print(f"\nAnomaly Detection Performance:")
print(f" Precision: {precision:.4f}")
print(f" Recall: {recall:.4f}")
print(f" F1-Score: {f1:.4f}")解説: オンラインPCAは、多変量プロセスデータの主成分を逐次的に更新し、再構成誤差（SPE）に基づいて異常を検知します。プロセスの正常変動に適応しながら、異常パターンを検出できます。
5.4 リアルタイムダッシュボード
コード例5: Plotly Dashによるリアルタイムダッシュボード
import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import plotly.graph_objs as go
import numpy as np
import pandas as pd
from collections import deque
from datetime import datetime
import random
# データバッファ（最新1000ポイントを保持）
max_length = 1000
time_buffer = deque(maxlen=max_length)
temp_buffer = deque(maxlen=max_length)
pressure_buffer = deque(maxlen=max_length)
flow_buffer = deque(maxlen=max_length)
# Dashアプリケーション
app = dash.Dash(__name__)
app.layout = html.Div([
html.H1("Chemical Process Real-Time Dashboard",
style={'textAlign': 'center', 'color': '#11998e'}),
html.Div([
html.Div([
html.H3("Temperature Monitor", style={'textAlign': 'center'}),
dcc.Graph(id='temperature-graph')
], className='six columns'),
html.Div([
html.H3("Pressure Monitor", style={'textAlign': 'center'}),
dcc.Graph(id='pressure-graph')
], className='six columns'),
], className='row'),
html.Div([
html.Div([
html.H3("Flow Rate Monitor", style={'textAlign': 'center'}),
dcc.Graph(id='flow-graph')
], className='six columns'),
html.Div([
html.H3("Process Statistics", style={'textAlign': 'center'}),
html.Div(id='stats-display', style={
'fontSize': '18px',
'padding': '20px',
'backgroundColor': '#f0f0f0',
'borderRadius': '10px',
'marginTop': '20px'
})
], className='six columns'),
], className='row'),
# 更新インターバル（1秒ごと）
dcc.Interval(
id='interval-component',
interval=1000, # ミリ秒
n_intervals=0
)
])
@app.callback(
[Output('temperature-graph', 'figure'),
Output('pressure-graph', 'figure'),
Output('flow-graph', 'figure'),
Output('stats-display', 'children')],
[Input('interval-component', 'n_intervals')]
)
def update_dashboard(n):
"""
ダッシュボードのリアルタイム更新
"""
# 新しいデータを生成（実際にはセンサーデータを取得）
current_time = datetime.now()
temperature = 180 + 5 * np.sin(n * 0.1) + np.random.randn() * 1
pressure = 3.0 + 0.2 * np.cos(n * 0.08) + np.random.randn() * 0.1
flow_rate = 50 + 5 * np.random.randn()
# バッファに追加
time_buffer.append(current_time)
temp_buffer.append(temperature)
pressure_buffer.append(pressure)
flow_buffer.append(flow_rate)
# 温度グラフ
temp_figure = {
'data': [
go.Scatter(
x=list(time_buffer),
y=list(temp_buffer),
mode='lines',
name='Temperature',
line=dict(color='#11998e', width=2)
)
],
'layout': go.Layout(
xaxis=dict(title='Time'),
yaxis=dict(title='Temperature [°C]', range=[170, 190]),
margin=dict(l=50, r=50, t=20, b=50),
hovermode='closest'
)
}
# 圧力グラフ
pressure_figure = {
'data': [
go.Scatter(
x=list(time_buffer),
y=list(pressure_buffer),
mode='lines',
name='Pressure',
line=dict(color='#38ef7d', width=2)
)
],
'layout': go.Layout(
xaxis=dict(title='Time'),
yaxis=dict(title='Pressure [bar]', range=[2.5, 3.5]),
margin=dict(l=50, r=50, t=20, b=50),
hovermode='closest'
)
}
# 流量グラフ
flow_figure = {
'data': [
go.Scatter(
x=list(time_buffer),
y=list(flow_buffer),
mode='lines',
name='Flow Rate',
line=dict(color='orange', width=2)
)
],
'layout': go.Layout(
xaxis=dict(title='Time'),
yaxis=dict(title='Flow Rate [L/h]', range=[30, 70]),
margin=dict(l=50, r=50, t=20, b=50),
hovermode='closest'
)
}
# 統計情報
if len(temp_buffer) > 0:
stats = html.Div([
html.P(f"Temperature: {temperature:.2f} °C"),
html.P(f" Mean: {np.mean(temp_buffer):.2f} °C"),
html.P(f" Std Dev: {np.std(temp_buffer):.2f} °C"),
html.Br(),
html.P(f"Pressure: {pressure:.2f} bar"),
html.P(f" Mean: {np.mean(pressure_buffer):.2f} bar"),
html.P(f" Std Dev: {np.std(pressure_buffer):.2f} bar"),
html.Br(),
html.P(f"Flow Rate: {flow_rate:.2f} L/h"),
html.P(f" Mean: {np.mean(flow_buffer):.2f} L/h"),
html.P(f" Std Dev: {np.std(flow_buffer):.2f} L/h"),
])
else:
stats = html.P("Waiting for data...")
return temp_figure, pressure_figure, flow_figure, stats
# 注意：このコードはJupyter Notebookでは動作しません
# ターミナルで実行してください：python dashboard.py
if __name__ == '__main__':
print("Starting Real-Time Dashboard...")
print("Open browser and navigate to: http://127.0.0.1:8050/")
app.run_server(debug=True, port=8050)実行方法:
# dashboard.pyとして保存後、実行
pip install dash plotly pandas
python dashboard.py
# ブラウザで http://127.0.0.1:8050/ を開く解説: Plotly Dashは、Pythonでインタラクティブなリアルタイムダッシュボードを構築できるフレームワークです。dcc.Intervalにより定期的に更新され、ブラウザで動的なプロセス監視が可能です。本番環境では、DCSやMESからのデータストリームと統合します。
5.5 データベースストリーミング
コード例6: InfluxDBによる時系列データストリーミング
# InfluxDB 2.x用のクライアント
from influxdb_client import InfluxDBClient, Point, WriteOptions
from influxdb_client.client.write_api import SYNCHRONOUS
import numpy as np
from datetime import datetime, timedelta
import time
# 注意：このコードを実行するには、InfluxDBサーバーが必要です
# Dockerで簡単に起動可能：
# docker run -d -p 8086:8086 influxdb:2.0
class InfluxDBStreamer:
"""
InfluxDBへのリアルタイムデータストリーミング
Parameters:
-----------
url : str
InfluxDBのURL
token : str
認証トークン
org : str
組織名
bucket : str
バケット名
"""
def __init__(self, url="http://localhost:8086", token="your-token",
org="your-org", bucket="process-data"):
self.url = url
self.token = token
self.org = org
self.bucket = bucket
# クライアント接続
try:
self.client = InfluxDBClient(url=url, token=token, org=org)
self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
self.query_api = self.client.query_api()
print("Connected to InfluxDB successfully")
except Exception as e:
print(f"Failed to connect to InfluxDB: {e}")
self.client = None
def write_data_point(self, measurement, tags, fields, timestamp=None):
"""
データポイントを書き込み
Parameters:
-----------
measurement : str
測定名（例: "temperature", "pressure"）
tags : dict
タグ（メタデータ、例: {"sensor_id": "T101", "location": "reactor"}）
fields : dict
フィールド（実データ、例: {"value": 180.5}）
timestamp : datetime
タイムスタンプ（Noneの場合は現在時刻）
"""
if self.client is None:
print("InfluxDB client not initialized")
return
if timestamp is None:
timestamp = datetime.utcnow()
# データポイント作成
point = Point(measurement)
for key, value in tags.items():
point = point.tag(key, value)
for key, value in fields.items():
point = point.field(key, value)
point = point.time(timestamp)
# 書き込み
try:
self.write_api.write(bucket=self.bucket, org=self.org, record=point)
except Exception as e:
print(f"Failed to write data: {e}")
def query_recent_data(self, measurement, time_range="-1h"):
"""
最近のデータをクエリ
Parameters:
-----------
measurement : str
測定名
time_range : str
時間範囲（例: "-1h", "-30m", "-10s"）
Returns:
--------
data : list
クエリ結果
"""
if self.client is None:
return []
# Fluxクエリ
query = f'''
from(bucket: "{self.bucket}")
|> range(start: {time_range})
|> filter(fn: (r) => r["_measurement"] == "{measurement}")
'''
try:
result = self.query_api.query(org=self.org, query=query)
data = []
for table in result:
for record in table.records:
data.append({
'time': record.get_time(),
'value': record.get_value(),
'field': record.get_field(),
'tags': record.values
})
return data
except Exception as e:
print(f"Failed to query data: {e}")
return []
def close(self):
"""クライアントを閉じる"""
if self.client:
self.client.close()
# 使用例（シミュレーション）
def simulate_streaming():
"""
プロセスデータのストリーミングシミュレーション
"""
# InfluxDBストリーマー初期化
# 注意：実際の環境に合わせてパラメータを変更してください
streamer = InfluxDBStreamer(
url="http://localhost:8086",
token="your-influxdb-token", # 実際のトークンを設定
org="my-org",
bucket="process-data"
)
print("Starting data streaming simulation...")
print("Press Ctrl+C to stop")
try:
for i in range(100):
# センサーデータ生成
temperature = 180 + 5 * np.sin(i * 0.1) + np.random.randn()
pressure = 3.0 + 0.2 * np.cos(i * 0.08) + np.random.randn() * 0.1
flow_rate = 50 + 5 * np.random.randn()
# データ書き込み
streamer.write_data_point(
measurement="temperature",
tags={"sensor_id": "T101", "unit": "reactor"},
fields={"value": temperature},
timestamp=datetime.utcnow()
)
streamer.write_data_point(
measurement="pressure",
tags={"sensor_id": "P201", "unit": "reactor"},
fields={"value": pressure},
timestamp=datetime.utcnow()
)
streamer.write_data_point(
measurement="flow_rate",
tags={"sensor_id": "F301", "unit": "feed"},
fields={"value": flow_rate},
timestamp=datetime.utcnow()
)
print(f"Sample {i+1}: T={temperature:.2f}°C, P={pressure:.2f}bar, F={flow_rate:.2f}L/h")
time.sleep(1) # 1秒間隔
except KeyboardInterrupt:
print("\nStreaming stopped by user")
finally:
# 最近のデータをクエリ
print("\nQuerying recent temperature data...")
recent_data = streamer.query_recent_data("temperature", time_range="-5m")
print(f"Retrieved {len(recent_data)} data points")
if len(recent_data) > 0:
print("\nSample data:")
for i, point in enumerate(recent_data[:5]):
print(f" {i+1}. Time: {point['time']}, Value: {point['value']:.2f}")
streamer.close()
# 注意：InfluxDBサーバーが起動している場合のみ実行可能
# simulate_streaming()
print("""
InfluxDB Streaming Example Code
To run this example:
1. Install InfluxDB: docker run -d -p 8086:8086 influxdb:2.0
2. Create an API token in InfluxDB UI (http://localhost:8086)
3. Update the connection parameters in the code
4. Install client: pip install influxdb-client
5. Run: python influxdb_streaming.py
Key Features:
- Real-time data ingestion
- Tag-based indexing (sensor_id, location, etc.)
- Powerful time-series queries (Flux language)
- Automatic data downsampling and retention policies
- Integration with Grafana for visualization
""")解説: InfluxDBは時系列データに特化したデータベースで、高速な書き込みとクエリが可能です。タグによるメタデータ管理、自動ダウンサンプリング、Grafanaとの統合により、産業用リアルタイム監視システムの基盤として広く使用されています。
コード例7: WebSocketによるリアルタイムデータ配信
import asyncio
import websockets
import json
import numpy as np
from datetime import datetime
# WebSocketサーバー（データ配信側）
class ProcessDataServer:
"""
WebSocketサーバーでプロセスデータを配信
"""
def __init__(self, host='localhost', port=8765):
self.host = host
self.port = port
self.clients = set()
async def register_client(self, websocket):
"""クライアント登録"""
self.clients.add(websocket)
print(f"Client connected: {websocket.remote_address}")
async def unregister_client(self, websocket):
"""クライアント登録解除"""
self.clients.remove(websocket)
print(f"Client disconnected: {websocket.remote_address}")
async def send_data_to_clients(self, data):
"""全クライアントにデータを送信"""
if self.clients:
message = json.dumps(data)
await asyncio.gather(
*[client.send(message) for client in self.clients],
return_exceptions=True
)
async def handler(self, websocket, path):
"""
クライアント接続ハンドラ
"""
await self.register_client(websocket)
try:
# クライアントからのメッセージを待機（接続維持）
async for message in websocket:
print(f"Received from client: {message}")
except websockets.exceptions.ConnectionClosed:
pass
finally:
await self.unregister_client(websocket)
async def generate_and_broadcast_data(self):
"""
定期的にデータを生成してブロードキャスト
"""
iteration = 0
while True:
# プロセスデータ生成
temperature = 180 + 5 * np.sin(iteration * 0.1) + np.random.randn()
pressure = 3.0 + 0.2 * np.cos(iteration * 0.08) + np.random.randn() * 0.1
flow_rate = 50 + 5 * np.random.randn()
data = {
'timestamp': datetime.now().isoformat(),
'temperature': round(temperature, 2),
'pressure': round(pressure, 2),
'flow_rate': round(flow_rate, 2),
'iteration': iteration
}
# 全クライアントに送信
await self.send_data_to_clients(data)
iteration += 1
await asyncio.sleep(1) # 1秒間隔
async def start(self):
"""サーバー起動"""
# WebSocketサーバー開始
server = await websockets.serve(self.handler, self.host, self.port)
print(f"WebSocket server started on ws://{self.host}:{self.port}")
# データ生成・配信タスク開始
await self.generate_and_broadcast_data()
# WebSocketクライアント（データ受信側）
class ProcessDataClient:
"""
WebSocketクライアントでデータを受信
"""
def __init__(self, uri='ws://localhost:8765'):
self.uri = uri
self.data_buffer = []
async def receive_data(self):
"""
データを受信して処理
"""
async with websockets.connect(self.uri) as websocket:
print(f"Connected to server: {self.uri}")
try:
async for message in websocket:
data = json.loads(message)
self.data_buffer.append(data)
# リアルタイム処理
print(f"Received: T={data['temperature']}°C, "
f"P={data['pressure']}bar, "
f"F={data['flow_rate']}L/h")
# 異常検知の例
if data['temperature'] > 185:
print(f" WARNING: High temperature detected!")
# バッファサイズ制限
if len(self.data_buffer) > 1000:
self.data_buffer.pop(0)
except websockets.exceptions.ConnectionClosed:
print("Connection closed by server")
# サーバー起動スクリプト
async def run_server():
server = ProcessDataServer(host='localhost', port=8765)
await server.start()
# クライアント起動スクリプト
async def run_client():
client = ProcessDataClient(uri='ws://localhost:8765')
await client.receive_data()
# 使用例
print("""
WebSocket Real-Time Data Streaming Example
Server Side:
-----------
python
import asyncio
from your_module import run_server
asyncio.run(run_server())
Client Side:
-----------
python
import asyncio
from your_module import run_client
asyncio.run(run_client())
Key Features:
- Low-latency bidirectional communication
- Multiple clients can connect simultaneously
- Automatic reconnection handling
- JSON data serialization
- Real-time anomaly detection on client side
Installation:
pip install websockets
""")
# 実行例（コメント解除して使用）
# asyncio.run(run_server()) # サーバー起動
# asyncio.run(run_client()) # クライアント起動解説: WebSocketは双方向通信プロトコルで、HTTPよりも低レイテンシでリアルタイムデータを配信できます。DCSからのデータをWebSocketサーバーで配信し、複数のクライアント（ダッシュボード、分析ツール）が同時に受信できます。
5.6 アラートシステムと自動化
コード例8: ルールベースアラートシステム
import numpy as np
import pandas as pd
from datetime import datetime
from enum import Enum
import smtplib
from email.mime.text import MIMEText
class AlertSeverity(Enum):
"""アラート重要度"""
INFO = 1
WARNING = 2
CRITICAL = 3
EMERGENCY = 4
class Alert:
"""アラート情報"""
def __init__(self, severity, message, value, threshold, timestamp=None):
self.severity = severity
self.message = message
self.value = value
self.threshold = threshold
self.timestamp = timestamp if timestamp else datetime.now()
def __str__(self):
return (f"[{self.severity.name}] {self.timestamp.strftime('%Y-%m-%d %H:%M:%S')} - "
f"{self.message} (Value: {self.value:.2f}, Threshold: {self.threshold:.2f})")
class ProcessAlertSystem:
"""
プロセス監視アラートシステム
Parameters:
-----------
rules : list of dict
アラートルールのリスト
例: {'variable': 'temperature', 'condition': '>', 'threshold': 185, 'severity': AlertSeverity.WARNING}
"""
def __init__(self, rules):
self.rules = rules
self.alerts = []
self.alert_counts = {severity: 0 for severity in AlertSeverity}
def evaluate_rules(self, data):
"""
ルールを評価してアラートを生成
Parameters:
-----------
data : dict
監視データ（変数名: 値）
Returns:
--------
alerts : list of Alert
生成されたアラートのリスト
"""
current_alerts = []
for rule in self.rules:
variable = rule['variable']
condition = rule['condition']
threshold = rule['threshold']
severity = rule['severity']
if variable not in data:
continue
value = data[variable]
# 条件評価
triggered = False
if condition == '>':
triggered = value > threshold
elif condition == '<':
triggered = value < threshold
elif condition == '>=':
triggered = value >= threshold
elif condition == '<=':
triggered = value <= threshold
elif condition == '==':
triggered = abs(value - threshold) < 1e-6
elif condition == '!=':
triggered = abs(value - threshold) >= 1e-6
if triggered:
message = f"{variable} {condition} {threshold}"
alert = Alert(severity, message, value, threshold)
current_alerts.append(alert)
self.alerts.append(alert)
self.alert_counts[severity] += 1
return current_alerts
def send_alert_notification(self, alert):
"""
アラート通知を送信（メール、SMS、等）
Parameters:
-----------
alert : Alert
送信するアラート
"""
print(f"\n*** ALERT NOTIFICATION ***")
print(alert)
print("*" * 50)
# 実際の実装では、メール送信やSMS送信を行う
# if alert.severity == AlertSeverity.CRITICAL:
# self._send_email(alert)
# self._send_sms(alert)
def _send_email(self, alert):
"""メール送信（実装例）"""
# 注意：実際の環境に合わせて設定してください
smtp_server = "smtp.example.com"
smtp_port = 587
sender = "alerts@example.com"
recipient = "operator@example.com"
password = "your-password"
msg = MIMEText(str(alert))
msg['Subject'] = f"[{alert.severity.name}] Process Alert"
msg['From'] = sender
msg['To'] = recipient
try:
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(sender, password)
server.send_message(msg)
print(f"Email sent to {recipient}")
except Exception as e:
print(f"Failed to send email: {e}")
def get_alert_summary(self):
"""アラートサマリーを取得"""
return {
'total_alerts': len(self.alerts),
'by_severity': self.alert_counts,
'recent_alerts': self.alerts[-10:] if len(self.alerts) > 0 else []
}
# シミュレーション
np.random.seed(42)
# アラートルール定義
alert_rules = [
{'variable': 'temperature', 'condition': '>', 'threshold': 185, 'severity': AlertSeverity.WARNING},
{'variable': 'temperature', 'condition': '>', 'threshold': 190, 'severity': AlertSeverity.CRITICAL},
{'variable': 'temperature', 'condition': '<', 'threshold': 170, 'severity': AlertSeverity.WARNING},
{'variable': 'pressure', 'condition': '>', 'threshold': 3.5, 'severity': AlertSeverity.WARNING},
{'variable': 'pressure', 'condition': '<', 'threshold': 2.5, 'severity': AlertSeverity.WARNING},
{'variable': 'flow_rate', 'condition': '<', 'threshold': 40, 'severity': AlertSeverity.INFO},
]
# アラートシステム初期化
alert_system = ProcessAlertSystem(rules=alert_rules)
# データストリーム処理
print("Starting Process Monitoring...")
print("=" * 70)
for i in range(100):
# プロセスデータ生成（時々異常値を含む）
if i in [25, 50, 75]:
# 異常データ
temperature = 192 + np.random.randn()
pressure = 3.8 + np.random.randn() * 0.1
else:
# 正常データ
temperature = 180 + 5 * np.sin(i * 0.1) + np.random.randn() * 2
pressure = 3.0 + 0.2 * np.cos(i * 0.08) + np.random.randn() * 0.1
flow_rate = 50 + 5 * np.random.randn()
data = {
'temperature': temperature,
'pressure': pressure,
'flow_rate': flow_rate
}
# ルール評価
current_alerts = alert_system.evaluate_rules(data)
# アラート通知
for alert in current_alerts:
alert_system.send_alert_notification(alert)
# アラートサマリー
summary = alert_system.get_alert_summary()
print("\n\n" + "=" * 70)
print("ALERT SUMMARY")
print("=" * 70)
print(f"Total Alerts: {summary['total_alerts']}")
print("\nBy Severity:")
for severity, count in summary['by_severity'].items():
print(f" {severity.name:12s}: {count}")
print("\nRecent Alerts (last 10):")
for alert in summary['recent_alerts']:
print(f" {alert}")解説: ルールベースのアラートシステムは、閾値や条件に基づいてリアルタイムに異常を検知し、重要度に応じて通知します。複数の通知チャネル（メール、SMS、SCADA画面）への統合が可能です。
コード例9: 機械学習ベースの異常検知とアラート
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
class MLAnomalyDetector:
"""
機械学習ベースの異常検知システム
Parameters:
-----------
contamination : float
異常の割合（0.0〜0.5）
"""
def __init__(self, contamination=0.05):
self.model = IsolationForest(
contamination=contamination,
random_state=42,
n_estimators=100
)
self.scaler = StandardScaler()
self.is_fitted = False
self.anomaly_scores = []
self.predictions = []
def fit(self, X_normal):
"""
正常データで学習
Parameters:
-----------
X_normal : array-like, shape (n_samples, n_features)
正常データ
"""
X_scaled = self.scaler.fit_transform(X_normal)
self.model.fit(X_scaled)
self.is_fitted = True
def predict(self, X):
"""
異常検知
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
データ
Returns:
--------
is_anomaly : array, shape (n_samples,)
異常フラグ（1: 異常, -1: 正常）
anomaly_score : array, shape (n_samples,)
異常スコア（低いほど異常）
"""
if not self.is_fitted:
return np.zeros(X.shape[0]), np.zeros(X.shape[0])
X_scaled = self.scaler.transform(X)
predictions = self.model.predict(X_scaled) # 1: normal, -1: anomaly
anomaly_scores = self.model.score_samples(X_scaled) # 異常スコア（低いほど異常）
self.predictions.extend(predictions.tolist())
self.anomaly_scores.extend(anomaly_scores.tolist())
return predictions, anomaly_scores
# シミュレーション
np.random.seed(42)
# 正常データで学習
n_train = 500
X_train = np.random.multivariate_normal(
mean=[180, 3.0, 50],
cov=[[4, 0.1, 0.2], [0.1, 0.04, 0.05], [0.2, 0.05, 25]],
size=n_train
)
# 異常検知器の学習
detector = MLAnomalyDetector(contamination=0.05)
detector.fit(X_train)
# テストデータ（正常 + 異常）
n_test = 300
X_test = []
true_labels = []
for i in range(n_test):
if i % 50 == 49: # 50サンプルごとに異常を注入
# 異常データ
X_test.append([195, 3.8, 35]) # 温度高、圧力高、流量低
true_labels.append(-1)
else:
# 正常データ
sample = np.random.multivariate_normal(
mean=[180, 3.0, 50],
cov=[[4, 0.1, 0.2], [0.1, 0.04, 0.05], [0.2, 0.05, 25]]
)
X_test.append(sample)
true_labels.append(1)
X_test = np.array(X_test)
true_labels = np.array(true_labels)
# リアルタイム異常検知
print("Real-Time ML Anomaly Detection")
print("=" * 70)
detected_anomalies = []
for i in range(n_test):
X_current = X_test[i:i+1]
prediction, score = detector.predict(X_current)
if prediction[0] == -1:
print(f"Time {i}: ANOMALY DETECTED (Score: {score[0]:.3f})")
print(f" Data: T={X_current[0,0]:.2f}°C, P={X_current[0,1]:.2f}bar, F={X_current[0,2]:.2f}L/h")
detected_anomalies.append(i)
# 可視化
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# 異常スコアの時系列
axes[0, 0].plot(detector.anomaly_scores, color='#11998e', linewidth=1.5)
axes[0, 0].axhline(y=np.percentile(detector.anomaly_scores, 5), color='red',
linestyle='--', linewidth=2, label='Anomaly Threshold (5%ile)')
for idx in detected_anomalies:
axes[0, 0].axvline(x=idx, color='red', linestyle=':', linewidth=1, alpha=0.5)
axes[0, 0].set_xlabel('Time [samples]')
axes[0, 0].set_ylabel('Anomaly Score')
axes[0, 0].set_title('Anomaly Score Over Time', fontweight='bold')
axes[0, 0].legend()
axes[0, 0].grid(alpha=0.3)
# 温度 vs 圧力（異常の可視化）
normal_idx = np.array(detector.predictions) == 1
anomaly_idx = np.array(detector.predictions) == -1
axes[0, 1].scatter(X_test[normal_idx, 0], X_test[normal_idx, 1],
color='green', alpha=0.5, label='Normal', s=30)
axes[0, 1].scatter(X_test[anomaly_idx, 0], X_test[anomaly_idx, 1],
color='red', alpha=0.8, label='Anomaly', s=100, marker='x', linewidths=2)
axes[0, 1].set_xlabel('Temperature [°C]')
axes[0, 1].set_ylabel('Pressure [bar]')
axes[0, 1].set_title('Anomalies in Temperature-Pressure Space', fontweight='bold')
axes[0, 1].legend()
axes[0, 1].grid(alpha=0.3)
# 圧力 vs 流量
axes[1, 0].scatter(X_test[normal_idx, 1], X_test[normal_idx, 2],
color='green', alpha=0.5, label='Normal', s=30)
axes[1, 0].scatter(X_test[anomaly_idx, 1], X_test[anomaly_idx, 2],
color='red', alpha=0.8, label='Anomaly', s=100, marker='x', linewidths=2)
axes[1, 0].set_xlabel('Pressure [bar]')
axes[1, 0].set_ylabel('Flow Rate [L/h]')
axes[1, 0].set_title('Anomalies in Pressure-Flow Space', fontweight='bold')
axes[1, 0].legend()
axes[1, 0].grid(alpha=0.3)
# 異常スコアのヒストグラム
axes[1, 1].hist(np.array(detector.anomaly_scores)[normal_idx], bins=30,
color='green', alpha=0.6, label='Normal', edgecolor='black')
axes[1, 1].hist(np.array(detector.anomaly_scores)[anomaly_idx], bins=30,
color='red', alpha=0.6, label='Anomaly', edgecolor='black')
axes[1, 1].set_xlabel('Anomaly Score')
axes[1, 1].set_ylabel('Frequency')
axes[1, 1].set_title('Distribution of Anomaly Scores', fontweight='bold')
axes[1, 1].legend()
axes[1, 1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
# 検知性能の評価
from sklearn.metrics import classification_report
predicted_labels = np.array(detector.predictions)
print("\n\nAnomaly Detection Performance:")
print("=" * 70)
print(classification_report(true_labels, predicted_labels,
target_names=['Anomaly', 'Normal']))解説: 機械学習ベースの異常検知は、ルールベースでは捉えられない複雑な異常パターンを検出できます。Isolation Forestは教師なし学習で、正常データのみから学習し、外れ値を効率的に検出します。プロセスの非線形な関係や多変量相関も考慮できます。
5.7 本番環境への展開
コード例10: 本番監視システムの統合実装
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from collections import deque
import logging
import threading
import time
# ロギング設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('process_monitoring.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger('ProcessMonitor')
class ProductionMonitoringSystem:
"""
本番環境用プロセス監視システム
Parameters:
-----------
sampling_interval : float
サンプリング間隔 [秒]
buffer_size : int
データバッファサイズ
alert_threshold : dict
アラート閾値の辞書
"""
def __init__(self, sampling_interval=1.0, buffer_size=1000, alert_threshold=None):
self.sampling_interval = sampling_interval
self.buffer_size = buffer_size
self.alert_threshold = alert_threshold or {}
# データバッファ
self.data_buffer = {
'timestamp': deque(maxlen=buffer_size),
'temperature': deque(maxlen=buffer_size),
'pressure': deque(maxlen=buffer_size),
'flow_rate': deque(maxlen=buffer_size)
}
# 統計情報
self.statistics = {}
self.alert_count = 0
self.uptime_start = datetime.now()
# 制御フラグ
self.is_running = False
self.monitoring_thread = None
logger.info("Production Monitoring System initialized")
def read_sensor_data(self):
"""
センサーデータを読み取る
実際の実装では、DCS、PLC、OPC-UAサーバー等から読み取り
Returns:
--------
data : dict
センサーデータ
"""
# シミュレーション用のダミーデータ
# 実環境では、OPC-UA、Modbus、MQTT等で実データを取得
timestamp = datetime.now()
temperature = 180 + 5 * np.sin(time.time() * 0.1) + np.random.randn()
pressure = 3.0 + 0.2 * np.cos(time.time() * 0.08) + np.random.randn() * 0.1
flow_rate = 50 + 5 * np.random.randn()
return {
'timestamp': timestamp,
'temperature': temperature,
'pressure': pressure,
'flow_rate': flow_rate
}
def update_buffer(self, data):
"""
データバッファを更新
Parameters:
-----------
data : dict
新しいセンサーデータ
"""
for key, value in data.items():
self.data_buffer[key].append(value)
def calculate_statistics(self):
"""
現在のデータバッファから統計量を計算
Returns:
--------
stats : dict
統計量の辞書
"""
stats = {}
for variable in ['temperature', 'pressure', 'flow_rate']:
data_array = np.array(self.data_buffer[variable])
if len(data_array) > 0:
stats[variable] = {
'mean': np.mean(data_array),
'std': np.std(data_array),
'min': np.min(data_array),
'max': np.max(data_array),
'latest': data_array[-1]
}
self.statistics = stats
return stats
def check_alerts(self, data):
"""
アラート条件をチェック
Parameters:
-----------
data : dict
現在のセンサーデータ
Returns:
--------
alerts : list
発生したアラートのリスト
"""
alerts = []
# 温度アラート
if data['temperature'] > self.alert_threshold.get('temperature_high', 185):
alert_msg = f"HIGH TEMPERATURE: {data['temperature']:.2f}°C"
alerts.append(alert_msg)
logger.warning(alert_msg)
if data['temperature'] < self.alert_threshold.get('temperature_low', 170):
alert_msg = f"LOW TEMPERATURE: {data['temperature']:.2f}°C"
alerts.append(alert_msg)
logger.warning(alert_msg)
# 圧力アラート
if data['pressure'] > self.alert_threshold.get('pressure_high', 3.5):
alert_msg = f"HIGH PRESSURE: {data['pressure']:.2f}bar"
alerts.append(alert_msg)
logger.warning(alert_msg)
if data['pressure'] < self.alert_threshold.get('pressure_low', 2.5):
alert_msg = f"LOW PRESSURE: {data['pressure']:.2f}bar"
alerts.append(alert_msg)
logger.warning(alert_msg)
# 流量アラート
if data['flow_rate'] < self.alert_threshold.get('flow_rate_low', 40):
alert_msg = f"LOW FLOW RATE: {data['flow_rate']:.2f}L/h"
alerts.append(alert_msg)
logger.warning(alert_msg)
self.alert_count += len(alerts)
return alerts
def monitoring_loop(self):
"""
メイン監視ループ
"""
logger.info("Monitoring loop started")
while self.is_running:
try:
# センサーデータ読み取り
data = self.read_sensor_data()
# バッファ更新
self.update_buffer(data)
# 統計量計算
self.calculate_statistics()
# アラートチェック
alerts = self.check_alerts(data)
# データログ（定期的にDBに保存する等）
if len(self.data_buffer['timestamp']) % 60 == 0: # 1分ごと
logger.info(f"System running normally. "
f"T={data['temperature']:.2f}°C, "
f"P={data['pressure']:.2f}bar, "
f"F={data['flow_rate']:.2f}L/h")
# 次のサンプリングまで待機
time.sleep(self.sampling_interval)
except Exception as e:
logger.error(f"Error in monitoring loop: {e}", exc_info=True)
time.sleep(5) # エラー時は5秒待機して再試行
def start(self):
"""監視システムを開始"""
if not self.is_running:
self.is_running = True
self.monitoring_thread = threading.Thread(target=self.monitoring_loop, daemon=True)
self.monitoring_thread.start()
logger.info("Production monitoring system started")
def stop(self):
"""監視システムを停止"""
if self.is_running:
self.is_running = False
if self.monitoring_thread:
self.monitoring_thread.join(timeout=5)
logger.info("Production monitoring system stopped")
def get_system_status(self):
"""
システムステータスを取得
Returns:
--------
status : dict
システムステータス情報
"""
uptime = datetime.now() - self.uptime_start
return {
'is_running': self.is_running,
'uptime': str(uptime),
'buffer_size': len(self.data_buffer['timestamp']),
'total_alerts': self.alert_count,
'statistics': self.statistics,
'latest_data': {
'temperature': self.data_buffer['temperature'][-1] if len(self.data_buffer['temperature']) > 0 else None,
'pressure': self.data_buffer['pressure'][-1] if len(self.data_buffer['pressure']) > 0 else None,
'flow_rate': self.data_buffer['flow_rate'][-1] if len(self.data_buffer['flow_rate']) > 0 else None
}
}
# 使用例
if __name__ == "__main__":
# 監視システムの初期化
alert_thresholds = {
'temperature_high': 185,
'temperature_low': 170,
'pressure_high': 3.5,
'pressure_low': 2.5,
'flow_rate_low': 40
}
monitor = ProductionMonitoringSystem(
sampling_interval=1.0,
buffer_size=1000,
alert_threshold=alert_thresholds
)
# 監視開始
monitor.start()
try:
# 60秒間実行
for i in range(60):
time.sleep(1)
# 10秒ごとにステータス表示
if i % 10 == 0:
status = monitor.get_system_status()
print(f"\n--- System Status (t={i}s) ---")
print(f"Running: {status['is_running']}")
print(f"Uptime: {status['uptime']}")
print(f"Buffer: {status['buffer_size']}/{monitor.buffer_size}")
print(f"Alerts: {status['total_alerts']}")
if status['latest_data']['temperature']:
print(f"Latest Data:")
print(f" T: {status['latest_data']['temperature']:.2f}°C")
print(f" P: {status['latest_data']['pressure']:.2f}bar")
print(f" F: {status['latest_data']['flow_rate']:.2f}L/h")
except KeyboardInterrupt:
print("\nShutting down...")
finally:
# 監視停止
monitor.stop()
print("\n" + "=" * 70)
print("Production Monitoring System - Final Status")
print("=" * 70)
final_status = monitor.get_system_status()
print(f"Total uptime: {final_status['uptime']}")
print(f"Total alerts: {final_status['total_alerts']}")
print(f"Total samples: {final_status['buffer_size']}")解説: 本番環境の監視システムは、堅牢性、ロギング、エラーハンドリング、スレッド管理が重要です。この実装は、DCS/PLCからのデータ取得、リアルタイム解析、アラート生成、ログ記録を統合し、24時間365日の連続運用に対応します。
5.8 本章のまとめ
学んだこと
- ストリーミングデータ処理
- 移動窓（deque）による効率的なバッファリング
- EWMA（指数移動平均）によるリアルタイム統計監視
- オンライン機械学習
- 増分学習（SGDRegressor）による適応的モデル
- オンラインPCAによる多変量異常検知
- リアルタイム可視化
- Plotly Dashによるインタラクティブダッシュボード
- 定期更新と動的グラフ生成
- データベース統合
- InfluxDBによる時系列データストリーミング
- WebSocketによる低レイテンシ配信
- アラートシステム
- ルールベース異常検知
- 機械学習ベース異常検知（Isolation Forest）
- 重要度に応じた通知
- 本番システム設計
- マルチスレッド処理
- ロギングとエラーハンドリング
- 連続運用とシステム監視
重要なポイント
- リアルタイム処理では、低レイテンシとメモリ効率が重要
- 移動窓とEWMAは、無限のデータストリームを効率的に処理できる
- オンライン学習は、概念ドリフト（プロセス変化）に適応可能
- Plotly DashとInfluxDBは、産業用監視システムの標準ツール
- アラートシステムは、ルールベースとMLベースを組み合わせると効果的
- 本番システムは、ロバスト性、ロギング、監視機能が不可欠
- WebSocketは双方向通信が必要な場合に有効
さらに学ぶには
- Apache Kafka : 大規模ストリーミングデータ処理
- Apache Flink : リアルタイムデータパイプライン
- Grafana : 高度な可視化ダッシュボード
- Prometheus : メトリクス収集と監視
- Docker/Kubernetes : コンテナ化と自動スケーリング
- OPC-UA : 産業機器との標準通信プロトコル