Ở bài trước, chúng ta đã làm được một việc rất quan trọng: Raspberry Pi không chỉ đọc tín hiệu từ cảm biến cửa từ và PIR, mà còn bắt đầu biến chúng thành event có ý nghĩa như:

• door_open
• door_closed
• motion_detected
• motion_idle

Nhưng nếu dừng ở đó, hệ thống vẫn mới chỉ là một bộ ghi nhận sự kiện. Nó biết chuyện gì vừa xảy ra, nhưng chưa biết:

• sự kiện nào là bình thường
• sự kiện nào cần cảnh báo
• khi nào nên gửi notify
• khi nào nên bỏ qua để tránh spam

Đó là lúc chúng ta cần một lớp logic mới:

👉 Rule Engine – nơi quyết định hành động của hệ thống dựa trên các event đầu vào.

Trong bài này, chúng ta sẽ xây dựng một bộ rule cảnh báo thông minh cho hệ thống giám sát cửa và chuyển động bằng Raspberry Pi, với các mục tiêu:

• phân biệt trạng thái armed / disarmed
• cảnh báo khi cửa mở bất thường
• cảnh báo khi PIR phát hiện chuyển động vào ban đêm
• giảm cảnh báo giả
• tránh gửi notify lặp lại quá nhiều

Đây là bước nâng hệ thống từ mức “đọc sensor” lên mức “giám sát thông minh”.

Bạn sẽ làm được gì sau bài này

Sau khi hoàn thành bài 24.4, bạn sẽ có thể:

• xây rule engine đơn giản bằng Python
• phân biệt event chỉ để log và event cần cảnh báo
• dùng trạng thái armed / disarmed để bật hoặc tắt giám sát
• thêm cooldown để chống spam cảnh báo
• giảm false positive từ PIR hoặc reed switch

Rule cảnh báo là gì

Hãy tưởng tượng hệ thống của bạn nhận được các event như sau:

08:00:10 door_open
08:00:12 door_closed
08:00:30 motion_detected
08:00:31 motion_idle

Nếu chỉ ghi log thì mọi event đều được lưu lại, nhưng chưa có “ý nghĩa hành động”.

Rule cảnh báo chính là lớp chuyển từ:

event thô -> quyết định

Ví dụ:

• nếu door_open xảy ra khi đang armed -> cảnh báo
• nếu motion_detected xảy ra vào ban đêm -> cảnh báo
• nếu vừa cảnh báo xong trong 10 giây trước -> bỏ qua lần tiếp theo
• nếu PIR rung liên tục trong thời gian ngắn -> gộp lại thành một cảnh báo

Nói ngắn gọn, rule engine giúp hệ thống biết:

👉 khi nào nên phản ứng, và phản ứng như thế nào

Mục tiêu của rule engine trong mini series này

Ở mức học tập và ứng dụng mini, chúng ta không cần xây một rule engine quá phức tạp.
Chỉ cần hệ thống làm được 4 việc chính là đã rất hữu ích:

1. Bật / tắt chế độ giám sát

Hệ thống cần biết khi nào đang theo dõi thật sự.

2. Nhận diện sự kiện bất thường

Ví dụ cửa mở khi không nên mở.

3. Giảm cảnh báo giả

Ví dụ PIR chập chờn hoặc cảm biến cửa rung nhẹ.

4. Tránh spam notify

Không gửi cùng một kiểu cảnh báo liên tục trong vài giây.

Đây chính là bộ khung rule đủ tốt cho một hệ thống giám sát nhà thông minh mini.

Khái niệm armed / disarmed

Đây là khái niệm rất quan trọng trong mọi hệ thống giám sát.

Armed

Nghĩa là:

• hệ thống đang ở chế độ theo dõi
• nếu có event phù hợp, hệ thống sẽ tạo cảnh báo

Disarmed

Nghĩa là:

• hệ thống vẫn có thể đọc sensor và log event
• nhưng không phát cảnh báo

Điều này rất hợp lý trong thực tế.

Ví dụ:

• ban ngày, bạn ở nhà -> hệ thống có thể ở chế độ disarmed
• ban đêm hoặc khi ra ngoài -> chuyển sang armed

Khi đó:

• mở cửa ban ngày có thể chỉ log
• mở cửa ban đêm khi armed thì phải cảnh báo

Tư duy rất quan trọng

Không phải mọi event đều là cảnh báo.
Nhiều event chỉ là hoạt động bình thường.

Rule engine chính là nơi phân biệt hai loại đó.

Những rule thực tế chúng ta sẽ xây trong bài này

Trong bài này, chúng ta sẽ tập trung vào 4 rule chính.

