Repository: home-assistant/core
File: homeassistant/components/tuya/models.py (Python)

Code before:
        if self.type_information.min <= new_value <= self.type_information.max:
            return new_value
        # Guarded by number validation
        # Safety net in case of future changes
        raise ValueError(
            f"Value `{new_value}` (converted from `{value}`) out of range:"
            f" ({self.type_information.min}-{self.type_information.max})"
        )


class DPCodeDeltaIntegerWrapper(DPCodeIntegerWrapper):
    """Wrapper for integer values with delta report accumulation.

    This wrapper handles sensors that report incremental (delta) values
    instead of cumulative totals. It accumulates the delta values locally
    to provide a running total.
    """

    _accumulated_value: float = 0
    _last_dp_timestamp: int | None = None

    def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
        """Init DPCodeDeltaIntegerWrapper."""
        super().__init__(dpcode, type_information)
        # Delta reports use TOTAL_INCREASING state class
        self.state_class = SensorStateClass.TOTAL_INCREASING

    def skip_update(
        self,
        device: CustomerDevice,
        updated_status_properties: list[str] | None,
        dp_timestamps: dict[str, int] | None,
    ) -> bool:
        """Override skip_update to process delta updates.

        Processes delta accumulation before determining if update should be skipped.
        """
        if super().skip_update(device, updated_status_properties, dp_timestamps):
            return True

        # Process delta update when not skipping
        current_timestamp = dp_timestamps.get(self.dpcode) if dp_timestamps else None

        # Skip duplicate updates with same timestamp
        if (
            current_timestamp is not None
            and current_timestamp == self._last_dp_timestamp
        ):
            _LOGGER.debug(
                "Skipping duplicate update for %s (same timestamp: %s)",
                self.dpcode,

Review comment (refactor):
Can we not move this to `DPCodeIntegerWrapper`?
```python
    if type_information.report_type in {"sum", "minux"}:
        self.state_class = SensorStateClass.TOTAL_INCREASING
```

Code after:
        if self.type_information.min <= new_value <= self.type_information.max:
            return new_value
        # Guarded by number validation
        # Safety net in case of future changes
        raise ValueError(
            f"Value `{new_value}` (converted from `{value}`) out of range:"
            f" ({self.type_information.min}-{self.type_information.max})"
        )


class DPCodeDeltaIntegerWrapper(DPCodeIntegerWrapper):
    """Wrapper for integer values with delta report accumulation.

    This wrapper handles sensors that report incremental (delta) values
    instead of cumulative totals. It accumulates the delta values locally
    to provide a running total.
    """

    _accumulated_value: float = 0
    _last_dp_timestamp: int | None = None

    def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
        """Init DPCodeDeltaIntegerWrapper."""
        super().__init__(dpcode, type_information)
        # Delta reports use TOTAL_INCREASING state class
        self.state_class = SensorStateClass.TOTAL_INCREASING

    def skip_update(
        self,
        device: CustomerDevice,
        updated_status_properties: list[str] | None,
        dp_timestamps: dict[str, int] | None,
    ) -> bool:
        """Override skip_update to process delta updates.

        Processes delta accumulation before determining if update should be skipped.
        """
        if (
            super().skip_update(device, updated_status_properties, dp_timestamps)
            or dp_timestamps is None
            or (current_timestamp := dp_timestamps.get(self.dpcode)) is None
            or current_timestamp == self._last_dp_timestamp
            or (raw_value := super().read_device_status(device)) is None
        ):
            return True

        delta = float(raw_value)
        self._accumulated_value += delta
        _LOGGER.debug(
            "Delta update for %s: +%s, total: %s",
            self.dpcode,