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

Code before:
        # Set state_class based on report_type
        if report_type := type_information.report_type:
            self.state_class = self._REPORT_TYPE_TO_STATE_CLASS.get(report_type)

    def initialize(self, device: CustomerDevice) -> None:
        """Initialize wrapper with device data.

        Called when the entity is added to Home Assistant.
        Initializes the accumulated value from the current device state.
        """
        raw_value = super().read_device_status(device)
        if raw_value is not None:
            self._accumulated_value = float(raw_value)
            _LOGGER.debug(
                "Initialized accumulated value from device for %s: %s",
                self.dpcode,
                self._accumulated_value,
            )

    def process_delta_update(
        self, device: CustomerDevice, dp_timestamps: dict | None = None
    ) -> None:
        """Process a delta update by accumulating the value.

        Called during state updates to accumulate incremental values.
        """
        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,
                current_timestamp,
            )
            return

        raw_value = super().read_device_status(device)
        if raw_value is None:
            return

        delta = float(raw_value)

        if self._accumulated_value is None:
            self._accumulated_value = delta
            _LOGGER.debug(
                "Initialized accumulated value for %s: %s",
                self.dpcode,

Review comment (suggestion):
I don't think this should be a separate method.
I think all the logic should be inside `skip_update`.

Code after:
        """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,
            delta,
            self._accumulated_value,
        )

        self._last_dp_timestamp = current_timestamp
        return False

    def read_device_status(self, device: CustomerDevice) -> float | None:
        """Read device status, returning accumulated value for delta reports."""
        return self._accumulated_value


class DPCodeRawWrapper(DPCodeTypeInformationWrapper[RawTypeInformation]):
    """Wrapper to extract information from a RAW/binary value."""

    _DPTYPE = RawTypeInformation


class DPCodeStringWrapper(DPCodeTypeInformationWrapper[StringTypeInformation]):
    """Wrapper to extract information from a STRING value."""

    _DPTYPE = StringTypeInformation


class DPCodeBitmapBitWrapper(DPCodeWrapper):
    """Simple wrapper for a specific bit in bitmap values."""

    def __init__(self, dpcode: str, mask: int) -> None:
        """Init DPCodeBitmapWrapper."""
        super().__init__(dpcode)
        self._mask = mask

    def read_device_status(self, device: CustomerDevice) -> bool | None: