script.py - master - nikitaselkov/Microcontrollers_labNo6

script.py
165 строк5 KB
nikitaselkov
upload files
17 дек 2025, 21:37
17 дек 2025, 21:37e7ab88d
import paho.mqtt.client as mqtt
from paho.mqtt.client import CallbackAPIVersion
import logging

# ===== НАСТРОЙКИ MQTT =====
BROKER_ADDRESS = 'm3.wqtt.ru'
BROKER_PORT = 14275
MQTT_USER = 'u_KGO7LO'
MQTT_PASSWORD = 'G2pyuUHz'

TOPIC_LUXMETER_DATA = 'luxmeter/data'
TOPIC_LED_CONTROL = 'led/command'
TOPIC_BRIGHTNESS = 'brightness/set'

# ===== ПОРОГОВЫЕ ЗНАЧЕНИЯ ОСВЕЩЕННОСТИ =====
THRESHOLD_RED = 700
THRESHOLD_YELLOW = 500
THRESHOLD_GREEN = 300

# ===== СОСТОЯНИЕ СВЕТОДИОДОВ =====
red_state = False
yellow_state = False
green_state = False

# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')


# ===== ФУНКЦИЯ: Подключение к MQTT =====
def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        logging.info('✅ MQTT connected!')
        client.subscribe(TOPIC_LUXMETER_DATA)
        logging.info(f'📡 Subscribed to: {TOPIC_LUXMETER_DATA}')
    else:
        logging.error(f'❌ MQTT failed: {reason_code}')


# ===== ФУНКЦИЯ: Обработка входящих сообщений =====
def on_message(client, userdata, msg):
    global red_state, yellow_state, green_state

    try:
        lux_value = int(msg.payload.decode())
        logging.info(f"📊 Light level: {lux_value}")

        if lux_value >= THRESHOLD_RED:
            if red_state or yellow_state or green_state:
                send_led_command('ALL_OFF')
                red_state = False
                yellow_state = False
                green_state = False
                logging.info("⚫⚫⚫ All OFF")

        elif lux_value < THRESHOLD_RED and lux_value >= THRESHOLD_YELLOW:
            if not red_state:
                send_led_command('RED_ON')
                red_state = True
                logging.info("🔴 RED ON")
            if yellow_state:
                send_led_command('YELLOW_OFF')
                yellow_state = False
            if green_state:
                send_led_command('GREEN_OFF')
                green_state = False

        elif lux_value < THRESHOLD_YELLOW and lux_value >= THRESHOLD_GREEN:
            if not red_state:
                send_led_command('RED_ON')
                red_state = True
            if not yellow_state:
                send_led_command('YELLOW_ON')
                yellow_state = True
                logging.info("🔴🟡 RED + YELLOW ON")
            if green_state:
                send_led_command('GREEN_OFF')
                green_state = False

        elif lux_value < THRESHOLD_GREEN:
            if not red_state:
                send_led_command('RED_ON')
                red_state = True
            if not yellow_state:
                send_led_command('YELLOW_ON')
                yellow_state = True
            if not green_state:
                send_led_command('GREEN_ON')
                green_state = True
                logging.info("🔴🟡🟢 ALL ON")

    except ValueError as e:
        logging.error(f"❌ Error: {e}")


# ===== ФУНКЦИЯ: Отправка команды =====
def send_led_command(command):
    result = client.publish(TOPIC_LED_CONTROL, command)
    logging.info(f"➡️  Sent: {command}")


# ===== ФУНКЦИЯ: Установка яркости =====
def set_brightness(brightness_value):
    if 0 <= brightness_value <= 255:
        client.publish(TOPIC_BRIGHTNESS, str(brightness_value))
        logging.info(f"💡 Brightness: {brightness_value}")
    else:
        logging.warning("⚠️  Brightness must be 0-255")


# ===== ФУНКЦИЯ: Интерактивное управление =====
def interactive_control():
    import threading

    def input_thread():
        print("\n" + "=" * 60)
        print("🎛️  BRIGHTNESS CONTROL")
        print("=" * 60)
        print("Enter brightness (0-255) or 'q' to quit")
        print("  0   = OFF")
        print("  128 = 50%")
        print("  255 = 100%")
        print("=" * 60 + "\n")

        while True:
            try:
                cmd = input("💡 Brightness: ")
                if cmd.lower() == 'q':
                    logging.info("Exiting...")
                    client.disconnect()
                    break
                else:
                    brightness = int(cmd)
                    set_brightness(brightness)
            except ValueError:
                logging.warning("⚠️  Enter 0-255 or 'q'")
            except:
                break

    thread = threading.Thread(target=input_thread, daemon=True)
    thread.start()


# ===== MAIN =====
if __name__ == '__main__':
    # Создаем клиент с новым API
    client = mqtt.Client(
        callback_api_version=CallbackAPIVersion.VERSION2,
        client_id="PythonClient"
    )

    client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
    client.on_connect = on_connect
    client.on_message = on_message

    try:
        logging.info("🚀 Connecting to MQTT...")
        client.connect(BROKER_ADDRESS, BROKER_PORT, 60)
        logging.info("✅ Connected!")
        interactive_control()
        client.loop_forever()
    except KeyboardInterrupt:
        logging.info("⛔ Interrupted")
        client.disconnect()
    except Exception as e:
        logging.error(f"❌ Error: {e}")