محرك التنبيهات

لا يفشل نظام المراقبة لأن عتبة موجودة; يفشل لأن ذروة واحدة تصبح 500 تنبيه متطابق. يبني هذا المشروع المحرك الصغير الأمين خلف ذلك الحكم: صف Rule يراقب نافذة متدحرجة من العينات, ويُطلق تنبيهًا فقط حين تصمد العتبة فعلًا, ثم يصمت خلال فترة تبريد حتى تُبلَّغ الحادثة المستمرة مرة واحدة بدل كل ثانية. تُسلسل الحالة إلى JSON فينجو المحرك من إعادة تشغيل في منتصف حادثة, وكل ذلك يعمل على دفق اصطناعي حتمي يمكنك إنجابه بالضبط. ينتج المحرك تنبيهين حقيقيين بالضبط من دفق من ثماني عينات مُقيَّد ، لا أكثر, لا أقل ، وستعرف لماذا.

هذا يفترض صفوفًا وطرائق وتقطيعًا زائد راحة مع JSON كبيانات. لا شيء هنا مُقيَّم؛ هذا المشروع اختياري وغير مُقيَّم ، راجع مشاريع من العالم الحقيقي للاطلاع على القائمة الكاملة والنامية.

🎯 ما ستفعله

1. تعريف صف Rule يحمل حالة المقياس والعامل والعتبة والنافذة والتبريد.
2. تمرير سلسلة زمنية اصطناعية عبر القاعدة والتنبؤ بأي العيّنتين ستنبهان.
3. تنفيذ التبريد الذي يحوّل الانفجارات إلى حوادث منفصلة.
4. أخذ لقطة لقاعدة واستعادتها من/إلى JSON دون فقدان حالتها في منتصف الحادثة.
5. تجميع تنبيهات كل قاعدة وطباعة سطر الملخص.

أين تُشغّل هذا

محليًا باستخدام uv هو المسار الموصى به ، المحرك Python نقي (يلزم json فقط), فالـuv init العادي يمنحك كل شيء.

Google Colab وKaggle Notebooks وBinder تشغّل كل خطوة دون تعديل ، لا تبعيات pip, والدفق الاصطناعي حتمي. لا شيء خاص بالمنصة يقف بين دفتر والمحرك الكامل.

الإعداد

كل ما يلزم قبل أن يعمل المحرك: مشروع في مجلد, ومفردات مشتركة لما هما «عيّنة» و«قاعدة».

أعِدَّ المشروع

uv init alerting-engine
cd alerting-engine

لا تبعيات. يقرأ المحرك دفقًا من عينات {"metric": value} وقائمة قواعد; كلاهما كائنات Python عادية.

✅ قائمة التحقق

• ✅ يُنشئ uv init alerting-engine المشروع وملف main.py.
• ✅ ينجح uv run python3 -c "import json" (json هو الاستيراد الوحيد).

🤔 سؤال (أسئلة) سقراطي(ة) :

• قاعدة بلا نافذة وبلا تبريد ما هي إلا مقارنة بنقطة واحدة. ما الذي ينكسر فعلًا في الإنتاج حين تُقيَّم عتبة على عينة واحدة بلا كبت ، وأي الآليتين (النافذة, التبريد) يصلح فشل «ذروة واحدة = 500 تنبيه»؟
• يغذّي المحرك سلسلة زمنية اصطناعية, حتمية عبر الأجهزة. لماذا يشتري لك ذلك شيئًا لا يستطيع دفق دائم البث إعطاءه ، وماذا تفقد لو استبدلت البذرة بتيار أجهزة استشعار حقيقي؟

الخطوة 1: عرّف صف القاعدة الجوهري

1.1 البنَّاء (constructor)

👟 تلميح البداية : اكتب Rule(metric, op, threshold, window=5, cooldown=3) يحمل وسائط القاعدة زائد حالتين تتغيران مع الزمن: history (العينات المتدحرجة) وlast_fired (زمن آخر تنبيه).