Rule 1: Cửa mở khi hệ thống đang armed

Nếu cửa mở trong khi hệ thống đang bật giám sát, tạo alert door_open_alert.

Rule 2: PIR phát hiện chuyển động khi hệ thống đang armed

Nếu PIR phát hiện chuyển động trong lúc giám sát đang bật, tạo alert motion_alert.

Rule 3: PIR phát hiện chuyển động vào ban đêm

Nếu thời gian hiện tại thuộc khung giờ ban đêm, có thể tăng mức ưu tiên cảnh báo.

Rule 4: Cooldown chống spam

Nếu vừa gửi cùng một loại cảnh báo cách đây chưa lâu, bỏ qua cảnh báo mới.

Ngoài ra, ta sẽ thêm một lớp đơn giản để giảm cảnh báo giả.

Kiến trúc

Từ kiến trúc ở bài trước:

Door Sensor + PIR -> Python Monitor -> Event Log

Bây giờ ta nâng cấp thành:

Door Sensor + PIR -> Event Monitor -> Rule Engine -> Alert Decision

Nói cách khác, chúng ta thêm một lớp ở giữa:

• monitor tạo event
• rule engine đọc event
• rule engine quyết định có alert hay không

Thiết kế dữ liệu event

Trước tiên, thay vì chỉ log string đơn giản, ta nên chuẩn hoá event thành object Python.

Ví dụ:

event = {
    "ts": "2026-04-11 21:30:00",
    "sensor": "door",
    "event": "door_open"
}

Hoặc:

event = {
    "ts": "2026-04-11 21:30:05",
    "sensor": "pir",
    "event": "motion_detected"
}

Cấu trúc này giúp rule engine xử lý dễ hơn nhiều.

Bắt đầu với một Rule Engine đơn giản

Ta sẽ viết một class RuleEngine để quản lý:

• trạng thái armed/disarmed
• thời điểm alert gần nhất
• quyết định alert cho từng event

Code khởi tạo

from datetime import datetime

class RuleEngine:
    def __init__(self):
        self.armed = True
        self.last_alert_times = {}
        self.cooldown_seconds = 10

Ở đây:

• armed là chế độ giám sát
• last_alert_times lưu thời điểm alert gần nhất theo từng loại
• cooldown_seconds là khoảng nghỉ để chống spam

Hàm kiểm tra ban đêm

Chúng ta sẽ cần biết event có xảy ra trong giờ nhạy cảm hay không.

Ví dụ: ban đêm từ 23:00 đến 05:00

def is_night_time():
    now = datetime.now().time()
    return now.hour >= 23 or now.hour < 5

Đây là cách đơn giản nhưng rất hữu ích để tăng độ “thông minh” cho rule.

Hàm kiểm tra cooldown

Nếu cùng một loại alert vừa xảy ra chưa lâu, ta bỏ qua cảnh báo mới.

from datetime import datetime

def should_alert(self, alert_type):
    now = datetime.now()

    if alert_type not in self.last_alert_times:
        self.last_alert_times[alert_type] = now
        return True

    seconds = (now - self.last_alert_times[alert_type]).total_seconds()

    if seconds >= self.cooldown_seconds:
        self.last_alert_times[alert_type] = now
        return True

    return False

Nhờ vậy:

• PIR rung liên tục 5 lần trong 3 giây
• hệ thống vẫn chỉ gửi 1 alert

Rule cho cảm biến cửa

Khi nhận event door_open, hệ thống sẽ xét:

• có đang armed không
• có đang trong thời gian cooldown không

Logic cơ bản

def handle_door_event(self, event):
    if event["event"] == "door_open":
        if self.armed and self.should_alert("door_open_alert"):
            return {
                "type": "door_open_alert",
                "message": "Door opened while system is armed"
            }

    return None

Nếu cửa đóng (door_closed) thì thường chỉ log, chưa cần alert.

Rule cho PIR

Với PIR, ta có thể làm nhiều mức hơn.

Logic cơ bản

def handle_pir_event(self, event):
    if event["event"] == "motion_detected":
        if self.armed and self.should_alert("motion_alert"):
            level = "high" if is_night_time() else "normal"

            return {
                "type": "motion_alert",
                "level": level,
                "message": "Motion detected while system is armed"
            }

    return None

Như vậy:

• cùng là motion_detected
• nhưng ban đêm có thể được đánh dấu high

Gộp thành một hàm xử lý chính

Ta sẽ viết một hàm chung nhận mọi event:

def process_event(self, event):
    if event["sensor"] == "door":
        return self.handle_door_event(event)

    if event["sensor"] == "pir":
        return self.handle_pir_event(event)

    return None

