מקצה לקצה – מסימולציה ועד מערכת Real-Time מקצועית המשלבת Industry 4.0
המטרה: הבנת האקו-סיסטם של מפעל מודרני מבוסס נתונים.
המטרה: יצירת "תאום דיגיטלי" (Digital Twin) המייצר נתונים פיזיקליים ריאליים.
timestamp, temperature, pressure, vibration
import random
import csv
from datetime import datetime, timedelta
def generate_sensor_data(num_samples=100):
data = []
start_time = datetime.now()
for i in range(num_samples):
timestamp = start_time + timedelta(seconds=i)
temperature = round(50 + random.gauss(0, 10), 2)
pressure = round(5.5 + random.gauss(0, 1.5), 2)
vibration = round(2.5 + random.gauss(0, 0.8), 2)
data.append({'timestamp': timestamp.strftime('%Y-%m-%d %H:%M:%S'), 'temperature': temperature, 'pressure': pressure, 'vibration': vibration})
return data
def save_to_csv(data, filename='sensor_data.csv'):
with open(filename, 'w', newline='', encoding='utf-8') as file:
fieldnames = ['timestamp', 'temperature', 'pressure', 'vibration']
writer = csv.DictWriter(file, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
print(f"✅ נשמרו {len(data)} דגימות")
if __name__ == "__main__":
save_to_csv(generate_sensor_data(100))
sensor_data.csv מוכן לניתוח.המטרה: הפעלת "בינה" על הנתונים הגולמיים וזיהוי חריגות (Anomaly Detection).
import pandas as pd
def analyze_sensor_data(filename='sensor_data.csv'):
df = pd.read_csv(filename)
print("\n📊 סטטיסטיקות מערכת:")
for col in ['temperature', 'pressure', 'vibration']:
print(f"{col.capitalize()}: Mean={df[col].mean():.2f}")
critical_temp = df[df['temperature'] > 70]
print(f"\n⚠️ התראות קריטיות: {len(critical_temp)}")
המטרה: יצירת פרוטוקול תקשורת מודרני מבוסס HTTP לאיסוף נתונים מיחידות הקצה (PLC/Edge).
from fastapi import FastAPI
from pydantic import BaseModel
import csv, datetime
app = FastAPI(title="Industrial Edge API")
class Reading(BaseModel):
temperature: float
pressure: float
vibration: float
@app.post("/sensor")
def receive_data(阅读: Reading):
ts = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
with open('realtime_data.csv', 'a', newline='') as f:
writer = csv.writer(f)
writer.writerow([ts, 阅读.temperature, 阅读.pressure, 阅读.vibration])
return {"status": "received"}
השלב הסופי והמרשים ביותר! יצירת ממשק ויזואלי המציג את נתוני המפעל בגרפים אינטראקטיביים.
import streamlit as st
import pandas as pd
import plotly.express as px
import time
st.set_page_config(page_title="Industrial Dashboard", layout="wide")
st.title("🏭 Real-Time Factory Monitor")
while True:
try:
df = pd.read_csv('realtime_data.csv', names=['time', 'temp', 'pres', 'vib'])
st.metric("Temperature", f"{df['temp'].iloc[-1]}°C")
st.plotly_chart(px.line(df.tail(20), x='time', y=['temp', 'pres', 'vib']))
except:
st.info("Waiting for data...")
time.sleep(1)
| רכיב בקורס | מקביל במפעל | תפקיד |
|---|---|---|
| 🔌 sensor_simulator.py | חיישן פיזי | מדידת ערכים והמרתם לדיגיטלי. |
| 🐍 Python Scripts | Edge Computer / IoT Gateway | עיבוד נתונים קרוב למקור. |
| 📊 Dashboard | SCADA / HMI | ממשק פיקוח ובקרה. |