# main.py
import json

class Rule:
    def __init__(self, metric, op, threshold, window=5, cooldown=3):
        self.metric = metric
        self.op = op
        self.threshold = threshold
        self.window = window
        self.cooldown = cooldown
        self.history = []
        self.last_fired = -10**9

البنَّاء هو إعداد القاعدة كله: أي مقياس تراقب, وأي اتجاه (gt أو lt), وأي حد يُعد خرقًا, ومقبضا الكبت الاثنان. الحقلان القابلان للتحوير ، history وlast_fired ، ليسا وسيطي بنَّاء عن قصد: يمثلان حالة القاعدة المتعلَّمة عبر الزمن, وهي بالضبط ما ستُسلسله الخطوة 4.

🎯 الناتج المتوقع : لا ناتج من البناء ، لكن r.metric == "load", وr.window == 5, وr.history == [] كلها true.

🩹 إذا لم يعمل : إن غاب metric, فمررت وسيطًا موضعيًا إلى حقل غير مدرج في __init__. إن كان window افتراضيًا 5 لكنك استدعيت Rule("load", "gt", 5.0, 4), فمررت 4 وسائط موضعية فقط ، يصبح window هو الموضع الرابع ويبقى cooldown على افتراضيه.

1.2 مثِّل الخرق

👟 تلميح البداية : أضف مساعدة _is_breach(value) تجيب عن «هل عيّنة واحدة فوق العتبة (لـgt) أو تحتها (لـlt)?» ، القرار الرياضي الوحيد للمحرك.

# main.py (continued)
    def _is_breach(self, value):
        if self.op == "gt":
            return value > self.threshold
        if self.op == "lt":
            return value < self.threshold
        raise ValueError(f"unknown op {self.op}")

print(Rule("a", "gt", 5.0)._is_breach(6.0))
print(Rule("a", "lt", 5.0)._is_breach(6.0))

_is_breach مسند نقي: نفس القيمة, نفس الإجابة, كل مرة. إبقاؤها طريقة منفصلة يعني أن منطق النافذة في الخطوة 2 لا يضطر أبدًا لمعرفة إن كان gt أو lt يعني «سيئ» ، يسأل هذه الطريقة فقط. raise على عامل مجهول هو الحارس الذي يفشل بسرعة فيلتقط التهجئة الخاطئة "LT" بدل الصمت وعدم التنبيه إلى الأبد.

🎯 الناتج المتوقع : True ثم False ، القاعدة الأولى تخترق على 6.0 > 5, والثانية لا تفعل لأن 6.0 < 5 خاطئة.

🩹 إذا لم يعمل : إن طبع كلاهما True, ففرع lt نسي <. إن ظهر ValueError, فاستدعيت البنَّاء بـop="lt" بحروف مختلفة عما يفحصه الأسلوب ، طبّع op.lower() في البنَّاء.

1.3 تحقّق من الصف

✅ قائمة التحقق

• ✅ Rule("load", "gt", 5.0) تملك window=5 وcooldown=3 وhistory فارغة وlast_fired في الماضي البعيد.
• ✅ تُعيد _is_breach قيمًا منطقية وترفع على عامل مجهول.
• ✅ تتصرف قاعدتا gt وlt عكسيًّا على نفس القيمة.

🤔 سؤال (أسئلة) سقراطي(ة) :

• last_fired = -10**9 هو حارس «منذ زمن بعيد». لماذا السالب حرفيًا «منذ زمن بعيد», لا مجرد «صفر» ، وكيف ستبدو نسخة last_fired = None من فحص التبريد؟
• تقرر _is_breach على عيّنة واحدة, لكن الخطوة 2 ترفع ذلك إلى نافذة. ما الفرق المفهومي بين «عيّنة واحدة تساوي 6.0» و«أقصى آخر 5 عينات لي يساوي 6.0» ، وأيهما تعريف أفضل للحادثة؟

الخطوة 2: راقب نافذة متدحرجة

