الرؤية الحاسوبية الصناعية (Edge AI): الدليل المعماري لدمج كاميرات الفحص الذكية مع أنظمة الـ ERP لأتمتة ضبط الجودة وخفض هوالك الإنتاج

قاعدة التكلفة المضاعفة للجودة (The 1-10-100 Quality Cost Rule)

توضح هذه القاعدة الاقتصادية الفارق الجذري بين الاكتشاف اللحظي للعيوب على خط الإنتاج والاكتشاف المتأخر في السوق:

import json
import paho.mqtt.client as mqtt
import numpy as np

# إعداد وسيط التراسل الصناعي للمصنع
mqtt_client = mqtt.Client(client_id="Edge_Camera_Line_04")
mqtt_client.connect("192.168.10.50", 1883, 60)

def evaluate_inspection_frame(frame, work_order_id, part_serial):
    """
    استدلال فحص الجودة على جهاز الحافة عبر TensorRT Engine
    إذا تم رصد عيب: نطلق إشارة PLC فيزيائية ونبث حدث MQTT للـ ERP
    """
    # استدعاء نموذج كشف الشذوذ (TensorRT FP16 Engine)
    # output: is_defective (bool), defect_type (str), confidence (float)
    is_defective, defect_type, confidence = run_tensorrt_inference(frame)

    if is_defective and confidence > 0.88:
        # أ. نبضة فيزيائية فورية لوحدة الـ PLC لطرد القطعة من السير (< 5ms)
        trigger_plc_ejector_pulse(line_pin=12)

        # ب. إعداد الحزمة اللوجستية لنظام الـ ERP
        payload = {
            "event": "QUALITY_DEFECT_DETECTED",
            "work_order_id": work_order_id,
            "part_serial": part_serial,
            "work_center": "CNC-Milling-02",
            "defect_type": defect_type,
            "confidence_score": round(confidence, 4),
            "action_taken": "PHYSICAL_REJECTED"
        }
        
        # بث البلاغ المشفر لنظام الـ ERP دون تعطيل حلقة معالجة الإطارات التالية
        mqtt_client.publish(
            "factory/line04/quality/defects",
            json.dumps(payload),
            qos=1
        )
        return False

    return True

from django.db import transaction
from rest_framework.views import APIView
from rest_framework.response import Response
from cv.models import WorkOrder, StockEntry, MaintenanceTicket, DefectLog

class EdgeQualityWebhookView(APIView):
    """
    استقبال أحداث كاميرات الذكاء الاصطناعي وترحيل الأثر المخزني والصياني بضمان الذرية المحاسبية
    """
    @transaction.atomic
    def post(self, request):
        data = request.data
        work_order = WorkOrder.objects.select_for_update().get(id=data["work_order_id"])

        # 1. توثيق سجل العيب الفني
        defect = DefectLog.objects.create(
            work_order=work_order,
            part_serial=data["part_serial"],
            work_center=data["work_center"],
            defect_type=data["defect_type"],
            confidence=data["confidence_score"]
        )

        # 2. نقل القطعة إلى مخزن الحجر الصحي (Quarantine Warehouse) لمنع بيعها
        StockEntry.objects.create(
            entry_type="Quarantine Transfer",
            from_warehouse="WIP-Floor",
            to_warehouse="Quarantine-Storage",
            part_serial=data["part_serial"],
            remarks=f"عزل آلي بواسطة كاميرا الفحص - عيب: {data['defect_type']}"
        )

        # 3. تحديث عداد الهوالك في أمر التشغيل
        work_order.scrap_count += 1
        work_order.save(update_fields=['scrap_count'])

        # 4. فحص شرط تكرار العيوب لإصدار أمر صيانة استباقي للماكينة
        recent_defects_count = DefectLog.objects.filter(
            work_center=data["work_center"],
            defect_type=data["defect_type"],
            created_at__gte=timezone.now() - timedelta(minutes=15)
        ).count()

        if recent_defects_count >= 3:
            MaintenanceTicket.objects.get_or_create(
                work_center=data["work_center"],
                status="Urgent",
                defaults={
                    "title": f"صيانة وقائية طارئة: تكرار عيب {data['defect_type']} 3 مرات متتالية",
                    "assigned_team": "Mechanical Maintenance",
                    "auto_generated": True
                }
            )

        return Response({"status": "SUCCESS", "message": "Quarantine & Scrap Processed"}, status=200)