from machine import Pin, I2C
from time import sleep, ticks_ms, ticks_diff
import onewire
import ds18x20
import network
import socket
import json
import _thread
import ssd1306
# =========================================================
# CONFIGURACIÓN
# =========================================================
CONFIG_PAGE = """\
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Lelit Anna</title>
</head>
<body>
<h1>Lelit Anna</h1>
<h2>Configuracion</h2>
<form method="POST" action="/config">
<label>
Tiempo de extraccion (segundos):
</label>
<br>
<input type="number"
name="extraction_time"
value="{extraction_time}">
<br><br>
<label>
Tiempo mostrando resultado (segundos):
</label>
<br>
<input type="number"
name="result_display_time"
value="{result_display_time}">
<br><br>
<label>
Actualizacion LCD (ms):
</label>
<br>
<input type="number"
name="lcd_update_interval"
value="{lcd_update_interval}">
<br><br>
<input type="submit" value="Guardar">
</form>
</body>
</html>
"""
STOP_PIN = 25
BUTTON_PIN = 27
RELAY_PIN = 26
CONFIG_FILE = "config.json"
DEFAULT_CONFIG = {
"extraction_time": 30,
"result_display_time": 5,
"lcd_update_interval": 500
}
# IDs de los sensores
BOILER_ID = bytes.fromhex("281111111111117e")
GROUP_ID = bytes.fromhex("28222222222222de")
# DEFINICIONES
def load_config():
try:
with open(CONFIG_FILE, "r") as file:
config = json.load(file)
print("Configuracion cargada:", config)
return config
except:
print("No existe ninguna configuracion. Usando valores por defecto.")
save_config(DEFAULT_CONFIG)
return DEFAULT_CONFIG.copy()
def save_config(config):
with open(CONFIG_FILE, "w") as file:
json.dump(config, file)
print("Configuracion guardada:", config)
def url_decode(value):
value = value.replace("+", " ")
result = ""
i = 0
while i < len(value):
if value[i] == "%" and i + 2 < len(value):
try:
result += chr(
int(value[i + 1:i + 3], 16)
)
i += 3
except:
result += value[i]
i += 1
else:
result += value[i]
i += 1
return result
def parse_form(body):
data = {}
parts = body.split("&")
for part in parts:
if "=" not in part:
continue
key, value = part.split("=", 1)
data[url_decode(key)] = url_decode(value)
return data
def send_response(client, content, status="200 OK"):
response = (
"HTTP/1.1 {}\r\n"
"Content-Type: text/html; charset=UTF-8\r\n"
"Connection: close\r\n"
"\r\n"
"{}"
).format(status, content)
client.send(response)
def handle_web_request(client):
request = client.recv(2048)
if not request:
client.close()
return
request_text = request.decode()
print("HTTP:", request_text.split("\r\n")[0])
# -----------------------------------------------------
# GET /
# -----------------------------------------------------
if request_text.startswith("GET /"):
page = CONFIG_PAGE.format(
extraction_time=config["extraction_time"],
result_display_time=config["result_display_time"],
lcd_update_interval=config["lcd_update_interval"]
)
send_response(client, page)
# -----------------------------------------------------
# POST /config
# -----------------------------------------------------
elif request_text.startswith("POST /config"):
body = request_text.split("\r\n\r\n", 1)[1]
data = parse_form(body)
try:
new_config = {
"extraction_time":
int(data["extraction_time"]),
"result_display_time":
int(data["result_display_time"]),
"lcd_update_interval":
int(data["lcd_update_interval"])
}
# Validaciones básicas
if new_config["extraction_time"] <= 0:
raise ValueError
if new_config["result_display_time"] < 0:
raise ValueError
if new_config["lcd_update_interval"] < 50:
raise ValueError
# Guardar
save_config(new_config)
# Actualizar configuración RAM
config.update(new_config)
# Actualizar variables utilizadas por el programa
global EXTRACTION_TIME
global RESULT_DISPLAY_TIME
global LCD_UPDATE_INTERVAL
EXTRACTION_TIME = config["extraction_time"] * 1000
RESULT_DISPLAY_TIME = config["result_display_time"] * 1000
LCD_UPDATE_INTERVAL = config["lcd_update_interval"]
print("Configuracion actualizada")
except:
print("Configuracion invalida")
# Volver a la página
page = CONFIG_PAGE.format(
extraction_time=config["extraction_time"],
result_display_time=config["result_display_time"],
lcd_update_interval=config["lcd_update_interval"]
)
send_response(client, page)
client.close()
def connect_wifi():
global wifi_connected
global wifi_ip
global wifi_status
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if wlan.isconnected():
wifi_connected = True
wifi_ip = wlan.ifconfig()[0]
wifi_status = "Conectado"
print("WiFi ya conectado")
print("IP:", wifi_ip)
return wlan
print("Conectando a WiFi...")
wifi_status = "Conectando"
wlan.connect("Wokwi-GUEST", "")
start = ticks_ms()
while not wlan.isconnected():
if ticks_diff(ticks_ms(), start) > 10000:
wifi_connected = False
wifi_ip = ""
wifi_status = "Error"
print("Timeout WiFi")
return wlan
sleep(0.1)
wifi_connected = True
wifi_ip = wlan.ifconfig()[0]
wifi_status = "Conectado"
print("WiFi conectado")
print("IP:", wifi_ip)
return wlan
def start_web_server():
server = socket.socket(
socket.AF_INET,
socket.SOCK_STREAM
)
server.setsockopt(
socket.SOL_SOCKET,
socket.SO_REUSEADDR,
1
)
server.bind(("0.0.0.0", 80))
server.listen(1)
server.settimeout(0.1)
print("Servidor web iniciado")
return server
def network_thread():
wlan = connect_wifi()
if not wlan.isconnected():
return
server = start_web_server()
while True:
try:
client, address = server.accept()
print("Conexion web:", address)
handle_web_request(client)
except OSError as error:
# ETIMEDOUT significa simplemente
# que no ha llegado ninguna conexión todavía.
if error.args[0] != 116:
print("Error servidor web:", error)
def lcd_show_wifi():
if wifi_connected:
oled_text("IP: {}".format(wifi_ip), 0, 24)
oled_show()
else:
oled_text("WIFI: {}".format(wifi_status), 0, 24)
oled_show()
# =========================================================
# LCD
# =========================================================
OLED_WIDTH = 128
OLED_HEIGHT = 64
OLED_ADDR = 0x3C
i2c = I2C(
0,
scl=Pin(22),
sda=Pin(21),
freq=400000
)
oled = ssd1306.SSD1306_I2C(
OLED_WIDTH,
OLED_HEIGHT,
i2c,
addr=OLED_ADDR
)
def oled_clear():
oled.fill(0)
oled.show()
def oled_text(text, x, y):
oled.text(
str(text),
x,
y
)
# =====================================================
# FUENTE PEQUEÑA 5x7
# =====================================================
SMALL_FONT = {
" ": [
0,0,0,0,0,0,0
],
"0": [
0x0E,0x11,0x13,0x15,0x19,0x11,0x0E
],
"1": [
0x04,0x0C,0x04,0x04,0x04,0x04,0x0E
],
"2": [
0x0E,0x11,0x01,0x02,0x04,0x08,0x1F
],
"3": [
0x1E,0x01,0x01,0x0E,0x01,0x01,0x1E
],
"4": [
0x02,0x06,0x0A,0x12,0x1F,0x02,0x02
],
"5": [
0x1F,0x10,0x10,0x1E,0x01,0x01,0x1E
],
"6": [
0x06,0x08,0x10,0x1E,0x11,0x11,0x0E
],
"7": [
0x1F,0x01,0x02,0x04,0x08,0x08,0x08
],
"8": [
0x0E,0x11,0x11,0x0E,0x11,0x11,0x0E
],
"9": [
0x0E,0x11,0x11,0x0F,0x01,0x02,0x0C
],
"A": [
0x0E,0x11,0x11,0x1F,0x11,0x11,0x11
],
"B": [
0x1E,0x11,0x11,0x1E,0x11,0x11,0x1E
],
"C": [
0x0E,0x11,0x10,0x10,0x10,0x11,0x0E
],
"D": [
0x1E,0x11,0x11,0x11,0x11,0x11,0x1E
],
"E": [
0x1F,0x10,0x10,0x1E,0x10,0x10,0x1F
],
"F": [
0x1F,0x10,0x10,0x1E,0x10,0x10,0x10
],
"G": [
0x0E,0x11,0x10,0x17,0x11,0x11,0x0F
],
"I": [
0x0E,0x04,0x04,0x04,0x04,0x04,0x0E
],
"K": [
0x11,0x12,0x14,0x18,0x14,0x12,0x11
],
"O": [
0x0E,0x11,0x11,0x11,0x11,0x11,0x0E
],
"W": [
0x11,0x11,0x11,0x15,0x15,0x15,0x0A
],
":": [
0,0x04,0x04,0,0x04,0x04,0
],
".": [
0,0,0,0,0,0x06,0x06
],
"-": [
0,0,0,0x1F,0,0,0
],
"/": [
0x01,0x02,0x02,0x04,0x08,0x08,0x10
],
"°": [
0x06,0x09,0x06,0,0,0,0
]
}
def oled_small_text(text, x, y):
text = str(text).upper()
for char in text:
if char not in SMALL_FONT:
char = " "
bitmap = SMALL_FONT[char]
for row in range(7):
if row >= len(bitmap):
continue
bits = bitmap[row]
for col in range(5):
if bits & (1 << (4 - col)):
oled.pixel(
x + col,
y + row,
1
)
x += 6
def oled_show():
oled.show()
def oled_progress(progress):
if progress < 0:
progress = 0
if progress > 1:
progress = 1
x = 0
y = 52
width = 128
height = 10
oled.fill_rect(
x,
y,
width,
height,
0
)
oled.rect(
x,
y,
width,
height,
1
)
fill_width = int(
(width - 2) * progress
)
if fill_width > 0:
oled.fill_rect(
x + 1,
y + 1,
fill_width,
height - 2,
1
)
def oled_draw():
oled.fill(0)
# =====================================================
# CABECERA - WIFI
# =====================================================
if wifi_connected:
oled_small_text(
"WiFi: OK",
0,
0
)
oled_small_text(
wifi_ip,
72,
0
)
else:
oled_small_text(
"WiFi: {}".format(wifi_status),
0,
0
)
# =====================================================
# TEMPERATURAS
# =====================================================
oled_small_text(
"B:{:.1f}C".format(boiler_temp),
0,
12
)
oled_small_text(
"G:{:.1f}C".format(group_temp),
72,
12
)
# =====================================================
# SEPARADOR
# =====================================================
oled.line(
0,
23,
127,
23,
1
)
# =====================================================
# ESTADO
# =====================================================
if state == "idle":
oled.text(
"LISTA",
48,
32
)
elif state == "extraction":
oled.text(
"EXTRAYENDO",
32,
32
)
elapsed = ticks_diff(
ticks_ms(),
extraction_start
)
progress = elapsed / EXTRACTION_TIME
oled_progress(progress)
elif state == "result":
oled.text(
"COMPLETADO",
28,
32
)
final_seconds = ticks_diff(
extraction_end,
extraction_start
) / 1000
oled.text(
"{:.1f} s".format(final_seconds),
48,
44
)
# =====================================================
# ACTUALIZAR OLED
# =====================================================
oled.show()
def oled_header():
oled.text(
"WiFi: OK",
0,
0
)
oled.text(
wifi_ip,
72,
0
)
oled.text(
"B:{:.1f}C".format(boiler_temp),
0,
12
)
oled.text(
"G:{:.1f}C".format(group_temp),
72,
12
)
def oled_status(text):
oled.text(
text,
0,
28
)
# INICIALIZAR OLED
oled_text("Inicializando...", 0, 0)
oled_text("Componentes", 0, 12)
oled_show()
sleep(1)
# BOTÓN + RELÉ
button = Pin(
BUTTON_PIN,
Pin.IN,
Pin.PULL_UP
)
stop = Pin(
STOP_PIN,
Pin.IN,
Pin.PULL_UP
)
relay = Pin(
RELAY_PIN,
Pin.OUT
)
relay.value(0)
oled_clear()
oled_text("Inicializando...", 0, 0)
oled_text("Sensores", 0, 12)
oled_show()
sleep(1)
# SENSORES
sensor_pin = Pin(4)
ow = onewire.OneWire(sensor_pin)
ds = ds18x20.DS18X20(ow)
sensors = ds.scan()
for sensor in sensors:
print ("Nuevo sensor: {}".format(bytearray(sensor)))
oled_clear()
oled_text("Inicializando...", 0, 0)
oled_text("Variables", 0, 12)
oled_show()
sleep(1)
# VARIABLES DEL SISTEMA
state = "idle"
extraction_start = 0
extraction_end = 0
result_start = 0
last_button = 1
stop_button_last = 1
boiler_temp = 0
group_temp = 0
wifi_connected = False
wifi_ip = ""
wifi_status = "Desconectado"
wweb_server_running = False
oled_clear()
oled_text("Inicializando...", 0, 0)
oled_text("Configuracion", 0, 12)
oled_show()
sleep(1)
config = load_config()
EXTRACTION_TIME = config["extraction_time"] * 1000 # 30 segundos
RESULT_DISPLAY_TIME = config["result_display_time"] * 1000 # mostrar resultado 5 segundos
LCD_UPDATE_INTERVAL = config["lcd_update_interval"]
if config == DEFAULT_CONFIG:
oled_clear()
oled_text("Inicializando...", 0, 0)
oled_text("Cargando predeterminada...", 0, 12)
oled_show()
sleep(1)
oled_clear()
oled_text("Inicializando...", 0, 0)
oled_text("Red", 0, 12)
oled_show()
_thread.start_new_thread(network_thread, ())
sleep(3)
oled_draw()
# =====================================================
# LEER TEMPERATURAS
# =====================================================
ds.convert_temp()
temperature_conversion_start = ticks_ms()
display_timer = ticks_ms()
# =========================================================
# BUCLE PRINCIPAL
# =========================================================
while True:
now = ticks_ms()
# =====================================================
# DETECTAR PULSACIÓN
# =====================================================
current_stop = stop.value()
stop_pressed = (
stop_button_last == 1 and
current_stop == 0
)
stop_button_last = current_stop
current_button = button.value()
button_pressed = (
last_button == 1 and
current_button == 0
)
last_button = current_button
if stop_pressed and state == "extraction":
print("EXTRACCION DETENIDA POR STOP")
relay.value(0)
extraction_end = now
result_start = now
state = "result"
# =====================================================
# ESTADO: ESPERA
# =====================================================
if state == "idle":
# Mostrar temperaturas normales
if ticks_diff(now, display_timer) >= LCD_UPDATE_INTERVAL:
oled_draw()
display_timer = now
# ¿Se ha pulsado el botón?
if button_pressed:
print("EXTRACCION INICIADA")
state = "extraction"
extraction_start = now
relay.value(1)
# =====================================================
# ESTADO: EXTRACCIÓN
# =====================================================
elif state == "extraction":
elapsed = ticks_diff(
now,
extraction_start
)
# Mostrar tiempo y temperatura
seconds = elapsed // 1000
if ticks_diff(now, display_timer) >= LCD_UPDATE_INTERVAL:
oled_draw()
display_timer = now
# Mostrar también en Serial
print(
"Extraccion:",
seconds,
"s | Group:",
group_temp
)
# ¿Ha terminado?
if elapsed >= EXTRACTION_TIME:
relay.value(0)
extraction_end = now
result_start = now
state = "result"
print("EXTRACCION TERMINADA")
# =====================================================
# ESTADO: RESULTADO
# =====================================================
elif state == "result":
elapsed_result = ticks_diff(
now,
result_start
)
# Tiempo final de extracción
final_seconds = ticks_diff(
extraction_end,
extraction_start
) // 1000
if ticks_diff(now, display_timer) >= LCD_UPDATE_INTERVAL:
oled_draw()
display_timer = now
# Después de 5 segundos volvemos al estado normal
if elapsed_result >= RESULT_DISPLAY_TIME:
state = "idle"
print("Volviendo a pantalla normal")
if ticks_diff(now, temperature_conversion_start) >= 800:
try:
new_boiler_temp = ds.read_temp(BOILER_ID)
new_group_temp = ds.read_temp(GROUP_ID)
if new_boiler_temp is not None:
boiler_temp = new_boiler_temp
if new_group_temp is not None:
group_temp = new_group_temp
except Exception as error:
print("Error leyendo temperatura:", error)
ds.convert_temp()
temperature_conversion_start = nowLoading
esp32-devkit-c-v4
esp32-devkit-c-v4
Loading
ssd1306
ssd1306