راسبيري باي خادمًا لوحدات ESP32

23.1 صورة النظام – תמונת המערכת (System Overview)

في الفصل 22 رأينا أن المتحكّم أدق توقيتا، وأن الباي أقوى في التخزين والشبكة. فنوزّع الأدوار على هذا الأساس:

23.2 الوسيط Mosquitto – המתווך Mosquitto

في الفصل 16 جرّبنا وسيطا عاما على الإنترنت. الآن يصبح الوسيط عندنا، على الشبكة المحلية، فلا يرى رسائلنا أحد غيرنا. أوامر النظام والبرامج غير المُحاكاة في هذا الفصل لم تُنفَّذ في بناء الكتاب، لأنها تحتاج إلى باي حقيقي؛ ونذكر ذلك عند كل برنامج.

sudo apt update
sudo apt install mosquitto mosquitto-clients
sudo mosquitto_passwd -c /etc/mosquitto/passwd esp32
sudo mosquitto_passwd /etc/mosquitto/passwd pi-store
sudo chown mosquitto: /etc/mosquitto/passwd
sudo chmod 600 /etc/mosquitto/passwd
sudo nano /etc/mosquitto/conf.d/local.conf
sudo systemctl restart mosquitto
sudo systemctl enable mosquitto

listener 1883
allow_anonymous false
password_file /etc/mosquitto/passwd

نجرّب قبل كتابة أي برنامج. في نافذة SSH أولى ننتظر الرسائل، وفي الثانية ننشر رسالة، فتظهر في الأولى. كلمة المرور change-me في هذا الفصل مثال فقط: اختاروا غيرها.

mosquitto_sub -h localhost -u pi-store -P change-me -t "home/#" -v
mosquitto_pub -h localhost -u esp32 -P change-me \
    -t home/test/env -m '{"t": 21.5, "h": 40.0}'

23.3 الوحدة: ESP32 ينشر – היחידה מפרסמת (The Publisher)

التوصيل كما في القسم 13.1، والاتصال بالشبكة بالملف wifi.py من الفصل 14، وكلمة مرور الشبكة في secrets.py. أما الحمولة فنص JSON قصير، والموضوع يحمل اسم الوحدة.

from machine import Pin
from umqtt.simple import MQTTClient
import dht
import time
import board
import wifi

UNIT = "esp32-1"
BROKER = "192.168.1.50"
MQTT_USER = "esp32"
MQTT_PASS = "change-me"
TOPIC = "home/%s/env" % UNIT

sensor = dht.DHT22(Pin(board.DHT))
wifi.connect()
client = MQTTClient(UNIT, BROKER, user=MQTT_USER, password=MQTT_PASS)
client.connect()

while True:
    sensor.measure()
    t = sensor.temperature()
    h = sensor.humidity()
    payload = '{"t": %.1f, "h": %.1f}' % (t, h)
    client.publish(TOPIC, payload)
    print(TOPIC, payload)
    time.sleep(30)

home/esp32-1/env {"t": 24.5, "h": 60.0}
home/esp32-1/env {"t": 25.5, "h": 60.0}
home/esp32-1/env {"t": 25.5, "h": 60.0}
[mqtt_pub]
home/esp32-1/env {"t": 24.5, "h": 60.0}
home/esp32-1/env {"t": 25.5, "h": 60.0}
home/esp32-1/env {"t": 25.5, "h": 60.0}
...

البرنامج يستعمل board ويعمل كما هو على Pico W أيضا. ولكل وحدة نغيّر سطرا واحدا: UNIT. وفي مشروع يعمل طويلا نضيف إعادة الاتصال عند انقطاع الوسيط، كما في الفصل 16.

23.4 الباي يستقبل: paho-mqtt – הפאי מקבל (The Subscriber)

على الباي نستعمل مكتبة بايثون paho-mqtt، ونثبّتها مع Flask في البيئة الافتراضية كما تعلّمنا في القسم 22.2:

cd ~/project
python -m venv --system-site-packages env
source env/bin/activate
pip install paho-mqtt flask

import paho.mqtt.client as mqtt
from readings_db import open_db, save_message

db = open_db()


def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe("home/+/env")


def on_message(client, userdata, msg):
    ok = save_message(db, msg.topic, msg.payload)
    print(msg.topic, msg.payload, "saved" if ok else "rejected")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.username_pw_set("pi-store", "change-me")
client.on_connect = on_connect
client.on_message = on_message
client.connect("localhost", 1883)
client.loop_forever()

23.5 التخزين: SQLite – שמירה במסד נתונים (SQLite)

نجمع كل التعامل مع القاعدة في مكتبة واحدة، يستعملها المشترك وصفحة الويب والنسخ الاحتياطي:

# readings_db.py - sensor readings in one SQLite file, on the Pi.
# (c) 2026 Fahed Ghanem - MIT License - fahed-lab.com
import json
import sqlite3
import time

TABLE = """CREATE TABLE IF NOT EXISTS readings (
    id INTEGER PRIMARY KEY,
    unit TEXT NOT NULL,
    received REAL NOT NULL,
    temp REAL NOT NULL,
    hum REAL NOT NULL)"""
ADD = "INSERT INTO readings (unit, received, temp, hum) VALUES (?, ?, ?, ?)"


def open_db(path="readings.db"):
    db = sqlite3.connect(path)
    db.execute(TABLE)
    db.commit()
    return db


def save_message(db, topic, payload, now=None):
    try:
        unit = topic.split("/")[1]
        data = json.loads(payload)
        when = time.time() if now is None else now
        row = (unit, when, float(data["t"]), float(data["h"]))
    except (ValueError, KeyError, IndexError, TypeError):
        return False
    db.execute(ADD, row)
    db.commit()
    return True

LATEST = """SELECT unit, received, temp, hum FROM readings
    ORDER BY id DESC LIMIT ?"""
AVERAGES = """SELECT unit, COUNT(*), AVG(temp), MAX(temp) FROM readings
    WHERE received >= ? GROUP BY unit ORDER BY unit"""
PRUNE = "DELETE FROM readings WHERE received < ?"


def latest(db, count=10):
    return db.execute(LATEST, (count,)).fetchall()


def averages(db, since=0):
    return db.execute(AVERAGES, (since,)).fetchall()


def prune(db, before):
    deleted = db.execute(PRUNE, (before,)).rowcount
    db.commit()
    return deleted


def backup(db, path):
    copy = sqlite3.connect(path)
    db.backup(copy)
    copy.close()

الدالة save_message لا تثق بالرسالة: رسالة تالفة تعيد False، ولا توقف الخادم. والزمن من ساعة الباي عند الوصول، فلا تحتاج الوحدة إلى ساعة مضبوطة. والآن نجرّب المكتبة فعلا: الرسائل التالية هي ما نشرته عند بناء الكتاب وحدتان بالبرنامج نفسه من القسم 23.3، أُضيفت إليها رسالة تالفة. الزمن هنا ممرَّر يدويا حتى يكون المُخرَج واضحا.

from readings_db import open_db, save_message, latest
from readings_db import averages, prune, backup

MESSAGES = [
    ("home/esp32-1/env", b'{"t": 24.5, "h": 60.0}'),
    ("home/esp32-2/env", b'{"t": 21.5, "h": 48.0}'),
    ("home/esp32-1/env", b'{"t": 25.5, "h": 60.0}'),
    ("home/esp32-2/env", b'{"t": 22.0, "h": 48.0}'),
    ("home/esp32-1/env", b'{"t": 25.5, "h": 60.0}'),
    ("home/esp32-2/env", b"sensor error"),
]

db = open_db("readings.db")
for i, (topic, payload) in enumerate(MESSAGES):
    ok = save_message(db, topic, payload, now=15 * i)
    print("saved   " if ok else "rejected", topic)

print("-- newest 2")
for unit, when, temp, hum in latest(db, 2):
    print("%s at %d s: %.1f C, %.1f %%" % (unit, when, temp, hum))

print("-- per unit")
for unit, count, mean, top in averages(db):
    print("%s: %d rows, mean %.1f, max %.1f" % (unit, count, mean, top))

print("-- deleted:", prune(db, 30))
backup(db, "backup.db")
copy = open_db("backup.db")
print("-- rows in backup:", len(latest(copy, 100)))

saved    home/esp32-1/env
saved    home/esp32-2/env
saved    home/esp32-1/env
saved    home/esp32-2/env
saved    home/esp32-1/env
rejected home/esp32-2/env
-- newest 2
esp32-1 at 60 s: 25.5 C, 60.0 %
esp32-2 at 45 s: 22.0 C, 48.0 %
-- per unit
esp32-1: 3 rows, mean 25.2, max 25.5
esp32-2: 2 rows, mean 21.8, max 22.0
-- deleted: 2
-- rows in backup: 3

23.6 صفحة الويب: Flask – דף הקריאות האחרונות (Flask)

في الفصل 14 كتبنا خادم HTTP صغيرا على اللوحة. على الباي نستعمل المكتبة Flask، وهي تتولى الطلبات والقوالب: نربط عنوانا بدالة، والدالة تعيد الصفحة.

from flask import Flask, render_template_string
from readings_db import open_db, latest
import time

app = Flask(__name__)