عيّنة واحدة ضجيج; نافذة إشارة. تحوّل الخطوة 2 المسند النقي _is_breach إلى قرار بنوافذ ، لكن بحذر, كي يبقى «التبريد» من الخطوة 3 منفصلًا.

2.1 غذِّ القاعدة عيّناتك

👟 تلميح البداية : نفّذ evaluate(t, value) الذي يلحق بـhistory, ويقصّها إلى النافذة, ويعيد False عادةً ، منطق الإطلاق يأتي في الخطوة 3.

# main.py (continued)
    def evaluate(self, t, value):
        self.history.append(value)
        self.history = self.history[-self.window:]
        return False   # window check lives in Step 3

r = Rule("load", "gt", 5.0, window=4)
for t, v in enumerate([1.0, 2.0, 3.0, 6.0, 4.0, 1.0, 1.0, 9.0]):
    r.evaluate(t, v)
print(r.history)

self.history[-self.window:] هو تعبير النافذة المتدحرجة: يبقي آخر window عينة فقط, فتظل الذاكرة مقيدة مهما طال الدفق. قصّ الذيل هو قصة الصواب والكفاءة معًا بضربة واحدة. لاحظ أن evaluate ما زالت تُعيد False هنا ، يحدث حفظ النافذة أولًا, ويأتي القرار في الخطوة 3.

🎯 الناتج المتوقع : [4.0, 1.0, 1.0, 9.0] ، بعد 8 قيم بـwindow=4, احتفظ المحرك بالعيّنات الأربع الأخيرة بالضبط.

🩹 إذا لم يعمل : إن كان r.history أطول من 4, فاستُبدل التقطيع [-self.window:] بـ.append فقط. إن كان أقصر حين يكون الدفق قصيرًا, فهذا سلوك صحيح (لا تملك القاعدة تاريخ 4 عينات حتى ترى 4 عينات) ، لا خطأ.

2.2 أضف فحص الخرق بالعدد

👟 تلميح البداية : استبدل return False بالقرار الحقيقي: max(self.history) > self.threshold لقواعد gt, وmin(...) < self.threshold لـlt ، لكن فقط حين تكن النافذة ممتلئة.

# main.py (continued)
    def _window_holds(self):
        if len(self.history) < self.window:
            return False
        if self.op == "gt":
            return max(self.history) > self.threshold
        return min(self.history) < self.threshold

    def evaluate(self, t, value):
        self.history.append(value)
        self.history = self.history[-self.window:]
        return self._window_holds()

r = Rule("load", "gt", 5.0, window=4)
for t, v in enumerate([1.0, 2.0, 3.0, 6.0]):
    print(t, v, "window-holds?", r.evaluate(t, v))

تتطلب _window_holds أن تكون النافذة ممتلئة قبل الوثوق بـmax/min ، نافذة من عيّنة واحدة تجاوزت العتبة صدفة ليست حادثة بعد. فقط حين يبلغ history حجم window تعني مقارنة max/min «هذا مستمر عبر النافذة». هذه هي الخطوة التي يتحول فيها «ذروة واحدة» إلى «حادثة حقيقية تؤكدها النافذة».

🎯 الناتج المتوقع :

0 1.0 window-holds? False
1 2.0 window-holds? False
2 3.0 window-holds? False
3 6.0 window-holds? True

مع أن 6.0 تتجاوز 5.0, ينتظر الطاقم حتى يكتمل الجيران في النافذة ليعتبرها حادثة.

🩹 إذا لم يعمل : إن كانت window-holds? True في وقت مبكر, فحارس len(history) < window مفقود. إن كانت False عند t=3 والنافذة [1,2,3,6], فليست max أكبر من 5 لأن القيمة المغذّاة 6 لا 6.0, أو أن مقارنة العتبة معكوسة.

2.3 تحقّق من النافذة

✅ قائمة التحقق

• ✅ تبقى history على window عيّنة بالضبط حالما يتجاوزها الدفق.
• ✅ تُعيد _window_holds False حتى تمتلئ النافذة.
• ✅ نافذة ممتلئة يعبر max/min بها العتبة تُعيد True, والتي لا تعبر تُعيد False.

