جمع البيانات الصناعية: أول خطوة نحو الذكاء

لماذا تبدأ كل ثورة صناعية ذكية من البيانات؟

تخيّل مصنعاً للإسمنت فيه 200 حساس يقيس الحرارة والضغط والاهتزاز كل ثانية — هذا يعني 17 مليون قراءة في اليوم الواحد. لكن هل كل هذه البيانات مفيدة؟ الإجابة: لا، إلا إذا جمعتها بطريقة صحيحة، نظّفتها من الأخطاء، وخزّنتها بشكل يسمح بالتحليل.

جمع البيانات الصناعية هو الأساس الذي يُبنى عليه كل نظام ذكاء اصطناعي في المصنع. بدون بيانات نظيفة وموثوقة، حتى أذكى الخوارزميات ستعطي نتائج خاطئة — كما يقال: "قمامة تدخل، قمامة تخرج" (Garbage In, Garbage Out).

أنواع الحساسات الصناعية ومصادر البيانات

في أي مصنع حديث، تأتي البيانات من مصادر متنوعة جداً:

حساسات فيزيائية مباشرة:

• حرارة: Thermocouples (من النوع K أو J) لقياس حرارة الأفران حتى 1200°C
• ضغط: Pressure transducers لمراقبة ضغط الأنابيب والمراجل
• اهتزاز: Accelerometers لكشف أعطال المحامل والمحركات
• تدفق: Flow meters لقياس تدفق السوائل والغازات
• مستوى: Level sensors لخزانات المواد الخام والمنتجات

مصادر رقمية:

• أنظمة SCADA و PLC التي تتحكم بخطوط الإنتاج
• أنظمة ERP التي تسجل أوامر العمل والمخزون
• سجلات الجودة وتقارير الفحص
• كاميرات الرؤية الآلية (Machine Vision)

معدل أخذ العينات: كم مرة نقرأ الحساس؟

تصوّر أنك تراقب حرارة فرن يتغير ببطء — قراءة واحدة كل 10 ثوانٍ تكفي تماماً. لكن لو كنت تراقب اهتزاز محرك يدور بسرعة 3000 دورة/دقيقة، تحتاج آلاف القراءات في الثانية لالتقاط أي خلل.

نظرية نايكويست (Nyquist Theorem) تقول: يجب أن يكون معدل أخذ العينات على الأقل ضعف أعلى تردد في الإشارة. عملياً، نأخذ 5-10 أضعاف لضمان الدقة.

# حساب معدل أخذ العينات المطلوب
motor_rpm = 3000          # سرعة المحرك
fundamental_freq = motor_rpm / 60  # 50 Hz
# لكشف التوافقيات حتى الدرجة العاشرة
max_freq = fundamental_freq * 10   # 500 Hz
# تطبيق نظرية نايكويست مع هامش أمان
sampling_rate = max_freq * 2.56    # 1280 Hz (معيار ISO)

print(f"التردد الأساسي: {fundamental_freq} Hz")
print(f"أعلى تردد مطلوب: {max_freq} Hz")
print(f"معدل أخذ العينات: {sampling_rate} Hz")
print(f"عدد القراءات في الدقيقة: {sampling_rate * 60:,.0f}")

القاعدة العملية:

• عمليات بطيئة (حرارة، مستوى): 0.1 - 1 Hz
• عمليات متوسطة (ضغط، تدفق): 1 - 100 Hz
• عمليات سريعة (اهتزاز، صوت): 1 - 50 kHz

جودة البيانات: العدو الخفي

في الواقع الصناعي، البيانات نادراً ما تكون مثالية. تخيّل أن حساس حرارة يُرسل القيمة -999.0 فجأة — هل انخفضت الحرارة فعلاً أم أن الحساس انقطع؟

القيم المفقودة (Missing Values)

تحدث عندما ينقطع الاتصال بالحساس أو يتعطل مؤقتاً.

import numpy as np

# بيانات حرارة مع قيم مفقودة
readings = [78.5, 79.1, None, None, 80.3, 77.8, None, 81.2, 80.5, 79.9]

# الطريقة 1: الاستيفاء الخطي (Linear Interpolation)
def interpolate_missing(data):
    result = data.copy()
    for i, val in enumerate(result):
        if val is None:
            # البحث عن أقرب قيمتين صالحتين
            prev_val = next((result[j] for j in range(i-1, -1, -1) if result[j] is not None), None)
            next_val = next((result[j] for j in range(i+1, len(result)) if result[j] is not None), None)
            if prev_val and next_val:
                result[i] = (prev_val + next_val) / 2
            elif prev_val:
                result[i] = prev_val  # Forward fill
    return result

clean_data = interpolate_missing(readings)
print(f"البيانات الأصلية: {readings}")
print(f"بعد المعالجة:    {clean_data}")

القيم الشاذة (Outliers)

قراءة بعيدة جداً عن المتوقع — قد تكون خطأ حساس أو حدث حقيقي يستحق التحقيق.

def detect_outliers_iqr(data):
    """كشف القيم الشاذة باستخدام المدى الرُبعي IQR"""
    clean = [x for x in data if x is not None]
    q1 = np.percentile(clean, 25)
    q3 = np.percentile(clean, 75)
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr

    outliers = [(i, v) for i, v in enumerate(data)
                if v is not None and (v < lower_bound or v > upper_bound)]
    return outliers, lower_bound, upper_bound

# بيانات ضغط مع قيمة شاذة
pressure = [4.2, 4.3, 4.1, 4.4, 12.8, 4.2, 4.3, 4.1, 4.5, 4.2]
outliers, low, high = detect_outliers_iqr(pressure)
print(f"الحدود: [{low:.1f}, {high:.1f}]")
print(f"قيم شاذة: {outliers}")