Khi monitor tạo ra event mới, chỉ cần truyền vào process_event().

Code hoàn chỉnh cho Rule Engine cơ bản

from datetime import datetime

def is_night_time():
    now = datetime.now().time()
    return now.hour >= 23 or now.hour < 5

class RuleEngine:
    def __init__(self):
        self.armed = True
        self.last_alert_times = {}
        self.cooldown_seconds = 10

    def should_alert(self, alert_type):
        now = datetime.now()

        if alert_type not in self.last_alert_times:
            self.last_alert_times[alert_type] = now
            return True

        seconds = (now - self.last_alert_times[alert_type]).total_seconds()

        if seconds >= self.cooldown_seconds:
            self.last_alert_times[alert_type] = now
            return True

        return False

    def handle_door_event(self, event):
        if event["event"] == "door_open":
            if self.armed and self.should_alert("door_open_alert"):
                return {
                    "type": "door_open_alert",
                    "level": "high" if is_night_time() else "normal",
                    "message": "Door opened while system is armed"
                }
        return None

    def handle_pir_event(self, event):
        if event["event"] == "motion_detected":
            if self.armed and self.should_alert("motion_alert"):
                return {
                    "type": "motion_alert",
                    "level": "high" if is_night_time() else "normal",
                    "message": "Motion detected while system is armed"
                }
        return None

    def process_event(self, event):
        if event["sensor"] == "door":
            return self.handle_door_event(event)

        if event["sensor"] == "pir":
            return self.handle_pir_event(event)

        return None

Tích hợp rule engine với monitor

Giả sử ở bài trước bạn đã có logic monitor tạo event như sau:

event = {
    "ts": now_str(),
    "sensor": "door",
    "event": "door_open"
}

Ta chỉ cần:

rule_engine = RuleEngine()

alert = rule_engine.process_event(event)

if alert:
    print("ALERT:", alert)

Ví dụ luồng thực tế

Trường hợp 1: mở cửa khi hệ thống đang armed

Input event:

{
    "ts": "2026-04-11 23:10:00",
    "sensor": "door",
    "event": "door_open"
}

Output:

{
    "type": "door_open_alert",
    "level": "high",
    "message": "Door opened while system is armed"
}

Trường hợp 2: PIR phát hiện chuyển động ban ngày

Input:

{
    "ts": "2026-04-11 14:00:00",
    "sensor": "pir",
    "event": "motion_detected"
}

Output:

{
    "type": "motion_alert",
    "level": "normal",
    "message": "Motion detected while system is armed"
}

Trường hợp 3: PIR phát hiện liên tục nhiều lần

Nếu vừa có motion_alert cách đây 3 giây và cooldown là 10 giây, output sẽ là:

None

Tức là:

• vẫn log event
• nhưng không tạo cảnh báo mới

Đây là cơ chế chống spam rất quan trọng.

Cách bật / tắt armed mode

Trong giai đoạn đầu, bạn có thể bật tay trong code:

rule_engine.armed = True

Hoặc:

rule_engine.armed = False

Nhưng tốt hơn là tạo hàm rõ ràng:

def arm(self):
    self.armed = True

def disarm(self):
    self.armed = False

Ví dụ đầy đủ:

def arm(self):
    self.armed = True
    print("System armed")

def disarm(self):
    self.armed = False
    print("System disarmed")

Về sau, bạn có thể gắn việc arm/disarm với:

• nút bấm
• command từ dashboard
• lệnh Telegram
• MQTT command

Giảm false positive như thế nào

False positive là cảnh báo giả.
Đây là chuyện rất thường gặp, nhất là với PIR.

Ví dụ:

• PIR quá nhạy
• thay đổi nhiệt độ môi trường
• người đi ngang ngoài vùng mong muốn
• reed switch rung nhẹ khi cửa gần đóng

Một số cách giảm false positive

1. Debounce cho cảm biến cửa

Nếu thấy trạng thái cửa đổi quá nhanh trong vài chục mili giây, đừng alert ngay.

2. Chỉ alert khi trạng thái ổn định

Ví dụ cửa phải giữ trạng thái mở ít nhất 100ms hoặc 200ms.

3. Cooldown cho PIR

Nếu PIR phát hiện liên tiếp, chỉ gửi một alert trong mỗi 10–20 giây.

4. Dùng ngữ cảnh thời gian

Motion ban ngày có thể ít nghiêm trọng hơn ban đêm.

5. Kết hợp nhiều event

Ví dụ:

• chỉ cảnh báo mạnh khi door_open rồi sau đó có motion_detected
• nếu chỉ có motion_detected đơn lẻ thì cảnh báo nhẹ hơn