🤔 سؤال (أسئلة) سقراطي(ة) :

• النافذة حصريًا عن «هل العيّنة وسط جيران فوق العتبة». ماذا يحدث لقاعدة gt تراقب مقياسًا مرتفعًا دائمًا لكنه يزحف ببطء؟ هل ستنطلق _window_holds, وهل نافذة قائمة على max هي الأداة الصحيحة لانجراف بطيء؟
• self.history[-self.window:] تُسقط العينات القديمة تمامًا. إن أردت معرفة «كم مرة أطلقت هذه القاعدة في الشهر الماضي», فما الحالة الإضافية التي ستحتفظ بها ، ولماذا يتخلص تصميم المحرك الحالي منها عن قصد؟

الخطوة 3: أضف التبريد ، حادثة واحدة, لا عاصفة

تقول النافذة إن العتبة تصمد; يقول التبريد لا تقلها مجددًا بعد أن قلتها لتوّك. هذا هو المقبض الذي يحوّل الانفجار إلى مجموعة حوادث منفصلة.

3.1 افهم التبريد

👟 تلميح البداية : مدّد evaluate بحيث, بعد إطلاق, تبقى القاعدة صامتة cooldown خطوة زمنية ، if t - self.last_fired < self.cooldown: return False.

# main.py (continued)
    def evaluate(self, t, value):
        self.history.append(value)
        self.history = self.history[-self.window:]
        if t - self.last_fired < self.cooldown:
            return False                # still quiet from the last alert
        if self._window_holds():
            self.last_fired = t         # remember when this incident fired
            return True
        return False

r = Rule("load", "gt", 5.0, window=4, cooldown=3)
seq = [1.0, 2.0, 3.0, 6.0, 4.0, 1.0, 1.0, 9.0]
alerts = [t for t, v in enumerate(seq) if r.evaluate(t, v)]
print("base alerts:", alerts)

التبريد هو قلب المحرك: يُختم last_fired لحظة الإطلاق, ولكل خطوة زمنية من خطوات cooldown التالية تُكبَت كل عيّنة ، حتى التي ما زالت فوق العتبة. النتيجة هي نموذج الحادثة الكلاسيكي: طفرة 6.0 تنطلق مرة, والقيم المرتفعة اللاحقة والانخفاض القصير صامتة, وخرق جديد لاحق ينطلق مجددًا. تنبيهان منفصلان من نافذة 4 عينات, بالضبط.

🎯 الناتج المتوقع : base alerts: [3, 6] ، الخرق الأول عند t=3 والخرق المتجدد عند t=6, مع كبت عيّنتي t=4 وt=5 بالتبريد. (t=5 مكبوتة لأن 5 - 3 = 2 < 3.)

🩹 إذا لم يعمل : إن أظهر التنبيهات [3, 4, 5, 6, 7], فإما أن last_fired لا يُضبط (سطر self.last_fired = t مفقود) أو أن فحص التبريد ليس t - self.last_fired < self.cooldown (زلة < مقابل <= تغيّر الحدود). إن لم تكن تنبيهات إطلاقًا, فـlast_fired يُعاد ضبطه على كل عيّنة غير مُطلقة.

3.2 قاعدة lt تعكسها

👟 تلميح البداية : تراقب قاعدة lt عبر min(self.history) < self.threshold ، منطق التبريد متطابق; ينقلب المسند فقط.

# main.py (continued)
r = Rule("mem", "lt", 20.0, window=3, cooldown=2)
alerts_lt = [t for t, v in enumerate([90.0, 85.0, 88.0, 12.0, 18.0, 40.0, 30.0])
             if r.evaluate(t, v)]
print("low-mem alerts:", alerts_lt)