PAGE = """<!doctype html>
<meta charset="utf-8">
<meta http-equiv="refresh" content="30">
<title>Readings</title>
<table border="1">
<tr><th>Unit</th><th>Time</th><th>Temp</th><th>Humidity</th></tr>
{% for r in rows %}
<tr><td>{{ r[0] }}</td><td>{{ r[1] }}</td>
<td>{{ r[2] }}</td><td>{{ r[3] }}</td></tr>
{% endfor %}
</table>"""


def pretty(row):
    unit, received, temp, hum = row
    stamp = time.strftime("%H:%M:%S", time.localtime(received))
    return unit, stamp, "%.1f" % temp, "%.1f" % hum


@app.route("/")
def home():
    db = open_db()
    rows = [pretty(r) for r in latest(db, 20)]
    db.close()
    return render_template_string(PAGE, rows=rows)


app.run(host="0.0.0.0", port=5000)

• @app.route("/") يربط الصفحة الرئيسية بالدالة home.
• كل طلب يفتح اتصالا خاصا به بالقاعدة ويغلقه، لأن Flask قد تخدم الطلبات في خيوط مختلفة، واتصال sqlite3 مرتبط بخيطه.
• الوسم {{ }} في القالب يضع القيمة، ويحوّل الرموز الخاصة فيها إلى نص آمن.
• host="0.0.0.0" يجعل الصفحة متاحة من الشبكة، على العنوان http://mypi.local:5000.

23.7 من تجربة إلى نظام – ניהול שירות, כתובת קבועה וגיבוי (Services and Backup)

الخدمات. نحوّل البرنامجين إلى خدمتين كما في القسم 22.8. هذا ملف المشترك، وملف صفحة الويب مثله مع تغيير الوصف واسم البرنامج:

[Unit]
Description=Store MQTT readings
After=network-online.target mosquitto.service
Wants=network-online.target

[Service]
User=student
WorkingDirectory=/home/student/project
ExecStart=/home/student/project/env/bin/python -u mqtt_store.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

السطر After يبدأ الخدمة بعد الشبكة والوسيط، وRestart=always يعيدها إذا انهارت. نفعّل الخدمتين معا: sudo systemctl enable --now mqtt-store web-page.

عنوان ثابت. الوحدات تحتوي عنوان الباي في برنامجها، فلا يصح أن يتغيّر. الطريقة الأبسط هي حجز العنوان في الموجّه (ראוטר) لبطاقة شبكة الباي، فنبقي الإعداد في مكان واحد. والبديل ضبطه على الباي نفسه بالأداة nmtui في الإصدارات التي تدير الشبكة بـNetworkManager، مثل Bookworm. ولا نعتمد في الوحدة على الاسم mypi.local، لأن حلّ هذا الاسم غير مضمون على المتحكّم.

النسخة الاحتياطية (גיבוי). بطاقة microSD قد تتعطل، فنحتفظ بنسخة خارج الباي. برنامج صغير يحذف القديم ويأخذ نسخة كل ليلة، وهو لا ينسخ الملف بأمر cp لأن النسخ أثناء الكتابة قد يعطي ملفا تالفا، بينما backup يتولى ذلك بأمان، كما رأينا في المُخرَج أعلاه:

from readings_db import open_db, prune, backup
import time

DAY = 86400

db = open_db()
print("deleted", prune(db, time.time() - 30 * DAY))
backup(db, time.strftime("backup-%Y-%m-%d.db"))
db.close()

crontab -e
0 3 * * * cd ~/project && env/bin/python backup_db.py
scp "student@mypi.local:project/backup-*.db" .

ونحفظ مع القاعدة ملفي Mosquitto وبرامج المشروع، فهي كل ما نحتاج إليه لبناء الباي من جديد.

أهمّ أفكار الفصل:

• الوحدات تقيس وتنشر، والباي يستقبل ويخزّن ويعرض.
• Mosquitto الحديث يحتاج إلى listener حتى يقبل اتصالا من الشبكة، ونحميه بمستخدمين.
• الاشتراك داخل on_connect حتى يتجدد بعد كل انقطاع.
• لا نثق برسالة من الشبكة: نفحصها، ونستعمل علامات ? في SQL.
• البرنامجان يتشاركان ملف القاعدة، وليس بينهما اتصال مباشر.
• كل برنامج خدمة، وعنوان الباي ثابت، والنسخة الاحتياطية خارج الباي.

قاموس المصطلحات – מילון מונחים:

أسئلة للتمرين:

from flask import jsonify


@app.route("/api/latest")
def api_latest():
    db = open_db()
    rows = latest(db, 20)
    db.close()
    return jsonify(rows)