Решение для MQTT-управляемого термовключателя реле на базе Raspberry Pi Zero с датчиками DHT-11, BMP-180, двустрочным LCD дисплеем и кнопкой включения его подсветки.
- На интернет-сервере поднять MQTT-брокер (например, mosquitto). Отдельная песня, довольно простая.
- sudo raspi-config (включить I2C)
- Установить python3, pip и общие библиотеки: sudo apt install python3-pip i2c-tools, pip3 install paho-mqtt, sudo apt install python3-rpi-lgpio, sudo apt install libgpiod3
- Следующие модули могут не установиться в систему (типа опасно), поэтому дальше работаем в виртуальном окружении venv. 1) python3 -m venv .venv 2) source .venv/bin/activate 3) pip install adafruit-circuitpython-dht RPLCD smbus2 4) pip install circuitpython-bmp180
- Текст программы (ds_climate.py) см. ниже
- Управляющая строка для
mosquitto_pubдолжна содержать JSON с теми полями, которые программа ожидает: mosquitto_pub -h ip_address -p 1883 -u user -P password -t «home/pi-zero/control» \ -m ‘{«min_temp»: 18, «max_temp»: 24, «log_interval»: 30, «max_log_size»: 2048}’ - Для автоматического запуска программы с виртуальным окружением (
venv) после перезагрузки Raspberry Pi лучше всего использовать systemd. 1) sudo nano /etc/systemd/system/ds_climate.service 2) Вставить следующее содержимое: [Unit] Description=DS Climate Controller (DHT11 + BMP180 + MQTT) \n After=network-online.target \n Wants=network-online.target \n [Service] \n Type=simple \n User=papa \n WorkingDirectory=/home/papa \n ExecStart=/home/papa/venv/bin/python /home/papa/ds_climate.py \n Restart=always \n RestartSec=5 \n StandardOutput=journal \n StandardError=journal \n [Install] WantedBy=multi-user.target 3) sudo systemctl daemon-reload 4) sudo systemctl enable ds_climate.service 5) sudo systemctl start ds_climate.service - mosquitto_sub -h ip_address -t home/pi-zero/sensor -p 1883 -u user -P password , видим: {«temperature»: 23.0, «pressure»: 1005.6, «humidity»: null, «relay»: «OFF», «min_temp»: 20.0, «max_temp»: 25.0, «log_interval»: 60, «max_log_size»: 10240, «backlight»: «ON»}
- Всё это быстро устаревает вместе с прошивками и библиотеками.
import paho.mqtt.client as mqtt
import RPi.GPIO as GPIO
import json
import time
import os
import subprocess
from datetime import datetime
# --- LCD (RPLCD) ---
from RPLCD.i2c import CharLCD
# --- BMP180 (пакет circuitpython-bmp180) ---
import board
import busio
import bmp180
# --- DHT (adafruit-circuitpython-dht) ---
import adafruit_dht
# ================= НАСТРОЙКИ =================
WIFI_SSID = "mywifi"
WIFI_PASSWORD = "mywifipassword"
MQTT_BROKER = "192.168.0.101"
MQTT_PORT = 1883
MQTT_CLIENT_ID = "pi-zero-ctrl-01"
MQTT_USER = "mqttuser"
MQTT_PASSWORD = "mypassword"
TOPIC_SENSOR = "home/pi-zero/sensor"
TOPIC_CONTROL = "home/pi-zero/control"
DHT_PIN = board.D14 # GPIO14 для DHT11
RELAY_PIN = 17
RELAY_ACTIVE_LOW = True
BUTTON_PIN = 27
BUTTON_BOUNCE_MS = 300
LOG_FILE = "sensor_log.txt"
BACKUP_LOG_FILE = "sensor_log_backup.txt"
min_temp = 20.0
max_temp = 25.0
log_interval = 60
max_log_size = 10240
last_log_time = 0
last_relay_state = None
backlight_on = True
# --- LCD (RPLCD) ---
# port=1 соответствует /dev/i2c-1 на Raspberry Pi
lcd = CharLCD(i2c_expander='PCF8574', address=0x27, port=1,
cols=16, rows=2, auto_linebreaks=True)
lcd.clear()
lcd.backlight_enabled = backlight_on
# --- BMP180 ---
i2c = busio.I2C(board.SCL, board.SDA, frequency=100000)
bmp_sensor = bmp180.BMP180(i2c)
bmp_sensor.sea_level_pressure = 1013.25
# --- DHT11 (только влажность) ---
dht_device = adafruit_dht.DHT11(DHT_PIN)
# =============================================
# --- LCD helper ---
def lcd_write(line1="", line2=""):
"""Пишет две строки по 16 символов, очищая остатки старых значений."""
lcd.cursor_pos = (0, 0)
lcd.write_string(f"{line1:<16}"[:16])
lcd.cursor_pos = (1, 0)
lcd.write_string(f"{line2:<16}"[:16])
# --- Лог ---
def rotate_log_if_needed():
if os.path.exists(LOG_FILE):
size = os.path.getsize(LOG_FILE)
if size >= max_log_size:
try:
os.replace(LOG_FILE, BACKUP_LOG_FILE)
open(LOG_FILE, "w").close()
print(f"[ЛОГ] Ротация: {size} байт")
except Exception as e:
print(f"[ЛОГ] Ошибка ротации: {e}")
def write_log(line):
rotate_log_if_needed()
try:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception as e:
print(f"[ЛОГ] Ошибка записи: {e}")
def log_event(event_text):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
write_log(f"[{ts}] EVENT: {event_text}")
print(f"[СОБЫТИЕ] {event_text}")
def log_sensor_data(temperature, pressure, humidity, relay_on):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
relay_str = "ON" if relay_on else "OFF"
write_log(f"[{ts}] DATA: T={temperature}°C, P={pressure}hPa, H={humidity}%, Relay={relay_str}")
# --- Wi-Fi ---
def is_wifi_connected():
try:
result = subprocess.run(["iwgetid", "-r"], capture_output=True, text=True)
return bool(result.stdout.strip())
except Exception:
return False
def get_ip():
try:
ip = subprocess.run(["hostname", "-I"], capture_output=True, text=True).stdout.strip()
return ip.split()[0] if ip else "0.0.0.0"
except Exception:
return "0.0.0.0"
def connect_wifi(ssid, password, timeout=30):
log_event(f"Старт поиска Wi-Fi: SSID='{ssid}'")
if is_wifi_connected():
ip = get_ip()
print(f"[WiFi] Уже подключено. IP: {ip}")
log_event(f"Wi-Fi уже подключён. IP={ip}")
lcd_write("WiFi OK", ip)
time.sleep(5)
return True
try:
subprocess.run(["wpa_cli", "-i", "wlan0", "add_network"],
capture_output=True, text=True, check=True)
networks = subprocess.run(["wpa_cli", "-i", "wlan0", "list_networks"],
capture_output=True, text=True).stdout
lines = [l for l in networks.splitlines() if l and not l.startswith("network id")]
if not lines:
log_event("ОШИБКА: не удалось добавить сеть в wpa_cli")
lcd_write("WiFi: Error", "No network")
time.sleep(2)
return False
net_id = lines[-1].split("\t")[0]
subprocess.run(["wpa_cli", "-i", "wlan0", "set_network", net_id, "ssid", f'"{ssid}"'],
capture_output=True, text=True, check=True)
subprocess.run(["wpa_cli", "-i", "wlan0", "set_network", net_id, "psk", f'"{password}"'],
capture_output=True, text=True, check=True)
subprocess.run(["wpa_cli", "-i", "wlan0", "enable_network", net_id],
capture_output=True, text=True, check=True)
subprocess.run(["wpa_cli", "-i", "wlan0", "select_network", net_id],
capture_output=True, text=True, check=True)
subprocess.run(["wpa_cli", "-i", "wlan0", "save_config"],
capture_output=True, text=True, check=True)
log_event("Сеть добавлена в wpa_cli, начинаем ожидание подключения")
except subprocess.CalledProcessError as e:
log_event(f"ОШИБКА wpa_cli: {e}")
lcd_write("WiFi: Error", "wpa_cli failed")
time.sleep(2)
return False
except Exception as e:
log_event(f"ОШИБКА Wi-Fi: {e}")
lcd_write("WiFi: Error", "")
time.sleep(2)
return False
attempt = 0
while attempt < timeout:
attempt += 1
lcd_write("WiFi:", f"{ssid[:10]} #{attempt}")
time.sleep(1)
if is_wifi_connected():
ip = get_ip()
print(f"[WiFi] Подключено на попытке {attempt}. IP: {ip}")
log_event(f"Wi-Fi подключён с попытки {attempt}. IP={ip}")
lcd_write("WiFi OK", ip)
time.sleep(5)
return True
log_event(f"ОШИБКА: таймаут Wi-Fi после {timeout} попыток")
lcd_write("WiFi: Timeout", "")
time.sleep(2)
return False
# --- GPIO ---
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(RELAY_PIN, GPIO.OUT)
if RELAY_ACTIVE_LOW:
GPIO.output(RELAY_PIN, GPIO.HIGH)
else:
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
def toggle_backlight(channel=None):
global backlight_on
backlight_on = not backlight_on
lcd.backlight_enabled = backlight_on
log_event(f"Подсветка {'ВКЛ' if backlight_on else 'ВЫКЛ'} (кнопка)")
print(f"[КНОПКА] Подсветка {'ВКЛ' if backlight_on else 'ВЫКЛ'}")
GPIO.add_event_detect(BUTTON_PIN, GPIO.FALLING,
callback=toggle_backlight,
bouncetime=BUTTON_BOUNCE_MS)
def set_relay(on):
if RELAY_ACTIVE_LOW:
GPIO.output(RELAY_PIN, GPIO.LOW if on else GPIO.HIGH)
else:
GPIO.output(RELAY_PIN, GPIO.HIGH if on else GPIO.LOW)
def get_relay_state():
val = GPIO.input(RELAY_PIN)
if RELAY_ACTIVE_LOW:
return val == GPIO.LOW
return val == GPIO.HIGH
# --- LCD ---
def update_lcd(temperature, pressure, humidity):
line1 = f"T:{temperature:.1f}C P:{pressure:.0f}"
line2 = f"H:{humidity:.1f}%"
lcd_write(line1, line2)
# --- MQTT ---
def on_connect(client, userdata, flags, rc):
if rc == 0:
print(f"[MQTT] Подключено к {MQTT_BROKER}")
log_event(f"MQTT подключено к {MQTT_BROKER}")
client.subscribe(TOPIC_CONTROL)
else:
print(f"[MQTT] Ошибка подключения, код: {rc}")
log_event(f"ОШИБКА MQTT подключения, код={rc}")
def on_message(client, userdata, msg):
global min_temp, max_temp, log_interval, max_log_size
try:
payload = json.loads(msg.payload.decode())
print(f"[MQTT] Управление: {payload}")
if "min_temp" in payload:
min_temp = float(payload["min_temp"])
if "max_temp" in payload:
max_temp = float(payload["max_temp"])
if "log_interval" in payload:
log_interval = max(1, int(payload["log_interval"]))
if "max_log_size" in payload:
max_log_size = max(100, int(payload["max_log_size"]))
log_event(f"Получены настройки: min={min_temp}, max={max_temp}, "
f"log_interval={log_interval}, max_log_size={max_log_size}")
except Exception as e:
log_event(f"ОШИБКА парсинга MQTT: {e}")
# --- Основная логика ---
def main():
global last_log_time, last_relay_state
log_event("=== СТАРТ ПРОГРАММЫ ===")
lcd_write("Starting...", "")
time.sleep(1)
connect_wifi(WIFI_SSID, WIFI_PASSWORD)
client = mqtt.Client(client_id=MQTT_CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
if MQTT_USER and MQTT_PASSWORD:
client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
try:
client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
client.loop_start()
except Exception as e:
print(f"[MQTT] Не удалось подключиться: {e}")
log_event(f"ОШИБКА подключения к брокеру: {e}")
try:
while True:
# --- BMP180: температура и давление ---
bmp_temp = None
bmp_press = None
try:
bmp_temp = round(bmp_sensor.temperature, 1)
bmp_press = round(bmp_sensor.pressure, 1)
except Exception as e:
print(f"[BMP180] Ошибка чтения: {e}")
log_event(f"ОШИБКА чтения BMP180: {e}")
# --- DHT: только влажность ---
dht_hum = None
try:
hum_value = dht_device.humidity
if hum_value is not None:
dht_hum = round(hum_value, 1)
except RuntimeError as error:
print(f"[DHT] Ошибка чтения (RuntimeError): {error.args[0]}")
except Exception as e:
print(f"[DHT] Неизвестная ошибка: {e}")
log_event(f"ОШИБКА чтения DHT: {e}")
# --- Логика управления реле по температуре BMP180 ---
if bmp_temp is not None:
if bmp_temp < min_temp:
set_relay(True)
elif bmp_temp > max_temp:
set_relay(False)
relay_on = get_relay_state()
if dht_hum is not None:
update_lcd(bmp_temp, bmp_press if bmp_press else 0, dht_hum)
else:
lcd_write(f"T:{bmp_temp:.1f}C P:{bmp_press or 0:.0f}", "H: --")
if last_relay_state is not None and relay_on != last_relay_state:
log_event(f"Реле {'ВКЛ' if relay_on else 'ВЫКЛ'} (T={bmp_temp}°C)")
last_relay_state = relay_on
now = time.time()
if now - last_log_time >= log_interval:
log_sensor_data(bmp_temp, bmp_press or 0, dht_hum or 0, relay_on)
last_log_time = now
payload = {
"temperature": bmp_temp,
"pressure": bmp_press,
"humidity": dht_hum,
"relay": "ON" if relay_on else "OFF",
"min_temp": min_temp,
"max_temp": max_temp,
"log_interval": log_interval,
"max_log_size": max_log_size,
"backlight": "ON" if backlight_on else "OFF"
}
try:
client.publish(TOPIC_SENSOR, json.dumps(payload))
except Exception as e:
print(f"[MQTT] Ошибка публикации: {e}")
log_event(f"ОШИБКА публикации MQTT: {e}")
else:
lcd_write("BMP180 Error", "Check wiring")
log_event("ОШИБКА: BMP180 не вернул температуру")
time.sleep(5)
except KeyboardInterrupt:
print("\n[INFO] Остановка")
log_event("=== ОСТАНОВКА ПРОГРАММЫ ===")
finally:
GPIO.remove_event_detect(BUTTON_PIN)
try:
client.loop_stop()
client.disconnect()
except Exception:
pass
try:
dht_device.exit()
except Exception:
pass
lcd.clear()
lcd.backlight_enabled = False
GPIO.cleanup()
if __name__ == "__main__":
main()