حين تهبط الذاكرة الحرة تحت 20, فهذه حادثة ذاكرة منخفضة. يعمل التبريد بالطريقة ذاتها: 12.0 الأولى تنطلق, و18.0 التي تليها مباشرة مكبوتة, وخرب لاحق (نزول ثانٍ بعد التعافي, أو قراءة جديدة) يصبح تنبيهًا منفصلًا.

🎯 الناتج المتوقع : low-mem alerts: [3, 5] ، خرق عند t=3 (12.0), تُكبَت t=4 (التبريد: 4 − 3 = 1 < 2), وعند t=5 انتهى التبريد والنافذة [40, 18, 12] ما زالت تحمل 12.0 دون 20 فتنطلق ، 40.0 نفسها ليست دون 20, لكن بقاء 12.0 في النافذة هو ما يشعل. جرّب يدويًا خطوة بخطوة إن اختلف ناتجك.

🩹 إذا لم يعمل : إن اختلف alerts_lt عن تتبّعك اليدوي, فخطِّ القاعدة عيّنة بعيّنة واطبع history وmin(history) وlast_fired ، يتفاعل تقليم النافذة مع التبريد, وطباعة الاثنين تكشف أين يتباينان بالضبط.

3.3 تحقّق من التبريد

✅ قائمة التحقق

• ✅ تنتج قاعدة gt على دفق 8 عينات [3, 6] بالضبط ، حادثتان.
• ✅ بين إطلاقين, تمر على الأقل cooldown عيّنة بصمت.
• ✅ تتشارك قاعدتا gt وlt نفس آليات التبريد, ولا تختلفان إلا في المسند.

🤔 سؤال (أسئلة) سقراطي(ة) :

• يكبُت التبريد كل عيّنة لخطوات cooldown, حتى ذروة جديدة مضاعفة ×10 فعلًا. هل هذا هو المقايضة الصحيحة لجهاز تنبيه حقيقي, أم تريد «أكبر تنبيه يربح» بدل ذلك ، وأين يعيش هذا المنطق؟
• يُختم last_fired بالزمن t, لا بفهرس العيّنة. في نظام يعالج حزم عينات دفعة واحدة (يقفز t بمقدار 100), كيف يسيئ فحص t - last_fired التصرف, وماذا ستخزّن بدل ذلك؟

الخطوة 4: ثبّت الحالة واستعدها

محرك ينسى أنه أطلق فعلًا خلال إعادة تشغيل يعيد التنبيه على نفس الحادثة. تسلسل الخطوة 4 حالة كل قاعدة المتعلَّمة ، لا إعدادها فقط ، كي يصمد الاستمرار.

4.1 التقط لقطة لقاعدة

👟 تلميح البداية : أضف snapshot() تُعيد قاموس الإعداد زائد history وlast_fired, وfrom_snapshot يستعدهما.

# main.py (continued)
    def snapshot(self):
        return {"metric": self.metric, "op": self.op, "threshold": self.threshold,
                "window": self.window, "cooldown": self.cooldown,
                "history": self.history, "last_fired": self.last_fired}

    @classmethod
    def from_snapshot(cls, snap):
        r = cls(snap["metric"], snap["op"], snap["threshold"],
                snap["window"], snap["cooldown"])
        r.history = snap["history"]
        r.last_fired = snap["last_fired"]
        return r

r = Rule("load", "gt", 5.0, window=4, cooldown=3)
for t, v in enumerate([1.0, 2.0, 3.0, 6.0, 4.0, 1.0, 1.0, 9.0]):
    r.evaluate(t, v)
snap = json.dumps(r.snapshot())
print("saved", snap)

json.dumps للقطة هو عقد الاستمرار: كل حقل يلزم لاستئناف القاعدة أصبح الآن قاموسًا قابلًا للتسلسل إلى JSON. يعيد from_snapshot بناء صف Rule جديد وينسخ الحقلين المتعلَّمين, فساعة تبريد القاعدة المستعادة ونافذتها في الموضع الذي تركتهما العملية بالضبط.

🎯 الناتج المتوقع : سلسلة JSON تحوي "metric": "load" و"window": 4 و"history" و"last_fired": 6.

