Set generic type

homeassistant/components/tuya/alarm_control_panel.py

@@ -37,23 +37,23 @@ ALARM: dict[DeviceCategory, tuple[AlarmControlPanelEntityDescription, ...]] = {
 }
 
 
-class _AlarmChangedByWrapper(DPCodeRawWrapper):
+class _AlarmChangedByWrapper(DPCodeRawWrapper[str]):
     """Wrapper for changed_by.
 
     Decode base64 to utf-16be string, but only if alarm has been triggered.
     """
 
-    def read_device_status(self, device: CustomerDevice) -> str | None:  # type: ignore[override]
+    def read_device_status(self, device: CustomerDevice) -> str | None:
         """Read the device status."""
         if (
             device.status.get(DPCode.MASTER_STATE) != "alarm"
-            or (status := super().read_device_status(device)) is None
+            or (status := self._read_dpcode_value(device)) is None
         ):
             return None
         return status.decode("utf-16be")
 
 
-class _AlarmStateWrapper(DPCodeEnumWrapper):
+class _AlarmStateWrapper(DPCodeEnumWrapper[AlarmControlPanelState]):
     """Wrapper for the alarm state of a device.
 
     Handles alarm mode enum values and determines the alarm state,
@@ -84,7 +84,7 @@ class _AlarmStateWrapper(DPCodeEnumWrapper):
             ):
                 return AlarmControlPanelState.TRIGGERED
 
-        if (status := super().read_device_status(device)) is None:
+        if (status := self._read_dpcode_value(device)) is None:
             return None
         return self._STATE_MAPPINGS.get(status)
 
@@ -139,10 +139,10 @@ async def async_setup_entry(
                         action_wrapper=_AlarmActionWrapper(
                             master_mode.dpcode, master_mode
                         ),
-                        changed_by_wrapper=_AlarmChangedByWrapper.find_dpcode(
+                        changed_by_wrapper=_AlarmChangedByWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, DPCode.ALARM_MSG
                         ),
-                        state_wrapper=_AlarmStateWrapper(
+                        state_wrapper=_AlarmStateWrapper(  # type: ignore[arg-type]
                             master_mode.dpcode, master_mode
                         ),
                     )

homeassistant/components/tuya/binary_sensor.py

@@ -376,7 +376,7 @@ BINARY_SENSORS: dict[DeviceCategory, tuple[TuyaBinarySensorEntityDescription, ..
 }
 
 
-class _CustomDPCodeWrapper(DPCodeWrapper):
+class _CustomDPCodeWrapper(DPCodeWrapper[bool]):
     """Custom DPCode Wrapper to check for values in a set."""
 
     _valid_values: set[bool | float | int | str]

homeassistant/components/tuya/climate.py

@@ -54,18 +54,18 @@ TUYA_HVAC_TO_HA = {
 }
 
 
-class _RoundedIntegerWrapper(DPCodeIntegerWrapper):
+class _RoundedIntegerWrapper(DPCodeIntegerWrapper[int]):
     """An integer that always rounds its value."""
 
     def read_device_status(self, device: CustomerDevice) -> int | None:
         """Read and round the device status."""
-        if (value := super().read_device_status(device)) is None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return round(value)
 
 
 @dataclass(kw_only=True)
-class _SwingModeWrapper(DeviceWrapper):
+class _SwingModeWrapper(DeviceWrapper[str]):
     """Wrapper for managing climate swing mode operations across multiple DPCodes."""
 
     on_off: DPCodeBooleanWrapper | None = None
@@ -158,7 +158,7 @@ def _filter_hvac_mode_mappings(tuya_range: list[str]) -> dict[str, HVACMode | No
     return modes_in_range
 
 
-class _HvacModeWrapper(DPCodeEnumWrapper):
+class _HvacModeWrapper(DPCodeEnumWrapper[HVACMode]):
     """Wrapper for managing climate HVACMode."""
 
     # Modes that do not map to HVAC modes are ignored (they are handled by PresetWrapper)
@@ -173,11 +173,11 @@ class _HvacModeWrapper(DPCodeEnumWrapper):
 
     def read_device_status(self, device: CustomerDevice) -> HVACMode | None:
         """Read the device status."""
-        if (raw := super().read_device_status(device)) not in TUYA_HVAC_TO_HA:
+        if (raw := self._read_dpcode_value(device)) not in TUYA_HVAC_TO_HA:
             return None
         return TUYA_HVAC_TO_HA[raw]
 
-    def _convert_value_to_raw_value(  # type: ignore[override]
+    def _convert_value_to_raw_value(
         self,
         device: CustomerDevice,
         value: HVACMode,
@@ -205,7 +205,7 @@ class _PresetWrapper(DPCodeEnumWrapper):
 
     def read_device_status(self, device: CustomerDevice) -> str | None:
         """Read the device status."""
-        if (raw := super().read_device_status(device)) in TUYA_HVAC_TO_HA:
+        if (raw := self._read_dpcode_value(device)) in TUYA_HVAC_TO_HA:
             return None
         return raw
 
@@ -358,7 +358,7 @@ async def async_setup_entry(
                         device,
                         manager,
                         CLIMATE_DESCRIPTIONS[device.category],
-                        current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
+                        current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, DPCode.HUMIDITY_CURRENT
                         ),
                         current_temperature_wrapper=temperature_wrappers[0],
@@ -367,7 +367,7 @@ async def async_setup_entry(
                             (DPCode.FAN_SPEED_ENUM, DPCode.LEVEL, DPCode.WINDSPEED),
                             prefer_function=True,
                         ),
-                        hvac_mode_wrapper=_HvacModeWrapper.find_dpcode(
+                        hvac_mode_wrapper=_HvacModeWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, DPCode.MODE, prefer_function=True
                         ),
                         preset_wrapper=_PresetWrapper.find_dpcode(
@@ -378,7 +378,7 @@ async def async_setup_entry(
                         switch_wrapper=DPCodeBooleanWrapper.find_dpcode(
                             device, DPCode.SWITCH, prefer_function=True
                         ),
-                        target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
+                        target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, DPCode.HUMIDITY_SET, prefer_function=True
                         ),
                         temperature_unit=temperature_wrappers[2],

homeassistant/components/tuya/cover.py

@@ -35,7 +35,7 @@ from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 
 
-class _DPCodePercentageMappingWrapper(DPCodeIntegerWrapper):
+class _DPCodePercentageMappingWrapper(DPCodeIntegerWrapper[int]):
     """Wrapper for DPCode position values mapping to 0-100 range."""
 
     def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
@@ -47,7 +47,7 @@ class _DPCodePercentageMappingWrapper(DPCodeIntegerWrapper):
         """Check if the position and direction should be reversed."""
         return False
 
-    def read_device_status(self, device: CustomerDevice) -> float | None:
+    def read_device_status(self, device: CustomerDevice) -> int | None:
         if (value := device.status.get(self.dpcode)) is None:
             return None
 
@@ -87,7 +87,7 @@ class _InstructionBooleanWrapper(DPCodeBooleanWrapper):
     options = ["open", "close"]
     _ACTION_MAPPINGS = {"open": True, "close": False}
 
-    def _convert_value_to_raw_value(self, device: CustomerDevice, value: str) -> bool:  # type: ignore[override]
+    def _convert_value_to_raw_value(self, device: CustomerDevice, value: str) -> bool:
         return self._ACTION_MAPPINGS[value]
 
 
@@ -118,12 +118,12 @@ class _IsClosedInvertedWrapper(DPCodeBooleanWrapper):
     """Boolean wrapper for checking if cover is closed (inverted)."""
 
     def read_device_status(self, device: CustomerDevice) -> bool | None:
-        if (value := super().read_device_status(device)) is None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return not value
 
 
-class _IsClosedEnumWrapper(DPCodeEnumWrapper):
+class _IsClosedEnumWrapper(DPCodeEnumWrapper[bool]):
     """Enum wrapper for checking if state is closed."""
 
     _MAPPINGS = {
@@ -133,8 +133,8 @@ class _IsClosedEnumWrapper(DPCodeEnumWrapper):
         "fully_open": False,
     }
 
-    def read_device_status(self, device: CustomerDevice) -> bool | None:  # type: ignore[override]
-        if (value := super().read_device_status(device)) is None:
+    def read_device_status(self, device: CustomerDevice) -> bool | None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return self._MAPPINGS.get(value)
 
@@ -291,19 +291,19 @@ async def async_setup_entry(
                         device,
                         manager,
                         description,
-                        current_position=description.position_wrapper.find_dpcode(
+                        current_position=description.position_wrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.current_position
                         ),
-                        current_state_wrapper=description.current_state_wrapper.find_dpcode(
+                        current_state_wrapper=description.current_state_wrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.current_state
                         ),
                         instruction_wrapper=_get_instruction_wrapper(
                             device, description
                         ),
-                        set_position=description.position_wrapper.find_dpcode(
+                        set_position=description.position_wrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.set_position, prefer_function=True
                         ),
-                        tilt_position=description.position_wrapper.find_dpcode(
+                        tilt_position=description.position_wrapper.find_dpcode(  # type: ignore[arg-type]
                             device,
                             (DPCode.ANGLE_HORIZONTAL, DPCode.ANGLE_VERTICAL),
                             prefer_function=True,

homeassistant/components/tuya/event.py

@@ -29,19 +29,17 @@ from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 
 
-class _EventEnumWrapper(DPCodeEnumWrapper):
+class _EventEnumWrapper(DPCodeEnumWrapper[tuple[str, None]]):
     """Wrapper for event enum DP codes."""
 
-    def read_device_status(  # type: ignore[override]
-        self, device: CustomerDevice
-    ) -> tuple[str, None] | None:
+    def read_device_status(self, device: CustomerDevice) -> tuple[str, None] | None:
         """Return the event details."""
-        if (raw_value := super().read_device_status(device)) is None:
+        if (raw_value := self._read_dpcode_value(device)) is None:
             return None
         return (raw_value, None)
 
 
-class _AlarmMessageWrapper(DPCodeStringWrapper):
+class _AlarmMessageWrapper(DPCodeStringWrapper[tuple[str, dict[str, Any]]]):
     """Wrapper for a STRING message on DPCode.ALARM_MESSAGE."""
 
     def __init__(self, dpcode: str, type_information: Any) -> None:
@@ -49,16 +47,16 @@ class _AlarmMessageWrapper(DPCodeStringWrapper):
         super().__init__(dpcode, type_information)
         self.options = ["triggered"]
 
-    def read_device_status(
+    def read_device_status(  # type: ignore[override]
         self, device: CustomerDevice
     ) -> tuple[str, dict[str, Any]] | None:
         """Return the event attributes for the alarm message."""
-        if (raw_value := super().read_device_status(device)) is None:
+        if (raw_value := self._read_dpcode_value(device)) is None:
             return None
         return ("triggered", {"message": b64decode(raw_value).decode("utf-8")})
 
 
-class _DoorbellPicWrapper(DPCodeRawWrapper):
+class _DoorbellPicWrapper(DPCodeRawWrapper[tuple[str, dict[str, Any]]]):
     """Wrapper for a RAW message on DPCode.DOORBELL_PIC.
 
     It is expected that the RAW data is base64/utf8 encoded URL of the picture.
