Smartwave/shared/cookingState.py

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.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

    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()
        # self._notify_refresh(force=True)
        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._pause_started_at = None
        self.paused = False
        # self._notify_refresh(force=True)

    def toggle_pause(self):
        if self.paused:
            self.unpause()
        else:
            self.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()
        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.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()
        self.estimated_remaining_time = self.get_remaining_time_estimation()
        
        print(elapsed_time, self.cook_time, self.current_dish_temp, self.target_temp, self._paused_duration, self._pause_started_at, now)

        if self.current_dish_temp is not None:
            if elapsed_time < (self.cook_time / 2.0) and self.current_dish_temp >= self.target_temp and (self._paused_duration == None or self._paused_duration < 5):  # If the dish is heating too fast, we require stirring
                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 == None or self._paused_duration < 5): # If the dish is not heating up
                self.state = CookingStates.STIRRING_REQUIRED
                self.pause()
            elif self._pause_started_at != None and (self._pause_started_at + self._paused_duration) < (now - (self.cook_time * 0.75)): # If the dish had to be pause and it's been a long time, we stop the cooking
                self.state = CookingStates.DONE

            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
    
    @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"