🩹 إذا لم يعمل : إن فشل json.dumps على قيمة غير قابلة للتسلسل, فأصبح last_fired أو history نوع numpy ، لفّها بـint(...)/float(...) قبل الإغراق. إن أغفل الناتج history, فمفتاح القاموس ليس في snapshot().

4.2 استعد ولا تُعد التنبيه على نفس الحادثة

👟 تلميح البداية : ألغِ التسلسل, وأعد البناء, وغذِّ استمرار الدفق ، يجب أن تبقى القاعدة المستعادة صامتة على العيّنات التي ما زالت ضمن تبريد آخر زمن إطلاق.

# main.py (continued)
import json
restored = Rule.from_snapshot(json.loads(snap))
print("history carried:", restored.history, "last_fired:", restored.last_fired)
for t, v in enumerate([6.0, 7.0, 8.0, 5.0], start=6):
    print("t", t, "v", v, "->", "ALERT" if restored.evaluate(t, v) else "quiet")

استعادة القاعدة والاستمرار عند t=6 يعيد إنجاب الحالة الحية: 6.0, 7.0, 8.0 من الدفق كلها ضمن تبريد إطلاق t=6 (أو تُطلفه مرة ثم تصمت), ونزول جديد فعلًا يُطلق طازجًا. الخاصية المفتاحية: لا يعيد المحرك التنبيه على الحادثة التي أبلغ عنها فعلًا قبل إعادة التشغيل.

🎯 الناتج المتوقع : history carried: [4.0, 1.0, 1.0, 9.0] last_fired: 6 يتبعها أثر استمرار يُطلق مرة واحدة على الأكثر في نافذة التبريد.

🩹 إذا لم يعمل : إن أطلقت القاعدة المستعادة على أول عيّنة متواصلة, فلن يُنسخ last_fired عبر from_snapshot (عاد إلى -10**9). إن لم تنطلق أبدًا على النزول الطازج, فـhistory نُسخت بإفراط والنافذة ما زالت تحمل قيمة مرتفعة قديمة ، تحقق من طول النافذة بعد الاستعادة.

4.3 تحقّق من الاستمرار

✅ قائمة التحقق

• ✅ يُتمم json.dumps(r.snapshot()) رحلة ذهاب وإياب عبر loads وfrom_snapshot.
• ✅ تحمل الحالة المستعادة history وlast_fired معًا; ويطابق history ذيل ما قبل الحفظ.
• ✅ Rule.from_snapshot(json.loads(snap)) == Rule.from_snapshot(json.loads(snap)) سلوكيًا ، استعادتان من نفس الكتلة تتصرفان بنفس الطريقة.

🤔 سؤال (أسئلة) سقراطي(ة) :

• ينسخ from_snapshot last_fired لكن لا شيء آخر يتحور بين إعادة التشغيلين. ماذا يحدث لو غيّرت نسخة كود جديدة قيمة window واستعدت كتلة قديمة تاريخُها بطول مختلف؟ هل هذا قلق مخطط بيانات أم قلق نسخة كود؟
• اللقطة قاموس واحد. لو كان لديك 50 قاعدة, أتخزن 50 ملفًا, أم مصفوفة JSON واحدة, أم قاموسًا بمفاتيح؟ ما الذي يجعل كل خيار صحيحًا لمحرك صغير خاطئًا عند المقياس؟

الخطوة 5: شغّل الدفق ولخّص

اكتمل المحرك. تدمج الخطوة 5 قواعد متعددة بدفق مُقيَّد وتطبع حكم السطر الواحد الذي يقرؤه مشغّل متعب فعلًا: أي مقياس أطلق, وكم مرة.

5.1 مرّر الدفق عبر كل القواعد

👟 تلميح البداية : احتفظ بقائمة قواعد, وغذِّ كل عيّنة لكل قاعدة, واجمع نتائج الإطلاق مفاتيحها المقاييس.

