Raspberry Pi وIoT مع Python: 100 مثال من الصفر إلى الاحتراف
Raspberry Pi حاسوب صغير يستطيع قراءة المستشعرات والتحكم في المرحلات وإرسال البيانات عبر الشبكة، لذلك يُعد أساسًا شائعًا لمشاريع إنترنت الأشياء IoT. تأخذك هذه المقالة من أول برنامج Python إلى خط معالجة كامل للمستشعرات في 100 مثال. ستتعلم أساسيات Python، ومعالجة بيانات المستشعرات، والتحكم في الأطراف عبر GPIO، وقراءة المستشعرات عبر I2C وSPI وعبر 1-Wire، وإرسال البيانات عبر HTTP وMQTT، وبناء لوحة ويب صغيرة، وتخزين القراءات في SQLite، وتشغيل كل ذلك كخدمة تبدأ مع الإقلاع. ملاحظة للشفافية: الأمثلة التي تعمل على أي حاسوب نُفِّذت فعليًا وعُرضت مخرجاتها الحقيقية بعد الشيفرة. أما بقية الأمثلة فتحتاج إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغتها فقط ولا تظهر مخرجاتها.
ما الذي تحتاجه
جهاز Raspberry Pi من الطراز 3 أو 4 أو 5 يعمل بنظام Raspberry Pi OS، وبطاقة microSD، ومصدر طاقة، واتصال بالشبكة. يأتي Python 3.10 أو أحدث مثبتًا في Raspberry Pi OS. تحتاج الأمثلة المتعلقة بالعتاد إلى مصباح LED ومقاومة وزر ضغط وأسلاك توصيل ومستشعرات قليلة. تعمل معظم الأمثلة القابلة للتشغيل على أي حاسوب يحتوي على Python 3.10 أو أحدث، وهكذا اختُبرت الأمثلة القابلة للتشغيل.
كيفية تشغيل المثال
احفظ المثال باسم main.py وشغّله بالأمر python3 main.py. تعرض الأمثلة المعلَّمة بأنها قابلة للتشغيل المخرجات الحقيقية لتشغيلها على حاسوب. أمثلة bash ملفات إعداد لـ Pi ولم تُشغَّل هنا.
الموضوع 1: البدء مع Python على Pi
تُعد Python اللغة المعتادة لمشاريع Raspberry Pi لأنها سهلة القراءة ولها مكتبات لكل المستشعرات تقريبًا. تغطي الأمثلة الأولى الإخراج والمتغيرات والأنواع والحساب.
تشغيل أول برنامج
احفظ ملفًا بامتداد py. وشغّله بالأمر python3 على Pi. تكتب print النص في الطرفية، وتضع f-string المتغيرات داخل النص باستخدام {name}. تقرأ Python الملف من الأعلى إلى الأسفل، لذلك لترتيب الأسطر أهمية.
الأنواع ومعانيها
يمكن أن تكون الأرقام int (صحيحة) أو float (عشرية)، والنص str، وتكون القيم True وFalse من النوع bool، وتعني None عدم وجود قيمة. استخدم type() لمعرفة نوع القيمة. تكون قراءات المستشعرات عادةً من نوع float، بينما تكون أرقام الأطراف عادةً من نوع int.
الحساب لأغراض المستشعرات
يعطي / دائمًا قيمة float، ويقسم // ويحذف الجزء الكسري، ويعطي % الباقي. تحدد round() عدد المنازل العشرية المعروضة. استخدم هذه العوامل لتحويل القراءات الخام وتجميع القيم.
رسوم توضيحية
مثال 1: مرحبًا من Pi
أول برنامج Python لك على Raspberry Pi OS.
print("Hello from my Raspberry Pi!")
المخرجات:
Hello from my Raspberry Pi!
مثال 2: المتغيرات وf-strings
خزّن القيم واعرضها داخل النص.
sensor_name = "DHT22"
temperature = 27.5
print(f"{sensor_name} reads {temperature} °C")
المخرجات:
DHT22 reads 27.5 °C
مثال 3: أنواع البيانات
أنواع Python الأساسية وأسماؤها.
for value in (42, 3.14, "pi", True, None):
print(repr(value), type(value).__name__)
المخرجات:
42 int
3.14 float
'pi' str
True bool
None NoneType
مثال 4: الحساب والقسمة الصحيحة والتقريب
القسمة والباقي والتقريب في حسابات المستشعرات.
readings = 17
per_group = 5
print(readings / per_group)
print(readings // per_group, readings % per_group)
print(round(3.14159, 2))
المخرجات:
3.4
3 2
3.14
مثال 5: العمل مع النصوص
نظّف أسماء الأجهزة وتحقق من البادئات.
status = " ONLINE "
clean = status.strip().lower()
print(clean, clean.startswith("on"), len(clean))
المخرجات:
online True 6
مثال 6: تحويل الوحدات
حوّل الدرجات المئوية إلى فهرنهايت بدالة.
def celsius_to_fahrenheit(c):
return c * 9 / 5 + 32
for c in (0, 25, 37.5):
print(f"{c} C = {celsius_to_fahrenheit(c):.1f} F")
المخرجات:
0 C = 32.0 F
25 C = 77.0 F
37.5 C = 99.5 F
الموضوع 2: القرارات والحلقات
يتفاعل مشروع Pi مع مدخلاته. تحدد الشروط ما الذي سيحدث، وتكرر الحلقات العمل، مثل أخذ عينات من المستشعر كل بضع ثوانٍ.
if وelif وelse
تُفحص الشروط من الأعلى إلى الأسفل، ويُنفَّذ أول فرع يكون صحيحًا. ضع الفحص الأكثر تحديدًا أولًا. تفحص المقارنة المتسلسلة مثل 20 <= t <= 25 نطاقًا كاملًا في سطر واحد.
for وwhile وbreak وcontinue
تكرر for على مجال أو تسلسل. تكرر while حتى يتغير الشرط. تخرج break من الحلقة، وتتخطى continue الدورة الحالية، وهذا مفيد لتجاهل قراءة خاطئة ومتابعة العمل.
قيمة الصواب والخطأ وعبارة match
تُعد القيم الفارغة، مثل النص الفارغ أو القائمة الفارغة، خاطئة. وتطابق عبارة match، المتوفرة من Python 3.10، أمرًا مع عدة أنماط، وهي طريقة نظيفة للتعامل مع أوامر نصية قادمة من تطبيق هاتف.
مثال 7: if وelif وelse
اختر إجراءً حسب درجة الحرارة.
temperature = 31
if temperature >= 35:
print("Too hot: stop the fan test")
elif temperature >= 28:
print("Warm: run the fan")
else:
print("Comfortable")
المخرجات:
Warm: run the fan
مثال 8: حلقة for مع range
كرر عملًا عددًا محددًا من المرات.
for minute in range(0, 15, 5):
print(f"minute {minute}")
المخرجات:
minute 0
minute 5
minute 10
مثال 9: حلقة while
كرر حتى يتغير الشرط.
countdown = 3
while countdown > 0:
print("sampling in", countdown)
countdown -= 1
print("start")
المخرجات:
sampling in 3
sampling in 2
sampling in 1
start
مثال 10: break وcontinue
تجاوز القراءات الخاطئة وتوقف عند العطل.
for reading in [21, None, 23, 99, 24]:
if reading is None:
continue
if reading > 50:
print("sensor fault, stop")
break
print("ok", reading)
المخرجات:
ok 21
ok 23
sensor fault, stop
مثال 11: القيم الصحيحة والخاطئة
أي القيم تُعد صحيحة أو خاطئة.
for value in ("", "Pi", 0, 5, [], [1]):
print(repr(value), "is true" if value else "is false")
المخرجات:
'' is false
'Pi' is true
0 is false
5 is true
[] is false
[1] is true
مثال 12: المقارنات المتسلسلة
تحقق من أن القيمة داخل نطاق.
humidity = 55
print(40 <= humidity <= 60)
print(humidity > 60 or humidity < 40)
المخرجات:
True
False
مثال 13: التعبير الشرطي
if-else في سطر واحد يعيد قيمة.
reading = 12
state = "ON" if reading < 15 else "OFF"
print(state)
المخرجات:
ON
مثال 14: عبارة match
اربط أوامر النص بالإجراءات.
def describe(command):
match command:
case "on" | "1":
return "turn relay on"
case "off" | "0":
return "turn relay off"
case _:
return "unknown command"
for c in ("on", "0", "blink"):
print(c, "->", describe(c))
المخرجات:
on -> turn relay on
0 -> turn relay off
blink -> unknown command
الموضوع 3: الدوال والمكتبة القياسية
تعطي الدالة اسمًا لمهمة تكررها. وتحتوي المكتبة القياسية بالفعل على دوال الرياضيات والإحصاء والوقت، لذلك نادرًا ما تحتاج إلى كتابتها بنفسك.
تعريف الدوال
ينشئ def دالة، ويعيد return قيمة. تعطي الوسائط الافتراضية قيمًا معقولة عندما يحذفها المستدعي، وتجعل الوسائط المسماة الاستدعاء مقروءًا، مثل sample(4, unit='volts').
إرجاع قيم متعددة وتمريرها
يمكن للدالة أن تعيد عدة قيم كمجموعة (tuple) يفكها المستدعي. يقبل *args أي عدد من الوسائط الموضعية، ويقبل **kwargs أي عدد من الوسائط المسماة، وهذا مناسب لدوال التسجيل المساعدة.
lambda وmap وfilter والاستدعاء الذاتي
تعبير lambda دالة صغيرة بلا اسم، وهو مفيد مع map وfilter لتحويل قائمة. تستدعي الدالة المتكررة نفسها ويجب أن يكون لها شرط توقف، والمضروب مثال كلاسيكي.
مثال 15: دوال بقيم افتراضية
امنح الدالة قيمًا افتراضية معقولة.
def sample(pin, samples=3, unit="raw"):
return f"read pin {pin} x{samples} ({unit})"
print(sample(4))
print(sample(4, 10, unit="volts"))
المخرجات:
read pin 4 x3 (raw)
read pin 4 x10 (volts)
مثال 16: إرجاع عدة قيم
أعد القيمة الدنيا والعليا معًا.
def min_max(values):
return min(values), max(values)
low, high = min_max([19.5, 22.0, 18.2])
print(f"low {low}, high {high}")
المخرجات:
low 18.2, high 22.0
مثال 17: *args و**kwargs
استقبل أي عدد من القيم الموضعية والمسمّاة.
def log_event(*parts, **fields):
text = " ".join(str(p) for p in parts)
extra = ", ".join(f"{k}={v}" for k, v in fields.items())
print(text, "|", extra)
log_event("door", "opened", sensor="reed-1", room="kitchen")
المخرجات:
door opened | sensor=reed-1, room=kitchen
مثال 18: lambda وmap وfilter
حوّل قيم ADC الخام في سطر واحد.
raw = [512, 1023, 0, 768]
volts = list(map(lambda x: round(x * 3.3 / 1023, 2), raw))
nonzero = list(filter(lambda v: v > 0, volts))
print(volts)
print(nonzero)
المخرجات:
[1.65, 3.3, 0.0, 2.48]
[1.65, 3.3, 2.48]
مثال 19: المكتبة القياسية: math وstatistics
استخدم وحدات الرياضيات والإحصاء المدمجة.
import math
import statistics
samples = [2.0, 4.0, 4.5, 3.5]
print(round(math.sqrt(16), 2), math.pi > 3.14)
print(statistics.mean(samples), statistics.median(samples))
المخرجات:
4.0 True
3.5 3.75
مثال 20: الاستدعاء الذاتي
دالة تستدعي نفسها مع شرط توقف.
def factorial(n):
if n <= 1:
return 1
return n * factorial(n - 1)
print(factorial(5), factorial(10))
المخرجات:
120 3628800
الموضوع 4: بنى البيانات للأجهزة والقراءات
يحتفظ مشروع Pi بقوائم من القراءات، وقواميس لإعدادات الأجهزة، ومجموعات من الأسماء الفريدة. يجعل اختيار البنية المناسبة الشيفرة أقصر وأسرع.
القوائم والتعبيرات الاختصارية
القائمة تسلسل مرتب يمكن تغييره. يبني تعبير القائمة الاختصاري قائمة جديدة من قائمة قديمة في سطر واحد، مثل تحويل قائمة قيم مئوية إلى فهرنهايت.
القواميس للقيم المسماة
يربط القاموس مفتاحًا بقيمة، وهو مثالي لإعدادات جهاز واحد أو لقراءة واحدة مع حقولها. استخدم get() لقراءة مفتاح قد يكون مفقودًا، واستخدم sorted() لطباعة المفاتيح بالترتيب.
المجموعات والمجموعات الفرعية والتقطيع
المجموعة (tuple) مجموعة ثابتة من القيم. وتحتوي المجموعة (set) على عناصر فريدة وتدعم الاتحاد والتقاطع. يأخذ التقطيع مثل values[-2:] جزءًا من القائمة، ويعكسها [::-1].
مثال 21: القوائم
أضف عناصر وأدرجها واحذفها.
readings = [20.1, 20.4]
readings.append(20.9)
readings.insert(0, 19.8)
readings.remove(20.4)
print(readings, len(readings), readings[-1])
المخرجات:
[19.8, 20.1, 20.9] 3 20.9
مثال 22: تعابير إنشاء القوائم
أنشئ قائمة جديدة من قائمة قديمة.
celsius = [0, 10, 20, 30]
fahrenheit = [c * 9 / 5 + 32 for c in celsius if c != 10]
print(fahrenheit)
المخرجات:
[32.0, 68.0, 86.0]
مثال 23: المجموعات (Tuples) والفك
مجموعات ثابتة من القيم لا تتغير.
location = ("kitchen", 2, 3.5)
room, floor, height = location
print(room, floor, height)
try:
location[0] = "garage"
except TypeError as e:
print("tuples cannot change:", type(e).__name__)
المخرجات:
kitchen 2 3.5
tuples cannot change: TypeError
مثال 24: القواميس
خزّن قيمًا مسماة لجهاز.
device = {"name": "node-1", "pin": 17, "online": True}
device["pin"] = 22
device["room"] = "garage"
print(device.get("name"), device.get("temp", "n/a"), len(device))
المخرجات:
node-1 n/a 4
مثال 25: مخرجات القاموس المرتبة
اطبع عدد كل حدث بالترتيب.
counts = {"motion": 4, "door": 1, "temp": 9}
for name in sorted(counts):
print(f"{name:<8}{counts[name]:>3}")
المخرجات:
door 1
motion 4
temp 9
مثال 26: المجموعات (Sets)
قيم فريدة وعمليات المجموعات.
seen = {"kitchen", "garage"}
seen.add("kitchen")
new = {"garage", "attic"}
print(len(seen), sorted(seen | new), sorted(seen & new))
المخرجات:
2 ['attic', 'garage', 'kitchen'] ['garage']
مثال 27: التقطيع
خذ أجزاء من قائمة، واعكسها.
samples = [1, 2, 3, 4, 5, 6, 7]
print(samples[:3], samples[-2:], samples[::2])
print(samples[::-1])
المخرجات:
[1, 2, 3] [6, 7] [1, 3, 5, 7]
[7, 6, 5, 4, 3, 2, 1]
مثال 28: قوائم القواميس
صفِّ قائمة الأجهزة المتصلة.
devices = [
{"id": "a1", "online": True},
{"id": "b2", "online": False},
{"id": "c3", "online": True},
]
online = [d["id"] for d in devices if d["online"]]
print("online:", online)
المخرجات:
online: ['a1', 'c3']
الموضوع 5: النص والملفات وصيغ البيانات
يجب حفظ بيانات المستشعرات ومشاركتها. الملفات النصية وCSV وJSON هي الصيغ التي تستخدمها معظم مشاريع إنترنت الأشياء، وتقرأ المكتبة القياسية الصيغ الثلاث وتكتبها.
قراءة الملفات وكتابتها
افتح الملف ضمن كتلة with ليُغلق تلقائيًا، واستخدم encoding='utf-8' مع النصوص. تعمل كائنات Path من pathlib على أي نظام تشغيل وتسهّل ربط أسماء الملفات.
CSV وJSON
تكتب الوحدة csv صفوفًا مع رأس، وتفهمها برامج الجداول وأدوات البيانات. وتحوّل الوحدة json القواميس إلى نص لخدمات الويب ورسائل MQTT، وتعيدها مرة أخرى.
تنسيق الجداول والتواريخ
تُصفّ عروض التنسيق مثل {name:<10} الأعمدة في الطرفية. ينسّق datetime الطوابع الزمنية باستخدام strftime، ويضيف المدد الزمنية باستخدام timedelta، فيمكن تجميع القراءات حسب الوقت.
مثال 29: تنسيق مخرجات الجدول
رتّب الأعمدة بعرض التنسيق.
rows = [("Temp", 27.456), ("Humidity", 55.0), ("Pressure", 1013.2)]
for name, value in rows:
print(f"{name:<10}{value:>10.1f}")
المخرجات:
Temp 27.5
Humidity 55.0
Pressure 1013.2
مثال 30: قراءة الملفات النصية وكتابتها
احفظ قيم المستشعر في ملف واقرأها.
from pathlib import Path
log = Path("sensor_log.txt")
with log.open("w", encoding="utf-8") as f:
f.write("21.5\n22.0\n22.4\n")
with log.open("r", encoding="utf-8") as f:
values = [float(line) for line in f]
log.unlink()
print("read", len(values), "values, last", values[-1])
المخرجات:
read 3 values, last 22.4
مثال 31: ملفات CSV
اكتب جدولًا من القياسات واقرأه.
import csv
from pathlib import Path
path = Path("readings.csv")
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["time", "temp_c"])
writer.writerows([["08:00", 21.5], ["08:05", 21.9]])
with path.open(newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
print(row["time"], float(row["temp_c"]) + 1)
path.unlink()
المخرجات:
08:00 22.5
08:05 22.9
مثال 32: JSON
حوّل البيانات من وإلى نص JSON.
import json
payload = {"device": "node-1", "readings": [21.5, 21.9], "ok": True}
text = json.dumps(payload, indent=2)
print(text)
print(json.loads(text)["readings"][1])
المخرجات:
{
"device": "node-1",
"readings": [
21.5,
21.9
],
"ok": true
}
21.9
مثال 33: المسارات
تعامل مع أسماء الملفات والمجلدات بأمان.
from pathlib import PurePosixPath
path = PurePosixPath("/home/pi/projects/iot/config.json")
print(path.name, path.suffix, path.parent)
print(path.parent / "logs")
المخرجات:
config.json .json /home/pi/projects/iot
/home/pi/projects/iot/logs
مثال 34: التواريخ والأوقات
نسّق قيم التاريخ والوقت وأضف إليها.
from datetime import datetime, timedelta
start = datetime(2026, 1, 15, 8, 0)
later = start + timedelta(minutes=90)
print(start.strftime("%Y-%m-%d %H:%M"), "->", later.strftime("%H:%M"))
المخرجات:
2026-01-15 08:00 -> 09:30
الموضوع 6: الأخطاء والتسجيل والتوقيت
يعمل الجهاز لأشهر دون أن يراقبه أحد. يجب أن يتحمل القراءات الخاطئة، ويسجل ما حدث، ويقيس توقيته الخاص.
معالجة الأخطاء بـ try وexcept
ضع الشيفرة التي قد تفشل داخل try، والتقط الاستثناء المحدد، واستمر في العمل. تعمل finally سواء وقع خطأ أم لا، لذلك فهي المكان المناسب لإيقاف مخرج أو إغلاق ملف.
التسجيل بدلًا من print
تسجل وحدة logging الرسائل مع مستوى مثل INFO أو WARNING، وتختار المستوى الذي يظهر. يمكن إرسال السجلات إلى ملف، وهذا ضروري عندما يعمل Pi بلا شاشة.
الفحوص الدفاعية والتوقيت
ترفض assert القيم المستحيلة مبكرًا أثناء التطوير. يقيس time.perf_counter الزمن المنقضي بدقة، مما يساعدك على العثور على الشيفرة البطيئة على حاسوب صغير.
مثال 35: try وexcept
عالج القراءة الخاطئة دون أن يتوقف البرنامج.
def parse_temperature(text):
try:
return float(text)
except ValueError:
print(f"bad reading: {text!r}")
return None
print(parse_temperature("23.4"))
print(parse_temperature("23,4"))
المخرجات:
23.4
bad reading: '23,4'
None
مثال 36: finally والتنظيف
نظّف دائمًا حتى بعد حدوث خطأ.
from pathlib import Path
target = Path("demo_state.txt")
try:
with target.open("w", encoding="utf-8") as f:
f.write("on")
print("saved")
finally:
target.unlink(missing_ok=True)
print("cleaned up")
المخرجات:
saved
cleaned up
مثال 37: الاستثناءات المخصصة
أطلق خطأً واضحًا لقيمة مستشعر مستحيلة.
class SensorError(Exception):
"""Raised when a sensor gives an impossible value."""
def check(value):
if value < -40 or value > 85:
raise SensorError(f"value {value} is outside the sensor range")
return value
try:
check(120)
except SensorError as e:
print("sensor error:", e)
المخرجات:
sensor error: value 120 is outside the sensor range
مثال 38: التسجيل Logging
سجّل الرسائل بمستويات بدلًا من print.
import logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("node-1")
log.info("started")
log.warning("battery at 18%")
log.debug("this line is hidden at INFO level")
المخرجات:
مثال 39: الفحوص الدفاعية بـ assert
ارفض المدخلات غير الصالحة مبكرًا.
def set_pwm(duty):
assert 0 <= duty <= 100, "duty must be 0 to 100"
return duty
print(set_pwm(50))
try:
set_pwm(150)
except AssertionError as e:
print("rejected:", e)
المخرجات:
50
rejected: duty must be 0 to 100
مثال 40: قياس الزمن المنقضي
قِس زمن الشيفرة بساعة دقيقة.
import time
start = time.perf_counter()
total = sum(i * i for i in range(200_000))
elapsed = time.perf_counter() - start
print("sum of squares:", total)
print("took measurable time:", elapsed >= 0)
المخرجات:
sum of squares: 2666646666700000
took measurable time: True
الموضوع 7: معالجة بيانات المستشعرات
قيم المستشعرات الخام مليئة بالضوضاء وغالبًا تكون بوحدات خاطئة. تحوّل هذه الأمثلة الأرقام الخام إلى قراءات مفيدة: معايرة، وتنعيم، وفحص للقيم الشاذة، ومقارنة مع عتبات.
المعايرة وتحويل الوحدات
يعيد ADC عدادات خام. تعطي نقطتا مرجع معروفتان معايرة خطية، فتستطيع تحويل العدادات إلى فولت. احفظ المعايرة مع الجهاز، لأن كل مستشعر يختلف عن غيره.
التنعيم والتباطؤ Hysteresis
يزيل المتوسط المتحرك على آخر بضع قراءات القمم المفاجئة. يستخدم التباطؤ عتبتين، واحدة للتشغيل وأخرى للإيقاف، فلا يهتز المرحّل عندما تقع القيمة قرب أحد الحدود.
الإحصاء والقيم الشاذة
يصف المتوسط والوسيط والانحراف المعياري نافذة من القراءات. يتجاهل الوسيط القيم المتطرفة، لذلك فهو طريقة جيدة لإزالة قمة خاطئة قبل حساب المتوسط.
مثال 41: معايرة قيم ADC الخام
حوّل القيم الخام إلى فولت بنقطتي مرجع.
RAW_MIN, RAW_MAX = 102, 921 # readings at 0 V and at full scale
VOLT_MAX = 3.3
def to_volts(raw):
return (raw - RAW_MIN) * VOLT_MAX / (RAW_MAX - RAW_MIN)
for raw in (102, 512, 921):
print(raw, "->", round(to_volts(raw), 3), "V")
المخرجات:
102 -> 0.0 V
512 -> 1.652 V
921 -> 3.3 V
مثال 42: مرشح المتوسط المتحرك
نعّم القراءات المتذبذبة ضمن نافذة.
from collections import deque
window = deque(maxlen=3)
def smooth(value):
window.append(value)
return sum(window) / len(window)
for reading in [20.0, 25.0, 21.0, 40.0, 22.0]:
print(f"raw {reading:5.1f} smoothed {smooth(reading):6.2f}")
المخرجات:
raw 20.0 smoothed 20.00
raw 25.0 smoothed 22.50
raw 21.0 smoothed 22.00
raw 40.0 smoothed 28.67
raw 22.0 smoothed 27.67
مثال 43: التباطؤ Hysteresis لمروحة مرحّل
امنع المرحّل من التشغيل والإيقاف بسرعة.
def relay_state(temp, currently_on, on_at=24.0, off_at=22.0):
if not currently_on and temp >= on_at:
return True
if currently_on and temp <= off_at:
return False
return currently_on
state = False
for t in [21.0, 23.0, 24.5, 23.5, 22.5, 21.9]:
state = relay_state(t, state)
print(t, "fan", "ON" if state else "OFF")
المخرجات:
21.0 fan OFF
23.0 fan OFF
24.5 fan ON
23.5 fan ON
22.5 fan ON
21.9 fan OFF
مثال 44: وحدات الضغط والتقريب
حوّل hPa إلى inHg وقرّب بشكل صحيح.
def hpa_to_inhg(hpa):
return hpa * 0.02953
pressure = 1013.25
print(f"{pressure} hPa = {hpa_to_inhg(pressure):.2f} inHg")
print(round(2.5), round(3.5))
المخرجات:
1013.25 hPa = 29.92 inHg
2 4
مثال 45: الإحصاء داخل نافذة
المتوسط والانتشار والمدى لأحدث القراءات.
import statistics
window = [21.4, 21.6, 21.5, 21.9, 21.7]
print("mean", round(statistics.mean(window), 2))
print("stdev", round(statistics.pstdev(window), 3))
print("range", round(max(window) - min(window), 2))
المخرجات:
mean 21.62
stdev 0.172
range 0.5
مثال 46: إزالة القيم الشاذة بالوسيط
تجاهل القيم البعيدة عن القراءة المعتادة.
import statistics
def remove_outliers(values, tolerance=3.0):
med = statistics.median(values)
return [v for v in values if abs(v - med) <= tolerance]
raw = [21.5, 21.7, 85.0, 21.6, -40.0, 21.8]
print(remove_outliers(raw))
المخرجات:
[21.5, 21.7, 21.6, 21.8]
مثال 47: معدل التغيّر
مقدار ارتفاع القيمة في الدقيقة.
def rate_per_minute(prev_value, prev_minute, value, minute):
if minute == prev_minute:
return 0.0
return (value - prev_value) / (minute - prev_minute)
print(round(rate_per_minute(20.0, 0, 21.2, 4), 2), "C per minute")
المخرجات:
0.3 C per minute
مثال 48: القراءة ككلاس بيانات
صف قياسًا واحدًا بصنف صغير.
from dataclasses import dataclass, asdict
@dataclass
class Reading:
device: str
temperature_c: float
humidity_pct: float
r = Reading("node-1", 21.5, 48.0)
print(r)
print(asdict(r))
المخرجات:
Reading(device='node-1', temperature_c=21.5, humidity_pct=48.0)
{'device': 'node-1', 'temperature_c': 21.5, 'humidity_pct': 48.0}
الموضوع 8: GPIO: التحكم في LED والأزرار والمحركات والمرحّلات
تتيح مجموعة الأطراف ذات الأربعين طرفًا لـ Pi تشغيل المخرجات وقراءة المدخلات. تعرض الأمثلة مصابيح LED والأزرار وتعتيم الإضاءة بـ PWM والمحركات المؤازرة ومستشعر المسافة والمرحّلات. تحقق دائمًا من التوصيلات مقابل مخطط الأطراف قبل تشغيل الجهاز.
المخرجات والمدخلات ومقاومات Pull-up
اضبط الطرف كمخرج لتشغيل LED عبر مقاومة، أو كمدخل لقراءة زر. تُبقي مقاومة Pull-up المدخل مرتفعًا حتى يصله الزر بالأرضي، فلا يبقى الطرف عائمًا.
PWM والمحركات المؤازرة
تُشغّل تعديلات عرض النبضة الطرف والإيقاف بسرعة، وتحدد نسبة زمن التشغيل السطوع أو موضع المحرك المؤازر. تُخفي مكتبة gpiozero معظم عمل التوقيت.
المرحّلات والمستشعرات والإيقاف الآمن
يُشغّل المرحّل الأجهزة ذات القدرة الأعلى، مثل المضخة، لذلك يشغّل Pi ملف التحكم الصغير فقط. أطفئ المخرجات دائمًا داخل كتلة finally، واستدعِ cleanup، حتى لا يبقى المحرك يعمل إذا تعطل البرنامج.
مثال 49: إضاءة LED وإطفاؤها (GPIO)
شغّل LED وأطفئه من طرف GPIO في Raspberry Pi.
import time
import RPi.GPIO as GPIO
LED_PIN = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)
try:
for _ in range(5):
GPIO.output(LED_PIN, GPIO.HIGH)
time.sleep(0.5)
GPIO.output(LED_PIN, GPIO.LOW)
time.sleep(0.5)
finally:
GPIO.cleanup()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 50: قراءة زر مع Pull-up
اكتشف الضغط على الزر بأمان مع تأخير لإزالة الارتداد.
import time
import RPi.GPIO as GPIO
BUTTON_PIN = 27
GPIO.setmode(GPIO.BCM)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
try:
while True:
if GPIO.input(BUTTON_PIN) == GPIO.LOW: # pressed connects the pin to GND
print("button pressed")
time.sleep(0.2) # simple debounce
time.sleep(0.01)
finally:
GPIO.cleanup()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 51: التحكم بسطوع LED بتعديل PWM
غيّر سطوع LED بتعديل عرض النبضة.
import time
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(18, GPIO.OUT)
pwm = GPIO.PWM(18, 1000) # 1 kHz
pwm.start(0)
try:
for duty in range(0, 101, 10):
pwm.ChangeDutyCycle(duty)
time.sleep(0.1)
finally:
pwm.stop()
GPIO.cleanup()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 52: LED مع gpiozero
مكتبة GPIO أبسط وأكثر أمانًا للمصابيح.
from gpiozero import LED
from time import sleep
led = LED(17)
for _ in range(3):
led.on()
sleep(0.5)
led.off()
sleep(0.5)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 53: ردود الاتصال عند الضغط على الزر
تفاعل مع الضغط على الزر دون حلقة استطلاع.
from gpiozero import Button
from signal import pause
button = Button(27)
button.when_pressed = lambda: print("pressed")
button.when_released = lambda: print("released")
pause()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 54: التحكم في المحرك المؤازر (Servo)
حرّك المحرك المؤازر إلى مواضع محددة.
from gpiozero import Servo
from time import sleep
servo = Servo(12)
for position in (-1, 0, 1, 0): # -1 = left, 0 = centre, 1 = right
servo.value = position
sleep(1)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 55: مستشعر المسافة بالموجات فوق الصوتية
قِس المسافة بالمستشعر HC-SR04.
import time
import RPi.GPIO as GPIO
TRIG, ECHO = 23, 24
GPIO.setmode(GPIO.BCM)
GPIO.setup(TRIG, GPIO.OUT)
GPIO.setup(ECHO, GPIO.IN)
def distance_cm():
GPIO.output(TRIG, False)
time.sleep(0.05)
GPIO.output(TRIG, True)
time.sleep(0.00001)
GPIO.output(TRIG, False)
while GPIO.input(ECHO) == 0:
start = time.time()
while GPIO.input(ECHO) == 1:
end = time.time()
return (end - start) * 34300 / 2 # speed of sound, there and back
print(round(distance_cm(), 1), "cm")
GPIO.cleanup()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 56: مرحّل بقيمة افتراضية آمنة
شغّل المضخة بأمان وأطفئها دائمًا عند الانتهاء.
from gpiozero import OutputDevice
relay = OutputDevice(22, active_high=True, initial_value=False)
relay.on() # power the pump
relay.off() # always end with the relay off
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 57: أصوات جهاز التنبيه (Buzzer)
شغّل نغمات بسيطة على جهاز التنبيه.
from gpiozero import TonalBuzzer
from time import sleep
buzzer = TonalBuzzer(26)
for note in (0.0, 0.5, 1.0):
buzzer.play(note)
sleep(0.3)
buzzer.stop()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 58: إيقاف GPIO بأمان
اجعل المخرجات آمنة قبل إنهاء البرنامج.
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(17, GPIO.OUT)
try:
GPIO.output(17, GPIO.HIGH)
raise RuntimeError("simulated fault")
except RuntimeError as e:
print("error:", e)
finally:
GPIO.output(17, GPIO.LOW) # make the output safe before cleanup
GPIO.cleanup()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
الموضوع 9: المستشعرات عبر I2C وSPI و1-Wire والكاميرا
تتحدث كثير من المستشعرات مع Pi عبر ناقل. يستخدم I2C سلكين وعنوانًا، ويستخدم SPI أربعة أسلاك وإشارة اختيار الشريحة، ويستخدم 1-Wire سلك بيانات واحدًا. وحدة الكاميرا حالة خاصة لها مكتبتها.
I2C: المسح وقراءة السجلات
لكل جهاز I2C عنوان. امسح الناقل أولًا لتعرف ما هو موصول. اقرأ سجل معرّف الشريحة لتتأكد من المستشعر، ثم اكتب في سجل التحكم واقرأ بايتات البيانات.
SPI والمدخلات التناظرية
ليس لدى Pi مدخلات تناظرية، لذلك يحوّل ADC مثل MCP3008 الفولتيات عبر SPI. أرسل رقم القناة، ثم اجمع القيمة ذات الـ 10 بتات من البايتات المعادة.
1-Wire ومستشعرات DHT والكاميرا
يظهر DS18B20 على 1-Wire كملف في /sys/bus/w1، وعليك تفعيل الواجهة في raspi-config أولًا. تفشل مستشعرات DHT أحيانًا، لذلك أعد المحاولة. تلتقط وحدة الكاميرا الصور الثابتة والإطارات باستخدام picamera2.
مثال 59: فحص ناقل I2C
اعرف المستشعرات الموصولة.
from smbus2 import SMBus
with SMBus(1) as bus:
found = []
for address in range(0x03, 0x78):
try:
bus.read_byte(address)
found.append(hex(address))
except OSError:
pass
print("devices:", found)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 60: قراءة مستشعر الضغط BMP280
تحقق من معرّف الشريحة واقرأ بايتات الضغط الخام عبر I2C.
from smbus2 import SMBus
ADDRESS = 0x76
with SMBus(1) as bus:
chip_id = bus.read_byte_data(ADDRESS, 0xD0)
print("chip id", hex(chip_id), "(expect 0x58)")
bus.write_byte_data(ADDRESS, 0xF4, 0b00100111) # normal mode, oversampling x1
raw = bus.read_i2c_block_data(ADDRESS, 0xF7, 6)
print("raw bytes", raw)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 61: درجة الحرارة والرطوبة بـ DHT22
اقرأ DHT22 وأعد المحاولة عندما لا يكون جاهزًا.
import time
import board
import adafruit_dht
dht = adafruit_dht.DHT22(board.D4)
for _ in range(5):
try:
print(f"{dht.temperature:.1f} C {dht.humidity:.0f} %")
except RuntimeError as e:
print("sensor not ready:", e) # DHT sensors fail sometimes; just retry
time.sleep(2)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 62: مدخل تناظري بـ MCP3008
اقرأ الفولتية التناظرية عبر SPI.
import spidev
spi = spidev.SpiDev()
spi.open(0, 0) # bus 0, device 0
spi.max_speed_hz = 1_350_000
def read_channel(channel):
cmd = [1, (8 + channel) << 4, 0]
reply = spi.xfer2(cmd)
return ((reply[1] & 3) << 8) | reply[2] # 10-bit value, 0 to 1023
print("channel 0 raw:", read_channel(0))
spi.close()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 63: درجة حرارة DS18B20 عبر 1-Wire
اقرأ مستشعر الحرارة المقاوم للماء من النواة.
from pathlib import Path
import glob
device_dirs = glob.glob("/sys/bus/w1/devices/28-*")
if not device_dirs:
raise SystemExit("no DS18B20 found: enable 1-Wire in raspi-config")
lines = (Path(device_dirs[0]) / "w1_slave").read_text().splitlines()
if lines[0].endswith("YES"):
milli = int(lines[1].split("t=")[1])
print(f"{milli / 1000:.3f} C")
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 64: إعادة المحاولة للمستشعرات غير المستقرة
حاول القراءة مرة أخرى قبل الاستسلام.
import time
def read_with_retries(read_fn, attempts=3, delay=1.0):
for attempt in range(1, attempts + 1):
try:
return read_fn()
except OSError as e:
print(f"attempt {attempt} failed: {e}")
time.sleep(delay)
raise RuntimeError("sensor unavailable after retries")
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 65: حفظ المعايرة في ملف JSON
احتفظ بقيم المعايرة في ملف وأعد تحميلها.
import json
from pathlib import Path
CONFIG = Path("calibration.json")
DEFAULTS = {"offset_c": 0.0, "scale": 1.0}
def load():
if CONFIG.exists():
return {**DEFAULTS, **json.loads(CONFIG.read_text(encoding="utf-8"))}
return dict(DEFAULTS)
def save(cfg):
CONFIG.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
cfg = load()
cfg["offset_c"] = -0.4
save(cfg)
print("saved:", load())
CONFIG.unlink()
المخرجات:
saved: {'offset_c': -0.4, 'scale': 1.0}
مثال 66: التقاط صورة ثابتة
التقط صورة بوحدة كاميرا Pi.
from picamera2 import Picamera2
from time import sleep
cam = Picamera2()
cam.configure(cam.create_still_configuration())
cam.start()
sleep(2) # let the auto-exposure settle
cam.capture_file("photo.jpg")
cam.stop()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 67: كشف الحركة من إطارات الكاميرا
قارن الإطارات لاكتشاف الحركة.
import numpy as np
from picamera2 import Picamera2
cam = Picamera2()
cam.configure(cam.create_preview_configuration(main={"size": (320, 240), "format": "YUV420"}))
cam.start()
previous = None
for _ in range(20):
frame = cam.capture_array()[:240].astype(np.int16)
if previous is not None and np.mean(np.abs(frame - previous)) > 10:
print("motion")
previous = frame
cam.stop()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
الموضوع 10: الشبكات باستخدام المقابس و HTTP
يجب أن يتحدث جهاز إنترنت الأشياء مع أجهزة كمبيوتر أخرى. المقابس (sockets) هي الأساس منخفض المستوى، وHTTP هو أسلوب عمل خدمات الويب وواجهات API، والشيفرة المتينة تتعامل مع المهل الزمنية وإعادة المحاولة وفترات الانقطاع.
مقابس TCP وUDP
مقبس TCP اتصال موثوق، بينما يرسل UDP رسائل منفردة بلا اتصال. الربط بالمنفذ 0 يجعل نظام التشغيل يختار منفذًا فارغًا، فلا تتعارض الأمثلة.
طلبات HTTP مع المهلة الزمنية
تنفذ مكتبة requests استدعاءات HTTP في سطر واحد. اضبط دائمًا مهلة زمنية، وإلا فستتجمد البرنامج عند شبكة لا تستجيب. تحوّل raise_for_status الاستجابة الخاطئة إلى استثناء يمكنك معالجته.
إعادة المحاولة والتراجع والطوابير دون اتصال
أعد محاولة الطلبات الفاشلة بتراجع أُسّي، فتنتظر أطول في كل مرة، حتى لا تُغرق خادمًا مشغولًا. عندما تكون الشبكة مقطوعة، احتفظ بالقراءات في طابور وأرسلها عند عودتها.
مثال 68: خادم TCP للصدى وعميل
أرسل بايتات إلى الخادم واقرأ الرد.
import socket
import threading
def serve(server):
conn, _ = server.accept()
with conn:
data = conn.recv(1024)
conn.sendall(b"echo: " + data)
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", 0)) # port 0: let the OS choose a free port
server.listen(1)
port = server.getsockname()[1]
thread = threading.Thread(target=serve, args=(server,))
thread.start()
with socket.create_connection(("127.0.0.1", port)) as client:
client.sendall(b"temp=21.5")
print(client.recv(1024).decode())
thread.join()
server.close()
المخرجات:
echo: temp=21.5
مثال 69: رسائل UDP على عنوان الحلقة المحلية
أرسل حزمة بيانات دون إنشاء اتصال.
import socket
receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
receiver.bind(("127.0.0.1", 0))
addr = receiver.getsockname()
sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sender.sendto(b"humidity=48", addr)
data, _ = receiver.recvfrom(1024)
print("received:", data.decode())
sender.close()
receiver.close()
المخرجات:
received: humidity=48
مثال 70: إرسال القراءات إلى خدمة ويب بـ POST
أرسل قراءة المستشعر كـ JSON عبر HTTPS.
import requests
reading = {"device": "node-1", "temperature_c": 21.5}
response = requests.post(
"https://api.example.com/readings",
json=reading,
timeout=5,
)
response.raise_for_status()
print("server said:", response.status_code)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 71: جلب JSON من واجهة برمجية مع مهلة زمنية
اجلب البيانات وتعامل مع الفشل دون توقف.
import requests
try:
r = requests.get("https://api.example.com/forecast", params={"city": "Bangkok"}, timeout=5)
r.raise_for_status()
data = r.json()
print(data.get("temp_c"))
except requests.RequestException as e:
print("request failed:", e)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 72: إعادة المحاولة مع تأخير متزايد
انتظر أطول بعد كل محاولة فاشلة.
import random
import time
def with_backoff(fn, tries=4, base=0.1):
for n in range(tries):
try:
return fn()
except ConnectionError:
wait = base * (2 ** n)
print(f"try {n + 1} failed, waiting {wait:.1f}s")
time.sleep(wait)
raise ConnectionError("gave up")
random.seed(3)
calls = {"n": 0}
def flaky():
calls["n"] += 1
if calls["n"] < 3:
raise ConnectionError("network down")
return "uploaded"
print(with_backoff(flaky))
المخرجات:
try 1 failed, waiting 0.1s
try 2 failed, waiting 0.2s
uploaded
مثال 73: إرسال JSON بـ urllib
أرسل الطلب بالمكتبة القياسية فقط.
import json
import urllib.request
body = json.dumps({"device": "node-1", "humidity_pct": 48}).encode()
req = urllib.request.Request(
"https://api.example.com/readings",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
print("status", resp.status)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 74: خادم ويب لحالة الجهاز
قدّم صفحة حالة صغيرة على Pi.
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
class Status(BaseHTTPRequestHandler):
def do_GET(self):
body = json.dumps({"status": "ok", "uptime_s": 42}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(body)
HTTPServer(("0.0.0.0", 8000), Status).serve_forever()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 75: فحص الشبكة قبل الإرسال
احفظ البيانات في قائمة انتظار عند انقطاع الإنترنت.
import socket
def internet_ok(host="1.1.1.1", port=53, timeout=3):
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
print("online" if internet_ok() else "offline, queue the data")
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
الموضوع 11: MQTT: النشر والاشتراك
MQTT بروتوكول المراسلة القياسي لإنترنت الأشياء. تنشر الأجهزة القراءات إلى مواضيع، ويسلّمها الوسيط، ويستقبلها أي عدد من المشتركين. وهو خفيف بما يكفي للأجهزة الصغيرة ومتسامح مع الشبكات الضعيفة.
المواضيع والنشر والاشتراك
الموضوع مسار مثل home/kitchen/temperature. يرسل النشر رسالة إلى موضوع عبر الوسيط. وتستخدم الاشتراكات رموزًا عامة: + تطابق مستوى واحدًا، وتطابق # كل ما تحته.
جودة الخدمة والرسائل المحفوظة
تعني QoS 0 الإرسال دون ضمان، وتضمن QoS 1 التسليم مرة واحدة على الأقل، وتسلّم QoS 2 الرسالة مرة واحدة بالضبط. يحتفظ الوسيط بالرسالة المحفوظة، فيحصل المشترك الجديد فورًا على آخر قيمة.
وصية الاتصال والتشفير TLS والاكتشاف
ينشر الوسيط رسالة الوصية عندما ينقطع الجهاز بشكل غير متوقع، وهذا يكشف الأجهزة غير المتصلة. استخدم TLS وكلمة مرور من متغيرات البيئة مع وسيط على الإنترنت. يستطيع Home Assistant العثور على المستشعرات من خلال رسالة اكتشاف تُرسل إلى موضوع خاص.
مثال 76: MQTT: نشر القياسات
أرسل القراءات إلى الوسيط على موضوع محدد.
import json
import time
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.connect("broker.local", 1883, keepalive=60)
client.loop_start()
payload = {"temp_c": 21.5, "ts": int(time.time())}
client.publish("home/kitchen/temperature", json.dumps(payload), qos=1)
time.sleep(1)
client.loop_stop()
client.disconnect()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 77: MQTT: الاشتراك في الرسائل
استقبل القراءات من أجهزة كثيرة بالرموز العامة.
import paho.mqtt.client as mqtt
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe("home/+/temperature")
def on_message(client, userdata, msg):
print(msg.topic, "->", msg.payload.decode())
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.local", 1883)
client.loop_forever()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 78: MQTT: مستويات QoS والرسائل المحفوظة
اختر ضمانات التسليم واحتفظ بآخر قيمة.
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect("broker.local", 1883)
client.loop_start()
# QoS 0 = at most once, 1 = at least once, 2 = exactly once
client.publish("home/door/state", "closed", qos=1, retain=True) # last value is kept
client.loop_stop()
client.disconnect()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 79: MQTT: وصية الاتصال لاكتشاف الانقطاع
أخبر الجميع عندما ينقطع الجهاز.
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.will_set("home/node-1/status", payload="offline", qos=1, retain=True)
client.publish("home/node-1/status", "online", qos=1, retain=True) # set on connect
client.connect("broker.local", 1883)
client.loop_forever()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 80: MQTT عبر TLS
شفّر الاتصال بالوسيط.
import ssl
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.tls_set(ca_certs="/etc/ssl/certs/ca-certificates.crt",
tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.username_pw_set("node-1", "use-a-secret-from-env-not-code")
client.connect("broker.example.com", 8883)
client.loop_forever()
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 81: رسالة اكتشاف Home Assistant
انشر المستشعر ليجده Home Assistant.
import json
config = {
"name": "Kitchen temperature",
"unique_id": "node-1_kitchen_temp",
"state_topic": "home/kitchen/temperature",
"unit_of_measurement": "°C",
"device_class": "temperature",
"value_template": "{{ value_json.temp_c }}",
}
topic = "homeassistant/sensor/node-1_kitchen_temp/config"
print(topic)
print(json.dumps(config, ensure_ascii=False, indent=2))
المخرجات:
homeassistant/sensor/node-1_kitchen_temp/config
{
"name": "Kitchen temperature",
"unique_id": "node-1_kitchen_temp",
"state_topic": "home/kitchen/temperature",
"unit_of_measurement": "°C",
"device_class": "temperature",
"value_template": "{{ value_json.temp_c }}"
}
مثال 82: رموز المواضيع العامة في MQTT
تحقق من المواضيع التي يطابقها الاشتراك.
def topic_matches(pattern, topic):
p_parts = pattern.split("/")
t_parts = topic.split("/")
for i, part in enumerate(p_parts):
if part == "#":
return True
if i >= len(t_parts):
return False
if part != "+" and part != t_parts[i]:
return False
return len(p_parts) == len(t_parts)
tests = [
("home/+/temperature", "home/kitchen/temperature", True),
("home/#", "home/kitchen/humidity", True),
("home/+/temperature", "home/kitchen/humidity", False),
]
for pattern, topic, expected in tests:
matched = topic_matches(pattern, topic)
print(pattern, topic, "match" if matched else "no match", "(as expected)" if matched == expected else "(UNEXPECTED)")
المخرجات:
home/+/temperature home/kitchen/temperature match (as expected)
home/# home/kitchen/humidity match (as expected)
home/+/temperature home/kitchen/humidity no match (as expected)
الموضوع 12: لوحات الويب وواجهات API مع Flask
Flask إطار ويب صغير بلغة Python. بعدة أسطر يعرض صفحة حالة، أو يعيد JSON، أو يقبل قراءات من أجهزة أخرى، فتستطيع رؤية Pi من الهاتف أو المتصفح.
المسارات والاستجابات
يربط مزيّن المسار مثل @app.get('/') عنوان URL بدالة. تعيد الدالة نصًا أو صفحة أو JSON باستخدام jsonify. استخدم العنوان 0.0.0.0 لقبول الاتصالات من الشبكة المحلية.
استقبال البيانات مع التحقق
يقرأ مسار POST بيانات JSON من الطلب. تحقق من النوع والنطاق قبل حفظ أي شيء، وأعِد الحالة 400 عندما تكون البيانات خاطئة، حتى يعرف المرسل ما الذي يجب إصلاحه.
الخيوط الخلفية ولوحات المعلومات
تقرأ خيط خلفي المستشعر بينما يجيب خادم الويب على الطلبات. اجعل البيانات المشتركة بسيطة وآمنة. يحوّل قالب HTML صغير أحدث القراءات إلى صفحة يستطيع أي شخص في الشبكة عرضها.
مثال 83: نقطة hello في Flask
قدّم أول مسار ويب على Pi.
from flask import Flask
app = Flask(__name__)
@app.get("/")
def hello():
return "Raspberry Pi is alive"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 84: نقطة JSON في Flask
أعد آخر قراءة بصيغة JSON.
from flask import Flask, jsonify
app = Flask(__name__)
latest = {"temp_c": 21.5, "humidity_pct": 48}
@app.get("/api/latest")
def get_latest():
return jsonify(latest)
app.run(host="0.0.0.0", port=5000)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 85: Flask POST مع التحقق
اقبل القراءات وارفض القيم الخاطئة برمز خطأ.
from flask import Flask, request, jsonify
app = Flask(__name__)
readings = []
@app.post("/api/readings")
def add_reading():
data = request.get_json(silent=True) or {}
temp = data.get("temp_c")
if not isinstance(temp, (int, float)) or not -40 <= temp <= 85:
return jsonify(error="temp_c must be a number from -40 to 85"), 400
readings.append(temp)
return jsonify(saved=len(readings)), 201
app.run(port=5000)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 86: لوحة معلومات HTML صغيرة
اعرض صفحة تظهر آخر درجة حرارة.
from flask import Flask, render_template_string
app = Flask(__name__)
PAGE = """
<h1>Kitchen</h1>
<p>Temperature: {{ temp }} °C</p>
"""
@app.get("/")
def home():
return render_template_string(PAGE, temp=21.5)
app.run(port=5000)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 87: Flask مع خيط مستشعر في الخلفية
اقرأ المستشعر باستمرار أثناء عمل خادم الويب.
import threading
import time
from flask import Flask, jsonify
app = Flask(__name__)
state = {"temp_c": None}
def sensor_loop():
while True:
state["temp_c"] = 21.5 # replace with the real sensor read
time.sleep(5)
threading.Thread(target=sensor_loop, daemon=True).start()
@app.get("/api/temp")
def temp():
return jsonify(state)
app.run(port=5000)
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
الموضوع 13: تخزين القراءات وتلخيصها
يحتاج جهاز يسجل لعام كامل إلى تخزين بسيط لا يملأ بطاقة SD. SQLite قاعدة بيانات بملف واحد تأتي مع Python، وملفات CSV مناسبة للسجل البسيط.
جداول SQLite والإدراج
تخزن SQLite الجداول في ملف واحد. استخدم العناصر النائبة ? في عبارات INSERT، فهذا يحمي من حقن SQL ويتعامل مع علامات الاقتباس. استدعِ commit بعد الكتابة حتى تُحفظ البيانات.
التجميع وفئات الوقت
تلخص GROUP BY مع AVG وCOUNT القراءات لكل جهاز. تجميع الطوابع الزمنية بالدقيقة، مثل أخذ أول خمسة أحرف من وقت ISO، يعطي متوسطات بسيطة لكل دقيقة.
الاحتفاظ بالبيانات وسجلات الإلحاق فقط
احذف الصفوف القديمة وفق جدول زمني حتى يبقى التخزين صغيرًا. يضيف ملف CSV للإلحاق فقط سطرًا لكل قراءة ويكتب الرأس مرة واحدة، وهو بسيط ومتين.
مثال 88: SQLite: إنشاء جدول وإدراج
خزّن القراءات في ملف قاعدة بيانات.
import sqlite3
conn = sqlite3.connect(":memory:") # use "readings.db" to keep data on the Pi
conn.execute("""CREATE TABLE readings (
ts TEXT NOT NULL, device TEXT NOT NULL, temp_c REAL NOT NULL)""")
rows = [("2026-01-15T08:00", "node-1", 21.5), ("2026-01-15T08:05", "node-1", 21.9)]
conn.executemany("INSERT INTO readings VALUES (?, ?, ?)", rows)
conn.commit()
print(conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0], "rows stored")
conn.close()
المخرجات:
2 rows stored
مثال 89: SQLite: استعلامات التجميع
احسب المتوسط والعدد لكل جهاز.
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (device TEXT, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)",
[("node-1", 21.5), ("node-1", 22.5), ("node-2", 19.0)])
for device, avg, n in conn.execute(
"SELECT device, AVG(temp_c), COUNT(*) FROM readings GROUP BY device ORDER BY device"):
print(f"{device}: average {avg:.2f} from {n} readings")
conn.close()
المخرجات:
node-1: average 22.00 from 2 readings
node-2: average 19.00 from 1 readings
مثال 90: مسجّل سجل CSV
أضف القراءات إلى ملف السجل مع رأس الأعمدة.
import csv
from pathlib import Path
LOG = Path("sensor_history.csv")
FIELDS = ["ts", "temp_c"]
def log_reading(ts, temp_c):
new_file = not LOG.exists()
with LOG.open("a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=FIELDS)
if new_file:
writer.writeheader()
writer.writerow({"ts": ts, "temp_c": temp_c})
log_reading("08:00", 21.5)
log_reading("08:05", 21.9)
print(LOG.read_text(encoding="utf-8"))
LOG.unlink()
المخرجات:
ts,temp_c
08:00,21.5
08:05,21.9
مثال 91: تجميع القراءات بالدقيقة
اجمع القراءات في متوسطات لكل دقيقة.
from collections import defaultdict
from statistics import mean
readings = [("08:00:10", 21.4), ("08:00:40", 21.6), ("08:01:05", 21.9)]
buckets = defaultdict(list)
for ts, value in readings:
buckets[ts[:5]].append(value) # keep HH:MM
for minute in sorted(buckets):
print(minute, round(mean(buckets[minute]), 2))
المخرجات:
08:00 21.5
08:01 21.9
مثال 92: الاحتفاظ بالبيانات
احذف الصفوف القديمة لإبقاء تخزين Pi صغيرًا.
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (day INTEGER, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)", [(d, 20.0 + d / 10) for d in range(1, 11)])
keep_from = 6
deleted = conn.execute("DELETE FROM readings WHERE day < ?", (keep_from,)).rowcount
conn.commit()
remaining = conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0]
print("deleted", deleted, "kept", remaining)
conn.close()
المخرجات:
deleted 5 kept 5
الموضوع 14: التشغيل كخدمة والإعدادات
لا يكون المشروع مفيدًا إلا إذا بدأ من تلقاء نفسه بعد انقطاع الكهرباء واستمر في العمل. يشغّل systemd الخدمات عند الإقلاع، ويشغّل cron المهام وفق جدول، وتُبقي متغيرات البيئة الإعدادات خارج الشيفرة.
خدمات systemd
يصف ملف الوحدة طريقة تشغيل البرنامج ومجلد العمل والمستخدم وما يُفعل عند الفشل باستخدام Restart=on-failure. فعّل الخدمة لتبدأ عند الإقلاع، واقرأ سجلاتها باستخدام journalctl.
المهام المجدولة بـ cron
يشغّل سطر في crontab أمرًا في أوقات محددة، مثل كل خمس دقائق. وجّه المخرجات إلى ملف سجل حتى تعرف ما الذي حدث وأنت بعيد.
الإعدادات من البيئة
اقرأ إعدادات مثل معرّف الجهاز وفترة الإرسال وعنوان الوسيط من متغيرات البيئة مع قيم افتراضية آمنة. يمكنك تغييرها في ملف الخدمة دون تعديل الشيفرة.
مثال 93: خدمة systemd عند الإقلاع
شغّل المسجّل تلقائيًا عند بدء تشغيل Pi.
# /etc/systemd/system/sensor-logger.service
[Unit]
Description=Sensor logger
After=network-online.target
Wants=network-online.target
[Service]
ExecStart=/usr/bin/python3 /home/pi/iot/logger.py
WorkingDirectory=/home/pi/iot
Restart=on-failure
RestartSec=5
User=pi
[Install]
WantedBy=multi-user.target
# enable and start it:
# sudo systemctl daemon-reload
# sudo systemctl enable --now sensor-logger.service
# journalctl -u sensor-logger -f
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 94: المهام المجدولة بـ cron
شغّل سكربتًا كل خمس دقائق.
# edit the schedule with: crontab -e
# every 5 minutes, record the time and the readings file size
*/5 * * * * /usr/bin/python3 /home/pi/iot/snapshot.py >> /home/pi/iot/cron.log 2>&1
# every day at 02:30, remove logs older than 30 days
30 2 * * * find /home/pi/iot/logs -name "*.csv" -mtime +30 -delete
يحتاج هذا المثال إلى Raspberry Pi أو خدمة شبكية أو خدمة نظام، لذلك فُحصت صياغته فقط ولا تظهر مخرجاته.
مثال 95: ملف نبض لمراقب الكلب الحارس
اكتب طابعًا زمنيًا ليعرف المراقب أن البرنامج يعمل.
import time
from pathlib import Path
HEARTBEAT = Path("heartbeat.txt")
def beat(count):
HEARTBEAT.write_text(f"{count} {time.time():.0f}", encoding="utf-8")
for i in range(3):
beat(i)
print("beat", i, "written")
time.sleep(0.05)
print("last:", HEARTBEAT.read_text(encoding="utf-8").split()[0])
HEARTBEAT.unlink()
المخرجات:
beat 0 written
beat 1 written
beat 2 written
last: 2
مثال 96: الإعدادات من متغيرات البيئة
اقرأ الإعدادات مع قيم افتراضية آمنة.
import os
def get_config():
return {
"device_id": os.environ.get("DEVICE_ID", "node-1"),
"interval_s": int(os.environ.get("INTERVAL_S", "60")),
"broker": os.environ.get("MQTT_HOST", "localhost"),
}
print(get_config())
المخرجات:
{'device_id': 'node-1', 'interval_s': 60, 'broker': 'localhost'}
الموضوع 15: السجلات والأمان وخط المعالجة الكامل
تجمع الموضوعات الأخيرة كل شيء معًا: سجلات دوّارة حتى لا تمتلئ بطاقة SD، وأسرار تبقى خارج الشيفرة، والتحقق من البيانات الواردة، وخط معالجة كامل من المستشعر إلى الإنذار.
السجلات الدوّارة ونبضات القلب
يحد معالج الملفات الدوّار من حجم كل ملف سجل، ويحتفظ ببضعة ملفات قديمة. يسمح ملف نبضات القلب بطابع زمني لمراقب الكلب الحارس بالتحقق من أن البرنامج ما زال يعمل.
الأسرار والتحقق
احفظ كلمات المرور والرموز في متغيرات البيئة، ولا تطبعها أبدًا. تحقق من كل قيمة واردة في النوع والنطاق قبل أن تصل إلى التخزين أو الإنذار، لأن المستشعرات والشبكات قد ترسل أي شيء.
تجميع خط المعالجة
يقرأ خط المعالجة المستشعر، ويُنعّم القيمة بنافذة، ويقارنها بعتبة، ويرفع علامة إنذار، ويخزن كل خطوة. كل مرحلة صغيرة وقابلة للاختبار، وهكذا تُنظَّم شيفرة إنترنت الأشياء الاحترافية.
مثال 97: ملفات السجل المتداولة
امنع السجلات من ملء بطاقة SD.
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
path = Path("iot.log")
handler = RotatingFileHandler(path, maxBytes=200, backupCount=2, encoding="utf-8")
log = logging.getLogger("iot")
log.setLevel(logging.INFO)
log.addHandler(handler)
for i in range(20):
log.info("reading %d: temp=21.%d", i, i % 10)
handler.close()
print("files:", sorted(p.name for p in Path(".").glob("iot.log*")))
for p in Path(".").glob("iot.log*"):
p.unlink()
المخرجات:
files: ['iot.log', 'iot.log.1', 'iot.log.2']
مثال 98: الأسرار من متغيرات البيئة
حمّل كلمات المرور من البيئة ولا تطبعها أبدًا.
import os
def require_secret(name):
value = os.environ.get(name)
if not value:
raise SystemExit(f"set {name} in the environment (do not write it in the code)")
return value
def mask(secret):
return secret[:2] + "*" * (len(secret) - 2)
os.environ["MQTT_PASSWORD"] = "example-secret" # demo only
token = require_secret("MQTT_PASSWORD")
print("loaded secret:", mask(token))
المخرجات:
loaded secret: ex************
مثال 99: التحقق من بيانات المستشعر الواردة
تحقق من النوع والنطاق قبل التخزين.
from dataclasses import dataclass
@dataclass
class Sample:
device: str
temp_c: float
def validate(raw):
errors = []
if not isinstance(raw.get("device"), str) or not raw["device"]:
errors.append("device is required")
temp = raw.get("temp_c")
if not isinstance(temp, (int, float)):
errors.append("temp_c must be a number")
elif not -40 <= temp <= 85:
errors.append("temp_c out of sensor range")
if errors:
return None, errors
return Sample(raw["device"], float(temp)), []
print(validate({"device": "node-1", "temp_c": 21.5}))
print(validate({"device": "", "temp_c": 300}))
المخرجات:
(Sample(device='node-1', temp_c=21.5), [])
(None, ['device is required', 'temp_c out of sensor range'])
مثال 100: خط معالجة صغير كامل
اقرأ، ونعّم، واكتشف الإنذار، وخزّن النتائج.
import sqlite3
from collections import deque
def fake_sensor(n):
base = [21.0, 21.2, 21.1, 35.0, 21.3, 21.4] # 35.0 is a spike
return base[n % len(base)]
window = deque(maxlen=3)
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (tick INTEGER, raw REAL, smooth REAL, alarm INTEGER)")
for tick in range(6):
raw = fake_sensor(tick)
window.append(raw)
smooth = sum(window) / len(window)
alarm = 1 if smooth > 25 else 0
conn.execute("INSERT INTO readings VALUES (?, ?, ?, ?)", (tick, raw, smooth, alarm))
print(f"tick {tick}: raw {raw:5.1f} smooth {smooth:5.2f} alarm {alarm}")
alarms = conn.execute("SELECT COUNT(*) FROM readings WHERE alarm = 1").fetchone()[0]
print("alarm ticks:", alarms)
conn.close()
المخرجات:
tick 0: raw 21.0 smooth 21.00 alarm 0
tick 1: raw 21.2 smooth 21.10 alarm 0
tick 2: raw 21.1 smooth 21.10 alarm 0
tick 3: raw 35.0 smooth 25.77 alarm 1
tick 4: raw 21.3 smooth 25.80 alarm 1
tick 5: raw 21.4 smooth 25.90 alarm 1
alarm ticks: 3
مشروع تدريبي: مراقب الغرفة
ابنِ مراقبًا واحدًا يستخدم أفكارًا من كل قسم. اكتبه بنفسك أولًا، ثم قارنه بالأمثلة. اختبر أجزاء Python وحدها على حاسوبك قبل توصيل العتاد.
1. اكتب دالة تحوّل القراءات الخام إلى درجة حرارة باستخدام نقطتي معايرة، واختبرها بثلاث قيم.
2. نعّم القراءات بمتوسط متحرك لآخر خمس قيم، وأضف تباطؤًا Hysteresis لمروحة بعتبتين 24 و22.
3. خزّن كل قراءة في جدول SQLite، واطبع متوسط كل دقيقة بعد عشر قراءات.
4. انشر آخر قراءة إلى موضوع MQTT كل 60 ثانية، وعندما يكون الوسيط غير متصل ضع القراءات في طابور.
5. أنشئ خدمة systemd تشغّل المراقب عند الإقلاع، واقرأ سجلاتها باستخدام journalctl.
الخطوات التالية
بعد ذلك، تعلّم منصة لإنترنت الأشياء مثل Home Assistant أو Node-RED، وأضف هوية آمنة للأجهزة والتحديثات اللاسلكية، وادرس Raspberry Pi Pico لأعمال المتحكمات الدقيقة. اختبر كل مثال عتادي أولًا بتوصيلاتك، وراجع وثائق Raspberry Pi الرسمية لأرقام الأطراف في نموذجك.
الخلاصة
يصبح Raspberry Pi جهازًا حقيقيًا لإنترنت الأشياء عندما يقرأ المستشعرات بموثوقية، ويعالج البيانات، ويخزنها، ويرسلها عبر الشبكة، ويعمل من تلقاء نفسه. عبر 100 مثال، غطت هذه المقالة أساسيات Python، والقرارات والحلقات، والدوال، وبنى البيانات، والملفات وصيغها، ومعالجة الأخطاء والتسجيل، ومعالجة بيانات المستشعرات، وGPIO وI2C وSPI و1-Wire والكاميرا، والمقابس وHTTP، وMQTT، وFlask، وSQLite، وsystemd وcron، وخط المعالجة الكامل. ابنِ مراقب الغرفة في التدريب، وغيّر شيئًا واحدًا في كل مرة لتتعلم كيف يتصرف كل جزء.