diff --git a/.gitignore b/.gitignore index 2a8c0e3..f1bfd12 100644 --- a/.gitignore +++ b/.gitignore @@ -39,4 +39,6 @@ testing_scripts/api_test_editor.py testing_scripts/yamltest.yaml testing_scripts/yamltest copy.yaml creds.json -docstest/* \ No newline at end of file +docstest/* +/.claude +.mcp.json 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/auth.py b/src/pyvesync/auth.py index 5104faf..899ebc9 100644 --- a/src/pyvesync/auth.py +++ b/src/pyvesync/auth.py @@ -236,6 +236,7 @@ async def save_credentials_to_file(self, file_path: str | Path | None = None) -> 'token': self._token, 'account_id': self._account_id, 'country_code': self._country_code, + 'current_region': self.current_region, } try: data = orjson.dumps(credentials).decode('utf-8') diff --git a/src/pyvesync/base_devices/fryer_base.py b/src/pyvesync/base_devices/fryer_base.py index 875b142..c48f4eb 100644 --- a/src/pyvesync/base_devices/fryer_base.py +++ b/src/pyvesync/base_devices/fryer_base.py @@ -1,11 +1,24 @@ -"""Air Purifier Base Class.""" +"""Air Fryer Base Class.""" from __future__ import annotations import logging +from datetime import datetime, timezone from typing import TYPE_CHECKING from pyvesync.base_devices.vesyncbasedevice import DeviceState, VeSyncBaseDevice +from pyvesync.const import ( + AIRFRYER_PID_MAP, + COOK_STATUSES, + PREHEAT_STATUSES, + RESUMABLE_STATUSES, + RUNNING_STATUSES, + AirFryerCookStatus, + AirFryerFeatures, + AirFryerPresetRecipe, + TemperatureUnits, + TimeUnits, +) if TYPE_CHECKING: from pyvesync import VeSync @@ -19,11 +32,63 @@ 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. + Each FryerState instance represents a single cooking chamber. For single + chamber fryers, ``state`` is set to ``state_chamber_1``. For dual chamber + fryers, ``state_chamber_1`` and ``state_chamber_2`` are used independently. + + Time units are in SECONDS. They are automatically converted from the API + response via the device's ``convert_time_for_state`` method. The + ``last_timestamp`` attribute is set when the fryer enters a running state + (cooking or heating) and is used to calculate remaining time. + + Args: + device (VeSyncFryer): The device object. + details (ResponseDeviceDetailsModel): The device details. + feature_map (AirFryerMap): The feature map for the device. + + Attributes: + active_time (int): Active time of device, defaults to None. + connection_status (str): Connection status of device. + device (VeSyncFryer): Device object. + device_status (str): Device status. + features (dict): Features of device. + time_units (TimeUnits): The time units used by the device. + cook_status (AirFryerCookStatus | None): The current cooking status. + cook_mode (str | None): The current cooking mode. + current_temp (int | None): The current temperature of the fryer. + cook_set_temp (int | None): The set cooking temperature. + cook_set_time (int | None): The set cooking time in seconds. + cook_last_time (int | None): The remaining cooking time in seconds. + last_timestamp (datetime | None): The timestamp of the last status update. + preheat_set_time (int | None): The set preheating time in seconds. + preheat_last_time (int | None): The remaining preheating time in seconds. + recipe (str | None): The current recipe or cooking mode. + ready_start (bool): Whether the fryer is ready to start cooking. + + Note: + Use convenience state methods like ``set_standby``, ``set_cook_stop_state``, + ``set_preheat_stop_state``, and ``set_preheat_resume_state`` to set the + state attributes when changing cooking states. The ``set_state`` method + can be used to set all state attributes at once when updating from the API + response. The ``last_timestamp`` attribute will be automatically updated + when setting a running state (cooking or heating). """ - __slots__ = () + __slots__ = ( + '_cook_status', + 'cook_last_time', + 'cook_mode', + 'cook_set_temp', + 'cook_set_time', + 'current_temp', + 'last_timestamp', + 'preheat_last_time', + 'preheat_set_time', + 'ready_start', + 'recipe', + 'sync_chambers', + 'time_units', + ) def __init__( self, @@ -31,23 +96,358 @@ def __init__( details: ResponseDeviceDetailsModel, feature_map: AirFryerMap, ) -> None: - """Initialize FryerState. + """Initialize FryerState.""" + 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.sync_chambers: bool = False + # Cooking state attributes + self._cook_status: AirFryerCookStatus | None = None + self.current_temp: int | None = None + self.cook_mode: str | None = None + self.cook_set_temp: int | None = None + self.cook_set_time: int | None = None + self.cook_last_time: int | None = None + self.preheat_set_time: int | None = None + self.preheat_last_time: int | None = None + self.last_timestamp: datetime | None = None + self.recipe: str | None = None + self.ready_start: bool = False + + @property + def is_in_preheat_mode(self) -> bool: + """Return True if the fryer has preheat feature.""" + return self.cook_status in PREHEAT_STATUSES 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 COOK_STATUSES or ( + self.cook_status == AirFryerCookStatus.PULL_OUT + and self.preheat_last_time is None + ) + + @property + def is_cooking(self) -> bool: + """Return True if the fryer is currently running in cook mode.""" + return self.cook_status == AirFryerCookStatus.COOKING + + @property + def is_preheating(self) -> bool: + """Return True if the fryer is currently running in preheat mode.""" + 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 RESUMABLE_STATUSES + + @property + def cook_status(self) -> AirFryerCookStatus | None: + """Return the current cooking status.""" + return self._cook_status + + @cook_status.setter + def cook_status(self, value: AirFryerCookStatus | None) -> None: + """Set the current cooking status. + + Automatically updates ``last_timestamp`` when setting a running state. Args: - device (VeSyncFryer): The device object. - details (ResponseDeviceDetailsModel): The device details. - feature_map (AirFryerMap): The feature map for the device. + value (AirFryerCookStatus | None): The cooking status to set. + """ + if value in RUNNING_STATUSES: + self.last_timestamp = datetime.now(tz=timezone.utc) + self._cook_status = value + + @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((datetime.now(timezone.utc) - self.last_timestamp).total_seconds()), + ) + 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((datetime.now(timezone.utc) - self.last_timestamp).total_seconds()), + ) + 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. """ - super().__init__(device, details, feature_map) - self.device: VeSyncFryer = device - self.features: list[str] = feature_map.features + 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.recipe = None + self._clear_preheat() + + def set_cook_stop_state(self, cook_set_time: int | None = None) -> None: + """Set the fryer state to cook stopped. + + Args: + cook_set_time (int | None): The cooking time in seconds. + """ + self._clear_preheat() + self.cook_status = AirFryerCookStatus.COOK_STOP + if cook_set_time is not None: + self.cook_set_time = cook_set_time + self.cook_last_time = self.cook_time_remaining + self.last_timestamp = None + + def set_preheat_stop_state(self, preheat_set_time: int | None = None) -> None: + """Set the fryer state to preheat stopped.""" + self.cook_status = AirFryerCookStatus.PREHEAT_STOP + if preheat_set_time is not None: + self.preheat_set_time = self.device.convert_time_for_state(preheat_set_time) + self.preheat_last_time = self.preheat_time_remaining + self.last_timestamp = None + + def set_preheat_resume_state(self, preheat_set_time: int | None = None) -> None: + """Set the fryer state to preheat resumed.""" + self.cook_status = AirFryerCookStatus.HEATING + if preheat_set_time is not None: + self.preheat_set_time = self.device.convert_time_for_state(preheat_set_time) + self.preheat_last_time = self.preheat_time_remaining + self.last_timestamp = datetime.now(timezone.utc) + + def set_cooking_state( + self, *, recipe: str, cook_set_time: int, cook_temp: int, cook_mode: str + ) -> None: + """Set the fryer state to cooking. + + Args: + recipe (str): The recipe name or cooking mode. + cook_set_time (int): The cooking time in device units. + cook_temp (int): The cooking temperature. + cook_mode (str): The cooking mode. + """ + self._clear_preheat() + self.cook_status = AirFryerCookStatus.COOKING + self.recipe = recipe + self.cook_set_time = self.device.convert_time_for_state(cook_set_time) + self.cook_last_time = self.cook_set_time + self.cook_set_temp = cook_temp + self.cook_mode = cook_mode + + def set_preheating_state( + self, + *, + recipe: str, + cook_temp: int, + cook_mode: str, + preheat_set_time: int | None = None, + ) -> None: + """Set the fryer state to preheating. + + Args: + recipe (str): The recipe name or cooking mode. + preheat_set_time (int | None): The preheating time in device units. + cook_temp (int): The cooking temperature. + cook_mode (str): The cooking mode. + """ + self.cook_status = AirFryerCookStatus.HEATING + self.preheat_set_time = self.device.convert_time_for_state( + preheat_set_time or 300 + ) + self.preheat_last_time = self.preheat_set_time + self.cook_set_temp = cook_temp + self.cook_mode = cook_mode + self.recipe = recipe + + def set_state( # noqa: PLR0912, PLR0913, C901 + self, + *, + cook_status: AirFryerCookStatus, + 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_set_time: int | None = None, + preheat_last_time: int | None = None, + current_temp: int | None = None, + recipe: str | None = None, + ) -> None: + """Set the cook state parameters from an API response or action. + + All parameters that are part of the current cook state must be passed. + This should be primarily used when updating from the API response to + ensure all state attributes are updated together. If updating state + from a user action, use the specific state methods like ``set_standby``, + ``set_cook_stop_state``, ``set_cooking_state``, etc. + + Time values are expected in device units and will be converted to + seconds automatically. + + Args: + cook_status (AirFryerCookStatus): The cooking status. + cook_time (int | None): The cooking time in device units. + cook_last_time (int | None): The remaining cooking time in device units. + 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_set_time (int | None): The preheating time in device units. + preheat_last_time (int | None): The remaining preheat time in device units. + current_temp (int | None): The current temperature. + recipe (str | None): The recipe name or cooking mode. + """ + # Stop/standby states delegate to helpers and return early + if cook_status == AirFryerCookStatus.STANDBY: + self.set_standby() + return + + if cook_status == AirFryerCookStatus.COOK_STOP: + self.set_cook_stop_state( + cook_set_time=self.device.convert_time_for_state(cook_time) + if cook_time is not None + else None + ) + return + + if cook_status == AirFryerCookStatus.PREHEAT_STOP: + self.set_preheat_stop_state(preheat_set_time=preheat_set_time) + return + + # Set cook status (setter auto-updates last_timestamp for running statuses) + self.cook_status = cook_status + + # Handle preheat attributes + if preheat_set_time is not None: + self.preheat_set_time = self.device.convert_time_for_state(preheat_set_time) + self.preheat_last_time = ( + self.device.convert_time_for_state(preheat_last_time) + if preheat_last_time is not None + else self.preheat_set_time + ) + elif cook_status not in PREHEAT_STATUSES: + self._clear_preheat() + + # Handle cook time attributes + if cook_time is not None: + self.cook_set_time = self.device.convert_time_for_state(cook_time) + if cook_last_time is not None: + self.cook_last_time = self.device.convert_time_for_state(cook_last_time) + elif cook_status == AirFryerCookStatus.HEATING: + self.cook_last_time = None + + # Set remaining attributes + 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 recipe is not None: + self.recipe = recipe class VeSyncFryer(VeSyncBaseDevice): - """Base class for VeSync Air Fryer devices.""" + """Base class for VeSync Air Fryer devices. + + Args: + details (ResponseDeviceDetailsModel): The device details. + manager (VeSync): The VeSync manager. + feature_map (AirFryerMap): The feature map for the device. - __slots__ = () + Attributes: + state (FryerState): Device state object. For single chamber fryers, + this is set to ``state_chamber_1``. + last_response (ResponseInfo): Last response from API call. + manager (VeSync): Manager object for API calls. + device_name (str): Name of device. + device_image (str): URL for device image. + cid (str): Device ID. + connection_type (str): Connection type of device. + device_type (str): Type of device. + type (str): Type of device. + uuid (str): UUID of device, not always present. + config_module (str): Configuration module of device. + mac_id (str): MAC ID of device. + current_firm_version (str): Current firmware version of device. + latest_firm_version (str | None): Latest firmware version of device. + device_region (str): Region of device. (US, EU, etc.) + pid (str): Product ID of device, pulled by some devices on update. + sub_device_no (int): Sub-device number of device. + product_type (str): Product type of device. + features (dict): Features of device. + status_map (MappingProxyType): Mapping of API status strings to + AirFryerCookStatus enums. + cook_modes (dict[str, str]): The available cooking modes and their API values. + default_preset (AirFryerPresetRecipe): The default preset recipe for the fryer. + min_temp_f (int): The minimum temperature in Fahrenheit. + max_temp_f (int): The maximum temperature in Fahrenheit. + min_temp_c (int): The minimum temperature in Celsius. + max_temp_c (int): The maximum temperature in Celsius. + state_chamber_1 (FryerState): The state object for chamber 1. + state_chamber_2 (FryerState): The state object for chamber 2 (if dual chamber). + sync_chambers (bool): Whether the chambers are synced for cooking. + temperature_interval (int): The available temperature step interval. + time_units (TimeUnits): The time units used by the device (seconds or minutes). + """ + + __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', + 'status_map', + 'sync_chambers', + 'temperature_interval', + 'time_units', + ) def __init__( self, @@ -55,14 +455,194 @@ def __init__( manager: VeSync, feature_map: AirFryerMap, ) -> None: - """Initialize VeSyncFryer. + """Initialize VeSyncFryer.""" + 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.configModule, 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 + self.status_map = feature_map.status_map + + # 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 + + # Set state to primary chamber (chamber 1) for base class compatibility + 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: - details (ResponseDeviceDetailsModel): The device details. - manager (VeSync): The VeSync manager. - feature_map (AirFryerMap): The feature map for the device. + 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_for_api(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 + + def convert_time_for_state(self, time_in_device_units: int) -> int: + """Convert time in device's time units to seconds. + + Args: + time_in_device_units (int): The time in device's time units. + + Returns: + int: The time converted to seconds. + """ + if self.time_units == TimeUnits.MINUTES: + return time_in_device_units * 60 + return time_in_device_units + + 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. Note: - This is a bare class as there is only one supported air fryer model. + See [AirFryerPresetRecipe][pyvesync.const.AirFryerPresetRecipe] for + more details on how to create a preset recipe. """ - super().__init__(details, manager, feature_map) + 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..61aaa98 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,179 @@ 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' + GRILL = 'grill' + 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' + COMPLETED = 'completed' + HEATING = 'heating' + UNKNOWN = 'unknown' + STANDBY = 'standby' + PREHEAT_END = 'preheatEnd' + PREHEAT_STOP = 'preheatStop' + + +PREHEAT_STATUSES = [ + AirFryerCookStatus.PREHEAT_END, + AirFryerCookStatus.PREHEAT_STOP, + AirFryerCookStatus.HEATING, +] + +COOK_STATUSES = [ + AirFryerCookStatus.COOKING, + AirFryerCookStatus.COOK_STOP, + AirFryerCookStatus.COOK_END, +] + +RESUMABLE_STATUSES = [ + AirFryerCookStatus.COOK_STOP, + AirFryerCookStatus.PREHEAT_STOP, +] + +RUNNING_STATUSES = [ + AirFryerCookStatus.COOKING, + AirFryerCookStatus.HEATING, +] + + # Thermostat Constants @@ -833,15 +1067,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..c40fc6d 100644 --- a/src/pyvesync/device_map.py +++ b/src/pyvesync/device_map.py @@ -61,10 +61,15 @@ from dataclasses import dataclass, field from itertools import chain -from types import ModuleType +from types import MappingProxyType, ModuleType from typing import Union from pyvesync.const import ( + AirFryerCookModes, + AirFryerCookStatus, + AirFryerFeatures, + AirFryerPresetRecipe, + AirFryerPresets, BulbFeatures, ColorMode, EnergyIntervals, @@ -86,6 +91,7 @@ ThermostatHoldOptions, ThermostatRoutineTypes, ThermostatWorkModes, + TimeUnits, ) from pyvesync.devices import ( vesyncbulb, @@ -135,8 +141,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 +247,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 +277,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 +308,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 +336,21 @@ 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 = field( + default_factory=lambda: AirFryerPresets.custom + ) + cook_modes: dict[str, str] = field(default_factory=dict) + default_cook_mode: str = AirFryerCookModes.AIRFRY + status_map: MappingProxyType[str, AirFryerCookStatus] = field( + default_factory=lambda: MappingProxyType({}) + ) @dataclass(kw_only=True) @@ -373,6 +389,7 @@ class ThermostatMap(DeviceMapTemplate): ThermostatMap( dev_types=['LTM-A401S-WUS'], class_name='VeSyncAuraThermostat', + features=[], fan_modes=[ ThermostatFanModes.AUTO, ThermostatFanModes.CIRCULATE, @@ -407,6 +424,7 @@ class ThermostatMap(DeviceMapTemplate): ], setup_entry='LTM-A401S-WUS', model_display='LTM-A401S Series', + device_alias='Aura Thermostat', model_name='Aura Thermostat', ) ] @@ -418,6 +436,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 +446,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 +455,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 +465,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 +474,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 +1098,88 @@ 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, + status_map=MappingProxyType( + { + 'heating': AirFryerCookStatus.HEATING, + 'cooking': AirFryerCookStatus.COOKING, + 'cookStop': AirFryerCookStatus.COOK_STOP, + 'heatStop': AirFryerCookStatus.PREHEAT_STOP, + 'heatEnd': AirFryerCookStatus.PREHEAT_END, + 'standby': AirFryerCookStatus.STANDBY, + } + ), + ), + 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), + status_map=MappingProxyType( + { + 'ready': AirFryerCookStatus.COOK_STOP, + 'cooking': AirFryerCookStatus.COOKING, + 'heating': AirFryerCookStatus.HEATING, + 'cookStop': AirFryerCookStatus.COOK_STOP, + 'pullOut': AirFryerCookStatus.PULL_OUT, + 'cookEnd': AirFryerCookStatus.COOK_END, + } + ), + ), + AirFryerMap( + class_name='VeSyncDualAirFryer', + module=vesynckitchen, + dev_types=['CAF-TF101S-AEU'], + setup_entry='CAF-TF101S', + device_alias='Dual Air Fryer', + model_display='CAF-TF101S Series', + model_name='Cosori Dual Air Fryer', + temperature_step_f=5, + features=[AirFryerFeatures.DUAL_CHAMBER], + cook_modes={ + AirFryerCookModes.AIRFRY: 'AirFry', + AirFryerCookModes.BAKE: 'Bake', + AirFryerCookModes.ROAST: 'Roast', + AirFryerCookModes.GRILL: 'Grill', + AirFryerCookModes.DRY: 'Dry', + AirFryerCookModes.REHEAT: 'Reheat', + }, + default_cook_mode=AirFryerCookModes.AIRFRY, + default_preset=AirFryerPresets.air_fry, + time_units=TimeUnits.SECONDS, + temperature_range_f=(130, 450), + temperature_range_c=(55, 240), + status_map=MappingProxyType( + { + 'standby': AirFryerCookStatus.STANDBY, + 'ready': AirFryerCookStatus.COOK_STOP, + 'cooking': AirFryerCookStatus.COOKING, + 'cookStop': AirFryerCookStatus.COOK_STOP, + 'pullOut': AirFryerCookStatus.PULL_OUT, + } + ), + ), ] """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..1c69c35 100644 --- a/src/pyvesync/devices/vesynckitchen.py +++ b/src/pyvesync/devices/vesynckitchen.py @@ -12,322 +12,44 @@ 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 typing import TYPE_CHECKING, TypeVar +from dataclasses import replace +from typing import TYPE_CHECKING from typing_extensions import deprecated -from pyvesync.base_devices import FryerState, VeSyncFryer -from pyvesync.const import AIRFRYER_PID_MAP, ConnectionStatus, DeviceStatus +from pyvesync.base_devices import VeSyncFryer +from pyvesync.const import ( + AIRFRYER_PID_MAP, + AirFryerCookStatus, + AirFryerPresetRecipe, + ConnectionStatus, +) +from pyvesync.models import fryer_models as models +from pyvesync.utils.device_mixins import ( + BYPASS_V1_PATH, + 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 + from pyvesync.base_devices.fryer_base import FryerState from pyvesync.device_map import AirFryerMap from pyvesync.models.vesync_models import ResponseDeviceDetailsModel -T = TypeVar('T') - 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): """Cosori Air Fryer Class. @@ -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,28 @@ class VeSyncAirFryer158(VeSyncFryer): """ __slots__ = ( - 'cook_temps', 'last_update', 'ready_start', 'refresh_interval', ) + request_keys: tuple[str, ...] = ( + 'acceptLanguage', + 'appVersion', + 'phoneBrand', + 'phoneOS', + 'accountID', + 'cid', + 'configModule', + 'debugMode', + 'traceId', + 'timeZone', + 'token', + 'userCountryCode', + 'uuid', + 'pid', + ) + def __init__( self, details: ResponseDeviceDetailsModel, @@ -376,278 +114,711 @@ 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 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( + def _build_158_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, - } + request_model: type[models.Fryer158RequestModel], + update_dict: dict | None = None, + method: str = 'bypass', + ) -> models.Fryer158RequestModel: + """Build API request body for the Bypass V1 endpoint. + + Args: + request_model (type[models.Fryer158RequestModel]): The request model to use. + update_dict (dict | None): Additional keys to add on. + method (str): The method to use in the outer body, defaults to bypass. + + Returns: + models.Fryer158RequestModel: The request body for the Bypass V1 endpoint, + the correct model is determined from the models.Fryer158RequestModel + discriminator. + """ + body = Helpers.get_defaultvalues_attributes(self.request_keys).copy() + body.update(Helpers.get_manager_attributes(self.manager, self.request_keys)) + body.update(Helpers.get_device_attributes(self, self.request_keys)) + body['method'] = method + body.update(update_dict or {}) + return request_model.from_dict(body) + + async def call_158_api( + self, + request_model: type[models.Fryer158RequestModel], + update_dict: dict | None = None, + method: str = 'bypass', + endpoint: str = 'bypass', + ) -> dict | None: + """Send Cosori 158 APIrequest. + + This uses the `_build_158_request` method to send API requests + to the Cosori 158 API. + + Args: + request_model (type[models.Fryer158RequestModel]): The request model to use. + update_dict (dict | None): Additional keys to add on. + method (str): The method to use in the outer body. + endpoint (str | None): The last part of the url path, defaults to + `bypass`, e.g. `/cloud/v1/deviceManaged/bypass`. + + Returns: + bytes: The response from the API request. + """ + request = self._build_158_request(request_model, update_dict, method) + url_path = BYPASS_V1_PATH + endpoint + resp_dict, _ = await self.manager.async_call_api( + url_path, 'post', request, Helpers.req_header_bypass() ) - return body - @property - def temp_unit(self) -> str | None: - """Return temp unit.""" - return self.state.temp_unit + return resp_dict + + 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 + + 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'] = self.status_map[AirFryerCookStatus.COOKING] + 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'] = self.status_map[AirFryerCookStatus.HEATING] + 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 + jsoncmd = {'jsonCmd': cmd} + resp = await self.call_158_api(models.Fryer158RequestModel, update_dict=jsoncmd) + + if not isinstance(resp, dict) or 'result' not in resp: + logger.debug( + 'Invalid response for get_details for %s: %s', self.device_name, resp + ) self.state.connection_status = ConnectionStatus.OFFLINE - return + self.state.set_standby() + return None + + 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 + if return_status.cookStatus not in self.status_map: + logger.warning( + 'Unknown cook status %s for %s', + return_status.cookStatus, + self.device_name, + ) + self.state.set_standby() + return None + return self.state.set_state( + cook_status=self.status_map[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_set_time=return_status.preheatSetTime, + preheat_last_time=return_status.preheatLastTime, + current_temp=return_status.currentTemp, + recipe=return_status.customRecipe, + ) + + 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'}} + elif 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_158_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: + elif 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_158_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.set_preheat_stop_state() + elif self.state.is_in_cook_mode is True: + self.state.set_cook_stop_state() 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_158_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': + + if self.state.is_in_preheat_mode is True: + self.state.set_preheat_resume_state() + elif 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_158_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 + self.state.set_state( + cook_status=cook_status, + cook_time=recipe.cook_time, + cook_last_time=recipe.cook_time, + cook_temp=recipe.target_temp, + cook_mode=recipe.cook_mode, + preheat_set_time=recipe.preheat_time, + preheat_last_time=recipe.preheat_time, + ) + return True + + 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_for_api(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_for_api(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 - 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, + cmd = { + 'cookMode': { + 'mode': self.state.cook_mode, + 'accountId': self.manager.account_id, + 'cookStatus': AirFryerCookStatus.COOKING.value, + 'tempUnit': self.temp_unit.label, + } } + json_cmd = {'jsonCmd': cmd} + resp = await self.call_158_api(models.Fryer158RequestModel, update_dict=json_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 - @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, - } + +class VeSyncTurboBlazeFryer(BypassV2Mixin, VeSyncFryer): + """VeSync TurboBlaze Air Fryer Class.""" + + __slots__ = () + + 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, ) - return cmd - async def _set_cook( + 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] + if resp_model.cookStatus not in self.status_map: + logger.warning( + 'Unknown cook status %s for %s', + resp_model.cookStatus, + self.device_name, + ) + self.state.set_standby() + return + + self.state.set_state( + cook_status=self.status_map[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_set_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 + + 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_set_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, - set_temp: int | None = None, - set_time: int | None = None, - status: str = 'cooking', + cook_time: int, + cook_temp: int, + *, + preheat_time: int | None = None, + chamber: int = 1, ) -> bool: - if set_temp is not None and set_time is not None: - set_cmd = self._cmd_api_dict + del chamber # chamber not used for this air fryer + recipe = replace(self.default_preset) + recipe.cook_time = self.convert_time_for_api(cook_time) + recipe.target_temp = self.round_temperature(cook_temp) + if preheat_time is not None: + recipe.preheat_time = self.convert_time_for_api(preheat_time) + return await self.set_mode_from_recipe(recipe) + + +class VeSyncDualAirFryer(BypassV2Mixin, VeSyncFryer): + """Cosori Dual Air Fryer Class (CAF-TF101S). + + Supports dual-chamber cooking with three operating modes: + - Single chamber (left or right) + - Whole chamber (merged) + - Sync mode (both chambers with identical settings) + + The device has no preheat, pause, or resume API. Pausing is handled + physically by pulling the basket out, which triggers a ``cookStop`` + or ``pullOut`` state. Pushing the basket back in resumes cooking + automatically. + + Args: + details (ResponseDeviceDetailsModel): Device details. + manager (VeSync): Manager class. + feature_map (AirFryerMap): Device feature map. + + Attributes: + state_chamber_1 (FryerState): State for chamber 1 (left) or whole. + state_chamber_2 (FryerState): State for chamber 2 (right). + sync_chambers (bool): Whether chambers are synced for cooking. + """ + + __slots__ = () + + # workChamber API values + _WC_NONE = 0 + _WC_LEFT = 1 + _WC_RIGHT = 2 + _WC_WHOLE = 3 + _WC_SYNC = 4 + _SYNC_TYPE_SYNCED = 2 - set_cmd['cookSetTime'] = set_time - set_cmd['cookSetTemp'] = set_temp + def _get_chamber_state(self, chamber: int) -> FryerState: + """Return the FryerState for the given chamber number. + + Args: + chamber: Chamber number (1=left, 2=right, 3=whole maps to chamber 1). + """ + if chamber == self._WC_RIGHT: + return self.state_chamber_2 + return self.state_chamber_1 + + def _get_work_chamber(self, chamber: int) -> int: + """Return the API workChamber value for the given chamber. + + Args: + chamber: Chamber number (1, 2, or 3). + + Returns: + API workChamber value (1, 2, 3, or 4 for sync). + """ + if self.sync_chambers: + return self._WC_SYNC + return chamber + + def _get_sync_type(self) -> int: + """Return the API syncType value.""" + return self._SYNC_TYPE_SYNCED if self.sync_chambers else 0 + + async def get_details(self) -> None: + """Get device details from the API. + + Polls ``getAirfryerMultiStatus`` and updates state for each chamber. + """ + resp = await self.call_bypassv2_api( + payload_method='getAirfryerMultiStatus', + ) + resp_model = process_bypassv2_result( + self, + logger, + 'get_details', + resp, + models.FryerDualMultiStatusResult, + ) + + if resp_model is None: + self.state_chamber_1.set_standby() + self.state_chamber_2.set_standby() + return + + # Update sync state from API response + self.sync_chambers = resp_model.syncType == self._SYNC_TYPE_SYNCED + self.state_chamber_1.sync_chambers = self.sync_chambers + self.state_chamber_2.sync_chambers = self.sync_chambers + + # Track which chambers were updated + updated_chambers: set[int] = set() + + for status_item in resp_model.statusList: + ch_num = status_item.chamber + updated_chambers.add(ch_num) + chamber_state = self._get_chamber_state(ch_num) + + if ( + status_item.cookStatus not in self.status_map + or status_item.cookStatus == AirFryerCookStatus.STANDBY.value + ): + chamber_state.set_standby() + continue + + chamber_state.set_state( + cook_status=self.status_map[status_item.cookStatus], + cook_time=status_item.cookSetTime, + cook_last_time=status_item.currentRemainingTime, + cook_temp=status_item.cookTemp, + temp_unit=resp_model.tempUnit, + cook_mode=status_item.mode, + recipe=status_item.recipeName or None, + ) + + # Set any chambers not in the response to standby + if ( + self._WC_LEFT not in updated_chambers + and self._WC_WHOLE not in updated_chambers + ): + self.state_chamber_1.set_standby() + if ( + self._WC_RIGHT not in updated_chambers + and self._WC_WHOLE not in updated_chambers + ): + self.state_chamber_2.set_standby() + + async def end(self, chamber: int = 1) -> bool: + """End the current cooking session. + + Args: + chamber: Chamber to end (1=left, 2=right, 3=whole). + If sync mode is active, ends both chambers. + + Returns: + True if the command was successful. + """ + api_chamber = self._get_work_chamber(chamber) + resp = await self.call_bypassv2_api( + payload_method='endCook', + data={'chamber': api_chamber}, + ) + r = Helpers.process_dev_response(logger, 'end', self, resp) + if r is None: + return False + + if self.sync_chambers or api_chamber == self._WC_SYNC: + self.state_chamber_1.set_standby() + self.state_chamber_2.set_standby() + self.sync_chambers = False + elif chamber == self._WC_WHOLE: + self.state_chamber_1.set_standby() 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: + self._get_chamber_state(chamber).set_standby() + return True + + def _build_cook_configs( + self, + recipe: AirFryerPresetRecipe, + chamber: int, + ) -> list[models.FryerDualCookConfig]: + """Build cookConfigs list for startMultiCook request. + + Args: + recipe: The recipe to cook. + chamber: Chamber number (1, 2, or 3 for whole). + + Returns: + List of FryerDualCookConfig for the API request. + """ + config_dict = { + 'cookSetTime': recipe.cook_time, + 'cookTemp': recipe.target_temp, + 'mode': recipe.cook_mode, + 'recipeId': recipe.recipe_id, + 'recipeName': recipe.recipe_name, + 'recipeType': recipe.recipe_type, + } + + if self.sync_chambers: + return [ + models.FryerDualCookConfig.from_dict({**config_dict, 'chamber': 1}), + models.FryerDualCookConfig.from_dict({**config_dict, 'chamber': 2}), + ] + return [ + models.FryerDualCookConfig.from_dict({**config_dict, 'chamber': chamber}), + ] + + async def set_mode_from_recipe( + self, + recipe: AirFryerPresetRecipe, + *, + chamber: int = 1, + ) -> bool: + """Start cooking with a preset recipe. + + Args: + recipe: The preset recipe to use. + chamber: Chamber to cook in (1=left, 2=right, 3=whole). + If ``sync_chambers`` is True, both chambers cook in sync. + + Returns: + True if the command was successful. + """ + cook_configs = self._build_cook_configs(recipe, chamber) + work_chamber = self._get_work_chamber(chamber) + sync_type = self._get_sync_type() + + start_data = models.FryerDualStartCookData.from_dict( + { + 'accountId': self.manager.account_id, + 'cookConfigs': [c.to_dict() for c in cook_configs], + 'readyStart': True, + 'syncType': sync_type, + 'tempUnit': self.temp_unit.code, + 'workChamber': work_chamber, + } + ) + + resp = await self.call_bypassv2_api( + payload_method='startMultiCook', + data=start_data.to_dict(), + ) + r = Helpers.process_dev_response(logger, 'set_mode_from_recipe', self, resp) + if r is None: return False - self.last_update = int(time.time()) - self.state.status_request(json_cmd) - await self.update() + # Update state for affected chambers + chambers = ( + [self.state_chamber_1, self.state_chamber_2] + if self.sync_chambers + else [self._get_chamber_state(chamber)] + ) + for ch_state in chambers: + ch_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, + recipe=recipe.recipe_name, + ) return True + + 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: + """Set cooking mode with manual parameters. + + Args: + cook_time: Cooking time in seconds. + cook_temp: Cooking temperature. + preheat_time: Not used for this device. + cook_mode: Cooking mode string (e.g. 'AirFry'). + chamber: Chamber number (1=left, 2=right, 3=whole). + + Returns: + True if the command was successful. + """ + del preheat_time # not supported by this device + if not self.validate_temperature(cook_temp): + logger.warning('Invalid cook temperature for %s', self.device_name) + return False + cook_temp = self.round_temperature(cook_temp) + recipe = replace(self.default_preset) + recipe.cook_time = cook_time + recipe.target_temp = cook_temp + if cook_mode is not None: + recipe.cook_mode = cook_mode + return await self.set_mode_from_recipe(recipe, chamber=chamber) diff --git a/src/pyvesync/models/fryer_models.py b/src/pyvesync/models/fryer_models.py index bf14725..81c4ae6 100644 --- a/src/pyvesync/models/fryer_models.py +++ b/src/pyvesync/models/fryer_models.py @@ -2,33 +2,317 @@ 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.exceptions import MissingField +from mashumaro.types import Discriminator + +from pyvesync.models.base_models import RequestBaseModel, ResponseBaseModel +from pyvesync.models.bypass_models import ( + BypassV1Result, + BypassV2InnerResult, +) + + +@dataclass +class Fryer158RequestModel(RequestBaseModel): + """Request model for air fryer commands.""" + + acceptLanguage: str + accountID: str + appVersion: str + cid: str + configModule: str + debugMode: bool + method: str + phoneBrand: str + phoneOS: str + traceId: str + timeZone: str + token: str + userCountryCode: str + uuid: str + pid: str = field(default_factory=str) + jsonCmd: dict = field(default_factory=dict) + # deviceId: str = field(default_factory=str) + # configModel: str = field(default_factory=lambda: '') + + @classmethod + def __post_deserialize__(cls, obj: Fryer158RequestModel) -> Fryer158RequestModel: # type: ignore[reportIncompatibleMethodOverride] + """Validate required fields after deserialization.""" + if not obj.pid: + raise MissingField('pid', str, Fryer158RequestModel) + if not obj.jsonCmd: + raise MissingField('jsonCmd', dict, Fryer158RequestModel) + return obj + + # 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 + customRecipe: str | 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.""" + + cookSetTime: int + cookTemp: int + preheatTemp: int = 0 + shakeTime: int = 0 + + +# Dual Air Fryer Models (CAF-TF101S) + + +@dataclass +class FryerDualChamberStatusItem(ResponseBaseModel): + """Status item for a single chamber in dual air fryer status response.""" cookStatus: str + chamber: int + cookSetTime: int = 0 + cookTemp: int = 0 + mode: str = '' + currentRemainingTime: int = 0 + totalTimeRemaining: int = 0 + startTime: int = 0 + recipeType: int = 3 + recipeId: int = 0 + recipeName: str = '' + upc: str = '' + holdTime: int = 0 + + +@dataclass +class FryerDualMultiStatusResult(BypassV2InnerResult): + """Result model for dual air fryer getAirfryerMultiStatus response.""" + + statusList: list[FryerDualChamberStatusItem] + tempUnit: str = 'c' + syncType: int = 0 + workChamber: int = 0 + + +@dataclass +class FryerDualCookConfig(RequestBaseModel): + """Cook configuration for a single chamber in startMultiCook request.""" + + chamber: int + cookSetTime: int + cookTemp: int + mode: str + recipeId: int + recipeName: str + recipeType: int + + +@dataclass +class FryerDualStartCookData(RequestBaseModel): + """Request data for dual air fryer startMultiCook command.""" + + accountId: str + cookConfigs: list[FryerDualCookConfig] + readyStart: bool + syncType: int + tempUnit: str + workChamber: int + + +# 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/api/vesynckitchen/CAF-DC601S.yaml b/src/tests/api/vesynckitchen/CAF-DC601S.yaml new file mode 100644 index 0000000..3d32bab --- /dev/null +++ b/src/tests/api/vesynckitchen/CAF-DC601S.yaml @@ -0,0 +1,26 @@ +update: + headers: + Content-Type: application/json; charset=UTF-8 + User-Agent: okhttp/3.12.1 + json_object: + acceptLanguage: en + accountID: sample_id + appVersion: 5.6.60 + cid: CAF-DC601S-CID + configModel: ConfigModule + configModule: ConfigModule + debugMode: false + deviceId: CAF-DC601S-CID + method: bypassV2 + payload: + data: {} + method: getAirfyerStatus + source: APP + phoneBrand: pyvesync + phoneOS: Android + timeZone: America/New_York + token: sample_tk + traceId: TRACE_ID + userCountryCode: US + method: post + url: /cloud/v2/deviceManaged/bypassV2 diff --git a/src/tests/api/vesynckitchen/CAF-TF101S.yaml b/src/tests/api/vesynckitchen/CAF-TF101S.yaml new file mode 100644 index 0000000..3ea1ad5 --- /dev/null +++ b/src/tests/api/vesynckitchen/CAF-TF101S.yaml @@ -0,0 +1,26 @@ +update: + headers: + Content-Type: application/json; charset=UTF-8 + User-Agent: okhttp/3.12.1 + json_object: + acceptLanguage: en + accountID: sample_id + appVersion: 5.6.60 + cid: CAF-TF101S-CID + configModel: ConfigModule + configModule: ConfigModule + debugMode: false + deviceId: CAF-TF101S-CID + method: bypassV2 + payload: + data: {} + method: getAirfryerMultiStatus + source: APP + phoneBrand: pyvesync + phoneOS: Android + timeZone: America/New_York + token: sample_tk + traceId: TRACE_ID + userCountryCode: US + method: post + url: /cloud/v2/deviceManaged/bypassV2 diff --git a/src/tests/api/vesynckitchen/CS158-AF.yaml b/src/tests/api/vesynckitchen/CS158-AF.yaml new file mode 100644 index 0000000..56ec6b5 --- /dev/null +++ b/src/tests/api/vesynckitchen/CS158-AF.yaml @@ -0,0 +1,24 @@ +update: + headers: + Content-Type: application/json; charset=UTF-8 + User-Agent: okhttp/3.12.1 + json_object: + acceptLanguage: en + accountID: sample_id + appVersion: 5.6.60 + cid: CS158-AF-CID + configModule: WiFi_SKA_AirFryer158_US + debugMode: false + jsonCmd: + getStatus: status + method: bypass + phoneBrand: pyvesync + phoneOS: Android + pid: 2cl8hmafsthl65bd + timeZone: America/New_York + token: sample_tk + traceId: TRACE_ID + userCountryCode: US + uuid: CS158-AF-UUID + method: post + url: /cloud/v1/deviceManaged/bypass diff --git a/src/tests/call_json.py b/src/tests/call_json.py index cb9d6a6..52f31cf 100644 --- a/src/tests/call_json.py +++ b/src/tests/call_json.py @@ -1,5 +1,6 @@ import copy from typing import Any +import call_json_fryers import pyvesync.const as const from defaults import TestDefaults from pyvesync.device_map import DeviceMapTemplate @@ -26,7 +27,8 @@ *call_json_outlets.outlet_modules, *call_json_switches.switch_modules, *call_json_humidifiers.humidifier_modules, - *call_json_purifiers.purifier_modules + *call_json_purifiers.purifier_modules, + *call_json_fryers.air_fryer_modules ] ALL_DEVICE_MAP_DICT: dict[str, DeviceMapTemplate] = { @@ -384,6 +386,8 @@ def device_list_item(cls, module: DeviceMapTemplate): model_dict['deviceType'] = module.dev_types[0] if module.setup_entry == 'ESO15-TB': model_dict['subDeviceNo'] = 1 + if 'CS137-AF/CS158-AF' in module.dev_types: + model_dict['configModule'] = 'WiFi_SKA_AirFryer158_US' return model_dict @classmethod diff --git a/src/tests/call_json_fryers.py b/src/tests/call_json_fryers.py new file mode 100644 index 0000000..1ca53a3 --- /dev/null +++ b/src/tests/call_json_fryers.py @@ -0,0 +1,234 @@ +""" +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 ( + AirFryerPresetRecipe, + TemperatureUnits, + AirFryerCookStatus, +) +from defaults import ( + TestDefaults, + FunctionResponsesV2, + FunctionResponsesV1, + build_bypass_v1_response, + build_bypass_v2_response, +) + +FRYERS = [m.setup_entry for m in air_fryer_modules] +FRYERS_NUM = len(FRYERS) + + +class AirFryerDefaults: + temp_unit = TemperatureUnits.FAHRENHEIT + cook_time_m = 10 + cook_time_s = cook_time_m * 60 + cook_last_time_m = cook_time_m - 2 + cook_last_time_s = cook_last_time_m * 60 + preheat_time_m = 5 + preheat_time_s = preheat_time_m * 60 + cook_temp_f = 350 + cook_temp_c = 175 + cook_mode = "custom" + current_temp_f = 150 + cook_status = AirFryerCookStatus.COOKING + recipe = "Manual" + + +DC601_RECIPE = AirFryerPresetRecipe( + recipe_name="Air Fry", + cook_mode="AirFry", + recipe_id=14, + recipe_type=3, + target_temp=AirFryerDefaults.cook_temp_f, + temp_unit=AirFryerDefaults.temp_unit, + cook_time=AirFryerDefaults.cook_time_s, + preheat_time=AirFryerDefaults.preheat_time_s, +) + +TF101S_RECIPE = AirFryerPresetRecipe( + recipe_name="Air Fry", + cook_mode="AirFry", + recipe_id=14, + recipe_type=3, + target_temp=AirFryerDefaults.cook_temp_f, + temp_unit=AirFryerDefaults.temp_unit, + cook_time=AirFryerDefaults.cook_time_s, +) + + +# Raw detail data for each device (V1 result or V2 inner result) +AIR_FRYER_COOKING_DETAILS: dict[str, dict] = { + "CS158-AF": { + "returnStatus": { + "curentTemp": AirFryerDefaults.current_temp_f, + "cookSetTemp": AirFryerDefaults.cook_temp_f, + "mode": AirFryerDefaults.cook_mode, + "cookSetTime": AirFryerDefaults.cook_time_m, # Minutes + "cookLastTime": AirFryerDefaults.cook_last_time_m, # Minutes + "cookStatus": AirFryerDefaults.cook_status.value, + "tempUnit": AirFryerDefaults.temp_unit.label, + "accountId": TestDefaults.account_id, + "customRecipe": AirFryerDefaults.recipe, + } + }, + "CAF-DC601S": { + "stepArray": [ + { + "cookSetTime": AirFryerDefaults.cook_time_s, # Seconds + "cookTemp": AirFryerDefaults.cook_temp_f, + "mode": DC601_RECIPE.cook_mode, + "cookLastTime": AirFryerDefaults.cook_last_time_s, # Seconds + "shakeTime": 0, + "cookEndTime": 0, + "recipeName": DC601_RECIPE.recipe_name, + "recipeId": DC601_RECIPE.recipe_id, + "recipeType": DC601_RECIPE.recipe_type, + } + ], + "cookMode": "normal", + "tempUnit": AirFryerDefaults.temp_unit.label, + "stepIndex": 0, + "cookStatus": AirFryerDefaults.cook_status.value, + "preheatSetTime": 0, + "preheatLastTime": 0, + "preheatEndTime": 0, + "preheatTemp": 0, + "startTime": 1767318116, + "totalTimeRemaining": 1176, + "currentTemp": 89, + "shakeStatus": 0, + }, + "CAF-TF101S": { + "statusList": [ + { + "cookStatus": AirFryerDefaults.cook_status.value, + "chamber": 1, + "cookSetTime": AirFryerDefaults.cook_time_s, # Seconds + "cookTemp": AirFryerDefaults.cook_temp_f, + "mode": TF101S_RECIPE.cook_mode, + "currentRemainingTime": AirFryerDefaults.cook_last_time_s, + "totalTimeRemaining": AirFryerDefaults.cook_last_time_s, + "startTime": 1767318116, + "recipeType": TF101S_RECIPE.recipe_type, + "recipeId": TF101S_RECIPE.recipe_id, + "recipeName": TF101S_RECIPE.recipe_name, + "upc": "", + "holdTime": 0, + }, + { + "cookStatus": "standby", + "chamber": 2, + "cookSetTime": 0, + "cookTemp": 0, + "mode": "", + "currentRemainingTime": 0, + "totalTimeRemaining": 0, + "startTime": 0, + "recipeType": 3, + "recipeId": 0, + "recipeName": "", + "upc": "", + "holdTime": 0, + }, + ], + "tempUnit": AirFryerDefaults.temp_unit.label, + "syncType": 0, + "workChamber": 1, + }, +} + +AIR_FRYER_STANDBY_DETAILS: dict[str, dict] = { + "CS158-AF": {"returnStatus": {"cookStatus": "standby"}}, + "CAF-DC601S": { + "stepArray": [], + "cookMode": "normal", + "tempUnit": AirFryerDefaults.temp_unit.label, + "stepIndex": 0, + "cookStatus": "standby", + "preheatSetTime": 0, + "preheatLastTime": 0, + "preheatEndTime": 0, + "preheatTemp": 0, + "startTime": 0, + "totalTimeRemaining": 0, + "currentTemp": 43, + "shakeStatus": 0, + }, + "CAF-TF101S": { + "statusList": [ + {"cookStatus": "standby", "chamber": 1}, + {"cookStatus": "standby", "chamber": 2}, + ], + "tempUnit": AirFryerDefaults.temp_unit.label, + "syncType": 0, + "workChamber": 0, + }, +} + + +METHOD_RESPONSES = { + 'CS158-AF': deepcopy(FunctionResponsesV1), + 'CAF-DC601S': deepcopy(FunctionResponsesV2), + 'CAF-TF101S': deepcopy(FunctionResponsesV2), +} + + +DETAILS_RESPONSES_COOKING = { + "CS158-AF": build_bypass_v1_response( + result_dict=deepcopy(AIR_FRYER_COOKING_DETAILS["CS158-AF"]), + ), + "CAF-DC601S": build_bypass_v2_response( + inner_result=deepcopy(AIR_FRYER_COOKING_DETAILS["CAF-DC601S"]), + ), + "CAF-TF101S": build_bypass_v2_response( + inner_result=deepcopy(AIR_FRYER_COOKING_DETAILS["CAF-TF101S"]), + ), +} + + +DETAILS_RESPONSES_STANDBY = { + "CS158-AF": build_bypass_v1_response( + result_dict=deepcopy(AIR_FRYER_STANDBY_DETAILS["CS158-AF"]), + ), + "CAF-DC601S": build_bypass_v2_response( + inner_result=deepcopy(AIR_FRYER_STANDBY_DETAILS["CAF-DC601S"]), + ), + "CAF-TF101S": build_bypass_v2_response( + inner_result=deepcopy(AIR_FRYER_STANDBY_DETAILS["CAF-TF101S"]), + ), +} diff --git a/src/tests/test_fryers.py b/src/tests/test_fryers.py new file mode 100644 index 0000000..2cdb4af --- /dev/null +++ b/src/tests/test_fryers.py @@ -0,0 +1,219 @@ +""" +This tests requests for AIR FRYERS. + +All tests inherit from the TestBase class which contains the fixtures +and methods needed to run the tests. + +The tests are automatically parametrized by `pytest_generate_tests` in +conftest.py. The two methods that are parametrized are `test_details` +and `test_methods`. The class variables are used to build the list of +devices, test methods and arguments. + +The `helpers.call_api` method is patched to return a mock response. +The method, endpoint, headers and json arguments are recorded +in YAML files in the api directory, categorized in folders by +module and files by the class name. + +The default is to record requests that do not exist and compare requests +that already exist. If the API changes, set the overwrite argument to True +in order to overwrite the existing YAML file with the new request. + +See Also +-------- +`utils.TestBase` - Base class for all tests, containing mock objects +`confest.pytest_generate_tests` - Parametrizes tests based on + method names & class attributes +`call_json_fryers` - Contains API responses +""" + +import logging +import pytest +import pyvesync.const as const +from pyvesync.base_devices.fryer_base import VeSyncFryer +from base_test_cases import TestBase +from utils import assert_test, parse_args +import call_json_fryers + + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + + +DETAILS_PARAMS_STANDBY = [ + pytest.param( + const.AirFryerCookStatus.STANDBY, + "CS158-AF", + "update", + call_json_fryers.DETAILS_RESPONSES_STANDBY["CS158-AF"], + id="CS158-AF.update.standby", + ), + pytest.param( + const.AirFryerCookStatus.STANDBY, + "CAF-DC601S", + "update", + call_json_fryers.DETAILS_RESPONSES_STANDBY["CAF-DC601S"], + id="CAF-DC601S.update.standby", + ), + pytest.param( + const.AirFryerCookStatus.STANDBY, + "CAF-TF101S", + "update", + call_json_fryers.DETAILS_RESPONSES_STANDBY["CAF-TF101S"], + id="CAF-TF101S.update.standby", + ), +] + +DETAILS_PARAMS_COOKING = [ + pytest.param( + const.AirFryerCookStatus.COOKING, + "CS158-AF", + "update", + call_json_fryers.DETAILS_RESPONSES_COOKING["CS158-AF"], + id="CS158-AF.update.cooking", + ), + pytest.param( + const.AirFryerCookStatus.COOKING, + "CAF-DC601S", + "update", + call_json_fryers.DETAILS_RESPONSES_COOKING["CAF-DC601S"], + id="CAF-DC601S.update.cooking", + ), + pytest.param( + const.AirFryerCookStatus.COOKING, + "CAF-TF101S", + "update", + call_json_fryers.DETAILS_RESPONSES_COOKING["CAF-TF101S"], + id="CAF-TF101S.update.cooking", + ), +] + + +class TestFryers(TestBase): + """Fryer testing class. + + This class tests Fryer product details and methods. The methods are + parametrized from the class variables using `pytest_generate_tests`. + The call_json_fryers module contains the responses for the API requests. + The device is instantiated from the details provided by + `call_json_fryers.DeviceList.device_list_item()`. Inherits from `utils.TestBase`. + + Instance Attributes + ------------------- + self.manager : VeSync + Instantiated VeSync object + self.mock_api : Mock + Mock with patched `helpers.call_api` method + self.caplog : LogCaptureFixture + Pytest fixture for capturing logs + + Class Variables + --------------- + device : str + Name of product class - fryers + fryers : list + List of setup_entry's for fryers, this variable is named + after the device variable value + base_methods : List[List[str, Dict[str, Any]]] + List of common methods for all devices + device_methods : Dict[List[List[str, Dict[str, Any]]]] + Dictionary of methods specific to device types + + Methods + -------- + test_details() + Test the device details API request and response + test_methods() + Test device methods API request and response + + Examples + -------- + >>> device = 'fryers' + >>> fryers = call_json_fryers.FRYER_MODELS + >>> base_methods = [['turn_on'], ['turn_off'], ['update']] + >>> device_methods = { + 'ESWD16': [['method1'], ['method2', {'kwargs': 'value'}]] + } + + """ + + device = 'fryers' + fryers = call_json_fryers.FRYERS + base_methods = [] # type: ignore + device_methods = { # type: ignore + "CS158-AF": [], + "CAF-DC601S": [], + "CAF-TF101S": [], + } + + @pytest.mark.parametrize( + "cook_status, setup_entry, method, response_dict", + DETAILS_PARAMS_STANDBY + DETAILS_PARAMS_COOKING, + ) + def test_details_fryers(self, cook_status: const.AirFryerCookStatus, setup_entry: str, method: str, response_dict: dict): + """Test the device details API request and response. + + This method is automatically parametrized by `pytest_generate_tests` + based on class variables `device` (name of product class - fryers), + device name (fryers) list of setup_entry's. + + Example: + >>> device = 'fryers' + >>> fryers = call_json_fryers.FRYERS + + See Also + -------- + `utils.TestBase` class docstring + `call_json_fryers` module docstring + + Notes + ------ + The device is instantiated using the `call_json.DeviceList.device_list_item()` + method. The device details contain the default values set in `utils.Defaults` + """ + # Set return value for call_api based on call_json_fan.DETAILS_RESPONSES + return_dict = response_dict + self.mock_api.return_value = (return_dict, 200) + + # Instantiate device from device list return item + fryer_obj = self.get_device("air_fryers", setup_entry) + assert isinstance(fryer_obj, VeSyncFryer) + + method_call = getattr(fryer_obj, method) + self.run_in_loop(method_call) + + # Parse mock_api args tuple from arg, kwargs to kwargs + all_kwargs = parse_args(self.mock_api) + + if cook_status == const.AirFryerCookStatus.STANDBY: + assert fryer_obj.state_chamber_1.cook_status == cook_status + assert fryer_obj.state_chamber_1.cook_set_temp is None + assert fryer_obj.state_chamber_1.cook_set_time is None + assert fryer_obj.state_chamber_1.cook_mode is None + assert fryer_obj.state_chamber_1.current_temp is None + assert fryer_obj.state_chamber_1.preheat_last_time is None + assert fryer_obj.state_chamber_1.cook_last_time is None + assert fryer_obj.state_chamber_1.last_timestamp is None + if const.AirFryerFeatures.DUAL_CHAMBER in fryer_obj.features: + assert fryer_obj.state_chamber_2.cook_status == cook_status + assert fryer_obj.state_chamber_2.cook_set_temp is None + assert fryer_obj.state_chamber_2.cook_set_time is None + assert fryer_obj.state_chamber_2.cook_mode is None + assert fryer_obj.state_chamber_2.current_temp is None + assert fryer_obj.state_chamber_2.preheat_last_time is None + assert fryer_obj.state_chamber_2.cook_last_time is None + assert fryer_obj.state_chamber_2.last_timestamp is None + + elif cook_status == const.AirFryerCookStatus.COOKING: + assert fryer_obj.state_chamber_1.cook_status == cook_status + assert fryer_obj.state_chamber_1.cook_set_temp == call_json_fryers.AirFryerDefaults.cook_temp_f + assert fryer_obj.state_chamber_1.cook_set_time == call_json_fryers.AirFryerDefaults.cook_time_s + assert fryer_obj.state_chamber_1.cook_mode is not None + assert fryer_obj.state_chamber_1.cook_last_time == call_json_fryers.AirFryerDefaults.cook_last_time_s + if const.AirFryerFeatures.DUAL_CHAMBER in fryer_obj.features: + # Chamber 2 should be standby in cooking test data + assert fryer_obj.state_chamber_2.cook_status == const.AirFryerCookStatus.STANDBY + + # Assert request matches recorded request or write new records + assert assert_test( + method_call, all_kwargs, setup_entry, self.write_api, self.overwrite + )