# main.py (continued)
rules = [
    Rule("load", "gt", 5.0, window=4, cooldown=3),
    Rule("mem_free", "lt", 20.0, window=3, cooldown=2),
]
stream = [
    {"t": 0, "load": 1.0, "mem_free": 90.0},
    {"t": 1, "load": 2.0, "mem_free": 85.0},
    {"t": 2, "load": 3.0, "mem_free": 88.0},
    {"t": 3, "load": 6.0, "mem_free": 12.0},
    {"t": 4, "load": 4.0, "mem_free": 18.0},
    {"t": 5, "load": 1.0, "mem_free": 40.0},
    {"t": 6, "load": 1.0, "mem_free": 30.0},
    {"t": 7, "load": 9.0, "mem_free": 28.0},
]
alerts = {}
for sample in stream:
    for rule in rules:
        if rule.evaluate(sample["t"], sample[rule.metric]):
            alerts.setdefault(rule.metric, []).append(sample["t"])
print(alerts)

sample[rule.metric] هو توجيه المقياس: تسحب كل قاعدة قيمتها من عيّنة الدفق المشتركة, فتمريرة واحدة عبر الدفق تقود كل قاعدة. alerts.setdefault(rule.metric, []).append(...) يبني قائمة أزمنة إطلاق لكل مقياس دون فحصٍ صريح «هل بدأت هذه القائمة فعلًا».

🎯 الناتج المتوقع : {'load': [3, 6], 'mem_free': [3, 5]} ، حادثتا قاعدة الحِمل وحادثتا قاعدة الذاكرة, كلها من دفق واحد من 8 عينات.

🩹 إذا لم يعمل : إن غابت قائمة مقياس, فقاعدته لم تُطلق قط (تحقق من عتبتها/مسندها ضد الدفق) أو لم يحصل مفتاح setdefault على أول إلحاق. إن أطلق مقياس أكثر من المتوقع, فالتبريد أو النافذة عليه متعطل.

5.2 اطبع الملخص

👟 تلميح البداية : أضف summarize(alerts) مجهريًا كي يرى المشغّل عدّادات لا أزمنة خام.

# main.py (continued)
def summarize(alerts):
    return {metric: len(times) for metric, times in alerts.items()}

print("OPERATOR SUMMARY:", summarize(alerts))

len(times) هو الضغط الودِّي للمشغّل: دفق 200 عيّنة أنتج 3 حوادث لـload وواحدة لـmem_free يلخصه بـ{'load': 3, 'mem_free': 1} في سطر واحد. كل عدّ مشتق من قائمة الإطلاقات الفعلية, فالملخص لا يستطيع الكذب بشأن ما أبلغ عنه المحرك.

🎯 الناتج المتوقع : OPERATOR SUMMARY: {'load': 2, 'mem_free': 2}.

🩹 إذا لم يعمل : إن أظهر الملخص مقياسًا بـ0 مع أنه أطلق, فبُني alerts بـsetdefault طازج على متغير مختلف. إن أظهر أكثر من المتوقع, فغذّى الدفق قيم t مكررة وحسبها التبريد حوادث منفصلة.

5.3 تحقّق من المحرك من طرفٍ إلى طرف

✅ قائمة التحقق

• ✅ ينتج دفق 8 عينات {'load': [3, 6], 'mem_free': [3, 5]} ، أربعة تنبيهات بالضبط.
• ✅ يطبع الملخص عدّادات مشتقة من تلك القوائم.
• ✅ يعمل الدفق والقواعد دون تعديل في دفتر أو طرفية.

🤔 سؤال (أسئلة) سقراطي(ة) :

• يعدّ الملخص len(times). إن تمددت نفس الحادثة عبر إعادة تشغيل, فلن تُطلق القاعدة المستعادة (بصواب) مجددًا, فالعدّ أدنى مما توحي به العيّنات الخام. هل «عدّ الإطلاقات» هو نفسه «عدّ الحوادث» ، وماذا ستضيف كي يميّز الملخص بينهما؟
• ينطلق mem_free من الدفق عند t=3 وt=5. تتبّع هل t=5 حادثة جديدة (ذاكرة تتعافى ثم تهبط مجددًا) أم الحلقة نفسها تسطح عبر تبريد أقصر ، واذكر أي فرضية يرمز إليها قيمة التبريد 2.