Trong bài này, chúng ta mới làm phần nền tảng. Nhưng đây chính là hướng để hệ thống trở nên “thông minh” thật sự.

Rule nâng cao: kết hợp cửa và chuyển động

Một rule rất hay là theo dõi chuỗi sự kiện.

Ví dụ:

• cửa mở
• trong vòng 10 giây sau đó có chuyển động

Khi đó hệ thống có thể hiểu:

• có người vừa mở cửa và đi vào

Ta có thể lưu thời điểm cửa mở gần nhất:

self.last_door_open_time = None

Khi xử lý door_open:

self.last_door_open_time = datetime.now()

Khi xử lý motion_detected, kiểm tra:

if self.last_door_open_time:
    seconds = (datetime.now() - self.last_door_open_time).total_seconds()
    if seconds <= 10:
        # có tương quan với việc mở cửa

Từ đó, ta có thể sinh ra alert kiểu:

{
    "type": "entry_detected",
    "message": "Door opened followed by motion"
}

Đây là một hướng rất hay để mở rộng sau khi bạn đã ổn với bộ rule cơ bản.

Tổ chức code cho sạch hơn

Đến bài này, project đã bắt đầu có nhiều phần.
Bạn nên tách code như sau:

smart_home_monitor/
├── main.py
├── sensors/
│   ├── door_sensor.py
│   └── pir_sensor.py
├── monitor/
│   └── event_monitor.py
├── rules/
│   └── rule_engine.py
└── logs/
    └── event_logger.py

Lợi ích:

• dễ bảo trì
• dễ thêm rule mới
• dễ test từng phần
• chuẩn bị tốt cho bài 24.5

Tích hợp monitor + rule engine hoàn chỉnh

Ví dụ đơn giản trong main.py:

from datetime import datetime
from rule_engine import RuleEngine

rule_engine = RuleEngine()

def now_str():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def handle_event(sensor, event_name):
    event = {
        "ts": now_str(),
        "sensor": sensor,
        "event": event_name
    }

    print("EVENT:", event)

    alert = rule_engine.process_event(event)

    if alert:
        print("ALERT:", alert)

handle_event("door", "door_open")
handle_event("pir", "motion_detected")

Kết quả sẽ cho bạn thấy rõ:

• event nào chỉ được ghi nhận
• event nào tạo alert

Các lỗi thường gặp

Hệ thống alert quá nhiều

Nguyên nhân:

• không có cooldown
• PIR quá nhạy
• mở cửa một lần nhưng reed switch bounce nhiều lần

Cách xử lý:

• tăng cooldown_seconds
• thêm debounce
• chỉ alert khi trạng thái ổn định

Không có alert dù event đã đúng

Nguyên nhân:

• hệ thống đang disarmed
• event name không khớp với rule
• cooldown đang chặn

Cách xử lý:

• in debug self.armed
• kiểm tra event["event"]
• log last_alert_times

Alert ban ngày và ban đêm không khác nhau

Nguyên nhân:

• chưa gọi is_night_time()
• rule chưa dùng level

Cách xử lý:

• tách mức cảnh báo normal / high
• in log mức cảnh báo để kiểm tra

Rule ngày càng rối

Nguyên nhân:

• viết tất cả trong một if-else lớn

Cách xử lý:

• tách riêng từng loại rule
• dùng class hoặc function theo sensor / alert type

Kết nối với bài tiếp theo

Đến đây, hệ thống của bạn đã có đủ những mảnh ghép rất quan trọng:

• đọc cảm biến
• tạo event
• ghi log local
• xét rule cảnh báo
• phân biệt armed / disarmed
• chống spam alert

Bước tiếp theo là đưa hệ thống ra ngoài thế giới thật:

👉 gửi notify và đẩy event lên MQTT / dashboard

Ở bài 24.5, chúng ta sẽ làm hệ thống:

• gửi cảnh báo qua Telegram, email hoặc webhook
• publish event lên MQTT
• đồng bộ trạng thái lên dashboard

Đó là lúc project “smart home mini” trở nên hoàn chỉnh hơn rất nhiều.

Kết luận

Bài 24.4 là bước chuyển quan trọng từ một hệ thống “đọc sensor” thành một hệ thống “ra quyết định”.

Nhờ rule engine, Raspberry Pi không còn chỉ ghi nhận tín hiệu từ cửa và PIR, mà đã bắt đầu hiểu:

• khi nào hệ thống đang giám sát
• event nào cần cảnh báo
• mức độ cảnh báo là gì
• khi nào nên im lặng để tránh spam

Đây chính là phần tạo nên sự khác biệt giữa một demo phần cứng đơn giản và một hệ thống giám sát có logic thực tế.