roborock.devices.traits.v1.clean_summary

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