Microwave Screen + defrost mode
This commit is contained in:
@@ -0,0 +1,135 @@
|
||||
import time
|
||||
from shared.cookingState import CookingState, CookingStates
|
||||
from shared import config
|
||||
import framebuf
|
||||
|
||||
# Icon and bitmaps
|
||||
image_weather_frost_bits = bytearray(b'\x01\x00\x13\x901\x18s\x9c\x09 \x05@S\x94\xfe\xfeS\x94\x05@\x09 s\x9c1\x18\x13\x90\x01\x00\x00\x00')
|
||||
image_Alert_bits = bytearray(b'\x08\x00\x1c\x00\x14\x006\x006\x00\x7f\x00w\x00\xff\x80')
|
||||
image_CoolHiSmall_bits = bytearray(b'\x1f\xff\xc0?\xff\xe0\x7f\xff\xf0\xff\xdf\xf8\xffW\xf8\xfd\x8d\xf8\xf9\xdc\xf8\xfe\xdb\xf8\xf7Wx\xfb\x8e\xf8\xe0\x008\xfb\x8e\xf8\xf7Wx\xfe\xdb\xf8\xf9\xdc\xf8\xfd\x8d\xf8\xffW\xf8\xff\xdf\xf8\x7f\xff\xf0?\xff\xe0\x1f\xff\xc0')
|
||||
image_CoolLoSmall_bits = bytearray(b'\x1f\xff\xc0 \x00 @\x00\x10\x80 \x08\x80\xa8\x08\x82r\x08\x86#\x08\x81$\x08\x88\xa8\x88\x84q\x08\x9f\xff\xc8\x84q\x08\x88\xa8\x88\x81$\x08\x86#\x08\x82r\x08\x80\xa8\x08\x80 \x08@\x00\x10 \x00 \x1f\xff\xc0')
|
||||
image_music_sound_wave_bits = bytearray(b'\x00\x00\x00\x02\x00\x00\x02 \x00\x02 \x00\x02 \x00\x0a(\x00\x0a\xa8\x00*\xa8\x00\xaa\xaa\x80\x0a\xaa\x00\x0a\xa8\x00\x0a \x00\x02 \x00\x02 \x00\x02\x00\x00\x00\x00\x00')
|
||||
image_weather_temperature_bits = bytearray(b'\x1c\x00\x22\x00+\x00*\x00+\x00*\x00+\x00*\x00*\x00I\x00\x9c\x80\xae\x80\xbe\x80\x9c\x80A\x00>\x00')
|
||||
fb_image_weather_frost_bits = framebuf.FrameBuffer(image_weather_frost_bits, 15, 16, framebuf.MONO_HLSB)
|
||||
fb_image_CoolHiSmall_bits = framebuf.FrameBuffer(image_CoolHiSmall_bits, 21, 21, framebuf.MONO_HLSB)
|
||||
fb_image_weather_temperature_bits = framebuf.FrameBuffer(image_weather_temperature_bits, 9, 16, framebuf.MONO_HLSB)
|
||||
fb_image_Alert_bits = framebuf.FrameBuffer(image_Alert_bits, 9, 8, framebuf.MONO_HLSB)
|
||||
fb_image_music_sound_wave_bits = framebuf.FrameBuffer(image_music_sound_wave_bits, 17, 16, framebuf.MONO_HLSB)
|
||||
fb_image_CoolLoSmall_bits = framebuf.FrameBuffer(image_CoolLoSmall_bits, 21, 21, framebuf.MONO_HLSB)
|
||||
|
||||
# Constants
|
||||
CHARACTER_WIDTH = 6
|
||||
CHARACTER_GAP = 2
|
||||
CHARACTER_FULL_WIDTH = CHARACTER_WIDTH + CHARACTER_GAP
|
||||
CHARACTER_HEIGHT = 7
|
||||
|
||||
|
||||
class MicrowaveScreen:
|
||||
# Used for animation
|
||||
lastUpdateTime = 0
|
||||
|
||||
def __init__(self, display):
|
||||
self.display = display
|
||||
self.width = display.width
|
||||
self.height = display.height
|
||||
|
||||
def h_centered_text(self, text, y, color=1):
|
||||
"""Centers text horizontally
|
||||
|
||||
Args:
|
||||
text (str): _description_
|
||||
y (int): _description_
|
||||
color (int, optional): _description_. Defaults to 1.
|
||||
"""
|
||||
text = str(text)
|
||||
text_width = len(text) * CHARACTER_FULL_WIDTH
|
||||
|
||||
if config.DEBUG:
|
||||
if text_width > self.width:
|
||||
print(f"[MicrowaveScreen] Warning: Text '{text}' is too long to fit on the screen.")
|
||||
|
||||
x = (self.width - text_width) // 2
|
||||
self.display.text(text, x, y, color)
|
||||
|
||||
def flick(self):
|
||||
self.display.fill(0)
|
||||
|
||||
def show(self):
|
||||
self.display.show()
|
||||
|
||||
def bootScreen(self):
|
||||
self.display.text("SmartWave", 28, 28, 1)
|
||||
self.show()
|
||||
|
||||
def _progressBar(self, progress, x, y, width=87, height=13, color=1):
|
||||
# Box outline
|
||||
self.display.rect(x, y, width, height, color)
|
||||
|
||||
# Fill the progress bar based on the progress value (0.0 to 1.0)
|
||||
fill_width = int(progress * (width)) # Subtract 2
|
||||
|
||||
self.display.fill_rect(x, y, fill_width, height, color)
|
||||
|
||||
def update(self, cooking_state: CookingState | None, defrost_mode: bool = False):
|
||||
self.flick()
|
||||
|
||||
self.lastUpdateTime = time.time()
|
||||
|
||||
if cooking_state: # Is currently cooking
|
||||
# self.h_centered_text(cooking_state.state, 28, 1)
|
||||
|
||||
# Screen center
|
||||
if cooking_state.state == CookingStates.STIRRING_REQUIRED:
|
||||
self._message("Pls Stir", True)
|
||||
elif cooking_state.state == CookingStates.ALERT:
|
||||
self._message("Alert !", True)
|
||||
elif cooking_state.state == CookingStates.DONE:
|
||||
self._message("Done !", False)
|
||||
else:
|
||||
# Progress bar
|
||||
elapsed_time = cooking_state.get_elapsed_time()
|
||||
estimated_progress = elapsed_time / (elapsed_time + cooking_state.estimated_remaining_time)
|
||||
self._progressBar(estimated_progress, 21, 25)
|
||||
|
||||
# Estimated remaining time
|
||||
if cooking_state.paused:
|
||||
self.h_centered_text("Paused", 40, 1)
|
||||
else:
|
||||
remaining_time = cooking_state.estimated_remaining_time
|
||||
minutes = remaining_time // 60
|
||||
seconds = remaining_time % 60
|
||||
self.h_centered_text(f"{minutes:02}:{seconds:02}", 40, 1)
|
||||
|
||||
# Current dish temperature
|
||||
if cooking_state.current_dish_temp is not None:
|
||||
self.display.blit(fb_image_weather_temperature_bits, 27, 2)
|
||||
self.display.text(str(int(cooking_state.current_dish_temp)), 37, 6, 1)
|
||||
self.display.ellipse(54, 6, 1, 1, 1, False)
|
||||
self.display.text("C", 56, 6, 1)
|
||||
|
||||
# Power level
|
||||
if cooking_state.power_level is not None:
|
||||
self.display.blit(fb_image_music_sound_wave_bits, 74, 2)
|
||||
self.display.text(f"{cooking_state.power_level}W", 92, 6, 1)
|
||||
else:
|
||||
self.h_centered_text("Ready !", 28, 1)
|
||||
|
||||
if defrost_mode:
|
||||
self.display.blit(fb_image_CoolHiSmall_bits, 2, 2)
|
||||
else:
|
||||
self.display.blit(fb_image_CoolLoSmall_bits, 2, 2)
|
||||
|
||||
self.show()
|
||||
pass
|
||||
|
||||
def _message(self, text, alert):
|
||||
self.display.fill_rect(0, 27, self.width, CHARACTER_HEIGHT + 4, 1)
|
||||
if alert and self.lastUpdateTime % 2000 < 1000:
|
||||
self.display.blit(fb_image_Alert_bits, 10, 27)
|
||||
self.h_centered_text(text, 28, 0)
|
||||
|
||||
def message(self, text, alert):
|
||||
self.flick()
|
||||
self._message(text, alert)
|
||||
self.show()
|
||||
pass
|
||||
@@ -2,9 +2,9 @@ import gc
|
||||
import sys
|
||||
import time
|
||||
import _thread
|
||||
from lib.microwaveScreen import MicrowaveScreen
|
||||
import uasyncio as asyncio
|
||||
from machine import Pin, SoftI2C
|
||||
import framebuf
|
||||
import ssd1306
|
||||
|
||||
# Clean memory immediately
|
||||
@@ -30,7 +30,9 @@ data_queue = SafeQueue()
|
||||
lora = None
|
||||
uart_device = None
|
||||
magnetron_led = None
|
||||
display = None
|
||||
microwave_screen = None
|
||||
defrost_mode = False
|
||||
|
||||
|
||||
PING_PAYLOAD = {
|
||||
"id": DEVICE_ID,
|
||||
@@ -39,7 +41,7 @@ PING_PAYLOAD = {
|
||||
|
||||
def init_hardware():
|
||||
"""Initializes all hardware components."""
|
||||
global lora, uart_device, magnetron_led, display
|
||||
global lora, uart_device, magnetron_led, microwave_screen
|
||||
|
||||
print("[Main] Initializing hardware peripherals...")
|
||||
|
||||
@@ -48,6 +50,14 @@ def init_hardware():
|
||||
vext.value(0)
|
||||
time.sleep_ms(100)
|
||||
|
||||
# Init OLED Display
|
||||
scl_pin = Pin(18, Pin.OUT, pull=Pin.PULL_UP)
|
||||
sda_pin = Pin(17, Pin.OUT, pull=Pin.PULL_UP)
|
||||
display_i2c = SoftI2C(scl=scl_pin, sda=sda_pin, freq=100000)
|
||||
display = ssd1306.SSD1306_I2C(128, 64, display_i2c, addr=0x3C)
|
||||
microwave_screen = MicrowaveScreen(display)
|
||||
microwave_screen.bootScreen()
|
||||
|
||||
# Init LoRa
|
||||
lora = get_lora()
|
||||
lora.configure(freq=868.1, sf=7)
|
||||
@@ -57,14 +67,6 @@ def init_hardware():
|
||||
magnetron_led.color = RGBLED.WHITE_YELLOW
|
||||
magnetron_led.off()
|
||||
|
||||
# Init OLED Display
|
||||
scl_pin = Pin(18, Pin.OUT, pull=Pin.PULL_UP)
|
||||
sda_pin = Pin(17, Pin.OUT, pull=Pin.PULL_UP)
|
||||
display_i2c = SoftI2C(scl=scl_pin, sda=sda_pin, freq=100000)
|
||||
display = ssd1306.SSD1306_I2C(128, 64, display_i2c, addr=0x3C)
|
||||
display.text("Booting...", 1, 2, 1)
|
||||
display.show()
|
||||
|
||||
# Init UART
|
||||
uart_device = get_uart(uart_id=1, tx_pin=46, rx_pin=45)
|
||||
|
||||
@@ -114,15 +116,10 @@ def cooking_state_on_state_change(state):
|
||||
if lora:
|
||||
lora.send_reliable({"id": DEVICE_ID, "new_cooking_state": state.state})
|
||||
|
||||
# Update OLED display
|
||||
if display:
|
||||
display.fill(0)
|
||||
display.text(cookingState.CookingStates.get_state_name(state.state), 1, 2, 1)
|
||||
display.show()
|
||||
|
||||
def cooking_state_on_refresh(state):
|
||||
# TODO: Show screen information
|
||||
pass
|
||||
# Update OLED display
|
||||
if microwave_screen:
|
||||
microwave_screen.update(state, defrost_mode)
|
||||
|
||||
def cooking_state_on_pause(state):
|
||||
# If the cooking is unpaused and was in STIRRING_REQUIRED or ALERT state, we set the state back to COOKING.
|
||||
@@ -166,7 +163,7 @@ async def uart_polling_task():
|
||||
|
||||
async def lora_process_task():
|
||||
"""Consumes packets pushed to data_queue by the LoRa hardware thread."""
|
||||
global cooking_state
|
||||
global cooking_state, defrost_mode, microwave_screen
|
||||
|
||||
while True:
|
||||
while not data_queue.empty():
|
||||
@@ -190,6 +187,10 @@ async def lora_process_task():
|
||||
print("[LoRa Process] Cooking resumed via orchestrator command.")
|
||||
else:
|
||||
log("[LoRa Process] No active cooking state to toggle pause/resume.")
|
||||
elif data["action"] == LoraCommands.TOGGLE_DEFROST:
|
||||
print("[LoRa Process] Toggling defrost mode via orchestrator command.")
|
||||
defrost_mode = data["defrost_state"]
|
||||
microwave_screen.update(cooking_state, defrost_mode)
|
||||
|
||||
await asyncio.sleep_ms(50)
|
||||
|
||||
@@ -215,6 +216,7 @@ async def memory_cleanup_task():
|
||||
|
||||
# --- BOOTSTRAP ---
|
||||
async def main():
|
||||
global cooking_state, defrost_mode, microwave_screen
|
||||
print("[Main] Starting application...")
|
||||
|
||||
init_hardware()
|
||||
@@ -235,6 +237,8 @@ async def main():
|
||||
|
||||
print("[Main] All async tasks running concurrently!")
|
||||
|
||||
microwave_screen.update(None, defrost_mode)
|
||||
|
||||
# Keep main task alive indefinitely
|
||||
while True:
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
@@ -127,12 +127,13 @@ async def mqtt_listener_task():
|
||||
def button_callback():
|
||||
"""Button physical interrupt callback."""
|
||||
global button_state
|
||||
if microwave_states.get("2") == MicrowaveState.COOKING:
|
||||
if microwave_states.get("2") == MicrowaveState.COOKING or microwave_states.get("2") == MicrowaveState.DONE:
|
||||
print("[Button] Toggling pause/resume for microwave '2'.")
|
||||
lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_PAUSE})
|
||||
else:
|
||||
button_state = not button_state
|
||||
print(f"[Button] Defrost state toggled to: {button_state}")
|
||||
lora.send_reliable({"id": DEVICE_ID, "microwave_id": "2", "action": LoraCommands.TOGGLE_DEFROST, "defrost_state": button_state})
|
||||
|
||||
button.set_callback(button_callback)
|
||||
button.start_button_monitoring_thread()
|
||||
@@ -199,13 +200,15 @@ async def handle_new_dish(microwave_id, detected_height):
|
||||
ir_data_events.pop(microwave_id, None)
|
||||
return
|
||||
|
||||
# 5. Wait for MQTT IR data (if it already arrived, event.wait() returns instantly)
|
||||
# 5. Wait for MQTT IR data
|
||||
try:
|
||||
await asyncio.wait_for(event.wait(), timeout=10.0)
|
||||
ir_payload = ir_data_cache.get(microwave_id, {})
|
||||
sensors_data["ir_initial_temp"] = ir_payload.get("dish_temp")
|
||||
sensors_data["ir_ambient_temp"] = ir_payload.get("ambient_temp")
|
||||
print(f"[{microwave_id}] IR data synchronized successfully: {ir_payload}")
|
||||
# 6. Dispatch cloud request task
|
||||
asyncio.create_task(request_cloud_cooking_plan(microwave_id, sensors_data))
|
||||
except asyncio.TimeoutError:
|
||||
print(f"[{microwave_id}] ⚠️ Timeout waiting for MQTT IR data from ESP32.")
|
||||
sensors_data["ir_initial_temp"] = None
|
||||
@@ -213,9 +216,6 @@ async def handle_new_dish(microwave_id, detected_height):
|
||||
finally:
|
||||
ir_data_events.pop(microwave_id, None)
|
||||
|
||||
# 6. Dispatch cloud request task
|
||||
asyncio.create_task(request_cloud_cooking_plan(microwave_id, sensors_data))
|
||||
|
||||
async def request_cloud_cooking_plan(microwave_id, sensors_data):
|
||||
"""Sends all data to the cloud with up to 3 retries (30s interval)."""
|
||||
global cloud_alert
|
||||
@@ -259,8 +259,8 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
|
||||
cloud_alert = False # Reset alert flag on success
|
||||
microwave_states[microwave_id] = MicrowaveState.COOKING
|
||||
if config.DEBUG:
|
||||
c_time = min(c_time, 10) # Limit to 10s for debug
|
||||
c_temp = min(c_temp, 50) # Limit to 50°C for debug
|
||||
c_time = 10 # Limit to 10s for debug
|
||||
c_temp = 50 # Limit to 50°C for debug
|
||||
mqtt_client.publish(
|
||||
config.MQTT_TOPIC_COOKING,
|
||||
payloads.mqtt_cooking_config(microwave_id, c_time, c_power, c_temp),
|
||||
|
||||
@@ -425,3 +425,4 @@ class LoraCommands:
|
||||
PING = "ping"
|
||||
COOKING_STATE_UPDATE = "cooking_state_update"
|
||||
TOGGLE_PAUSE = "toggle_pause"
|
||||
TOGGLE_DEFROST = "toggle_defrost"
|
||||
Reference in New Issue
Block a user