@@ -69,11 +67,11 @@ class _DoorbellPicWrapper(DPCodeRawWrapper):
         super().__init__(dpcode, type_information)
         self.options = ["triggered"]
 
-    def read_device_status(  # type: ignore[override]
+    def read_device_status(
         self, device: CustomerDevice
     ) -> tuple[str, dict[str, Any]] | None:
         """Return the event attributes for the doorbell picture."""
-        if (status := super().read_device_status(device)) is None:
+        if (status := self._read_dpcode_value(device)) is None:
             return None
         return ("triggered", {"message": status.decode("utf-8")})
 

homeassistant/components/tuya/fan.py

@@ -59,7 +59,7 @@ class _DirectionEnumWrapper(DPCodeEnumWrapper):
 
     def read_device_status(self, device: CustomerDevice) -> str | None:
         """Read the device status and return the direction string."""
-        if (value := super().read_device_status(device)) and value in {
+        if (value := self._read_dpcode_value(device)) and value in {
             DIRECTION_FORWARD,
             DIRECTION_REVERSE,
         }:
@@ -80,12 +80,12 @@ def _has_a_valid_dpcode(device: CustomerDevice) -> bool:
     return any(get_dpcode(device, code) for code in properties_to_check)
 
 
-class _FanSpeedEnumWrapper(DPCodeEnumWrapper):
+class _FanSpeedEnumWrapper(DPCodeEnumWrapper[int]):
     """Wrapper for fan speed DP code (from an enum)."""
 
-    def read_device_status(self, device: CustomerDevice) -> int | None:  # type: ignore[override]
+    def read_device_status(self, device: CustomerDevice) -> int | None:
         """Get the current speed as a percentage."""
-        if (value := super().read_device_status(device)) is None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return ordered_list_item_to_percentage(self.options, value)
 
@@ -94,7 +94,7 @@ class _FanSpeedEnumWrapper(DPCodeEnumWrapper):
         return percentage_to_ordered_list_item(self.options, value)
 
 
-class _FanSpeedIntegerWrapper(DPCodeIntegerWrapper):
+class _FanSpeedIntegerWrapper(DPCodeIntegerWrapper[int]):
     """Wrapper for fan speed DP code (from an integer)."""
 
     def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
@@ -104,7 +104,7 @@ class _FanSpeedIntegerWrapper(DPCodeIntegerWrapper):
 
     def read_device_status(self, device: CustomerDevice) -> int | None:
         """Get the current speed as a percentage."""
-        if (value := super().read_device_status(device)) is None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return round(self._remap_helper.remap_value_to(value))
 
@@ -154,7 +154,7 @@ async def async_setup_entry(
                         oscillate_wrapper=DPCodeBooleanWrapper.find_dpcode(
                             device, _OSCILLATE_DPCODES, prefer_function=True
                         ),
-                        speed_wrapper=_get_speed_wrapper(device),
+                        speed_wrapper=_get_speed_wrapper(device),  # type: ignore[arg-type]
                         switch_wrapper=DPCodeBooleanWrapper.find_dpcode(
                             device, _SWITCH_DPCODES, prefer_function=True
                         ),

homeassistant/components/tuya/humidifier.py

@@ -29,12 +29,12 @@ from .entity import TuyaEntity
 from .util import ActionDPCodeNotFoundError, get_dpcode
 
 
-class _RoundedIntegerWrapper(DPCodeIntegerWrapper):
+class _RoundedIntegerWrapper(DPCodeIntegerWrapper[int]):
     """An integer that always rounds its value."""
 
     def read_device_status(self, device: CustomerDevice) -> int | None:
         """Read and round the device status."""
-        if (value := super().read_device_status(device)) is None:
+        if (value := self._read_dpcode_value(device)) is None:
             return None
         return round(value)
 
@@ -104,7 +104,7 @@ async def async_setup_entry(
                         device,
                         manager,
                         description,
-                        current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
+                        current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.current_humidity
                         ),
                         mode_wrapper=DPCodeEnumWrapper.find_dpcode(
@@ -115,7 +115,7 @@ async def async_setup_entry(
                             description.dpcode or description.key,
                             prefer_function=True,
                         ),
-                        target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
+                        target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.humidity, prefer_function=True
                         ),
                     )

homeassistant/components/tuya/light.py

@@ -41,7 +41,7 @@ from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode, WorkMode
 from .entity import TuyaEntity
 
 
-class _BrightnessWrapper(DPCodeIntegerWrapper):
+class _BrightnessWrapper(DPCodeIntegerWrapper[int]):
     """Wrapper for brightness DP code.
 
     Handles brightness value conversion between device scale and Home Assistant's
@@ -59,7 +59,7 @@ class _BrightnessWrapper(DPCodeIntegerWrapper):
         super().__init__(dpcode, type_information)
         self._remap_helper = RemapHelper.from_type_information(type_information, 0, 255)
 
-    def read_device_status(self, device: CustomerDevice) -> Any | None:
+    def read_device_status(self, device: CustomerDevice) -> int | None:
         """Return the brightness of this light between 0..255."""
         if (brightness := device.status.get(self.dpcode)) is None:
             return None
@@ -123,7 +123,7 @@ class _BrightnessWrapper(DPCodeIntegerWrapper):
         return round(self._remap_helper.remap_value_from(value))
 
 
-class _ColorTempWrapper(DPCodeIntegerWrapper):
+class _ColorTempWrapper(DPCodeIntegerWrapper[int]):
     """Wrapper for color temperature DP code."""
 
     def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
@@ -133,7 +133,7 @@ class _ColorTempWrapper(DPCodeIntegerWrapper):
             type_information, MIN_MIREDS, MAX_MIREDS
         )
 
-    def read_device_status(self, device: CustomerDevice) -> Any | None:
+    def read_device_status(self, device: CustomerDevice) -> int | None:
         """Return the color temperature value in Kelvin."""
         if (temperature := device.status.get(self.dpcode)) is None:
             return None
@@ -167,18 +167,18 @@ DEFAULT_V_TYPE_V2 = RemapHelper(
 )
 
 
-class _ColorDataWrapper(DPCodeJsonWrapper):
+class _ColorDataWrapper(DPCodeJsonWrapper[tuple[float, float, float]]):
     """Wrapper for color data DP code."""
 
     h_type = DEFAULT_H_TYPE
     s_type = DEFAULT_S_TYPE
     v_type = DEFAULT_V_TYPE
 
-    def read_device_status(  # type: ignore[override]
+    def read_device_status(
         self, device: CustomerDevice
     ) -> tuple[float, float, float] | None:
         """Return a tuple (H, S, V) from this color data."""
-        if (status := super().read_device_status(device)) is None:
+        if (status := self._read_dpcode_value(device)) is None:
             return None
         return (
             self.h_type.remap_value_to(status["h"]),
@@ -633,17 +633,17 @@ async def async_setup_entry(
                         manager,
                         description,
                         brightness_wrapper=(
-                            brightness_wrapper := _get_brightness_wrapper(
+                            brightness_wrapper := _get_brightness_wrapper(  # type: ignore[arg-type]
                                 device, description
                             )
                         ),
-                        color_data_wrapper=_get_color_data_wrapper(
+                        color_data_wrapper=_get_color_data_wrapper(  # type: ignore[arg-type]
                             device, description, brightness_wrapper
                         ),
                         color_mode_wrapper=DPCodeEnumWrapper.find_dpcode(
                             device, description.color_mode, prefer_function=True
                         ),
-                        color_temp_wrapper=_ColorTempWrapper.find_dpcode(
+                        color_temp_wrapper=_ColorTempWrapper.find_dpcode(  # type: ignore[arg-type]
                             device, description.color_temp, prefer_function=True
                         ),
                         switch_wrapper=switch_wrapper,

homeassistant/components/tuya/manifest.json

@@ -44,7 +44,7 @@
   "iot_class": "cloud_push",
   "loggers": ["tuya_sharing"],
   "requirements": [
-    "tuya-device-handlers==0.0.10",
+    "tuya-device-handlers==0.0.11",
     "tuya-device-sharing-sdk==0.2.8"
   ]
 }

homeassistant/components/tuya/vacuum.py

@@ -25,7 +25,7 @@ from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 
 
-class _VacuumActivityWrapper(DeviceWrapper):
+class _VacuumActivityWrapper(DeviceWrapper[VacuumActivity]):
     """Wrapper for the state of a device."""
 
     _TUYA_STATUS_TO_HA = {

requirements_all.txt

@@ -3121,7 +3121,7 @@ ttls==1.8.3
 ttn_client==1.2.3
 
 # homeassistant.components.tuya
-tuya-device-handlers==0.0.10
+tuya-device-handlers==0.0.11
 
 # homeassistant.components.tuya
 tuya-device-sharing-sdk==0.2.8

requirements_test_all.txt

@@ -2624,7 +2624,7 @@ ttls==1.8.3
 ttn_client==1.2.3
 
 # homeassistant.components.tuya
-tuya-device-handlers==0.0.10
+tuya-device-handlers==0.0.11
 
 # homeassistant.components.tuya
 tuya-device-sharing-sdk==0.2.8