roborock.devices.traits.v1.clean_summary

  1import logging
  2
  3from roborock.data import CleanRecord, CleanSummaryWithDetail, RoborockBase
  4from roborock.devices.traits.v1 import common
  5from roborock.exceptions import RoborockParsingException
  6from roborock.roborock_typing import RoborockCommand
  7from roborock.util import unpack_list
  8
  9_LOGGER = logging.getLogger(__name__)
 10
 11
 12class CleanSummaryConverter(common.V1TraitDataConverter):
 13    """Converter for CleanSummaryWithDetail objects."""
 14
 15    def convert(self, response: common.V1ResponseData) -> RoborockBase:
 16        """Parse the response from the device into a CleanSummary."""
 17        if isinstance(response, dict):
 18            return CleanSummaryWithDetail.from_dict(response)
 19        elif isinstance(response, list):
 20            clean_time, clean_area, clean_count, records = unpack_list(response, 4)
 21            return CleanSummaryWithDetail(
 22                clean_time=clean_time,
 23                clean_area=clean_area,
 24                clean_count=clean_count,
 25                records=records,
 26            )
 27        elif isinstance(response, int):
 28            return CleanSummaryWithDetail(clean_time=response)
 29        raise ValueError(f"Unexpected clean summary format: {response!r}")
 30
 31
 32class CleanRecordConverter(common.V1TraitDataConverter):
 33    """Convert server responses to a CleanRecord."""
 34
 35    def convert(self, response: common.V1ResponseData) -> CleanRecord:
 36        """Parse the response from the device into a CleanRecord."""
 37        if isinstance(response, list) and len(response) == 1:
 38            response = response[0]
 39        if isinstance(response, dict):
 40            return CleanRecord.from_dict(response)
 41        if isinstance(response, list):
 42            if isinstance(response[-1], dict):
 43                records = [CleanRecord.from_dict(rec) for rec in response]
 44                final_record = records[-1]
 45                try:
 46                    # This code is semi-presumptuous - so it is put in a try finally to be safe.
 47                    final_record.begin = records[0].begin
 48                    final_record.begin_datetime = records[0].begin_datetime
 49                    final_record.start_type = records[0].start_type
 50                    for rec in records[0:-1]:
 51                        final_record.duration = (final_record.duration or 0) + (rec.duration or 0)
 52                        final_record.area = (final_record.area or 0) + (rec.area or 0)
 53                        final_record.avoid_count = (final_record.avoid_count or 0) + (rec.avoid_count or 0)
 54                        final_record.wash_count = (final_record.wash_count or 0) + (rec.wash_count or 0)
 55                        final_record.square_meter_area = (final_record.square_meter_area or 0) + (
 56                            rec.square_meter_area or 0
 57                        )
 58                    return final_record
 59                except Exception:  # noqa: BLE001
 60                    # Return final record when an exception occurred
 61                    return final_record
 62            # There are still a few unknown variables in this.
 63            begin, end, duration, area = unpack_list(response, 4)
 64            return CleanRecord(begin=begin, end=end, duration=duration, area=area)
 65        raise ValueError(f"Unexpected clean record format: {response!r}")
 66
 67
 68class CleanSummaryTrait(CleanSummaryWithDetail, common.V1TraitMixin):
 69    """Trait for managing the clean summary of Roborock devices."""
 70
 71    command = RoborockCommand.GET_CLEAN_SUMMARY
 72    converter = CleanSummaryConverter()
 73    clean_record_converter = CleanRecordConverter()
 74
 75    async def refresh(self) -> None:
 76        """Refresh the clean summary data and last clean record.
 77
 78        Assumes that the clean summary has already been fetched.
 79        """
 80        await super().refresh()
 81        if not self.records:
 82            _LOGGER.debug("No clean records available in clean summary.")
 83            self.last_clean_record = None
 84            return
 85        # Record IDs are start timestamps; devices may return them in either order.
 86        last_record_id = max(self.records)
 87        self.last_clean_record = await self.get_clean_record(last_record_id)
 88
 89    async def get_clean_record(self, record_id: int) -> CleanRecord:
 90        """Load a specific clean record by ID."""
 91        response = await self.rpc_channel.send_command(RoborockCommand.GET_CLEAN_RECORD, params=[record_id])
 92        try:
 93            return self.clean_record_converter.convert(response)
 94        except (TypeError, ValueError) as err:
 95            raise RoborockParsingException(
 96                trait_name=type(self).__name__,
 97                command=RoborockCommand.GET_CLEAN_RECORD,
 98                payload=response,
 99                inner_error=err,
100            ) from err