100天跟着CP学PostgreSQL+AI,第27天 :设备预测性维护：时序数据 + AI 模型落地三步法

CREATE EXTENSION timescaledb;
CREATE TABLE motor_sensor_data (
    time TIMESTAMP NOT NULL,
    device_id INTEGER NOT NULL,
    speed FLOAT,
    temperature FLOAT,
    vibration FLOAT,
    PRIMARY KEY (device_id, time)
);
SELECT create_hypertable('motor_sensor_data', 'time');

import pandas as pd
from datetime import datetime, timedelta
def clean_timestamps(df):
    # 将时间戳字符串转换为datetime对象
    df['time'] = pd.to_datetime(df['time'], errors='coerce')
    # 去除时间戳异常的数据
    df = df.dropna(subset=['time'])
    # 按设备ID和时间排序
    df = df.sort_values(by=['device_id', 'time'])
    # 处理时间间隔异常的数据，使用前一个时间戳进行补全
    for device_id in df['device_id'].unique():
        device_df = df[df['device_id'] == device_id]
        time_diff = device_df['time'].diff()
        invalid_indices = time_diff > timedelta(seconds=10)  # 假设最大允许时间间隔为10秒
        for i in invalid_indices[invalid_indices].index:
            if i > 0:
                df.at[i, 'time'] = df.at[i-1, 'time'] + timedelta(seconds=1)
    return df

def create_dataset(data, window_size):
    X, y = [], []
    for i in range(window_size, len(data)):
        X.append(data[i-window_size:i, :])
        y.append(data[i, -1])  # 假设最后一个特征是故障标签
    return np.array(X), np.array(y)

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
# 假设X_train和y_train是训练数据，X_test和y_test是测试数据
model = Sequential()
model.add(LSTM(units=50, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2])))
model.add(LSTM(units=50))
model.add(Dense(1, activation='sigmoid'))
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
history = model.fit(X_train, y_train, batch_size=32, epochs=50, validation_split=0.1, verbose=1)
# 模型评估
loss, accuracy = model.evaluate(X_test, y_test, verbose=0)
print(f'Test Loss: {loss}, Test Accuracy: {accuracy}')

model.save('motor_fault_prediction_model.h5')

from flask import Flask, request, jsonify
import numpy as np
from tensorflow.keras.models import load_model
app = Flask(__name__)
model = load_model('motor_fault_prediction_model.h5')
@app.route('/predict', methods=['POST'])
def predict():
    data = request.json
    # 对输入数据进行预处理和归一化
    input_data = np.array(data['sensor_data']).reshape(1, -1, len(data['sensor_data'][0]))
    # 进行预测
    prediction = model.predict(input_data)
    return jsonify({'fault_probability': float(prediction[0][0])})
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

import psycopg2
from datetime import datetime, timedelta
def get_latest_sensor_data(device_id, window_size):
    conn = psycopg2.connect(
        dbname='industrial_db',
        user='postgres',
        password='password',
        host='localhost'
    )
    cur = conn.cursor()
    # 查询最近window_size个时间点的传感器数据
    query = f"""
        SELECT speed, temperature, vibration
        FROM motor_sensor_data
        WHERE device_id = {device_id}
        ORDER BY time DESC
        LIMIT {window_size}
    """
    cur.execute(query)
    data = cur.fetchall()
    cur.close()
    conn.close()
    # 将数据转换为适合模型输入的格式
    input_data = np.array(data[::-1]).reshape(1, window_size, 3)
    return input_data
# 进行实时预测
device_id = 1
window_size = 100
input_data = get_latest_sensor_data(device_id, window_size)
prediction = model.predict(input_data)
print(f"Device {device_id} fault probability: {prediction[0][0]}")