257 lines
9.1 KiB
Python
257 lines
9.1 KiB
Python
import time
|
|
|
|
|
|
class CookingState:
|
|
TEMPERATURE_TOLERANCE = 1.0
|
|
MIN_SIGNIFICANT_HEATING_RATE = 0.05 # °C/s threshold to consider valid heating
|
|
|
|
def __init__(self, cook_time: int, power_level: int, target_temp: float, temperature_provider=None, on_state_change=None, on_refresh=None):
|
|
self.cook_time = cook_time
|
|
self.power_level = power_level
|
|
self.target_temp = target_temp
|
|
|
|
self.start_time = time.time()
|
|
self.temperature_provider = temperature_provider
|
|
self.on_state_change = on_state_change
|
|
self.on_refresh = on_refresh
|
|
self.on_pause = None
|
|
|
|
self.state = CookingStates.COOKING
|
|
self.paused = False
|
|
self._pause_started_at = None
|
|
self._paused_duration = 0.0
|
|
|
|
self.current_dish_temp = None
|
|
self.current_ambient_temp = None
|
|
self.estimated_remaining_time = float(cook_time)
|
|
self._last_temperature_sample = None
|
|
self._last_refresh_signature = None
|
|
self._stirred = False
|
|
|
|
# Moving average filter for heating rate (°C / sec)
|
|
self._smoothed_heating_rate = 0.0
|
|
|
|
def set_temperature_provider(self, temperature_provider):
|
|
self.temperature_provider = temperature_provider
|
|
|
|
def set_state_change_callback(self, callback):
|
|
self.on_state_change = callback
|
|
|
|
def set_refresh_callback(self, callback):
|
|
self.on_refresh = callback
|
|
|
|
def set_pause_callback(self, callback):
|
|
self.on_pause = callback
|
|
|
|
def pause(self):
|
|
if self.paused:
|
|
return
|
|
|
|
self.paused = True
|
|
self._pause_started_at = time.time()
|
|
if self.on_pause:
|
|
self.on_pause(self)
|
|
|
|
def unpause(self):
|
|
if not self.paused:
|
|
return
|
|
|
|
now = time.time()
|
|
if self._pause_started_at is not None:
|
|
self._paused_duration += now - self._pause_started_at
|
|
|
|
self.paused = False
|
|
|
|
def toggle_pause(self):
|
|
if self.paused:
|
|
self.unpause()
|
|
else:
|
|
self.pause()
|
|
if self.on_pause:
|
|
self.on_pause(self)
|
|
|
|
def set_state(self, state):
|
|
if self.state == state:
|
|
return
|
|
|
|
self.state = state
|
|
self._notify_state_change()
|
|
self._notify_refresh(force=True)
|
|
|
|
def get_elapsed_time(self) -> float:
|
|
now = time.time()
|
|
elapsed = now - self.start_time - self._paused_duration
|
|
|
|
if self.paused and self._pause_started_at is not None:
|
|
elapsed -= now - self._pause_started_at
|
|
|
|
return max(0.0, elapsed)
|
|
|
|
def get_remaining_time(self) -> int:
|
|
"""Returns the estimated remaining cooking time in seconds."""
|
|
return int(max(0.0, self.get_remaining_time_estimation()))
|
|
|
|
def get_remaining_time_estimation(self) -> float:
|
|
elapsed_time = self.get_elapsed_time()
|
|
|
|
# 1. Base timer remaining based on standard cook time
|
|
timer_remaining = max(0.0, float(self.cook_time) - elapsed_time)
|
|
|
|
# If temperature is unavailable or already met target, rely on standard timer
|
|
if self.current_dish_temp is None or self.current_dish_temp >= (self.target_temp - self.TEMPERATURE_TOLERANCE):
|
|
return timer_remaining
|
|
|
|
# 2. Prevent hitting 00:00 before stirring trigger:
|
|
# If we passed half cook_time and temp is far from target, extend expected base time to 1.25x cook_time
|
|
temp_progress = max(0.0, self.current_dish_temp) / max(1.0, self.target_temp)
|
|
if elapsed_time > (self.cook_time * 0.5) and temp_progress < 0.8:
|
|
adjusted_cook_time = self.cook_time * 1.25
|
|
timer_remaining = max(0.0, adjusted_cook_time - elapsed_time)
|
|
|
|
# 3. Estimate using heating rate
|
|
heating_rate = self._estimate_heating_rate()
|
|
temp_needed = self.target_temp - self.current_dish_temp
|
|
|
|
if heating_rate > 0.01:
|
|
rate_based_remaining = temp_needed / heating_rate
|
|
else:
|
|
# If flat/slow, project remaining time based on remaining missing temperature fraction
|
|
temp_ratio = max(0.1, temp_needed / self.target_temp)
|
|
rate_based_remaining = max(timer_remaining, self.cook_time * temp_ratio * 1.25)
|
|
|
|
# 4. Strict dynamic cap: Never exceed maximum possible execution window (2.0x cook_time total)
|
|
max_possible_remaining = max(0.0, (self.cook_time * 2.0) - elapsed_time)
|
|
bounded_remaining = min(rate_based_remaining, max_possible_remaining)
|
|
|
|
# Return the larger of the adjusted timer or the bounded prediction
|
|
return max(timer_remaining, bounded_remaining)
|
|
|
|
def _estimate_heating_rate(self) -> float:
|
|
"""Calculates heating rate in °C/sec over time interval."""
|
|
if self._last_temperature_sample is None or self.current_dish_temp is None:
|
|
return 0.0
|
|
|
|
last_time, last_temp = self._last_temperature_sample
|
|
now = time.time()
|
|
delta_time = now - last_time
|
|
|
|
if delta_time < 0.8: # Skip micro-ticks
|
|
return 0.0
|
|
|
|
delta_temp = self.current_dish_temp - last_temp
|
|
if delta_temp <= 0:
|
|
return 0.0
|
|
|
|
return delta_temp / delta_time
|
|
|
|
def _read_temperatures(self):
|
|
if self.temperature_provider is None:
|
|
return None, None
|
|
|
|
temperatures = self.temperature_provider()
|
|
if temperatures is None:
|
|
return None, None
|
|
|
|
if isinstance(temperatures, (list, tuple)) and len(temperatures) >= 2:
|
|
return temperatures[0], temperatures[1]
|
|
|
|
raise ValueError("temperature_provider must return a pair: (dish_temp, ambient_temp)")
|
|
|
|
def _update_heating_rate(self):
|
|
"""Calculates instantaneous rate and updates the Exponential Moving Average."""
|
|
now = time.time()
|
|
if self._last_temperature_sample is None:
|
|
self._last_temperature_sample = (now, self.current_dish_temp)
|
|
return
|
|
|
|
last_time, last_temp = self._last_temperature_sample
|
|
delta_time = now - last_time
|
|
|
|
if delta_time <= 0.5 or self.current_dish_temp is None:
|
|
return
|
|
|
|
instant_rate = (self.current_dish_temp - last_temp) / delta_time
|
|
self._last_temperature_sample = (now, self.current_dish_temp)
|
|
|
|
# Exponential Moving Average (EMA) - alpha smooths out sensor noise
|
|
alpha = 0.2
|
|
self._smoothed_heating_rate = (alpha * instant_rate) + ((1.0 - alpha) * self._smoothed_heating_rate)
|
|
|
|
def _notify_state_change(self):
|
|
if self.on_state_change is None:
|
|
return
|
|
self.on_state_change(self)
|
|
|
|
def _notify_refresh(self, force=False):
|
|
if self.on_refresh is None:
|
|
return
|
|
|
|
signature = (
|
|
int(self.get_elapsed_time()),
|
|
int(self.get_remaining_time_estimation()),
|
|
self.current_dish_temp,
|
|
self.current_ambient_temp,
|
|
self.state,
|
|
self.paused,
|
|
)
|
|
|
|
if not force and signature == self._last_refresh_signature:
|
|
return
|
|
|
|
self._last_refresh_signature = signature
|
|
self.on_refresh(self)
|
|
|
|
def update_tick(self):
|
|
if self.state == CookingStates.IDLE:
|
|
return self.state
|
|
if self.paused:
|
|
self._notify_refresh()
|
|
return self.state
|
|
|
|
previous_state = self.state
|
|
previous_temperature = self.current_dish_temp
|
|
|
|
try:
|
|
self.current_dish_temp, self.current_ambient_temp = self._read_temperatures()
|
|
except Exception:
|
|
self.current_dish_temp = previous_temperature
|
|
|
|
now = time.time()
|
|
elapsed_time = self.get_elapsed_time()
|
|
|
|
if self.current_dish_temp is not None:
|
|
self._update_heating_rate()
|
|
|
|
if elapsed_time < (self.cook_time / 2.0) and self.current_dish_temp >= self.target_temp and (self._paused_duration is None or self._paused_duration < 5):
|
|
self.state = CookingStates.STIRRING_REQUIRED
|
|
self.pause()
|
|
elif elapsed_time >= self.cook_time and self.current_dish_temp >= (self.target_temp - self.TEMPERATURE_TOLERANCE):
|
|
self.state = CookingStates.DONE
|
|
elif elapsed_time >= self.cook_time * 1.25 and (self._paused_duration is None or self._paused_duration < 5):
|
|
self.state = CookingStates.STIRRING_REQUIRED
|
|
self.pause()
|
|
elif self._pause_started_at is not None and (self._pause_started_at + self._paused_duration) < (now - (self.cook_time * 0.75)):
|
|
self.state = CookingStates.DONE
|
|
|
|
self.estimated_remaining_time = self.get_remaining_time_estimation()
|
|
|
|
if self.state != previous_state:
|
|
self._notify_state_change()
|
|
|
|
self._notify_refresh()
|
|
return self.state
|
|
|
|
|
|
class CookingStates:
|
|
COOKING = 0
|
|
STIRRING_REQUIRED = 1
|
|
DONE = 2
|
|
ALERT = 3 # Microwave is too hot internally or other alerts
|
|
IDLE = 4 # Waiting for cooking parameters to be set, or after cooking is done
|
|
|
|
@staticmethod
|
|
def get_state_name(state_val):
|
|
for key, value in CookingStates.__dict__.items():
|
|
if value == state_val and not key.startswith('__'):
|
|
return key
|
|
return "UNKNOWN" |