الضوضاء (Noise)

التقلبات العشوائية الصغيرة في قراءات الحساس. نستخدم مرشحات (Filters) لتنعيم الإشارة:

def moving_average_filter(data, window=5):
    """مرشح المتوسط المتحرك لتنعيم البيانات"""
    filtered = []
    for i in range(len(data)):
        start = max(0, i - window // 2)
        end = min(len(data), i + window // 2 + 1)
        window_data = [x for x in data[start:end] if x is not None]
        filtered.append(sum(window_data) / len(window_data) if window_data else None)
    return filtered

تسمية البيانات (Data Labeling)

لتدريب نماذج الذكاء الاصطناعي، نحتاج بيانات مُعلّمة — أي كل قراءة أو صورة مرفقة بوصف يقول "هذا طبيعي" أو "هذا عطل".

تصوّر أنك تدرّب نموذجاً لكشف أعطال المحامل من بيانات الاهتزاز. تحتاج أن تقول للنموذج: "هذه الإشارة = محمل سليم" و"هذه الإشارة = محمل متآكل".

طرق التسمية:

1. يدوية: خبير يراجع البيانات ويصنفها — دقيقة لكن بطيئة ومكلفة
2. شبه آلية: الخبير يُسمّي عينة صغيرة، ثم نموذج أولي يُسمّي الباقي
3. آلية بالقواعد: مثلاً: "إذا تجاوز الاهتزاز 10 mm/s = عطل"
4. من سجلات الصيانة: ربط أعطال مُسجّلة مع بيانات الحساسات في نفس الوقت

عملية ETL: استخراج، تحويل، تحميل

ETL هي اختصار Extract-Transform-Load — خط الأنابيب الذي ينقل البيانات من مصادرها الخام إلى قاعدة البيانات التحليلية.

┌──────────┐    ┌──────────────┐    ┌──────────────┐
│ Extract  │───▶│  Transform   │───▶│    Load      │
│ استخراج  │    │  تحويل       │    │  تحميل       │
├──────────┤    ├──────────────┤    ├──────────────┤
│ PLC/SCADA│    │ تنظيف        │    │ Historian DB │
│ Sensors  │    │ توحيد الوحدات│    │ Time-series  │
│ ERP      │    │ حساب مؤشرات  │    │ Data Lake    │
│ CSV/API  │    │ كشف شذوذ أولي│    │              │
└──────────┘    └──────────────┘    └──────────────┘

# مثال مبسّط لخط أنابيب ETL صناعي
class IndustrialETL:
    def extract(self, source):
        """استخراج البيانات من مصادر متعددة"""
        if source == "opc_ua":
            # قراءة من خادم OPC-UA (معيار الاتصال الصناعي)
            return self.read_opc_ua()
        elif source == "modbus":
            return self.read_modbus()
        elif source == "csv":
            return self.read_csv_files()

    def transform(self, raw_data):
        """تنظيف وتحويل البيانات"""
        data = self.remove_duplicates(raw_data)
        data = self.fill_missing_values(data)
        data = self.convert_units(data)       # توحيد: PSI→bar, °F→°C
        data = self.flag_outliers(data)
        data = self.calculate_kpis(data)      # OEE, استهلاك طاقة
        return data

    def load(self, clean_data, target):
        """تحميل إلى قاعدة البيانات الهدف"""
        if target == "historian":
            self.write_to_historian(clean_data)
        elif target == "data_lake":
            self.write_parquet(clean_data)     # تخزين عمودي مضغوط

قواعد بيانات المؤرخ (Historian Databases)

في عالم المصانع، لا نستخدم قواعد بيانات عادية مثل MySQL لتخزين بيانات الحساسات — نستخدم قواعد بيانات السلاسل الزمنية (Time-Series Databases) المُحسّنة لهذا الغرض.

أشهر قواعد بيانات المؤرخ:

• InfluxDB: مفتوحة المصدر، ممتازة للمشاريع المتوسطة
• TimescaleDB: مبنية على PostgreSQL، تجمع بين SQL والسلاسل الزمنية
• OSIsoft PI: المعيار الصناعي في المصانع الكبرى
• Siemens Historian: مدمج مع أنظمة سيمنز

من الحساس إلى القرار: الصورة الكاملة

تصوّر الرحلة الكاملة لقراءة حساس واحدة:

حساس حرارة → PLC → OPC-UA Server → ETL Pipeline
    → تنظيف وفحص جودة → Historian DB
    → لوحة مراقبة حية (Dashboard)
    → نموذج AI يتنبأ بالأعطال
    → تنبيه صيانة تلقائي

الزمن من القراءة إلى القرار:

• مراقبة حية: < 1 ثانية
• كشف شذوذ: 1-5 ثوانٍ
• صيانة تنبؤية: دقائق إلى ساعات (تحليل دُفعات)

نصائح عملية للمهندس السوري

1. ابدأ بالبيانات الموجودة: معظم المصانع السورية لديها PLCs تسجل بيانات — ابحث عنها قبل شراء حساسات جديدة
2. الجودة قبل الكمية: 100 قراءة نظيفة أفضل من مليون قراءة مليئة بالأخطاء
3. وثّق كل شيء: أي حساس، أي وحدة قياس، أي معدل قراءة — لأنك ستنسى بعد 6 أشهر
4. خطط للتخزين: بيانات الاهتزاز بمعدل 10 kHz تملأ القرص بسرعة — استخدم الضغط وسياسات الحذف
5. اختبر خط الأنابيب على بيانات حقيقية: لأن بيانات المحاكاة لا تحتوي على المشاكل التي ستواجهك في الواقع