This chapter covers Real. You will learn streaming data processing, real-time statistical monitoring (moving windows), and Create real-time dashboards with Plotly Dash.
Learning Objectives
By reading this chapter, you will master the following:
- ✅ Understand streaming data processing and buffering techniques
- ✅ Implement real-time statistical monitoring (moving windows)
- ✅ Build online machine learning models (incremental learning)
- ✅ Create real-time dashboards with Plotly Dash
- ✅ Utilize database streaming (InfluxDB, TimescaleDB)
- ✅ Implement alert systems and automated anomaly detection
- ✅ Design and optimize monitoring systems for production environments
5.1 Fundamentals of Real-Time Processing
Challenges in Real-Time Data Processing
Real-time monitoring of chemical processes requires addressing the following technical challenges:
| Challenge | Description | Solution |
|---|---|---|
| Low Latency | Response in milliseconds to seconds | Efficient algorithms, buffering |
| High Throughput | Data from numerous sensors | Parallel processing, distributed systems |
| Data Quality | Missing values, noise, outliers | Real-time filtering |
| Memory Constraints | Infinite data streams | Moving windows, memory-efficient data structures |
| Model Adaptation | Changes in process conditions | Online learning, adaptive algorithms |
5.2 Streaming Data Processing
Code Example 1: Fundamentals of Streaming Data Processing with pandas
# Requirements:
# - Python 3.9+
# - matplotlib>=3.7.0
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from collections import deque
import time
class StreamingDataProcessor:
"""
Real-time data stream processing class
Parameters:
-----------
window_size : int
Size of moving window (number of data points)
sampling_rate : float
Sampling rate [Hz]
"""
def __init__(self, window_size=100, sampling_rate=1.0):
self.window_size = window_size
self.sampling_rate = sampling_rate
# Data buffer (using fast deque)
self.buffer = deque(maxlen=window_size)
# Statistics history
self.timestamps = []
self.means = []
self.stds = []
self.mins = []
self.maxs = []
def add_data_point(self, value, timestamp=None):
"""
Add a new data point
Parameters:
-----------
value : float
Data value
timestamp : float
Timestamp (current time if None)
"""
if timestamp is None:
timestamp = time.time()
self.buffer.append(value)
# Calculate statistics (data within window)
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):
"""
Get statistics for current window
Returns:
--------
stats : dict
Dictionary of statistics
"""
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):
"""
Anomaly detection (moving average ± threshold_std × standard deviation)
Parameters:
-----------
threshold_std : float
Threshold as multiple of standard deviation
Returns:
--------
is_anomaly : bool
Whether anomaly exists
"""
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-sigma rule
is_anomaly = abs(latest_value - mean) > threshold_std * std
return is_anomaly
# Simulation: Real-time data stream processing
np.random.seed(42)
# Initialize processor
processor = StreamingDataProcessor(window_size=50, sampling_rate=10.0)
# Data generation and streaming processing
n_points = 300
anomaly_indices = [100, 150, 250] # Positions to inject anomalies
for i in range(n_points):
# Normal data + anomalies
if i in anomaly_indices:
value = 180 + np.random.randn() * 15 # Abnormally large fluctuation
else:
value = 180 + 2 * np.sin(0.1 * i) + np.random.randn() * 1
# Add data point
processor.add_data_point(value, timestamp=i)
# Real-time anomaly detection
if processor.detect_anomaly(threshold_std=3.0) and i > 50:
print(f"Anomaly detected at time {i}: value = {value:.2f}")
# Visualization
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
# Raw data and moving average
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)
# Mark anomaly positions
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)
# Moving standard deviation (variability monitoring)
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 statistics
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}")Explanation: The moving window using deque efficiently manages a fixed-size buffer and can add new data with O(1) time complexity. Real-time anomaly detection is based on the 3-sigma rule, detecting statistical outliers.
Code Example 2: Real-Time Statistical Monitoring (EWMA)
# Requirements:
# - Python 3.9+
# - matplotlib>=3.7.0
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
class EWMAMonitor:
"""
Real-time monitoring using Exponentially Weighted Moving Average (EWMA)
Parameters:
-----------
alpha : float
Smoothing parameter (0 < alpha <= 1)
Smaller values place more weight on past data
"""
def __init__(self, alpha=0.1):
self.alpha = alpha
self.ewma = None
self.ewmvar = None # EWMA variance
# History
self.values = []
self.ewma_values = []
self.ewmvar_values = []
self.control_limits_upper = []
self.control_limits_lower = []
def update(self, new_value):
"""
Update EWMA with new data point
Parameters:
-----------
new_value : float
New data value
"""
# Initialization
if self.ewma is None:
self.ewma = new_value
self.ewmvar = 0
else:
# EWMA update
self.ewma = self.alpha * new_value + (1 - self.alpha) * self.ewma
# EWMA variance update
deviation = new_value - self.ewma
self.ewmvar = self.alpha * (deviation ** 2) + (1 - self.alpha) * self.ewmvar
# Control limits (UCL/LCL: Upper/Lower Control Limit)
ewm_std = np.sqrt(self.ewmvar)
ucl = self.ewma + 3 * ewm_std
lcl = self.ewma - 3 * ewm_std
# Save to history
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):
"""
Determine if control limit is exceeded
Returns:
--------
bool : Whether out of control
"""
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
# Simulation
np.random.seed(42)
n_points = 300
# Process shift (mean shifts at time 100 and 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 # Mean shift (+5°C)
else:
value = 180 + np.random.randn() * 3 # Increased variability
data.append(value)
# EWMA monitoring
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}")
# Visualization
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(len(data))
# EWMA control chart
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')
# Mark process shift positions
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 variance (variability monitoring)
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}")Explanation: EWMA (Exponentially Weighted Moving Average) is a real-time monitoring method that can quickly detect small process shifts. It places more weight on recent data than simple moving averages and has low latency, making it suitable for chemical process control.
5.3 Online Machine Learning
Code Example 3: Online Linear Regression (Incremental Learning)
# Requirements:
# - Python 3.9+
# - matplotlib>=3.7.0
# - numpy>=1.24.0, <2.0.0
import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import SGDRegressor
from sklearn.preprocessing import StandardScaler
class OnlineLinearRegressor:
"""
Linear regression model using online learning
Parameters:
-----------
learning_rate : str
Learning rate schedule ('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 # Enable incremental learning
)
self.scaler_X = StandardScaler()
self.scaler_y = StandardScaler()
self.is_fitted = False
# History
self.predictions = []
self.actuals = []
self.errors = []
def partial_fit(self, X, y):
"""
Partial fit with new data (incremental learning)
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
Features
y : array-like, shape (n_samples,)
Target variable
"""
X = np.atleast_2d(X)
y = np.atleast_1d(y)
# Scaling
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()
# Incremental learning
self.model.partial_fit(X_scaled, y_scaled)
def predict(self, X):
"""
Prediction
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
Features
Returns:
--------
y_pred : array-like
Predicted values
"""
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):
"""
Predict then learn (typical pattern for online learning)
Parameters:
-----------
X : array-like
Features
y : float
Actual value
Returns:
--------
y_pred : float
Predicted value
"""
# Prediction
y_pred = self.predict(X)
# Learning
self.partial_fit(X, y)
# Update history
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
# Simulation: Online regression
np.random.seed(42)
n_samples = 500
# Time-varying linear relationship (concept drift)
X_stream = []
y_stream = []
for i in range(n_samples):
# Features
x1 = np.random.randn()
x2 = np.random.randn()
X_stream.append([x1, x2])
# True relationship (changes over time)
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 (relationship changes)
y_true = 2*x1 + 4*x2
else:
# Phase 3: y = 1*x1 + 5*x2 (further change)
y_true = 1*x1 + 5*x2
y_stream.append(y_true + np.random.randn() * 0.5) # Add noise
X_stream = np.array(X_stream)
y_stream = np.array(y_stream)
# Online learning
online_model = OnlineLinearRegressor(learning_rate='constant', eta0=0.01)
# Initial learning (first 10 samples)
online_model.partial_fit(X_stream[:10], y_stream[:10])
# Streaming learning
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)
# Visualization
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(10, n_samples)
# Prediction vs Actual
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)
# Prediction error (evaluate adaptability of online learning)
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 for each phase
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}")Explanation: Online learning (incremental learning) updates the model as data arrives and adapts to concept drift (changes in data distribution). Unlike batch learning, it doesn’t need to retain all data and is memory-efficient.
Code Example 4: Adaptive Anomaly Detection with Online PCA
# Requirements:
# - Python 3.9+
# - matplotlib>=3.7.0
# - numpy>=1.24.0, <2.0.0
import numpy as np
import matplotlib.pyplot as plt
from sklearn.decomposition import IncrementalPCA
class OnlineAnomalyDetector:
"""
Adaptive anomaly detection using online PCA
Parameters:
-----------
n_components : int
Number of principal components
threshold : float
Anomaly detection threshold (multiple of SPE standard deviation)
"""
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):
"""
Fit PCA with initial data
Parameters:
-----------
X_initial : array-like, shape (n_samples, n_features)
Initial training data
"""
self.ipca.partial_fit(X_initial)
self.is_fitted = True
def detect(self, X):
"""
Anomaly detection
Parameters:
-----------
X : array-like, shape (n_features,) or (n_samples, n_features)
Data to monitor
Returns:
--------
is_anomaly : bool or array
Whether anomaly exists
spe : float or array
SPE (Squared Prediction Error)
"""
if not self.is_fitted:
return False, 0.0
X = np.atleast_2d(X)
# Projection to principal component space and reconstruction
X_transformed = self.ipca.transform(X)
X_reconstructed = self.ipca.inverse_transform(X_transformed)
# SPE (Squared Prediction Error) calculation
spe = np.sum((X - X_reconstructed) ** 2, axis=1)
# Threshold determination
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 # No threshold in initial phase
is_anomaly = spe > threshold_value
# Update history
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
# Simulation: Multivariate process data
np.random.seed(42)
n_samples = 500
n_features = 10
# Generate normal data (correlated multivariate data)
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 # Positive definite symmetric matrix
X_normal = np.random.multivariate_normal(mean_normal, cov_normal, size=400)
# Inject anomalous data (large deviations)
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:
# Last 100 samples have anomalies injected
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)
# Online PCA anomaly detection
detector = OnlineAnomalyDetector(n_components=3, threshold=3.0)
detector.fit(X_stream[:50]) # Initial learning
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)
# Periodic model update
if i % 10 == 0:
detector.ipca.partial_fit(X_stream[i-10:i])
# Visualization
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
time_axis = np.arange(50, n_samples)
# SPE values and threshold
axes[0].plot(time_axis, spe_values, color='#11998e', linewidth=1.5, label='SPE')
# Dynamic threshold
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σ)')
# Mark true anomaly positions
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)
# Visualization of detection results
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 anomalies
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()
# Evaluation of detection accuracy
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}")Explanation: Online PCA sequentially updates principal components of multivariate process data and detects anomalies based on reconstruction error (SPE). It can detect anomalous patterns while adapting to normal process variations.
5.4 Real-Time Dashboard
Code Example 5: Real-Time Dashboard with Plotly Dash
# Requirements:
# - Python 3.9+
# - dash>=2.11.0
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
# - plotly>=5.14.0
"""
Example: Code Example 5: Real-Time Dashboard with Plotly Dash
Purpose: Demonstrate data visualization techniques
Target: Intermediate
Execution time: 2-5 seconds
Dependencies: None
"""
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
# Data buffer (holds latest 1000 points)
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 application
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'),
# Update interval (every 1 second)
dcc.Interval(
id='interval-component',
interval=1000, # milliseconds
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):
"""
Real-time dashboard update
"""
# Generate new data (in practice, acquire sensor data)
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()
# Add to buffer
time_buffer.append(current_time)
temp_buffer.append(temperature)
pressure_buffer.append(pressure)
flow_buffer.append(flow_rate)
# Temperature graph
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 graph
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 rate graph
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'
)
}
# Statistics information
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
# Note: This code does not work in Jupyter Notebook
# Run in terminal: 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)How to run:
# Save as dashboard.py and run
pip install dash plotly pandas
python dashboard.py
# Open browser at http://127.0.0.1:8050/Explanation: Plotly Dash is a framework for building interactive real-time dashboards in Python. dcc.Interval enables periodic updates, allowing dynamic process monitoring in the browser. In production environments, it integrates with data streams from DCS or MES.
5.5 Database Streaming
Code Example 6: Time Series Data Streaming with InfluxDB
# Requirements:
# - Python 3.9+
# - numpy>=1.24.0, <2.0.0
# InfluxDB 2.x client
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
# Note: Running this code requires an InfluxDB server
# Can be easily started with Docker:
# docker run -d -p 8086:8086 influxdb:2.0
class InfluxDBStreamer:
"""
Real-time data streaming to InfluxDB
Parameters:
-----------
url : str
InfluxDB URL
token : str
Authentication token
org : str
Organization name
bucket : str
Bucket name
"""
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
# Client connection
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):
"""
Write a data point
Parameters:
-----------
measurement : str
Measurement name (e.g., "temperature", "pressure")
tags : dict
Tags (metadata, e.g., {"sensor_id": "T101", "location": "reactor"})
fields : dict
Fields (actual data, e.g., {"value": 180.5})
timestamp : datetime
Timestamp (current time if None)
"""
if self.client is None:
print("InfluxDB client not initialized")
return
if timestamp is None:
timestamp = datetime.utcnow()
# Create data point
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)
# Write
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"):
"""
Query recent data
Parameters:
-----------
measurement : str
Measurement name
time_range : str
Time range (e.g., "-1h", "-30m", "-10s")
Returns:
--------
data : list
Query results
"""
if self.client is None:
return []
# Flux query
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):
"""Close client"""
if self.client:
self.client.close()
# Usage example (simulation)
def simulate_streaming():
"""
Process data streaming simulation
"""
# Initialize InfluxDB streamer
# Note: Adjust parameters for your environment
streamer = InfluxDBStreamer(
url="http://localhost:8086",
token="your-influxdb-token", # Set your actual token
org="my-org",
bucket="process-data"
)
print("Starting data streaming simulation...")
print("Press Ctrl+C to stop")
try:
for i in range(100):
# Generate sensor data
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()
# Write data
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 second interval
except KeyboardInterrupt:
print("\nStreaming stopped by user")
finally:
# Query recent data
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()
# Note: Can only run when InfluxDB server is running
# 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
""")Explanation: InfluxDB is a database specialized for time series data, enabling high-speed writes and queries. With metadata management via tags, automatic downsampling, and Grafana integration, it is widely used as the foundation for industrial real-time monitoring systems.
Code Example 7: Real-Time Data Distribution with WebSocket
# Requirements:
# - Python 3.9+
# - numpy>=1.24.0, <2.0.0
import asyncio
import websockets
import json
import numpy as np
from datetime import datetime
# WebSocket server (data distribution side)
class ProcessDataServer:
"""
Distribute process data via WebSocket server
"""
def __init__(self, host='localhost', port=8765):
self.host = host
self.port = port
self.clients = set()
async def register_client(self, websocket):
"""Register client"""
self.clients.add(websocket)
print(f"Client connected: {websocket.remote_address}")
async def unregister_client(self, websocket):
"""Unregister client"""
self.clients.remove(websocket)
print(f"Client disconnected: {websocket.remote_address}")
async def send_data_to_clients(self, data):
"""Send data to all clients"""
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):
"""
Client connection handler
"""
await self.register_client(websocket)
try:
# Wait for messages from client (maintain connection)
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):
"""
Periodically generate data and broadcast
"""
iteration = 0
while True:
# Generate process data
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
}
# Send to all clients
await self.send_data_to_clients(data)
iteration += 1
await asyncio.sleep(1) # 1 second interval
async def start(self):
"""Start server"""
# Start WebSocket server
server = await websockets.serve(self.handler, self.host, self.port)
print(f"WebSocket server started on ws://{self.host}:{self.port}")
# Start data generation/distribution task
await self.generate_and_broadcast_data()
# WebSocket client (data receiving side)
class ProcessDataClient:
"""
Receive data via WebSocket client
"""
def __init__(self, uri='ws://localhost:8765'):
self.uri = uri
self.data_buffer = []
async def receive_data(self):
"""
Receive and process data
"""
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)
# Real-time processing
print(f"Received: T={data['temperature']}°C, "
f"P={data['pressure']}bar, "
f"F={data['flow_rate']}L/h")
# Anomaly detection example
if data['temperature'] > 185:
print(f" WARNING: High temperature detected!")
# Buffer size limit
if len(self.data_buffer) > 1000:
self.data_buffer.pop(0)
except websockets.exceptions.ConnectionClosed:
print("Connection closed by server")
# Server startup script
async def run_server():
server = ProcessDataServer(host='localhost', port=8765)
await server.start()
# Client startup script
async def run_client():
client = ProcessDataClient(uri='ws://localhost:8765')
await client.receive_data()
# Usage example
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
""")
# Execution example (uncomment to use)
# asyncio.run(run_server()) # Start server
# asyncio.run(run_client()) # Start clientExplanation: WebSocket is a bidirectional communication protocol that can distribute real-time data with lower latency than HTTP. Data from DCS can be distributed via a WebSocket server, and multiple clients (dashboards, analysis tools) can receive simultaneously.
5.6 Alert Systems and Automation
Code Example 8: Rule-Based Alert System
# Requirements:
# - Python 3.9+
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
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):
"""Alert severity"""
INFO = 1
WARNING = 2
CRITICAL = 3
EMERGENCY = 4
class Alert:
"""Alert information"""
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:
"""
Process monitoring alert system
Parameters:
-----------
rules : list of dict
List of alert rules
Example: {'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):
"""
Evaluate rules and generate alerts
Parameters:
-----------
data : dict
Monitoring data (variable name: value)
Returns:
--------
alerts : list of Alert
List of generated alerts
"""
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]
# Evaluate condition
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):
"""
Send alert notification (email, SMS, etc.)
Parameters:
-----------
alert : Alert
Alert to send
"""
print(f"\n*** ALERT NOTIFICATION ***")
print(alert)
print("*" * 50)
# In actual implementation, send email or SMS
# if alert.severity == AlertSeverity.CRITICAL:
# self._send_email(alert)
# self._send_sms(alert)
def _send_email(self, alert):
"""Email sending (implementation example)"""
# Note: Configure for your environment
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):
"""Get alert summary"""
return {
'total_alerts': len(self.alerts),
'by_severity': self.alert_counts,
'recent_alerts': self.alerts[-10:] if len(self.alerts) > 0 else []
}
# Simulation
np.random.seed(42)
# Define alert rules
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},
]
# Initialize alert system
alert_system = ProcessAlertSystem(rules=alert_rules)
# Process data stream
print("Starting Process Monitoring...")
print("=" * 70)
for i in range(100):
# Generate process data (occasionally contains abnormal values)
if i in [25, 50, 75]:
# Abnormal data
temperature = 192 + np.random.randn()
pressure = 3.8 + np.random.randn() * 0.1
else:
# Normal data
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
}
# Evaluate rules
current_alerts = alert_system.evaluate_rules(data)
# Send alert notifications
for alert in current_alerts:
alert_system.send_alert_notification(alert)
# Alert summary
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}")Explanation: Rule-based alert systems detect anomalies in real-time based on thresholds and conditions, and notify according to severity. Integration with multiple notification channels (email, SMS, SCADA screens) is possible.
Code Example 9: Machine Learning-Based Anomaly Detection and Alerts
# Requirements:
# - Python 3.9+
# - matplotlib>=3.7.0
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
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:
"""
Machine learning-based anomaly detection system
Parameters:
-----------
contamination : float
Proportion of anomalies (0.0 to 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):
"""
Train with normal data
Parameters:
-----------
X_normal : array-like, shape (n_samples, n_features)
Normal data
"""
X_scaled = self.scaler.fit_transform(X_normal)
self.model.fit(X_scaled)
self.is_fitted = True
def predict(self, X):
"""
Anomaly detection
Parameters:
-----------
X : array-like, shape (n_samples, n_features)
Data
Returns:
--------
is_anomaly : array, shape (n_samples,)
Anomaly flag (1: normal, -1: anomaly)
anomaly_score : array, shape (n_samples,)
Anomaly score (lower is more anomalous)
"""
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) # Anomaly score (lower is more anomalous)
self.predictions.extend(predictions.tolist())
self.anomaly_scores.extend(anomaly_scores.tolist())
return predictions, anomaly_scores
# Simulation
np.random.seed(42)
# Train with normal data
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
)
# Train anomaly detector
detector = MLAnomalyDetector(contamination=0.05)
detector.fit(X_train)
# Test data (normal + anomalies)
n_test = 300
X_test = []
true_labels = []
for i in range(n_test):
if i % 50 == 49: # Inject anomaly every 50 samples
# Anomalous data
X_test.append([195, 3.8, 35]) # High temp, high pressure, low flow
true_labels.append(-1)
else:
# Normal data
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)
# Real-time anomaly detection
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)
# Visualization
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# Time series of anomaly scores
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)
# Temperature vs Pressure (anomaly visualization)
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)
# Pressure vs Flow
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)
# Histogram of anomaly scores
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()
# Evaluate detection performance
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']))Explanation: Machine learning-based anomaly detection can detect complex anomalous patterns that rule-based methods cannot capture. Isolation Forest uses unsupervised learning, training only on normal data and efficiently detecting outliers. It can also consider nonlinear relationships and multivariate correlations in processes.
5.7 Production Environment Deployment
Code Example 10: Integrated Implementation of Production Monitoring System
# Requirements:
# - Python 3.9+
# - numpy>=1.24.0, <2.0.0
# - pandas>=2.0.0, <2.2.0
"""
Example: Code Example 10: Integrated Implementation of Production Mon
Purpose: Demonstrate simulation and statistical methods
Target: Advanced
Execution time: 5-15 seconds
Dependencies: None
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from collections import deque
import logging
import threading
import time
# Logging configuration
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:
"""
Production environment process monitoring system
Parameters:
-----------
sampling_interval : float
Sampling interval [seconds]
buffer_size : int
Data buffer size
alert_threshold : dict
Dictionary of alert thresholds
"""
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 {}
# Data buffer
self.data_buffer = {
'timestamp': deque(maxlen=buffer_size),
'temperature': deque(maxlen=buffer_size),
'pressure': deque(maxlen=buffer_size),
'flow_rate': deque(maxlen=buffer_size)
}
# Statistics information
self.statistics = {}
self.alert_count = 0
self.uptime_start = datetime.now()
# Control flags
self.is_running = False
self.monitoring_thread = None
logger.info("Production Monitoring System initialized")
def read_sensor_data(self):
"""
Read sensor data
In actual implementation, read from DCS, PLC, OPC-UA server, etc.
Returns:
--------
data : dict
Sensor data
"""
# Dummy data for simulation
# In real environment, acquire actual data via OPC-UA, Modbus, MQTT, etc.
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):
"""
Update data buffer
Parameters:
-----------
data : dict
New sensor data
"""
for key, value in data.items():
self.data_buffer[key].append(value)
def calculate_statistics(self):
"""
Calculate statistics from current data buffer
Returns:
--------
stats : dict
Dictionary of statistics
"""
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):
"""
Check alert conditions
Parameters:
-----------
data : dict
Current sensor data
Returns:
--------
alerts : list
List of triggered alerts
"""
alerts = []
# Temperature 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)
# Pressure alerts
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)
# Flow rate alerts
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):
"""
Main monitoring loop
"""
logger.info("Monitoring loop started")
while self.is_running:
try:
# Read sensor data
data = self.read_sensor_data()
# Update buffer
self.update_buffer(data)
# Calculate statistics
self.calculate_statistics()
# Check alerts
alerts = self.check_alerts(data)
# Data logging (periodically save to DB, etc.)
if len(self.data_buffer['timestamp']) % 60 == 0: # Every minute
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")
# Wait until next sampling
time.sleep(self.sampling_interval)
except Exception as e:
logger.error(f"Error in monitoring loop: {e}", exc_info=True)
time.sleep(5) # Wait 5 seconds on error and retry
def start(self):
"""Start monitoring system"""
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):
"""Stop monitoring system"""
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):
"""
Get system status
Returns:
--------
status : dict
System status information
"""
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
}
}
# Usage example
if __name__ == "__main__":
# Initialize monitoring system
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
)
# Start monitoring
monitor.start()
try:
# Run for 60 seconds
for i in range(60):
time.sleep(1)
# Display status every 10 seconds
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:
# Stop monitoring
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']}")Explanation: Production environment monitoring systems require robustness, logging, error handling, and thread management. This implementation integrates data acquisition from DCS/PLC, real-time analysis, alert generation, and log recording, supporting 24/7 continuous operation.
5.8 Chapter Summary
What We Learned
1. Streaming Data Processing — Efficient buffering with moving windows using deque, and real-time statistical monitoring with EWMA (Exponentially Weighted Moving Average).
2. Online Machine Learning — Adaptive models through incremental learning with SGDRegressor, and multivariate anomaly detection with online PCA.
3. Real-Time Visualization — Interactive dashboards with Plotly Dash, featuring periodic updates and dynamic graph generation.
4. Database Integration — Time series data streaming with InfluxDB and low-latency distribution with WebSocket.
5. Alert Systems — Rule-based anomaly detection, machine learning-based anomaly detection using Isolation Forest, and notifications according to severity levels.
6. Production System Design — Multi-threaded processing, logging and error handling, and continuous operation with system monitoring.
Key Points
Low latency and memory efficiency are critical in real-time processing. Moving windows and EWMA can efficiently process infinite data streams. Online learning can adapt to concept drift as processes change over time. Plotly Dash and InfluxDB are standard tools for industrial monitoring systems. Alert systems are most effective when combining rule-based and ML-based approaches. Production systems require robustness, logging, and monitoring capabilities. WebSocket is effective when bidirectional communication is needed.
Further Learning
For continued development, consider exploring Apache Kafka for large-scale streaming data processing, Apache Flink for real-time data pipelines, Grafana for advanced visualization dashboards, Prometheus for metrics collection and monitoring, Docker/Kubernetes for containerization and auto-scaling, and OPC-UA as the standard communication protocol for industrial equipment.