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:
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        last_record_id = self.records[0]
86        self.last_clean_record = await self.get_clean_record(last_record_id)
87
88    async def get_clean_record(self, record_id: int) -> CleanRecord:
89        """Load a specific clean record by ID."""
90        response = await self.rpc_channel.send_command(RoborockCommand.GET_CLEAN_RECORD, params=[record_id])
91        try:
92            return self.clean_record_converter.convert(response)
93        except (TypeError, ValueError) as err:
94            raise RoborockParsingException(
95                trait_name=type(self).__name__,
96                command=RoborockCommand.GET_CLEAN_RECORD,
97                payload=response,
98                inner_error=err,
99            ) from err