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