Premise: Conveyor belt system in industrial environments What needs to be done: simulate the interaction between motor performance (speed, temperature) and physical health (vibration)
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def generate_conveyor_data(rows=1000):
start_time = datetime.now()
# 5-second intervals
timestamps = [start_time + timedelta(seconds=i*5) for i in range(rows)]
# 1. Belt Speed: Nominal 1.5 m/s with 5% variance
speed = 1.5 + np.random.normal(0, 0.05, rows)
# 2. Vibration: Higher vibration indicates bearing wear or misalignment
vibration = 1.2 + np.abs(np.random.normal(0, 0.4, rows))
# 3. Temperature: Base of 45C with a slow upward drift as the motor runs
temp_drift = np.linspace(0, 5, rows)
temperature = 45.0 + temp_drift + np.random.normal(0, 0.5, rows)
# 4. Current: Amperage draw increases with speed and load
current = (speed * 4) + np.random.normal(0, 0.2, rows)
# Labeling logic for Predictive Maintenance
status = ["Warning" if (v > 2.8 or t > 53) else "Normal" for v, t in zip(vibration, temperature)]
df = pd.DataFrame({
'timestamp': timestamps,
'belt_speed_mps': np.round(speed, 2),
'motor_temp_c': np.round(temperature, 1),
'vibration_rms': np.round(vibration, 3),
'motor_current_a': np.round(current, 2),
'load_percentage': np.random.randint(30, 95, rows),
'status': status
})
return df
# Generate and save
df = generate_conveyor_data(5000)
df.to_csv('conveyor_iot_data.csv', index=False)
Adding failure events turns a simple monitoring dataset into a powerful training tool for Predictive Maintenance (PdM). In industrial systems, failures aren't always instant; they often leave "signatures" in the data before the machine stops.
Types of Simulated Failures: Bearing Degradation (The "Slow Burn"): Signature: Vibration increases steadily over 2-3 minutes. Friction causes motor_temp_c to rise. Finally, the belt slows down (belt_speed_mps drops) as the motor struggles to turn. Use Case: Training AI to predict failure hours before it happens. Belt Snap (The "Sudden Stop"): Signature: motor_current_a drops to near zero instantly (the motor is spinning with no resistance). Speed falls to zero.Use Case: Testing real-time emergency stop triggers.
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def generate_conveyor_failure_data(rows=1000):
np.random.seed(42)
start_time = datetime.now()
timestamps = [start_time + timedelta(seconds=i*5) for i in range(rows)]
# Baseline normal behavior
speed = 1.5 + np.random.normal(0, 0.05, rows)
vibration = 1.2 + np.abs(np.random.normal(0, 0.2, rows))
temp_drift = np.linspace(0, 3, rows)
temperature = 45.0 + temp_drift + np.random.normal(0, 0.3, rows)
current = (speed * 4) + np.random.normal(0, 0.1, rows)
status = ["Normal"] * rows
# --- Injection: Bearing Failure (Index 300-350) ---
for i in range(300, 350):
vibration[i] += (i - 300) * 0.15 # Climbing vibration
temperature[i] += (i - 300) * 0.2 # Rising heat
if i > 340:
speed[i] *= 0.5 # Jamming starts
status[i] = "Critical: Bearing Failure"
else:
status[i] = "Warning: High Friction"
# --- Injection: Belt Snap (Index 700-710) ---
for i in range(700, 710):
current[i] = 0.5 # No load current
speed[i] = 0.0
vibration[i] = 0.1
status[i] = "Critical: Belt Snap"
return pd.DataFrame({
'timestamp': timestamps,
'belt_speed_mps': np.round(speed, 2),
'motor_temp_c': np.round(temperature, 1),
'vibration_rms': np.round(vibration, 3),
'motor_current_a': np.round(current, 2),
'status': status
})
df = generate_conveyor_failure_data(1000)
df.to_csv('industrial_failure_dataset.csv', index=False)
Adding noise and network drops makes the dataset "dirty," which is essential for testing the robustness of AI models. Real-world IIoT data is rarely perfect.
The "Messy" Conveyor Dataset This version introduces three common industrial data problems: Network Drops (Packet Loss): Random gaps in the timestamp sequence (notice the missing rows like index 3 and 4 in the snippet). Sensor Noise (Jitter): Extreme vibration spikes that aren't failures, just electrical interference. Null Values (NaNs): Simulating a sensor that occasionally fails to report temperature.
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def generate_messy_iot_data(rows=2000):
start_time = datetime.now()
# 1. Base Generation
df = pd.DataFrame({
'timestamp': [start_time + timedelta(seconds=i*5) for i in range(rows)],
'belt_speed_mps': 1.5 + np.random.normal(0, 0.05, rows),
'motor_temp_c': 45.0 + np.random.normal(0, 0.3, rows),
'vibration_rms': 1.2 + np.abs(np.random.normal(0, 0.2, rows)),
'status': ["Normal"] * rows
})
# 2. Add Failures (Bearing & Snap)
df.loc[300:320, 'vibration_rms'] += 5.0
df.loc[300:320, 'status'] = "Warning: Bearing Wear"
# 3. ADD NOISE: Random vibration spikes (Interference)
noise_idx = df.sample(frac=0.03).index
df.loc[noise_idx, 'vibration_rms'] += np.random.uniform(5, 10, len(noise_idx))
# 4. ADD DROPS: Delete 10% of rows (Packet loss)
df = df.drop(df.sample(frac=0.10).index)
# 5. ADD NULLS: Randomly strip temperature readings
null_idx = df.sample(frac=0.02).index
df.loc[null_idx, 'motor_temp_c'] = np.nan
return df
df = generate_messy_iot_data()
df.to_csv('messy_conveyor_data.csv', index=False)
To clean industrial IoT data, need to address gaps in time, missing sensor values, and electrical "noise" (spikes) that don't represent real physical failures. The Data Cleaning Pipeline Resampling (Fixing Network Drops): Use resample to fill in missing timestamps. This ensures the AI model sees a steady heartbeat (e.g., one record every 5 seconds). Imputation (Filling NaNs): Forward Fill (ffill): Good for temperature, as it doesn't change instantly. Linear Interpolation: Best for speed or vibration to create a smooth transition between known points. Rolling Median (Noise Removal): A rolling window helps "smooth over" a single-second sensor glitch while keeping actual failure trends intact.
import pandas as pd
import numpy as np
# Load your messy CSV
df = pd.read_csv('messy_conveyor_data.csv')
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 1. Resample to 5-second intervals to recover 'dropped' packets
# This creates new rows for missing times with NaN values
df_cleaned = df.set_index('timestamp').resample('5s').asfreq()
# 2. Impute missing values
# Forward fill temperature (keeps the last known state)
df_cleaned['motor_temp_c'] = df_cleaned['motor_temp_c'].ffill()
# Interpolate speed and vibration for a smooth data stream
df_cleaned['belt_speed_mps'] = df_cleaned['belt_speed_mps'].interpolate()
df_cleaned['vibration_rms'] = df_cleaned['vibration_rms'].interpolate()
# 3. Remove electrical noise spikes
# If vibration jumps 5x above the median instantly, cap it
median_vibe = df_cleaned['vibration_rms'].median()
df_cleaned.loc[df_cleaned['vibration_rms'] > (median_vibe * 5), 'vibration_rms'] = median_vibe
# 4. Moving Average (Optional smoothing for dashboarding)
df_cleaned['vibe_smoothed'] = df_cleaned['vibration_rms'].rolling(window=3).mean()
print(df_cleaned.head())
belt_speed_mps motor_temp_c vibration_rms status \
timestamp
2026-05-12 15:28:40 NaN NaN NaN NaN
2026-05-12 15:28:45 NaN NaN NaN NaN
2026-05-12 15:28:50 NaN NaN NaN NaN
2026-05-12 15:28:55 NaN NaN NaN NaN
2026-05-12 15:29:00 NaN NaN NaN NaN
vibe_smoothed
timestamp
2026-05-12 15:28:40 NaN
2026-05-12 15:28:45 NaN
2026-05-12 15:28:50 NaN
2026-05-12 15:28:55 NaN
2026-05-12 15:29:00 NaN
The model ignores the noise but detects the 5-minute upward trend that signifies a bearing is actually dying.
up next: use streamlit to create a dashboard that displays real-time metrics, a smoothed vibration trend (to ignore noise), and an alert log.