Beginning of cooking cycle

shared/cookingState.py

@@ -0,0 +1,199 @@
+import time
+
+
+class CookingState:
+    TEMPERATURE_TOLERANCE = 1.0
+
+    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.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
+
+    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 pause(self):
+        if self.paused:
+            return
+
+        self.paused = True
+        self._pause_started_at = time.time()
+        self._notify_refresh(force=True)
+
+    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._pause_started_at = None
+        self.paused = False
+        self._notify_refresh(force=True)
+
+    def toggle_pause(self):
+        if self.paused:
+            self.unpause()
+        else:
+            self.pause()
+
+    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()
+        timer_remaining = max(0.0, float(self.cook_time) - elapsed_time)
+
+        if self.current_dish_temp is None:
+            return timer_remaining
+
+        if self.current_dish_temp >= self.target_temp:
+            return timer_remaining
+
+        heating_rate = self._estimate_heating_rate()
+        if heating_rate <= 0:
+            return timer_remaining
+
+        target_remaining = (self.target_temp - self.current_dish_temp) / heating_rate
+        return max(timer_remaining, max(0.0, target_remaining))
+
+    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 _estimate_heating_rate(self):
+        if self._last_temperature_sample is None:
+            return 0.0
+
+        last_time, last_temp = self._last_temperature_sample
+        now = time.time()
+        current_temp = self.current_dish_temp
+
+        if current_temp is None:
+            return 0.0
+
+        delta_time = now - last_time
+        if delta_time <= 0:
+            return 0.0
+
+        return (current_temp - last_temp) / delta_time
+
+    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.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()
+        self.estimated_remaining_time = self.get_remaining_time_estimation()
+
+        if self.current_dish_temp is not None:
+            if elapsed_time < (self.cook_time / 2.0) and self.current_dish_temp >= self.target_temp:
+                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 self.state == CookingStates.DONE and self.current_dish_temp < (self.target_temp - self.TEMPERATURE_TOLERANCE):
+                self.state = CookingStates.COOKING
+            elif elapsed_time >= self.cook_time * 1.25: # If the dish is not heating up
+                self.state = CookingStates.STIRRING_REQUIRED
+                self.pause()
+
+            self._last_temperature_sample = (now, self.current_dish_temp)
+
+        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