⚠️ مآزق شائعة

• نسيان حارس النافذة الكاملة. تقييم max(history) على نافذة من 2 عيّنة قبل بلوغها window يعامل ذروة صغيرة كحادثة. اشترط len(history) == window (أو >=) قبل الوثوق بـmax/min.
• تبريد لا يُطلق أبدًا مجددًا. إذا ضُبط last_fired على كل عيّنة (لا عند الإطلاق فقط), تبقى القاعدة صامتة إلى الأبد. اختم last_fired داخل فرع if self._window_holds(): فقط.
• < مقابل <= في التبريد. يكبُت t - last_fired < cooldown لخطوات cooldown بالضبط; يكبُت <= عيّنة أقل. اختر واحدًا وافهم طرف النافذة الخاص بك.
• تقطيع الطرف الخاطئ. self.history[-self.window:] يُبقي الذيل; [:self.window] يُبقي الرأس وسيراقب ماضي الدفق بعد زوال أهميته بزمن طويل.
• عدم معالجة الأنواع غير القابلة للتسلسل. json.dumps للقطة بها عدد صحيح numpy (last_fired من np.arange) يرفع. اجعلها int/float عادية قبل التثبيت.
• إعادة التنبيه بعد إعادة تشغيل. استعادة قاعدة مع نسيان نسخ last_fired تجعل المحرك المستعاد يعيد إطلاق الحادثة التي قد أبلغ عنها. استعد دائمًا history وlast_fired معًا.

ما بنيته للتو

محرك مراقبة بحالة: قواعد تراقب نوافذ متدحرجة, وتبريد يحوّل الانفجارات إلى حوادث منفصلة, واستمرار JSON ينجو من إعادة التشغيل, وملخص مشغّل من سطر واحد. البصيرة الجوهرية هي أن التنبيه قرار بحالة, لا مقارنة ، تجيب النافذة عن «هل هذا مستمر؟», ويجيب التبريد عن «ألم أقله فعلًا؟», وlast_fired هي الذاكرة التي تربطهما. الانقسام الثلاثي ذاك ينتقل إلى محددات المعدل, وتراجع إعادة المحاولة, وارتداد الزر (debouncing), وأي كود يجب أن يقرر متى يتكلم مقابل متى يبقى صامتًا.

إلى أين تذهب من هنا

• أضف تأثيرًا جانبيًا email() يطبع سطر التنبيه فقط حين تُطلق قاعدة, وخُبّه خلف نفس التبريد كي تنتج الحادثة بريدًا واحدًا لا واحدًا لكل عيّنة.
• مدّد Rule بحقل severity واجعل summarize تعدّ حوادث الحرِجة منفصلة عن التحذيرية.
• ثبّت قائمة rules كلها مع alerts معًا في كتلة JSON واحدة كي تستعيد إعادة تشغيل كاملة الإعداد ولوحة المشغّل معًا.
• غذِّ المحرك بيانات CPU/ذاكرة حية من psutil وقارن عدّاد تنبيهاته عبر يوم بدفق الاصطناع.

شارك مشروعك مع الصف

بنيت شيئًا تفخر به؟ examples/student-projects/ معرض لمشاريع قدّمها طلاب آخرون ، وREADME الخاص به يحوي شرحًا كاملًا وودودًا للمبتدئين لإضافة مشروعك عبر pull request, حتى لو لم تستخدم git من قبل قط: عمل fork للمستودع, وإنشاء فرع, وتثبيت ملفاتك, وفتح الـ PR, خطوة بخطوة. لا يُفترض أي خبرة سابقة بـ git.

مرحبًا بك في كتابة Python خارج المتصفح. 🎓