|
1 | 1 | import logging |
2 | | -from typing import Self |
3 | 2 |
|
4 | | -from roborock.data import CleanRecord, CleanSummaryWithDetail |
| 3 | +from roborock.data import CleanRecord, CleanSummaryWithDetail, RoborockBase |
5 | 4 | from roborock.devices.traits.v1 import common |
6 | 5 | from roborock.roborock_typing import RoborockCommand |
7 | 6 | from roborock.util import unpack_list |
8 | 7 |
|
9 | 8 | _LOGGER = logging.getLogger(__name__) |
10 | 9 |
|
11 | 10 |
|
12 | | -class 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) |
| 11 | +class CleanSummaryConverter(common.V1TraitDataConverter): |
| 12 | + """Converter for CleanSummaryWithDetail objects.""" |
29 | 13 |
|
30 | | - @classmethod |
31 | | - def _parse_type_response(cls, response: common.V1ResponseData) -> Self: |
| 14 | + def convert(self, response: common.V1ResponseData) -> RoborockBase: |
32 | 15 | """Parse the response from the device into a CleanSummary.""" |
33 | 16 | if isinstance(response, dict): |
34 | | - return cls.from_dict(response) |
| 17 | + return CleanSummaryWithDetail.from_dict(response) |
35 | 18 | elif isinstance(response, list): |
36 | 19 | clean_time, clean_area, clean_count, records = unpack_list(response, 4) |
37 | | - return cls( |
| 20 | + return CleanSummaryWithDetail( |
38 | 21 | clean_time=clean_time, |
39 | 22 | clean_area=clean_area, |
40 | 23 | clean_count=clean_count, |
41 | 24 | records=records, |
42 | 25 | ) |
43 | 26 | elif isinstance(response, int): |
44 | | - return cls(clean_time=response) |
| 27 | + return CleanSummaryWithDetail(clean_time=response) |
45 | 28 | raise ValueError(f"Unexpected clean summary format: {response!r}") |
46 | 29 |
|
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 | 30 |
|
52 | | - @classmethod |
53 | | - def _parse_clean_record_response(cls, response: common.V1ResponseData) -> CleanRecord: |
| 31 | +class CleanRecordConverter(common.V1TraitDataConverter): |
| 32 | + """Convert server responses to a CleanRecord.""" |
| 33 | + |
| 34 | + def convert(self, response: common.V1ResponseData) -> CleanRecord: |
54 | 35 | """Parse the response from the device into a CleanRecord.""" |
55 | 36 | if isinstance(response, list) and len(response) == 1: |
56 | 37 | response = response[0] |
@@ -81,3 +62,29 @@ def _parse_clean_record_response(cls, response: common.V1ResponseData) -> CleanR |
81 | 62 | begin, end, duration, area = unpack_list(response, 4) |
82 | 63 | return CleanRecord(begin=begin, end=end, duration=duration, area=area) |
83 | 64 | raise ValueError(f"Unexpected clean record format: {response!r}") |
| 65 | + |
| 66 | + |
| 67 | +class 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) |
0 commit comments