diff --git a/ruff.toml b/ruff.toml index b82d11c..7eca042 100644 --- a/ruff.toml +++ b/ruff.toml @@ -50,7 +50,7 @@ ignore = [ "EXE002", # Use of exec - IGNORE "DTZ005", # Use of datetime.now() without tz - IGNORE # "Q000", # Quotes - # "ERA001", # Commented out code + "ERA001", # Commented out code ] diff --git a/src/pyvesync/base_devices/fryer_base.py b/src/pyvesync/base_devices/fryer_base.py index 875b142..7c6e766 100644 --- a/src/pyvesync/base_devices/fryer_base.py +++ b/src/pyvesync/base_devices/fryer_base.py @@ -3,9 +3,18 @@ from __future__ import annotations import logging +import time from typing import TYPE_CHECKING from pyvesync.base_devices.vesyncbasedevice import DeviceState, VeSyncBaseDevice +from pyvesync.const import ( + AIRFRYER_PID_MAP, + AirFryerCookStatus, + AirFryerFeatures, + AirFryerPresetRecipe, + TemperatureUnits, + TimeUnits, +) if TYPE_CHECKING: from pyvesync import VeSync @@ -19,11 +28,24 @@ class FryerState(DeviceState): """State class for Air Fryer devices. - Note: This class is a placeholder for future functionality and does not currently - implement any specific features or attributes. + Time units are in seconds by default. They are automatically converted + from the API response. """ - __slots__ = () + __slots__ = ( + '_time_conv', + 'cook_last_time', + 'cook_mode', + 'cook_set_temp', + 'cook_set_time', + 'cook_status', + 'current_temp', + 'last_timestamp', + 'preheat_last_time', + 'preheat_set_time', + 'ready_start', + 'time_units', + ) def __init__( self, @@ -42,12 +64,194 @@ def __init__( super().__init__(device, details, feature_map) self.device: VeSyncFryer = device self.features: list[str] = feature_map.features + self.time_units: TimeUnits = feature_map.time_units + self.ready_start: bool = False + self.cook_status: str | None = None + self.cook_mode: str | None = None + self.current_temp: int | None = None + self.cook_set_temp: int | None = None + self.cook_set_time: int | None = None + self.cook_last_time: int | None = None + self.last_timestamp: int | None = None + self.preheat_set_time: int | None = None + self.preheat_last_time: int | None = None + self._time_conv: float = 60 if feature_map.time_units == TimeUnits.MINUTES else 1 + + @property + def is_in_preheat_mode(self) -> bool: + """Return True if the fryer has preheat feature.""" + return self.cook_status in [ + AirFryerCookStatus.HEATING, + AirFryerCookStatus.PREHEAT_STOP, + ] or ( + self.cook_status == AirFryerCookStatus.PULL_OUT + and self.preheat_set_time is not None + ) + + @property + def is_in_cook_mode(self) -> bool: + """Return True if the fryer is in cook mode.""" + return self.cook_status in [ + AirFryerCookStatus.COOKING, + AirFryerCookStatus.COOK_STOP, + ] or ( + self.cook_status == AirFryerCookStatus.PULL_OUT + and self.cook_set_time is not None + ) + + @property + def is_cooking(self) -> bool: + """Return True if the fryer is currently cooking (not preheating).""" + return self.cook_status == AirFryerCookStatus.COOKING + + @property + def is_preheating(self) -> bool: + """Return True if the fryer is currently preheating.""" + return self.cook_status == AirFryerCookStatus.HEATING + + @property + def is_running(self) -> bool: + """Return True if the fryer is running (cooking or preheating).""" + return self.is_cooking or self.is_preheating + + @property + def can_resume(self) -> bool: + """Return True if the fryer can resume cooking.""" + return self.cook_status in [ + AirFryerCookStatus.PREHEAT_STOP, + AirFryerCookStatus.COOK_STOP, + ] + + @property + def preheat_time_remaining(self) -> int | None: + """Return the remaining preheat time in seconds.""" + if not self.is_in_preheat_mode: + return None + if self.cook_status in [ + AirFryerCookStatus.PREHEAT_STOP, + AirFryerCookStatus.PULL_OUT, + ]: + return self.preheat_last_time + if self.preheat_last_time is not None and self.last_timestamp is not None: + return max( + 0, + self.preheat_last_time + - int((self.last_timestamp - time.time()) * self._time_conv), + ) + return None + + @property + def cook_time_remaining(self) -> int | None: + """Return the remaining cook time in seconds.""" + if not self.is_in_cook_mode: + return None + if self.cook_status in [ + AirFryerCookStatus.PULL_OUT, + AirFryerCookStatus.COOK_STOP, + ]: + return self.cook_last_time + + if self.cook_last_time is not None and self.last_timestamp is not None: + return max( + 0, + self.cook_last_time + - int((self.last_timestamp - time.time()) * self._time_conv), + ) + return None + + def _clear_preheat(self) -> None: + """Clear preheat status.""" + self.preheat_set_time = None + self.preheat_last_time = None + + def set_standby(self) -> None: + """Set the fryer state to standby and clear all state attributes. + + This is to be called by device classes before updating the state from + the API response to prevent stale data. The get_details API responses + do not include all keys in every response depending on the status. + """ + self.cook_status = AirFryerCookStatus.STANDBY + self.current_temp = None + self.cook_set_temp = None + self.cook_set_time = None + self.cook_last_time = None + self.last_timestamp = None + self._clear_preheat() + + def set_state( # noqa: PLR0913, C901 + self, + *, + cook_status: str, + cook_time: int | None = None, + cook_last_time: int | None = None, + cook_temp: int | None = None, + temp_unit: str | None = None, + cook_mode: str | None = None, + preheat_time: int | None = None, + preheat_last_time: int | None = None, + current_temp: int | None = None, + ) -> None: + """Set the cook state parameters. + + Args: + cook_status (str): The cooking status. + cook_time (int | None): The cooking time in seconds. + cook_last_time (int | None): The last cooking time in seconds. + cook_temp (int | None): The cooking temperature. + temp_unit (str | None): The temperature units (F or C). + cook_mode (str | None): The cooking mode. + preheat_time (int | None): The preheating time in seconds. + preheat_last_time (int | None): The remaining preheat time in seconds. + current_temp (int | None): The current temperature. + """ + if cook_status == AirFryerCookStatus.STANDBY: + self.set_standby() + return + + self.preheat_set_time = preheat_time + self.preheat_last_time = preheat_last_time + + if cook_status is not None: + self.cook_status = AirFryerCookStatus(cook_status) + if cook_time is not None: + self.cook_set_time = cook_time + if cook_temp is not None: + self.cook_set_temp = cook_temp + if cook_mode is not None: + self.cook_mode = cook_mode + if current_temp is not None: + self.current_temp = current_temp + if temp_unit is not None: + self.device.temp_unit = TemperatureUnits.from_string(temp_unit) + if preheat_time is not None: + self.preheat_set_time = preheat_time + if cook_last_time is not None: + self.cook_last_time = cook_last_time + if cook_status in [ + AirFryerCookStatus.COOKING, + AirFryerCookStatus.HEATING, + ]: + self.last_timestamp = int(time.time()) class VeSyncFryer(VeSyncBaseDevice): """Base class for VeSync Air Fryer devices.""" - __slots__ = () + __slots__ = ( + '_temp_unit', + 'cook_modes', + 'default_preset', + 'max_temp_c', + 'max_temp_f', + 'min_temp_c', + 'min_temp_f', + 'state_chamber_1', + 'state_chamber_2', + 'sync_chambers', + 'temperature_interval', + 'time_units', + ) def __init__( self, @@ -66,3 +270,175 @@ def __init__( This is a bare class as there is only one supported air fryer model. """ super().__init__(details, manager, feature_map) + self.cook_modes: dict[str, str] = feature_map.cook_modes + self.pid: str | None = AIRFRYER_PID_MAP.get(details.deviceType, None) + self.default_preset: AirFryerPresetRecipe = feature_map.default_preset + self.state_chamber_1: FryerState = FryerState(self, details, feature_map) + self.state_chamber_2: FryerState = FryerState(self, details, feature_map) + self.sync_chambers: bool = False + self.min_temp_f: int = feature_map.temperature_range_f[0] + self.max_temp_f: int = feature_map.temperature_range_f[1] + self.min_temp_c: int = feature_map.temperature_range_c[0] + self.max_temp_c: int = feature_map.temperature_range_c[1] + self.temperature_interval: int = feature_map.temperature_step_f + self.time_units: TimeUnits = feature_map.time_units + + # attempt to set temp unit from country code before first update + self._temp_unit: TemperatureUnits = TemperatureUnits.CELSIUS + if self.manager.measure_unit and self.manager.measure_unit.lower() == 'imperial': + self._temp_unit = TemperatureUnits.FAHRENHEIT + + # Use single state attribute if not dual chamber fryer for compatibility + if AirFryerFeatures.DUAL_CHAMBER not in self.features: + self.state = self.state_chamber_1 + + @property + def temp_unit(self) -> TemperatureUnits: + """Return the temperature unit (F or C).""" + return self._temp_unit + + @temp_unit.setter + def temp_unit(self, value: TemperatureUnits) -> None: + """Set the temperature unit. + + Args: + value (TemperatureUnits): The temperature unit (F or C). + """ + self._temp_unit = TemperatureUnits.from_string(value) + + def validate_temperature(self, temperature: int) -> bool: + """Validate the temperature is within the allowed range. + + Args: + temperature (int): The temperature to validate. + + Returns: + bool: True if the temperature is valid, False otherwise. + """ + if self.temp_unit == TemperatureUnits.FAHRENHEIT: + return self.min_temp_f <= temperature <= self.max_temp_f + return self.min_temp_c <= temperature <= self.max_temp_c + + def round_temperature(self, temperature: int) -> int: + """Round the temperature to the nearest valid step. + + Args: + temperature (int): The temperature to round. + + Returns: + int: The rounded temperature. + """ + if self.temp_unit == TemperatureUnits.FAHRENHEIT: + step: float = self.temperature_interval + return int(round(temperature / step) * step) + step = self.temperature_interval * 5 / 9 + return int(round(temperature / step) * step) + + def convert_time(self, time_in_seconds: int) -> int: + """Convert time in seconds to the device's time units. + + Args: + time_in_seconds (int): The time in seconds. + + Returns: + int: The time converted to the device's time units. + """ + if self.time_units == TimeUnits.MINUTES: + return int(time_in_seconds / 60) + return time_in_seconds + + async def end(self, chamber: int = 1) -> bool: + """End the current cooking or preheating session. + + Arguments: + chamber (int): The chamber number to end for. Default is 1. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del chamber + logger.info('end not configured for this fryer.') + return False + + async def stop(self, chamber: int = 1) -> bool: + """Stop (Pause) the current cooking or preheating session. + + Arguments: + chamber (int): The chamber number to stop for. Default is 1. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del chamber + logger.info('stop not supported by this fryer.') + return False + + async def resume(self, chamber: int = 1) -> bool: + """Resume a paused cooking or preheating session. + + Arguments: + chamber (int): The chamber number to resume for. Default is 1. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del chamber + logger.info('resume not supported by this fryer.') + return False + + async def set_mode( + self, + cook_time: int, + cook_temp: int, + *, + preheat_time: int | None = None, + chamber: int = 1, + ) -> bool: + """Set the cooking mode. + + Args: + cook_time (int): The cooking time in seconds. + cook_temp (int): The cooking temperature. + preheat_time (int | None): The preheating time in seconds, if any. + chamber (int): The chamber number to set cooking for. Default is 1. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del cook_time, cook_temp, chamber, preheat_time + logger.warning('set_mode method not implemented for base fryer class.') + return False + + async def set_mode_from_recipe( + self, + recipe: AirFryerPresetRecipe, + ) -> bool: + """Set the cooking mode from a preset recipe. + + Args: + recipe (AirFryerPresetRecipe): The preset recipe to use. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del recipe + logger.warning( + 'set_mode_from_recipe method not implemented for base fryer class.' + ) + return False + + async def cook_from_preheat(self, chamber: int = 1) -> bool: + """Start cooking after preheating, cookStatus must be preheatEnd. + + Args: + chamber (int): The chamber number to start cooking for. Default is 1. + + Returns: + bool: True if the command was successful, False otherwise. + """ + del chamber + if AirFryerFeatures.PREHEAT not in self.features: + logger.info('Preheat feature not supported on this fryer.') + return False + logger.info('cook_from_preheat not configured for this fryer.') + return False diff --git a/src/pyvesync/const.py b/src/pyvesync/const.py index a0571b4..8d04c20 100644 --- a/src/pyvesync/const.py +++ b/src/pyvesync/const.py @@ -28,6 +28,7 @@ import platform import uuid +from dataclasses import dataclass from enum import Enum, IntEnum, StrEnum from random import randint from types import MappingProxyType @@ -75,6 +76,20 @@ KELVIN_MAX = 6500 +class TimeUnits(StrEnum): + """Time units for VeSync devices. + + Attributes: + MINUTES: Time in minutes. + SECONDS: Time in seconds. + HOURS: Time in hours. + """ + + MINUTES = 'minutes' + SECONDS = 'seconds' + HOURS = 'hours' + + class ProductLines(StrEnum): """High level product line.""" @@ -303,6 +318,52 @@ def from_bool(cls, value: bool | None) -> ConnectionStatus: return cls.ONLINE if value else cls.OFFLINE +class TemperatureUnits(StrEnum): + """Temperature units for VeSync devices. + + Attributes: + CELSIUS: Temperature in Celsius. + FAHRENHEIT: Temperature in Fahrenheit. + """ + + CELSIUS = 'c' + FAHRENHEIT = 'f' + + @property + def code(self) -> str: + """Return the code for the temperature unit 'f' or 'c'.""" + return self.value + + @property + def label(self) -> str: + """Return the label for the temperature unit.""" + return self.name.lower() + + @classmethod + def from_string(cls, value: str) -> TemperatureUnits: + """Convert string value to corresponding TemperatureUnit.""" + if value.lower() == 'c' or value.lower() == 'celsius': + return cls.CELSIUS + if value.lower() == 'f' or value.lower() == 'fahrenheit': + return cls.FAHRENHEIT + exc_msg = f'Invalid temperature unit: {value} value' + raise ValueError(exc_msg) + + @classmethod + def to_celsius(cls, value: float, unit: TemperatureUnits) -> float: + """Convert temperature to Celsius.""" + if unit == cls.FAHRENHEIT: + return (value - 32) * 5.0 / 9.0 + return value + + @classmethod + def to_fahrenheit(cls, value: float, unit: TemperatureUnits) -> float: + """Convert temperature to Fahrenheit.""" + if unit == cls.CELSIUS: + return (value * 9.0 / 5.0) + 32 + return value + + class NightlightModes(StrEnum): """Nightlight modes. @@ -690,6 +751,157 @@ class FanModes(StrEnum): """PID's for VeSync Air Fryers based on ConfigModule.""" +CUSTOM_RECIPE_ID = 1 +CUSTOM_RECIPE_TYPE = 3 +CUSTOM_RECIPE_NAME = 'Manual Cook' +CUSTOM_COOK_MODE = 'custom' + + +@dataclass +class AirFryerPresetRecipe: + """Preset recipe for VeSync Air Fryers. + + Set preheat_time to enable preheat mode. + + Attributes: + recipe_id (int): Recipe ID. + recipe_type (int): Recipe type. + recipe_name (str): Recipe name. + cook_mode (str): Cooking mode. + target_temp (int): Target temperature. + temp_unit (str): Temperature unit ('f' or 'c'). + cook_time (int): Cooking time in seconds. + preheat_time (int | None): Preheating time in seconds, if any. + """ + + recipe_name: str + cook_mode: str + recipe_id: int + recipe_type: int + target_temp: int + temp_unit: str + cook_time: int + preheat_time: int | None = None + + +class AirFryerPresets: + """Preset recipes for VeSync Air Fryers. + + Attributes: + custom (AirFryerPresetRecipe): Custom preset recipe. + """ + + custom: AirFryerPresetRecipe = AirFryerPresetRecipe( + cook_mode='Custom', + recipe_name='Manual Cook', + recipe_id=1, + recipe_type=3, + target_temp=350, + temp_unit='f', + cook_time=10 * 60, + ) + air_fry: AirFryerPresetRecipe = AirFryerPresetRecipe( + cook_mode='AirFry', + recipe_name='AirFry', + recipe_id=14, + recipe_type=3, + target_temp=400, + temp_unit='f', + cook_time=10 * 60, + ) + + +AIRFRYER_PRESET_MAP = { + 'custom': AirFryerPresetRecipe( + cook_mode='Custom', + recipe_name='Manual Cook', + recipe_id=1, + recipe_type=3, + target_temp=350, + temp_unit='f', + cook_time=20, + ), + 'airfry': AirFryerPresetRecipe( + cook_mode='AirFry', + recipe_name='AirFry', + recipe_id=4, + recipe_type=3, + target_temp=400, + temp_unit='f', + cook_time=25, + ), +} + + +class AirFryerCookModes(StrEnum): + """Cooking modes for VeSync Air Fryers. + + Attributes: + CUSTOM: Custom cooking mode. + FRY: Fry cooking mode. + ROAST: Roast cooking mode. + BAKE: Bake cooking mode. + REHEAT: Reheat cooking mode. + DEHYDRATE: Dehydrate cooking mode. + """ + + CUSTOM = 'custom' + ROAST = 'roast' + BAKE = 'bake' + REHEAT = 'reheat' + DEHYDRATE = 'dehydrate' + FROZEN = 'frozen' + PROOF = 'proof' + BROIL = 'broil' + WARM = 'warm' + AIRFRY = 'airfry' + DRY = 'dry' + PREHEAT = 'preheat' + + +class AirFryerFeatures(Features): + """VeSync Air Fryer features. + + Attributes: + ONOFF: Device on/off status. + TEMP: Temperature status. + TIME: Time status. + COOK_MODE: Cooking mode status. + PRESET_RECIPE: Preset recipe status. + CUSTOM_RECIPE: Custom recipe status. + PAUSE: Pause status. + """ + + DUAL_BLAZE = 'dual_blaze' + PREHEAT = 'preheat' + DUAL_CHAMBER = 'dual_chamber' + RESUMABLE = 'resumable' + + +class AirFryerCookStatus(StrEnum): + """Cooking status for VeSync Air Fryers. + + Attributes: + COOKING: Device is cooking. + PAUSED: Device is paused. + COMPLETED: Cooking is completed. + UNKNOWN: Cooking status is unknown. + """ + + COOKING = 'cooking' + COOK_STOP = 'cookStop' + COOK_END = 'cookEnd' + PULL_OUT = 'pullOut' + PAUSED = 'paused' + COMPLETED = 'completed' + HEATING = 'heating' + STOPPED = 'stopped' + UNKNOWN = 'unknown' + STANDBY = 'standby' + PREHEAT_END = 'preheatEnd' + PREHEAT_STOP = 'preheatStop' + + # Thermostat Constants @@ -833,15 +1045,6 @@ class ThermostatConst: WorkStatus = ThermostatWorkStatusCodes -# ------------------- AIR FRYER CONST ------------------ # - - -CUSTOM_RECIPE_ID = 1 -CUSTOM_RECIPE_TYPE = 3 -CUSTOM_RECIPE_NAME = 'Manual Cook' -CUSTOM_COOK_MODE = 'custom' - - # ------------------- OUTLET CONST ------------------ # diff --git a/src/pyvesync/device_map.py b/src/pyvesync/device_map.py index 320c298..23efc1b 100644 --- a/src/pyvesync/device_map.py +++ b/src/pyvesync/device_map.py @@ -65,6 +65,10 @@ from typing import Union from pyvesync.const import ( + AirFryerCookModes, + AirFryerFeatures, + AirFryerPresetRecipe, + AirFryerPresets, BulbFeatures, ColorMode, EnergyIntervals, @@ -86,6 +90,7 @@ ThermostatHoldOptions, ThermostatRoutineTypes, ThermostatWorkModes, + TimeUnits, ) from pyvesync.devices import ( vesyncbulb, @@ -135,8 +140,8 @@ class DeviceMapTemplate: setup_entry: str model_display: str model_name: str - device_alias: str | None = None - features: list[str] = field(default_factory=list) + features: list[str] + device_alias: str @dataclass(kw_only=True) @@ -241,11 +246,11 @@ class FanMap(DeviceMapTemplate): set_mode_method (str): Method to set the mode for the device. """ + modes: dict[str, str] + fan_levels: list[int] product_line: str = ProductLines.WIFI_AIR product_type: str = ProductTypes.FAN module: ModuleType = vesyncfan - fan_levels: list[int] = field(default_factory=list) - modes: dict[str, str] = field(default_factory=dict) sleep_preferences: list[str] = field(default_factory=list) set_mode_method: str = '' @@ -271,9 +276,9 @@ class HumidifierMap(DeviceMapTemplate): warm_mist_levels (list[int | str]): List of warm mist levels for the device. """ + mist_modes: dict[str, str] + mist_levels: list[int] product_line: str = ProductLines.WIFI_AIR - mist_modes: dict[str, str] = field(default_factory=dict) - mist_levels: list[int] = field(default_factory=list) product_type: str = ProductTypes.HUMIDIFIER module: ModuleType = vesynchumidifier target_minmax: tuple[int, int] = (30, 80) @@ -302,11 +307,11 @@ class PurifierMap(DeviceMapTemplate): auto_preferences (list[str]): List of auto preferences for the device. """ + fan_levels: list[int] + modes: list[str] product_line: str = ProductLines.WIFI_AIR product_type: str = ProductTypes.PURIFIER module: ModuleType = vesyncpurifier - fan_levels: list[int] = field(default_factory=list) - modes: list[str] = field(default_factory=list) nightlight_modes: list[str] = field(default_factory=list) auto_preferences: list[str] = field(default_factory=list) @@ -330,11 +335,16 @@ class AirFryerMap(DeviceMapTemplate): module (ModuleType): Module for the device. """ + time_units: TimeUnits = TimeUnits.MINUTES temperature_range_f: tuple[int, int] = (200, 400) temperature_range_c: tuple[int, int] = (75, 200) + temperature_step_f: int = 10 product_line: str = ProductLines.WIFI_KITCHEN product_type: str = ProductTypes.AIR_FRYER module: ModuleType = vesynckitchen + default_preset: AirFryerPresetRecipe = AirFryerPresets.custom + cook_modes: dict[str, str] = field(default_factory=dict) + default_cook_mode: str = AirFryerCookModes.AIRFRY @dataclass(kw_only=True) @@ -373,6 +383,7 @@ class ThermostatMap(DeviceMapTemplate): ThermostatMap( dev_types=['LTM-A401S-WUS'], class_name='VeSyncAuraThermostat', + features=[], fan_modes=[ ThermostatFanModes.AUTO, ThermostatFanModes.CIRCULATE, @@ -407,6 +418,7 @@ class ThermostatMap(DeviceMapTemplate): ], setup_entry='LTM-A401S-WUS', model_display='LTM-A401S Series', + device_alias='Aura Thermostat', model_name='Aura Thermostat', ) ] @@ -418,6 +430,7 @@ class ThermostatMap(DeviceMapTemplate): class_name='VeSyncOutlet7A', features=[OutletFeatures.ENERGY_MONITOR], model_name='WiFi Outlet US/CA', + device_alias='Round 7A WiFi Outlet', model_display='ESW01-USA Series', setup_entry='wifi-switch-1.3', ), @@ -427,6 +440,7 @@ class ThermostatMap(DeviceMapTemplate): features=[], model_name='10A WiFi Outlet USA', model_display='ESW10-USA Series', + device_alias='10A Round WiFi Outlet', setup_entry='ESW10-USA', ), OutletMap( @@ -435,6 +449,7 @@ class ThermostatMap(DeviceMapTemplate): features=[OutletFeatures.ENERGY_MONITOR], model_name='ESW03 10A WiFi Outlet', model_display='ESW01/03 USA/EU', + device_alias='10A Round WiFi Outlet', setup_entry='ESW03', ), OutletMap( @@ -444,6 +459,7 @@ class ThermostatMap(DeviceMapTemplate): nightlight_modes=[NightlightModes.ON, NightlightModes.OFF, NightlightModes.AUTO], model_name='15A WiFi Outlet US/CA', model_display='ESW15-USA Series', + device_alias='15A Rectangular WiFi Outlet', setup_entry='ESW15-USA', ), OutletMap( @@ -452,6 +468,7 @@ class ThermostatMap(DeviceMapTemplate): features=[OutletFeatures.ENERGY_MONITOR], model_name='Outdoor Plug', model_display='ESO15-TB Series', + device_alias='Outdoor Smart Plug', setup_entry='ESO15-TB', ), OutletMap( @@ -1075,8 +1092,35 @@ class ThermostatMap(DeviceMapTemplate): device_alias='Air Fryer', model_display='CS158/159/168/169-AF Series', model_name='Smart/Pro/Pro Gen 2 5.8 Qt. Air Fryer', - setup_entry='CS137-AF/CS158-AF', - ) + setup_entry='CS158-AF', + temperature_step_f=10, + features=[AirFryerFeatures.PREHEAT, AirFryerFeatures.RESUMABLE], + cook_modes={ + AirFryerCookModes.AIRFRY: 'custom', + }, + default_preset=AirFryerPresets.custom, + default_cook_mode=AirFryerCookModes.CUSTOM, + time_units=TimeUnits.MINUTES, + ), + AirFryerMap( + class_name='VeSyncTurboBlazeFryer', + module=vesynckitchen, + dev_types=['CAF-DC601S-WUSR', 'CAF-DC601S-WUS'], + setup_entry='CAF-DC601S', + device_alias='TurboBlaze Air Fryer', + model_display='CAF-DC601S Series', + model_name='TurboBlaze 6 Qt. Air Fryer', + temperature_step_f=5, + features=[AirFryerFeatures.PREHEAT, AirFryerFeatures.RESUMABLE], + cook_modes={ + AirFryerCookModes.AIRFRY: 'AirFry', + }, + default_cook_mode=AirFryerCookModes.AIRFRY, + default_preset=AirFryerPresets.air_fry, + time_units=TimeUnits.SECONDS, + temperature_range_f=(90, 450), + temperature_range_c=(30, 230), + ), ] """List of ['AirFryerMap'][pyvesync.device_map.AirFryerMap] configuration for air fryer devices.""" diff --git a/src/pyvesync/devices/vesynckitchen.py b/src/pyvesync/devices/vesynckitchen.py index b1e5499..98fd6a2 100644 --- a/src/pyvesync/devices/vesynckitchen.py +++ b/src/pyvesync/devices/vesynckitchen.py @@ -12,32 +12,33 @@ necessary to maintain state, especially when trying to `pause` or `resume` the device. Defaults to 60 seconds but can be set via: -```python -# Change to 120 seconds before status is updated between calls -VeSyncAirFryer158.refresh_interval = 120 - -# Set status update before every call -VeSyncAirFryer158.refresh_interval = 0 - -# Disable status update before every call -VeSyncAirFryer158.refresh_interval = -1 -``` - """ from __future__ import annotations import logging -import time +from dataclasses import replace from typing import TYPE_CHECKING, TypeVar from typing_extensions import deprecated from pyvesync.base_devices import FryerState, VeSyncFryer -from pyvesync.const import AIRFRYER_PID_MAP, ConnectionStatus, DeviceStatus +from pyvesync.const import ( + AIRFRYER_PID_MAP, + AirFryerCookStatus, + AirFryerPresetRecipe, +) +from pyvesync.models import fryer_models as models +from pyvesync.utils.device_mixins import ( + BypassV1Mixin, + BypassV2Mixin, + process_bypassv1_result, + process_bypassv2_result, +) from pyvesync.utils.errors import VeSyncError from pyvesync.utils.helpers import Helpers -from pyvesync.utils.logs import LibraryLogger + +# from pyvesync.utils.logs import LibraryLogger if TYPE_CHECKING: from pyvesync import VeSync @@ -49,286 +50,7 @@ logger = logging.getLogger(__name__) -# Status refresh interval in seconds -# API calls outside of interval are automatically refreshed -# Set VeSyncAirFryer158.refresh_interval to 0 to refresh every call -# Set to None or -1 to disable auto-refresh -REFRESH_INTERVAL = 60 - -RECIPE_ID = 1 -RECIPE_TYPE = 3 -CUSTOM_RECIPE = 'Manual Cook' -COOK_MODE = 'custom' - - -class AirFryer158138State(FryerState): - """Dataclass for air fryer status. - - Attributes: - active_time (int): Active time of device, defaults to None. - connection_status (str): Connection status of device. - device (VeSyncBaseDevice): Device object. - device_status (str): Device status. - features (dict): Features of device. - last_update_ts (int): Last update timestamp of device, defaults to None. - ready_start (bool): Ready start status of device, defaults to False. - preheat (bool): Preheat status of device, defaults to False. - cook_status (str): Cooking status of device, defaults to None. - current_temp (int): Current temperature of device, defaults to None. - cook_set_temp (int): Cooking set temperature of device, defaults to None. - last_timestamp (int): Last timestamp of device, defaults to None. - preheat_set_time (int): Preheat set time of device, defaults to None. - preheat_last_time (int): Preheat last time of device, defaults to None. - _temp_unit (str): Temperature unit of device, defaults to None. - """ - - __slots__ = ( - '_temp_unit', - 'cook_last_time', - 'cook_set_temp', - 'cook_set_time', - 'cook_status', - 'current_temp', - 'last_timestamp', - 'max_temp_c', - 'max_temp_f', - 'min_temp_c', - 'min_temp_f', - 'preheat', - 'preheat_last_time', - 'preheat_set_time', - 'ready_start', - ) - - def __init__( - self, - device: VeSyncAirFryer158, - details: ResponseDeviceDetailsModel, - feature_map: AirFryerMap, - ) -> None: - """Init the Air Fryer 158 class.""" - super().__init__(device, details, feature_map) - self.device: VeSyncFryer = device - self.features: list[str] = feature_map.features - self.min_temp_f: int = feature_map.temperature_range_f[0] - self.max_temp_f: int = feature_map.temperature_range_f[1] - self.min_temp_c: int = feature_map.temperature_range_c[0] - self.max_temp_c: int = feature_map.temperature_range_c[1] - self.ready_start: bool = False - self.preheat: bool = False - self.cook_status: str | None = None - self.current_temp: int | None = None - self.cook_set_temp: int | None = None - self.cook_set_time: int | None = None - self.cook_last_time: int | None = None - self.last_timestamp: int | None = None - self.preheat_set_time: int | None = None - self.preheat_last_time: int | None = None - self._temp_unit: str | None = None - - @property - def is_resumable(self) -> bool: - """Return if cook is resumable.""" - if self.cook_status in ['cookStop', 'preheatStop']: - if self.cook_set_time is not None: - return self.cook_set_time > 0 - if self.preheat_set_time is not None: - return self.preheat_set_time > 0 - return False - - @property - def temp_unit(self) -> str | None: - """Return temperature unit.""" - return self._temp_unit - - @temp_unit.setter - def temp_unit(self, temp_unit: str) -> None: - """Set temperature unit.""" - if temp_unit.lower() in ['f', 'fahrenheit', 'fahrenheight']: # API TYPO - self._temp_unit = 'fahrenheit' - elif temp_unit.lower() in ['c', 'celsius']: - self._temp_unit = 'celsius' - else: - msg = f'Invalid temperature unit - {temp_unit}' - raise ValueError(msg) - - @property - def preheat_time_remaining(self) -> int: - """Return preheat time remaining.""" - if self.preheat is False or self.cook_status == 'preheatEnd': - return 0 - if self.cook_status in ['pullOut', 'preheatStop']: - if self.preheat_last_time is None: - return 0 - return int(self.preheat_last_time) - if self.preheat_last_time is not None and self.last_timestamp is not None: - return int( - max( - ( - self.preheat_last_time * 60 - - (int(time.time()) - self.last_timestamp) - ) - // 60, - 0, - ) - ) - return 0 - - @property - def cook_time_remaining(self) -> int: - """Returns the amount of time remaining if cooking.""" - if self.preheat is True or self.cook_status == 'cookEnd': - return 0 - if self.cook_status in ['pullOut', 'cookStop']: - if self.cook_last_time is None: - return 0 - return int(max(self.cook_last_time, 0)) - if self.cook_last_time is not None and self.last_timestamp is not None: - return int( - max( - (self.cook_last_time * 60 - (int(time.time()) - self.last_timestamp)) - // 60, - 0, - ) - ) - return 0 - - @property - def remaining_time(self) -> int: - """Return minutes remaining if cooking/heating.""" - if self.preheat is True: - return self.preheat_time_remaining - return self.cook_time_remaining - - @property - def is_running(self) -> bool: - """Return if cooking or heating.""" - return bool(self.cook_status in ['cooking', 'heating']) and bool( - self.remaining_time > 0 - ) - - @property - def is_cooking(self) -> bool: - """Return if cooking.""" - return self.cook_status == 'cooking' and self.remaining_time > 0 - - @property - def is_heating(self) -> bool: - """Return if heating.""" - return self.cook_status == 'heating' and self.remaining_time > 0 - - def status_request(self, json_cmd: dict) -> None: # noqa: C901 - """Set status from jsonCmd of API call.""" - self.last_timestamp = None - if not isinstance(json_cmd, dict): - return - self.preheat = False - preheat = json_cmd.get('preheat') - cook = json_cmd.get('cookMode') - if isinstance(preheat, dict): - self.preheat = True - if preheat.get('preheatStatus') == 'stop': - self.cook_status = 'preheatStop' - elif preheat.get('preheatStatus') == 'heating': - self.cook_status = 'heating' - self.last_timestamp = int(time.time()) - self.preheat_set_time = preheat.get( - 'preheatSetTime', self.preheat_set_time - ) - if preheat.get('preheatSetTime') is not None: - self.preheat_last_time = preheat.get('preheatSetTime') - self.cook_set_temp = preheat.get('targetTemp', self.cook_set_temp) - self.cook_set_time = preheat.get('cookSetTime', self.cook_set_time) - self.cook_last_time = None - elif preheat.get('preheatStatus') == 'end': - self.cook_status = 'preheatEnd' - self.preheat_last_time = 0 - elif isinstance(cook, dict): - self.clear_preheat() - if cook.get('cookStatus') == 'stop': - self.cook_status = 'cookStop' - elif cook.get('cookStatus') == 'cooking': - self.cook_status = 'cooking' - self.last_timestamp = int(time.time()) - self.cook_set_time = cook.get('cookSetTime', self.cook_set_time) - self.cook_set_temp = cook.get('cookSetTemp', self.cook_set_temp) - self.current_temp = cook.get('currentTemp', self.current_temp) - self.temp_unit = cook.get( - 'tempUnit', - self.temp_unit, # type: ignore[assignment] - ) - elif cook.get('cookStatus') == 'end': - self.set_standby() - self.cook_status = 'cookEnd' - - def clear_preheat(self) -> None: - """Clear preheat status.""" - self.preheat = False - self.preheat_set_time = None - self.preheat_last_time = None - - def set_standby(self) -> None: - """Clear cooking status.""" - self.cook_status = 'standby' - self.clear_preheat() - self.cook_last_time = None - self.current_temp = None - self.cook_set_time = None - self.cook_set_temp = None - self.last_timestamp = None - - def status_response(self, return_status: dict) -> None: - """Set status of Air Fryer Based on API Response.""" - self.last_timestamp = None - self.preheat = False - self.cook_status = return_status.get('cookStatus') - if self.cook_status == 'standby': - self.set_standby() - return - - # If drawer is pulled out, set standby if resp does not contain other details - if self.cook_status == 'pullOut': - self.last_timestamp = None - if 'currentTemp' not in return_status or 'tempUnit' not in return_status: - self.set_standby() - self.cook_status = 'pullOut' - return - if return_status.get('preheatLastTime') is not None or self.cook_status in [ - 'heating', - 'preheatStop', - 'preheatEnd', - ]: - self.preheat = True - - self.cook_set_time = return_status.get('cookSetTime', self.cook_set_time) - self.cook_last_time = return_status.get('cookLastTime') - self.current_temp = return_status.get('curentTemp') - self.cook_set_temp = return_status.get( - 'targetTemp', return_status.get('cookSetTemp') - ) - self.temp_unit = return_status.get( - 'tempUnit', - self.temp_unit, # type: ignore[assignment] - ) - self.preheat_set_time = return_status.get('preheatSetTime') - self.preheat_last_time = return_status.get('preheatLastTime') - - # Set last_time timestamp if cooking - if self.cook_status in ['cooking', 'heating']: - self.last_timestamp = int(time.time()) - - if self.cook_status == 'preheatEnd': - self.preheat_last_time = 0 - self.cook_last_time = None - if self.cook_status == 'cookEnd': - self.cook_last_time = 0 - - # If Cooking, clear preheat status - if self.cook_status in ['cooking', 'cookStop', 'cookEnd']: - self.clear_preheat() - - -class VeSyncAirFryer158(VeSyncFryer): +class VeSyncAirFryer158(BypassV1Mixin, VeSyncFryer): """Cosori Air Fryer Class. Args: @@ -338,7 +60,7 @@ class VeSyncAirFryer158(VeSyncFryer): Attributes: features (list[str]): List of features. - state (AirFryer158138State): Air fryer state. + state (FryerState): Air fryer state. last_update (int): Last update timestamp. refresh_interval (int): Refresh interval in seconds. cook_temps (dict[str, list[int]] | None): Cook temperatures. @@ -361,12 +83,13 @@ class VeSyncAirFryer158(VeSyncFryer): """ __slots__ = ( - 'cook_temps', 'last_update', 'ready_start', 'refresh_interval', ) + request_keys: tuple[str, ...] = (*BypassV1Mixin.request_keys, 'pid') + def __init__( self, details: ResponseDeviceDetailsModel, @@ -376,278 +99,375 @@ def __init__( """Init the VeSync Air Fryer 158 class.""" super().__init__(details, manager, feature_map) self.features: list[str] = feature_map.features - self.state: AirFryer158138State = AirFryer158138State(self, details, feature_map) - self.last_update: int = int(time.time()) - self.refresh_interval = 0 - self.ready_start = False - self.cook_temps: dict[str, list[int]] | None = None + self.ready_start = True + self.state: FryerState = FryerState(self, details, feature_map) if self.config_module not in AIRFRYER_PID_MAP: msg = ( 'Report this error as an issue - ' - f'{self.config_module} not found in PID map for {self}' + f'{self.config_module} not found in PID map for {self.device_type}' ) raise VeSyncError(msg) self.pid = AIRFRYER_PID_MAP[self.config_module] - self.request_keys = ( - 'acceptLanguage', - 'accountID', - 'appVersion', - 'cid', - 'configModule', - 'deviceRegion', - 'phoneBrand', - 'phoneOS', - 'timeZone', - 'token', - 'traceId', - 'userCountryCode', - 'method', - 'debugMode', - 'uuid', - 'pid', - ) @deprecated('There is no on/off function for Air Fryers.') async def toggle_switch(self, toggle: bool | None = None) -> bool: """Turn on or off the air fryer.""" return toggle if toggle is not None else not self.is_on - def _build_request( - self, - json_cmd: dict | None = None, - method: str | None = None, - ) -> dict: - """Return body of api calls.""" - req_dict = Helpers.get_defaultvalues_attributes(self.request_keys) - req_dict.update(Helpers.get_manager_attributes(self.manager, self.request_keys)) - req_dict.update(Helpers.get_device_attributes(self, self.request_keys)) - req_dict['method'] = method or 'bypass' - req_dict['jsonCmd'] = json_cmd or {} - return req_dict - - def _build_status_body(self, cmd_dict: dict) -> dict: - """Return body of api calls.""" - body = self._build_request() - body.update( - { - 'uuid': self.uuid, - 'configModule': self.config_module, - 'jsonCmd': cmd_dict, - 'pid': self.pid, - 'accountID': self.manager.account_id, - } - ) - return body + def _build_base_request( + self, cook_set_time: int, recipe: AirFryerPresetRecipe | None = None + ) -> dict[str, int | str | bool]: + """Build base cook or preheat request body. + + This allows a custom recipe to be passed, but defaults to manual + cooking. The cook_set_time argument is required and will override + the default time in the recipe. + """ + cook_base: dict[str, int | str | bool] = {} + cook_base['cookSetTime'] = cook_set_time + if recipe is None: + cook_base['recipeId'] = self.default_preset.recipe_id + cook_base['customRecipe'] = self.default_preset.recipe_name + cook_base['mode'] = self.default_preset.cook_mode + cook_base['recipeType'] = self.default_preset.recipe_type + else: + cook_base['recipeId'] = recipe.recipe_id + cook_base['customRecipe'] = recipe.recipe_name + cook_base['mode'] = recipe.cook_mode + cook_base['recipeType'] = recipe.recipe_type + + cook_base['accountId'] = self.manager.account_id + if self.temp_unit is not None: + cook_base['tempUnit'] = self.temp_unit.label + else: + cook_base['tempUnit'] = 'fahrenheit' + cook_base['readyStart'] = True + return cook_base - @property - def temp_unit(self) -> str | None: - """Return temp unit.""" - return self.state.temp_unit + def _build_cook_request( + self, + cook_time: int, + cook_temp: int, + recipe: AirFryerPresetRecipe | None = None, + ) -> dict[str, int | str | bool]: + """Internal command to build cookMode API command.""" + cook_mode = self._build_base_request(cook_time, recipe) + cook_mode['appointmentTs'] = 0 + cook_mode['cookSetTemp'] = cook_temp + cook_mode['cookStatus'] = AirFryerCookStatus.COOKING.value + return cook_mode + + def _build_preheat_request( + self, + cook_time: int, + cook_temp: int, + recipe: AirFryerPresetRecipe | None = None, + ) -> dict[str, int | str | bool]: + """Internal command to build preheat API command.""" + preheat_mode = self._build_base_request(cook_time, recipe) + preheat_mode['targetTemp'] = cook_temp + preheat_mode['preheatSetTime'] = cook_time + preheat_mode['preheatStatus'] = AirFryerCookStatus.HEATING.value + return preheat_mode async def get_details(self) -> None: - """Get Air Fryer Status and Details.""" cmd = {'getStatus': 'status'} - req_body = self._build_request(json_cmd=cmd) - url = '/cloud/v1/deviceManaged/bypass' - r_dict, _ = await self.manager.async_call_api(url, 'post', json_object=req_body) - resp = Helpers.process_dev_response(logger, 'get_details', self, r_dict) - if resp is None: - self.state.device_status = DeviceStatus.OFF - self.state.connection_status = ConnectionStatus.OFFLINE - return + resp = await self.call_bypassv1_api(models.Fryer158RequestModel, update_dict=cmd) + + resp_model = process_bypassv1_result( + self, + logger, + 'get_details', + resp, + models.Fryer158Result, + ) - return_status = resp.get('result', {}).get('returnStatus') - if return_status is None: - LibraryLogger.error_device_response_content( - logger, - self, - 'get_details', - msg='Return status not found in response', + if resp_model is None or resp_model.returnStatus is None: + logger.debug( + 'No returnStatus in get_details response for %s', self.device_name ) - return - self.state.status_response(return_status) - - async def check_status(self) -> None: - """Update status if REFRESH_INTERVAL has passed.""" - seconds_elapsed = int(time.time()) - self.last_update - logger.debug('Seconds elapsed between updates: %s', seconds_elapsed) - refresh = False - if self.refresh_interval is None: - refresh = bool(seconds_elapsed > REFRESH_INTERVAL) - elif self.refresh_interval == 0: - refresh = True - elif self.refresh_interval > 0: - refresh = bool(seconds_elapsed > self.refresh_interval) - if refresh is True: - logger.debug('Updating status, %s seconds elapsed', seconds_elapsed) - await self.update() - - async def end(self) -> bool: - """End the cooking process.""" - await self.check_status() - if self.state.preheat is False and self.state.cook_status in [ - 'cookStop', - 'cooking', - ]: + self.state.set_standby() + return None + + return_status = resp_model.returnStatus + return self.state.set_state( + cook_status=return_status.cookStatus, + cook_time=return_status.cookSetTime, + cook_last_time=return_status.cookLastTime, + cook_temp=return_status.cookSetTemp, + temp_unit=return_status.tempUnit, + cook_mode=return_status.mode, + preheat_time=return_status.preheatSetTime, + preheat_last_time=return_status.preheatLastTime, + current_temp=return_status.currentTemp, + ) + + async def end(self, chamber: int = 1) -> bool: + del chamber # chamber not used for this air fryer + if self.state.is_in_cook_mode is True: cmd = {'cookMode': {'cookStatus': 'end'}} - elif self.state.preheat is True and self.state.cook_status in [ - 'preheatStop', - 'heating', - ]: - cmd = {'preheat': {'cookStatus': 'end'}} + if self.state.is_in_preheat_mode is True: + cmd = {'preheat': {'preheatStatus': 'end'}} else: logger.debug( 'Cannot end %s as it is not cooking or preheating', self.device_name ) return False - - status_api = await self._status_api(cmd) - if status_api is False: + json_cmd = {'jsonCmd': cmd} + resp = await self.call_bypassv1_api( + models.Fryer158RequestModel, update_dict=json_cmd + ) + r = Helpers.process_dev_response(logger, 'end', self, resp) + if r is None: return False self.state.set_standby() return True - async def pause(self) -> bool: - """Pause the cooking process.""" - await self.check_status() - if self.state.cook_status not in ['cooking', 'heating']: - logger.debug( - 'Cannot pause %s as it is not cooking or preheating', self.device_name - ) - return False - if self.state.preheat is True: + async def stop(self, chamber: int = 1) -> bool: + del chamber # chamber not used for this air fryer + if self.state.is_in_preheat_mode is True: cmd = {'preheat': {'preheatStatus': 'stop'}} - else: + if self.state.is_in_cook_mode is True: cmd = {'cookMode': {'cookStatus': 'stop'}} - status_api = await self._status_api(cmd) - if status_api is True: - if self.state.preheat is True: - self.state.cook_status = 'preheatStop' - else: - self.state.cook_status = 'cookStop' - return True - return False - - def _validate_temp(self, set_temp: int) -> bool: - """Temperature validation.""" - if self.state.temp_unit == 'fahrenheit' and ( - set_temp < self.state.min_temp_f or set_temp > self.state.max_temp_f - ): - logger.debug('Invalid temperature %s for %s', set_temp, self.device_name) + else: + logger.debug( + 'Cannot stop %s as it is not cooking or preheating', self.device_name + ) return False - if self.state.temp_unit == 'celsius' and ( - set_temp < self.state.min_temp_c or set_temp > self.state.max_temp_c - ): - logger.debug('Invalid temperature %s for %s', set_temp, self.device_name) + json_cmd = {'jsonCmd': cmd} + resp = await self.call_bypassv1_api( + models.Fryer158RequestModel, update_dict=json_cmd + ) + r = Helpers.process_dev_response(logger, 'stop', self, resp) + if r is None: return False + if self.state.is_in_preheat_mode is True: + self.state.cook_status = AirFryerCookStatus.PREHEAT_STOP + if self.state.is_in_cook_mode is True: + self.state.cook_status = AirFryerCookStatus.COOK_STOP return True - async def cook(self, set_temp: int, set_time: int) -> bool: - """Set cook time and temperature in Minutes.""" - await self.check_status() - if self._validate_temp(set_temp) is False: - return False - return await self._set_cook(set_temp, set_time) - - async def resume(self) -> bool: - """Resume paused preheat or cook.""" - await self.check_status() - if self.state.cook_status not in ['preheatStop', 'cookStop']: - logger.debug('Cannot resume %s as it is not paused', self.device_name) - return False - if self.state.preheat is True: + async def resume(self, chamber: int = 1) -> bool: + del chamber # chamber not used for this air fryer + if self.state.is_in_preheat_mode is True: cmd = {'preheat': {'preheatStatus': 'heating'}} - else: + elif self.state.is_in_cook_mode is True: cmd = {'cookMode': {'cookStatus': 'cooking'}} - status_api = await self._status_api(cmd) - if status_api is True: - if self.state.preheat is True: - self.state.cook_status = 'heating' - else: - self.state.cook_status = 'cooking' - return True - return False - - async def set_preheat(self, target_temp: int, cook_time: int) -> bool: - """Set preheat mode with cooking time.""" - await self.check_status() - if self.state.cook_status not in ['standby', 'cookEnd', 'preheatEnd']: + else: logger.debug( - 'Cannot set preheat for %s as it is not in standby', self.device_name + 'Cannot resume %s as it is not cooking or preheating', self.device_name ) return False - if self._validate_temp(target_temp) is False: + json_cmd = {'jsonCmd': cmd} + resp = await self.call_bypassv1_api( + models.Fryer158RequestModel, update_dict=json_cmd + ) + r = Helpers.process_dev_response(logger, 'resume', self, resp) + if r is None: return False - cmd = self._cmd_api_dict - cmd['preheatSetTime'] = 5 - cmd['preheatStatus'] = 'heating' - cmd['targetTemp'] = target_temp - cmd['cookSetTime'] = cook_time - json_cmd = {'preheat': cmd} - return await self._status_api(json_cmd) - - async def cook_from_preheat(self) -> bool: - """Start Cook when preheat has ended.""" - await self.check_status() - if self.state.preheat is False or self.state.cook_status != 'preheatEnd': - logger.debug('Cannot start cook from preheat for %s', self.device_name) + + if self.state.is_in_preheat_mode is True: + self.state.cook_status = AirFryerCookStatus.HEATING + if self.state.is_in_cook_mode is True: + self.state.cook_status = AirFryerCookStatus.COOKING + return True + + async def set_mode_from_recipe( + self, + recipe: AirFryerPresetRecipe, + *, + chamber: int = 1, + ) -> bool: + del chamber # chamber not used for this air fryer + if recipe.preheat_time is not None and recipe.preheat_time > 0: + cook_status = AirFryerCookStatus.HEATING + preheat_req = self._build_preheat_request( + cook_time=recipe.preheat_time, cook_temp=recipe.target_temp, recipe=recipe + ) + cmd = {'preheat': preheat_req} + else: + cook_status = AirFryerCookStatus.COOKING + cook_req = self._build_cook_request( + cook_time=recipe.cook_time, cook_temp=recipe.target_temp, recipe=recipe + ) + cmd = {'cookMode': cook_req} + json_cmd = {'jsonCmd': cmd} + resp = await self.call_bypassv1_api( + models.Fryer158RequestModel, update_dict=json_cmd + ) + r = Helpers.process_dev_response(logger, 'set_mode_from_recipe', self, resp) + if r is None: return False - return await self._set_cook(status='cooking') - - async def update(self) -> None: - """Update the device details.""" - await self.get_details() - - @property - def _cmd_api_base(self) -> dict: - """Return Base api dictionary for setting status.""" - return { - 'mode': COOK_MODE, - 'accountId': self.manager.account_id, - } + self.state.set_state( + cook_status=cook_status, + cook_time=recipe.cook_time, + cook_temp=recipe.target_temp, + cook_mode=recipe.cook_mode, + preheat_time=recipe.preheat_time, + ) + return True - @property - def _cmd_api_dict(self) -> dict: - """Return API dictionary for setting status.""" - cmd = self._cmd_api_base - cmd.update( - { - 'appointmentTs': 0, - 'recipeId': RECIPE_ID, - 'readyStart': self.ready_start, - 'recipeType': RECIPE_TYPE, - 'customRecipe': CUSTOM_RECIPE, + async def set_mode( + self, + cook_time: int, + cook_temp: int, + *, + preheat_time: int | None = None, + cook_mode: str | None = None, + chamber: int = 1, + ) -> bool: + if self.validate_temperature(cook_temp) is False: + logger.warning('Invalid cook temperature for %s', self.device_name) + return False + cook_temp = self.round_temperature(cook_temp) + cook_time = self.convert_time(cook_time) + preset_recipe = replace(self.default_preset) + preset_recipe.cook_time = cook_time + preset_recipe.target_temp = cook_temp + if cook_mode is not None: + preset_recipe.cook_mode = cook_mode + if preheat_time is not None: + preset_recipe.preheat_time = self.convert_time(preheat_time) + return await self.set_mode_from_recipe(preset_recipe, chamber=chamber) + + async def cook_from_preheat(self, chamber: int = 1) -> bool: + del chamber # chamber not used for this air fryer + if self.state.cook_status != AirFryerCookStatus.PREHEAT_END: + logger.debug('Cannot start cook from preheat for %s', self.device_name) + return False + cmd = { + 'cookMode': { + 'mode': self.state.cook_mode, + 'accountId': self.manager.account_id, + 'cookStatus': 'cooking', } + } + json_cmd = {'jsonCmd': cmd} + resp = await self.call_bypassv1_api( + models.Fryer158RequestModel, update_dict=json_cmd ) - return cmd + r = Helpers.process_dev_response(logger, 'cook_from_preheat', self, resp) + if r is None: + return False + self.state.set_state(cook_status=AirFryerCookStatus.COOKING) + return True + + +class VeSyncTurboBlazeFryer(BypassV2Mixin, VeSyncFryer): + """VeSync TurboBlaze Air Fryer Class.""" + + __slots__ = () - async def _set_cook( + def __init__( self, - set_temp: int | None = None, - set_time: int | None = None, - status: str = 'cooking', - ) -> bool: - if set_temp is not None and set_time is not None: - set_cmd = self._cmd_api_dict + details: ResponseDeviceDetailsModel, + manager: VeSync, + feature_map: AirFryerMap, + ) -> None: + """Init the VeSync TurboBlaze Air Fryer class.""" + super().__init__(details, manager, feature_map) - set_cmd['cookSetTime'] = set_time - set_cmd['cookSetTemp'] = set_temp - else: - set_cmd = self._cmd_api_base - set_cmd['cookStatus'] = status - cmd = {'cookMode': set_cmd} - return await self._status_api(cmd) - - async def _status_api(self, json_cmd: dict) -> bool: - """Set API status with jsonCmd.""" - body = self._build_status_body(json_cmd) - url = '/cloud/v1/deviceManaged/bypass' - r_dict, _ = await self.manager.async_call_api(url, 'post', json_object=body) - resp = Helpers.process_dev_response(logger, 'set_status', self, r_dict) - if resp is None: + # Single chamber fryer state + self.state: FryerState = FryerState(self, details, feature_map) + + def _build_cook_request( + self, recipe: AirFryerPresetRecipe + ) -> models.FryerTurboBlazeRequestData: + cook_req: dict[str, int | str | bool | dict] = {} + cook_req['accountId'] = self.manager.account_id + if recipe.preheat_time is not None and recipe.preheat_time > 0: + cook_req['hasPreheat'] = int(True) + cook_req['hasWarm'] = False + cook_req['mode'] = recipe.cook_mode + cook_req['readyStart'] = True + cook_req['recipeId'] = recipe.recipe_id + cook_req['recipeName'] = recipe.recipe_name + cook_req['recipeType'] = recipe.recipe_type + cook_req['tempUnit'] = self.temp_unit.code + cook_req['startAct'] = { + 'cookSetTime': recipe.cook_time, + 'cookTemp': recipe.target_temp, + 'preheatTemp': recipe.target_temp if recipe.preheat_time else 0, + 'shakeTime': 0, + } + return models.FryerTurboBlazeRequestData.from_dict(cook_req) + + async def get_details(self) -> None: + resp = await self.call_bypassv2_api(payload_method='getAirfyerStatus') + resp_model = process_bypassv2_result( + self, + logger, + 'get_details', + resp, + models.FryerTurboBlazeDetailResult, + ) + + if ( + resp_model is None + or resp_model.cookStatus == AirFryerCookStatus.STANDBY.value + or not resp_model.stepArray + ): + self.state.set_standby() + return + + cook_step = resp_model.stepArray[resp_model.stepIndex] + + self.state.set_state( + cook_status=resp_model.cookStatus, + cook_time=cook_step.cookSetTime, + cook_last_time=cook_step.cookLastTime, + cook_temp=cook_step.cookTemp, + temp_unit=resp_model.tempUnit, + cook_mode=cook_step.mode, + preheat_time=resp_model.preheatSetTime, + preheat_last_time=resp_model.preheatLastTime, + current_temp=resp_model.currentTemp, + ) + + async def end(self, chamber: int = 1) -> bool: + del chamber # chamber not used for this air fryer + payload_method = 'endCook' + resp = await self.call_bypassv2_api(payload_method=payload_method) + r = Helpers.process_dev_response(logger, 'end', self, resp) + if r is None: return False + self.state.set_standby() + return True - self.last_update = int(time.time()) - self.state.status_request(json_cmd) - await self.update() + async def set_mode_from_recipe(self, recipe: AirFryerPresetRecipe) -> bool: + payload_method = 'startCook' + data = self._build_cook_request(recipe) + resp = await self.call_bypassv2_api( + payload_method=payload_method, + data=data.to_dict(), + ) + r = Helpers.process_dev_response(logger, 'set_mode_from_recipe', self, resp) + if r is None: + self.state.set_standby() + return False + self.state.set_state( + cook_status=AirFryerCookStatus.COOKING, + cook_time=recipe.cook_time, + cook_last_time=recipe.cook_time, + cook_temp=recipe.target_temp, + cook_mode=recipe.cook_mode, + preheat_time=recipe.preheat_time if recipe.preheat_time else None, + preheat_last_time=recipe.preheat_time if recipe.preheat_time else None, + ) return True + + async def set_mode( + self, + cook_time: int, + cook_temp: int, + *, + preheat_time: int | None = None, + chamber: int = 1, + ) -> bool: + del chamber # chamber not used for this air fryer + recipe = replace(self.default_preset) + recipe.cook_time = self.convert_time(cook_time) + recipe.target_temp = self.round_temperature(cook_temp) + if preheat_time is not None: + recipe.preheat_time = self.convert_time(preheat_time) + return await self.set_mode_from_recipe(recipe) diff --git a/src/pyvesync/models/fryer_models.py b/src/pyvesync/models/fryer_models.py index bf14725..ab0c966 100644 --- a/src/pyvesync/models/fryer_models.py +++ b/src/pyvesync/models/fryer_models.py @@ -2,33 +2,236 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import Annotated -from pyvesync.models.base_models import ResponseBaseModel +from mashumaro.types import Discriminator + +from pyvesync.models.base_models import RequestBaseModel, ResponseBaseModel +from pyvesync.models.bypass_models import ( + BypassV1Result, + BypassV2InnerResult, + RequestBypassV1, +) + + +@dataclass +class Fryer158RequestModel(RequestBypassV1): + """Request model for air fryer commands.""" + + pid: str # type: ignore[misc] # bug in mypy invalid argument ordering + jsonCmd: dict # type: ignore[misc] # bug in mypy invalid argument ordering + deviceId: str = field(default_factory=str) + configModel: str = field(default_factory=lambda: '') + + def __post_serialize__(self, d: dict) -> dict: + """Remove empty strings before serialization.""" + for attrs in ['deviceId', 'configModel']: + d.pop(attrs, None) + return d @dataclass -class ResultFryerDetails(ResponseBaseModel): +class Fryer158Result(BypassV1Result): """Result model for air fryer details.""" - returnStatus: FryerCookingReturnStatus | FryerBaseReturnStatus | None = None + returnStatus: Fryer158CookingReturnStatus @dataclass -class FryerCookingReturnStatus(ResponseBaseModel): +class Fryer158CookingReturnStatus(ResponseBaseModel): """Result returnStatus model for air fryer status.""" - currentTemp: int + cookStatus: str + currentTemp: int | None = None + cookSetTemp: int | None = None + mode: str | None = None + cookSetTime: int | None = None + cookLastTime: int | None = None + tempUnit: str | None = None + preheatLastTime: int | None = None + preheatSetTime: int | None = None + targetTemp: int | None = None + + +@dataclass +class Fryer158CookRequest(RequestBaseModel): + """Base request model for air fryer cooking commands.""" + + cookMode: Annotated[Fryer158CookModeBase, Discriminator(include_subtypes=True)] + + +@dataclass +class Fryer158PreheatRequest(RequestBaseModel): + """Base request model for air fryer preheat commands.""" + + preheat: Annotated[Fryer158PreheatModeBase, Discriminator(include_subtypes=True)] + + +@dataclass +class Fryer158CookModeBase(RequestBaseModel): + """Base model for air fryer cooking modes.""" + + +@dataclass +class Fryer158CookModeFromPreheat(Fryer158CookModeBase): + """Model for continuing a cooking mode.""" + + cookStatus: str + accountId: str + mode: str + + +@dataclass +class Fryer158CookModeChange(Fryer158CookModeBase): + """Model for stopping a cooking mode.""" + + cookStatus: str + + +@dataclass +class Fryer158CookModeStart(Fryer158CookModeBase): + """Model for starting a cooking mode.""" + + cookStatus: str + accountId: str + mode: str + tempUnit: str + readyStart: bool + cookSetTime: int cookSetTemp: int + appointmentTs: int = 0 + customRecipe: str = 'Manual Cooking' + recipeId: int = 1 + recipeType: int = 3 + + +@dataclass +class Fryer158PreheatModeBase(RequestBaseModel): + """Base model for air fryer preheat modes.""" + + +@dataclass +class Fryer158PreheatModeChange(Fryer158PreheatModeBase): + """Model for continuing a preheat mode.""" + + preheatStatus: str + + +@dataclass +class Fryer158PreheatModeStart(Fryer158PreheatModeBase): + """Model for starting a preheat mode.""" + + preheatStatus: str + accountId: str mode: str + tempUnit: str + readyStart: bool + preheatSetTime: int + targetTemp: int cookSetTime: int - cookLastTime: int + customRecipe: str = 'Manual' + recipeId: int = 1 + recipeType: int = 3 + + +@dataclass +class FryerTurboBlazeDetailResult(BypassV2InnerResult): + """Result model for TurboBlaze air fryer details.""" + + stepArray: list[FryerTurboBlazeStepItem] + cookMode: str + tempUnit: str + stepIndex: int cookStatus: str + preheatSetTime: int + preheatLastTime: int + preheatEndTime: int + preheatTemp: int + startTime: int + totalTimeRemaining: int + currentTemp: int + shakeStatus: int + + +@dataclass +class FryerTurboBlazeStepItem(ResponseBaseModel): + """Data model for TurboBlaze air fryer cooking steps.""" + + cookSetTime: int + cookTemp: int + mode: str + cookLastTime: int + shakeTime: int + cookEndTime: int + recipeName: str + recipeId: int + recipeType: int + + +@dataclass +class FryerTurboBlazeRequestData(RequestBaseModel): + """Request model for TurboBlaze air fryer cooking commands.""" + + accountId: str + hasPreheat: int + hasWarm: bool + readyStart: bool + recipeId: int + recipeName: str + recipeType: int tempUnit: str + startAct: list[FryerTurboBlazeStartActItem] @dataclass -class FryerBaseReturnStatus(ResponseBaseModel): - """Result returnStatus model for air fryer status.""" +class FryerTurboBlazeStartActItem(RequestBaseModel): + """Data model for TurboBlaze air fryer startAct items.""" - cookStatus: str + cookSetTime: int + cookTemp: int + preheatTemp: int = 0 + shakeTime: int = 0 + + +# a = { +# 'cookMode': { +# 'accountId': '1221391', +# 'appointmentTs': 0, +# 'cookSetTemp': 350, +# 'cookSetTime': 15, +# 'cookStatus': 'cooking', +# 'customRecipe': 'Manual Cooking', +# 'mode': 'custom', +# 'readyStart': True, +# 'recipeId': 1, +# 'recipeType': 3, +# 'tempUnit': 'fahrenheit', +# }, +# 'preheat': { +# 'customRecipe': 'Manual', +# 'readyStart': False, +# 'cookSetTime': 15, +# 'tempUnit': 'fahrenheit', +# 'mode': 'custom', +# 'accountId': '1221391', +# 'targetTemp': 350, +# 'preheatSetTime': 4, +# 'preheatStatus': 'heating', +# 'recipeId': 1, +# 'recipeType': 3, +# }, +# 'cookMode': { +# 'accountId': '1221391', +# 'cookSetTemp': 400, +# 'recipeId': 1, +# 'mode': 'custom', +# 'readyStart': False, +# 'appointmentTs': 0, +# 'cookStatus': 'cooking', +# 'customRecipe': 'Manual', +# 'cookSetTime': 15, +# 'tempUnit': 'fahrenheit', +# 'recipeType': 3, +# }, +# } diff --git a/src/pyvesync/utils/helpers.py b/src/pyvesync/utils/helpers.py index 8994851..47065e6 100644 --- a/src/pyvesync/utils/helpers.py +++ b/src/pyvesync/utils/helpers.py @@ -123,6 +123,16 @@ def temperature_celsius_to_fahrenheit(celsius: float) -> float: """Convert Celsius to Fahrenheit.""" return celsius * 9.0 / 5.0 + 32 + @staticmethod + def minutes_to_seconds(minutes: int) -> int: + """Convert minutes to seconds.""" + return minutes * 60 + + @staticmethod + def seconds_to_minutes(seconds: int) -> int: + """Convert seconds to minutes.""" + return seconds // 60 + class Helpers: """VeSync Helper Functions.""" diff --git a/src/pyvesync/vesync.py b/src/pyvesync/vesync.py index 3fda881..68cca90 100644 --- a/src/pyvesync/vesync.py +++ b/src/pyvesync/vesync.py @@ -62,6 +62,7 @@ class VeSync: # pylint: disable=function-redefined 'enabled', 'in_process', 'language', + 'measure_unit', 'session', 'time_zone', ) @@ -142,6 +143,7 @@ class is instantiated, call `await manager.login()` to log in to VeSync servers, self.language: str = 'en' self.enabled = False self.in_process = False + self.measure_unit: str | None = None self._device_container: DeviceContainer = DeviceContainer() # Initialize authentication manager diff --git a/src/tests/call_json_air_fryers.py b/src/tests/call_json_air_fryers.py new file mode 100644 index 0000000..db1a51e --- /dev/null +++ b/src/tests/call_json_air_fryers.py @@ -0,0 +1,128 @@ +""" +Air Fryer Device API Responses + +AIR_FRYER variable is a list of device types + +DETAILS_RESPONSES variable is a dictionary of responses from the API +for get_details() methods. The keys are the device types and the +values are the responses. The responses are tuples of (response, status) + +METHOD_RESPONSES variable is a defaultdict of responses from the API. This is +the FunctionResponse variable from the utils module in the tests dir. +The default response is a tuple with the value ({"code": 0, "msg": "success"}, 200). + +The values of METHOD_RESPONSES can be a function that takes a single argument or +a static value. The value is checked if callable at runtime and if so, it is called +with the provided argument. If not callable, the value is returned as is. + +METHOD_RESPONSES = { + 'CS158-AF': defaultdict( + lambda: ({"code": 0, "msg": "success"}, 200)) + ) +} + +# For a function to handle the response +def status_response(request_body=None): + # do work with request_body + return request_body, 200 + +METHOD_RESPONSES['CS158-AF']['set_cook_mode'] = status_response + +# To change the default value for a device type + +METHOD_RESPONSES['CS158-AF'].default_factory = lambda: ({"code": 0, "msg": "success"}, 200) + +""" + +from copy import deepcopy +from pyvesync.device_map import air_fryer_modules +from pyvesync.const import ( + DeviceStatus, + ConnectionStatus, + TemperatureUnits, + AirFryerCookStatus, +) +from defaults import ( + TestDefaults, + FunctionResponsesV2, + FunctionResponsesV1, + build_bypass_v1_response, + build_bypass_v2_response, +) + + +class AirFryerDefaults: + temp_unit = TemperatureUnits.FAHRENHEIT + cook_time_f = 10 + cook_temp_f = 350 + cook_mode = "custom" + current_temp_f = 150 + cook_status = AirFryerCookStatus.COOKING + recipe = "Manual" + + +AIR_FRYER_COOKING_DETAILS: dict[str, dict[str, list | str | float | dict | None]] = { + "CS158-AF": { + "returnStatus": { + "curentTemp": AirFryerDefaults.current_temp_f, + "cookSetTemp": AirFryerDefaults.cook_temp_f, + "mode": AirFryerDefaults.cook_mode, + "cookSetTime": AirFryerDefaults.cook_time_f, + "cookLastTime": AirFryerDefaults.cook_time_f - 2, + "cookStatus": AirFryerDefaults.cook_status.value, + "tempUnit": AirFryerDefaults.temp_unit.label, + "accountId": TestDefaults.account_id, + "customRecipe": AirFryerDefaults.recipe, + } + }, + "CAF-DC601S": { + "traceId": "1767318172645", + "code": 0, + "result": { + "stepArray": [ + { + "cookSetTime": 1200, + "cookTemp": 330, + "mode": "Bake", + "cookLastTime": 1176, + "shakeTime": 0, + "cookEndTime": 0, + "recipeName": "Bake", + "recipeId": 9, + "recipeType": 3, + } + ], + "cookMode": "normal", + "tempUnit": "f", + "stepIndex": 0, + "cookStatus": "cooking", + "preheatSetTime": 0, + "preheatLastTime": 0, + "preheatEndTime": 0, + "preheatTemp": 0, + "startTime": 1767318116, + "totalTimeRemaining": 1176, + "currentTemp": 89, + "shakeStatus": 0, + }, + }, +} + +AIR_FRYER_STANDYBY_DETAILS: dict[str, dict[str, list | str | float | dict | None]] = { + "CAF-DC601S": { + "stepArray": [], + "cookMode": "normal", + "tempUnit": "f", + "stepIndex": 0, + "cookStatus": "standby", + "preheatSetTime": 0, + "preheatLastTime": 0, + "preheatEndTime": 0, + "preheatTemp": 0, + "startTime": 0, + "totalTimeRemaining": 0, + "currentTemp": 43, + "shakeStatus": 0, + }, + "CS158-AF": {"returnStatus": {"cookStatus": "standby"}}, +}