# https://t.me/ZGQinc import machine import network import time import json import webrepl from umqtt.simple import MQTTClient WIFI_SSID = '' WIFI_PASSWORD = '' MQTT_BROKER = '' MQTT_USER = '' MQTT_PASSWORD = '' CLIENT_ID = 'esp32c3_water_ripple' LIGHT_CMD_TOPIC = b"water_ripple/light/set" LIGHT_STATE_TOPIC = b"water_ripple/light/state" MOTOR_CMD_TOPIC = b"water_ripple/motor/set" MOTOR_STATE_TOPIC = b"water_ripple/motor/state" TELEMETRY_TOPIC = b"water_ripple/tele/STATE" DISCOVERY_PREFIX = "homeassistant" LIGHT_DISCOVERY_TOPIC = f"{DISCOVERY_PREFIX}/light/{CLIENT_ID}/rgb/config".encode() MOTOR_DISCOVERY_TOPIC = f"{DISCOVERY_PREFIX}/fan/{CLIENT_ID}/motor/config".encode() pwm_r = machine.PWM(machine.Pin(1), freq=1000, duty_u16=0) pwm_g = machine.PWM(machine.Pin(2), freq=1000, duty_u16=0) pwm_b = machine.PWM(machine.Pin(3), freq=1000, duty_u16=0) pwm_motor = machine.PWM(machine.Pin(10), freq=20000, duty_u16=0) light_state = {"state": "OFF", "brightness": 255, "color_mode": "rgb", "color": {"r": 255, "g": 255, "b": 255}} motor_state = {"state": "OFF", "percentage": 100} current_r, current_g, current_b = 0.0, 0.0, 0.0 target_r, target_g, target_b = 0.0, 0.0, 0.0 step_r, step_g, step_b = 0.0, 0.0, 0.0 transition_steps_left = 0 wifi_connect_count = 0 mqtt_connect_count = 0 boot_time_str = "Unknown" cause = machine.reset_cause() if cause == machine.PWRON_RESET: reset_reason_str = "Power On" elif cause == machine.HARD_RESET: reset_reason_str = "Hard Reset" elif cause == machine.WDT_RESET: reset_reason_str = "Watchdog Reset" elif cause == machine.DEEPSLEEP_RESET: reset_reason_str = "Deep Sleep Wake" elif cause == machine.SOFT_RESET: reset_reason_str = "Soft Reset" else: reset_reason_str = f"Unknown ({cause})" def map_8bit_to_u16(value): return int((value / 255.0) * 65535) def calculate_target_rgb(): if light_state["state"] == "ON": factor = light_state["brightness"] / 255.0 return ( light_state["color"].get("r", 0) * factor, light_state["color"].get("g", 0) * factor, light_state["color"].get("b", 0) * factor ) return 0.0, 0.0, 0.0 def start_light_transition(duration_sec): global target_r, target_g, target_b, step_r, step_g, step_b, transition_steps_left tr, tg, tb = calculate_target_rgb() if duration_sec <= 0: global current_r, current_g, current_b current_r, current_g, current_b = tr, tg, tb transition_steps_left = 0 pwm_r.duty_u16(map_8bit_to_u16(int(current_r))) pwm_g.duty_u16(map_8bit_to_u16(int(current_g))) pwm_b.duty_u16(map_8bit_to_u16(int(current_b))) return total_steps = int(duration_sec / 0.02) if total_steps <= 0: total_steps = 1 target_r, target_g, target_b = tr, tg, tb step_r = (target_r - current_r) / total_steps step_g = (target_g - current_g) / total_steps step_b = (target_b - current_b) / total_steps transition_steps_left = total_steps def tick_light_transition(): global current_r, current_g, current_b, transition_steps_left if transition_steps_left > 0: current_r += step_r current_g += step_g current_b += step_b transition_steps_left -= 1 if transition_steps_left == 0: current_r, current_g, current_b = target_r, target_g, target_b pwm_r.duty_u16(map_8bit_to_u16(max(0, min(255, int(current_r))))) pwm_g.duty_u16(map_8bit_to_u16(max(0, min(255, int(current_g))))) pwm_b.duty_u16(map_8bit_to_u16(max(0, min(255, int(current_b))))) def update_motor_pwm(): if motor_state["state"] == "ON": ha_percent = motor_state["percentage"] actual_percent = 65.0 + (ha_percent / 100.0) * 35.0 duty = int((actual_percent / 100.0) * 65535) pwm_motor.duty_u16(duty) else: pwm_motor.duty_u16(0) def publish_discovery(client): dev_info = {"identifiers": [CLIENT_ID], "name": "水波纹灯", "model": "DIY Water Ripple", "manufacturer": "Custom"} light_config = { "name": "水波纹灯光", "unique_id": f"{CLIENT_ID}_rgb", "schema": "json", "command_topic": LIGHT_CMD_TOPIC.decode(), "state_topic": LIGHT_STATE_TOPIC.decode(), "brightness": True, "color_mode": True, "supported_color_modes": ["rgb"], "transition": True, "device": dev_info } motor_config = { "name": "水波纹电机", "unique_id": f"{CLIENT_ID}_motor", "command_topic": MOTOR_CMD_TOPIC.decode(), "command_template": "{\"state\": \"{{ value }}\"}", "state_topic": MOTOR_STATE_TOPIC.decode(), "state_value_template": "{{ value_json.state }}", "percentage_command_topic": MOTOR_CMD_TOPIC.decode(), "percentage_command_template": "{\"percentage\": {{ value }}}", "percentage_state_topic": MOTOR_STATE_TOPIC.decode(), "percentage_value_template": "{{ value_json.percentage }}", "speed_range_min": 1, "speed_range_max": 100, "device": dev_info } client.publish(LIGHT_DISCOVERY_TOPIC, json.dumps(light_config).encode(), retain=True) client.publish(MOTOR_DISCOVERY_TOPIC, json.dumps(motor_config).encode(), retain=True) tele_keys = [ ("IP", "ip", "mdi:ip-network", None), ("Last Restart Time", "last_restart", "mdi:clock", None), ("Restart Reason", "restart_reason", "mdi:information-outline", None), ("SSID", "ssid", "mdi:wifi-cog", None), ("WiFi Connect Count", "wifi_count", "mdi:counter", None), ("MQTT Connect Count", "mqtt_count", "mdi:counter", None), ("RSSI", "rssi", "mdi:wifi", "dBm"), ("Signal", "signal", "mdi:wifi", "%") ] for jk, k, ic, u in tele_keys: cfg = { "name": jk, "unique_id": f"{CLIENT_ID}_{k}", "state_topic": TELEMETRY_TOPIC.decode(), "value_template": "{{ value_json[\"" + jk + "\"] }}", "icon": ic, "entity_category": "diagnostic", "device": dev_info } if u: cfg["unit_of_measurement"] = u client.publish(f"{DISCOVERY_PREFIX}/sensor/{CLIENT_ID}/{k}/config".encode(), json.dumps(cfg).encode(), retain=True) def mqtt_callback(topic, msg): global light_state, motor_state try: payload = json.loads(msg) if topic == LIGHT_CMD_TOPIC: light_state_changed = False if "state" in payload: if light_state["state"] != payload["state"]: light_state["state"] = payload["state"] light_state_changed = True if "brightness" in payload: light_state["brightness"] = payload["brightness"] if "color" in payload: for k in ["r", "g", "b"]: if k in payload["color"]: light_state["color"][k] = payload["color"][k] duration = payload.get("transition", 0.5) start_light_transition(duration) client.publish(LIGHT_STATE_TOPIC, json.dumps(light_state)) if light_state_changed: motor_state["state"] = light_state["state"] update_motor_pwm() client.publish(MOTOR_STATE_TOPIC, json.dumps(motor_state)) elif topic == MOTOR_CMD_TOPIC: if "state" in payload: motor_state["state"] = payload["state"] if "percentage" in payload: motor_state["percentage"] = payload["percentage"] update_motor_pwm() client.publish(MOTOR_STATE_TOPIC, json.dumps(motor_state)) except Exception as e: pass def publish_telemetry(client): wlan = network.WLAN(network.STA_IF) try: rssi = wlan.status('rssi') except: rssi = -100 signal = max(0, min(100, 2 * (rssi + 100))) payload = { "IP": wlan.ifconfig()[0], "Last Restart Time": boot_time_str, "MQTT Connect Count": mqtt_connect_count, "Restart Reason": reset_reason_str, "RSSI": rssi, "Signal": signal, "SSID": WIFI_SSID, "WiFi Connect Count": wifi_connect_count } client.publish(TELEMETRY_TOPIC, json.dumps(payload).encode(), retain=True) def connect_wifi(): global wifi_connect_count, boot_time_str wlan = network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): wlan.connect(WIFI_SSID, WIFI_PASSWORD) while not wlan.isconnected(): time.sleep(0.5) wifi_connect_count += 1 if boot_time_str == "Unknown": try: import ntptime ntptime.host = 'ntp.aliyun.com' ntptime.settime() tm = time.localtime(time.time() + 8 * 3600) boot_time_str = "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(tm[0], tm[1], tm[2], tm[3], tm[4], tm[5]) except Exception as e: pass try: with open('webrepl_cfg.py', 'w') as f: f.write(f"PASS = '{WIFI_PASSWORD}'\n") webrepl.start() except Exception as e: pass connect_wifi() client = MQTTClient(CLIENT_ID, MQTT_BROKER, user=MQTT_USER, password=MQTT_PASSWORD, keepalive=60) client.set_callback(mqtt_callback) client.connect() mqtt_connect_count += 1 publish_discovery(client) client.subscribe(LIGHT_CMD_TOPIC) client.subscribe(MOTOR_CMD_TOPIC) client.publish(LIGHT_STATE_TOPIC, json.dumps(light_state)) client.publish(MOTOR_STATE_TOPIC, json.dumps(motor_state)) publish_telemetry(client) curr_r, curr_g, curr_b = calculate_target_rgb() pwm_r.duty_u16(map_8bit_to_u16(int(curr_r))) pwm_g.duty_u16(map_8bit_to_u16(int(curr_g))) pwm_b.duty_u16(map_8bit_to_u16(int(curr_b))) update_motor_pwm() last_ping = time.ticks_ms() last_telemetry = time.ticks_ms() try: while True: client.check_msg() tick_light_transition() current_time = time.ticks_ms() if time.ticks_diff(current_time, last_ping) > 30000: client.ping() last_ping = current_time if time.ticks_diff(current_time, last_telemetry) > 300000: publish_telemetry(client) last_telemetry = current_time time.sleep(0.02) except OSError as e: time.sleep(3) machine.reset()