pytoyoda.models.vehicle

  1"""Vehicle model."""
  2
  3import copy
  4import json
  5from collections.abc import Callable
  6from dataclasses import dataclass
  7from datetime import date, timedelta
  8from enum import Enum, auto
  9from functools import partial
 10from itertools import groupby
 11from operator import attrgetter
 12from typing import Any, TypeVar
 13
 14from arrow import Arrow
 15from loguru import logger
 16from pydantic import computed_field
 17
 18from pytoyoda.api import Api
 19from pytoyoda.exceptions import ToyotaApiError
 20from pytoyoda.models.climate import ClimateSettings, ClimateStatus
 21from pytoyoda.models.dashboard import Dashboard
 22from pytoyoda.models.electric_status import ElectricStatus
 23from pytoyoda.models.endpoints.command import CommandType
 24from pytoyoda.models.endpoints.common import StatusModel
 25from pytoyoda.models.endpoints.electric import (
 26    ElectricCommandResponseModel,
 27    NextChargeSettings,
 28)
 29from pytoyoda.models.endpoints.refresh_status import RefreshStatusResponseModel
 30from pytoyoda.models.endpoints.trips import _SummaryItemModel
 31from pytoyoda.models.endpoints.vehicle_guid import VehicleGuidModel
 32from pytoyoda.models.location import Location
 33from pytoyoda.models.lock_status import LockStatus
 34from pytoyoda.models.nofication import Notification
 35from pytoyoda.models.service_history import ServiceHistory
 36from pytoyoda.models.summary import Summary, SummaryType
 37from pytoyoda.models.trips import Trip
 38from pytoyoda.utils.helpers import add_with_none
 39from pytoyoda.utils.log_utils import censor_all
 40from pytoyoda.utils.models import CustomAPIBaseModel
 41
 42T = TypeVar(
 43    "T",
 44    bound=Api | VehicleGuidModel | bool,
 45)
 46
 47
 48class VehicleType(Enum):
 49    """Vehicle types."""
 50
 51    PLUG_IN_HYBRID = auto()
 52    FULL_HYBRID = auto()
 53    ELECTRIC = auto()
 54    FUEL_ONLY = auto()
 55
 56    @classmethod
 57    def from_vehicle_info(cls, info: VehicleGuidModel) -> "VehicleType":
 58        """Determine the vehicle type based on detailed vehicle fuel information.
 59
 60        Args:
 61            info (VehicleGuidModel): Vehicle information model
 62
 63        Returns:
 64            VehicleType: Determined vehicle type
 65
 66        """
 67        try:
 68            if info.fuel_type == "B":
 69                vehicle_type = cls.FULL_HYBRID
 70            elif info.fuel_type == "E":
 71                vehicle_type = cls.ELECTRIC
 72            elif info.fuel_type == "I":
 73                vehicle_type = cls.PLUG_IN_HYBRID
 74            else:
 75                vehicle_type = cls.FUEL_ONLY
 76        except AttributeError:
 77            return cls.FUEL_ONLY
 78        else:
 79            return vehicle_type
 80
 81
 82@dataclass
 83class EndpointDefinition:
 84    """Definition of an API endpoint."""
 85
 86    name: str
 87    capable: bool
 88    function: Callable
 89
 90
 91class Vehicle(CustomAPIBaseModel[type[T]]):
 92    """Vehicle data representation."""
 93
 94    def __init__(
 95        self,
 96        api: Api,
 97        vehicle_info: VehicleGuidModel,
 98        metric: bool = True,  # noqa: FBT001, FBT002
 99        **kwargs: dict,
100    ) -> None:
101        """Initialise the Vehicle data representation."""
102        data = {
103            "api": api,
104            "vehicle_info": vehicle_info,
105            "metric": metric,
106        }
107        super().__init__(data=data, **kwargs)  # type: ignore[reportArgumentType, arg-type]
108        self._api = api
109        self._vehicle_info = vehicle_info
110        self._metric = metric
111        self._endpoint_data: dict[str, Any] = {}
112
113        if self._vehicle_info.vin:
114            self._api_endpoints: list[EndpointDefinition] = [
115                EndpointDefinition(
116                    name="location",
117                    capable=(
118                        getattr(
119                            getattr(self._vehicle_info, "extended_capabilities", False),
120                            "last_parked_capable",
121                            False,
122                        )
123                        or getattr(
124                            getattr(self._vehicle_info, "features", False),
125                            "last_parked",
126                            False,
127                        )
128                    ),
129                    function=partial(
130                        self._api.get_location, vin=self._vehicle_info.vin
131                    ),
132                ),
133                EndpointDefinition(
134                    name="health_status",
135                    capable=True,
136                    function=partial(
137                        self._api.get_vehicle_health_status,
138                        vin=self._vehicle_info.vin,
139                    ),
140                ),
141                EndpointDefinition(
142                    name="electric_status",
143                    capable=getattr(
144                        getattr(self._vehicle_info, "extended_capabilities", False),
145                        "econnect_vehicle_status_capable",
146                        False,
147                    ),
148                    function=partial(
149                        self._api.get_vehicle_electric_status,
150                        vin=self._vehicle_info.vin,
151                    ),
152                ),
153                EndpointDefinition(
154                    name="telemetry",
155                    capable=getattr(
156                        getattr(self._vehicle_info, "extended_capabilities", False),
157                        "telemetry_capable",
158                        False,
159                    ),
160                    function=partial(
161                        self._api.get_telemetry, vin=self._vehicle_info.vin
162                    ),
163                ),
164                EndpointDefinition(
165                    name="notifications",
166                    capable=True,
167                    function=partial(
168                        self._api.get_notifications, vin=self._vehicle_info.vin
169                    ),
170                ),
171                EndpointDefinition(
172                    name="status",
173                    capable=getattr(
174                        getattr(self._vehicle_info, "extended_capabilities", False),
175                        "vehicle_status",
176                        False,
177                    ),
178                    function=partial(
179                        self._api.get_remote_status, vin=self._vehicle_info.vin
180                    ),
181                ),
182                EndpointDefinition(
183                    name="service_history",
184                    capable=getattr(
185                        getattr(self._vehicle_info, "features", False),
186                        "service_history",
187                        False,
188                    ),
189                    function=partial(
190                        self._api.get_service_history, vin=self._vehicle_info.vin
191                    ),
192                ),
193                EndpointDefinition(
194                    name="climate_settings",
195                    capable=getattr(
196                        getattr(self._vehicle_info, "features", False),
197                        "climate_start_engine",
198                        False,
199                    ),
200                    function=partial(
201                        self._api.get_climate_settings, vin=self._vehicle_info.vin
202                    ),
203                ),
204                EndpointDefinition(
205                    name="climate_status",
206                    capable=getattr(
207                        getattr(self._vehicle_info, "features", False),
208                        "climate_start_engine",
209                        False,
210                    ),
211                    function=partial(
212                        self._api.get_climate_status, vin=self._vehicle_info.vin
213                    ),
214                ),
215                EndpointDefinition(
216                    name="trip_history",
217                    capable=True,
218                    function=partial(
219                        self._api.get_trips,
220                        vin=self._vehicle_info.vin,
221                        from_date=(date.today() - timedelta(days=90)),  # noqa: DTZ011
222                        to_date=date.today(),  # noqa: DTZ011
223                        summary=True,
224                        limit=1,
225                        offset=0,
226                        route=False,
227                    ),
228                ),
229            ]
230        else:
231            raise ToyotaApiError(
232                logger.error(
233                    "The VIN (vehicle identification number) "
234                    "required for the end point request could not be determined"
235                )
236            )
237        self._endpoint_collect = [
238            (endpoint.name, endpoint.function)
239            for endpoint in self._api_endpoints
240            if endpoint.capable
241        ]
242
243    async def update(
244        self,
245        skip: list[str] | None = None,
246        only: list[str] | None = None,
247    ) -> None:
248        """Update the data for the vehicle.
249
250        Endpoint functions are awaited sequentially rather than in a single
251        asyncio.gather. Toyota's API gateway appears to rate-limit on bursts
252        of near-simultaneous requests: firing ~10 requests in the same event
253        loop tick reliably trips a 429 with `{"description": "Unauthorized"}`
254        response bodies, while the same requests serialised at poll cadence
255        succeed cleanly. See pytoyoda/ha_toyota#282 for measurement evidence.
256
257        Args:
258            skip: Endpoint names (matching EndpointDefinition.name values
259                like "status", "telemetry", etc.) to skip this cycle.
260                Skipped endpoints retain their previous _endpoint_data
261                entry, so consumers continue to see the last-known value.
262                Used by ha_toyota's smart-refresh strategy to skip
263                /v1/global/remote/status when a separate POST/GET cycle
264                handles it explicitly.
265            only: Inverse of skip - if provided, ONLY these endpoint names
266                will be fetched. Mutually exclusive with skip.
267                Used by ha_toyota's smart-refresh strategy to update just
268                /v1/global/remote/status after a wake POST without
269                re-hitting the other endpoints that are already fresh.
270
271        Returns:
272            None
273
274        Raises:
275            ValueError: If both skip and only are provided.
276
277        """
278        if skip is not None and only is not None:
279            msg = "update(): pass either skip or only, not both"
280            raise ValueError(msg)
281        skip_set = set(skip or [])
282        only_set = set(only) if only is not None else None
283        for name, function in self._endpoint_collect:
284            if only_set is not None and name not in only_set:
285                continue
286            if name in skip_set:
287                continue
288            self._endpoint_data[name] = await function()
289
290    @computed_field  # type: ignore[prop-decorator]
291    @property
292    def vin(self) -> str | None:
293        """Return the vehicles VIN number.
294
295        Returns:
296            Optional[str]: The vehicles VIN number
297
298        """
299        return self._vehicle_info.vin
300
301    @computed_field  # type: ignore[prop-decorator]
302    @property
303    def alias(self) -> str | None:
304        """Vehicle's alias.
305
306        Returns:
307            Optional[str]: Nickname of vehicle
308
309        """
310        return self._vehicle_info.nickname
311
312    @computed_field  # type: ignore[prop-decorator]
313    @property
314    def type(self) -> str | None:
315        """Returns the "type" of vehicle.
316
317        Returns:
318            Optional[str]: "fuel" if only fuel based
319                "mildhybrid" if hybrid
320                "phev" if plugin hybrid
321                "ev" if full electric vehicle
322
323        """
324        vehicle_type = VehicleType.from_vehicle_info(self._vehicle_info)
325        return vehicle_type.name.lower()
326
327    @computed_field  # type: ignore[prop-decorator]
328    @property
329    def dashboard(self) -> Dashboard | None:
330        """Returns the Vehicle dashboard.
331
332        The dashboard consists of items of information you would expect to
333        find on the dashboard. i.e. Fuel Levels.
334
335        Returns:
336            Optional[Dashboard]: A dashboard
337
338        """
339        # Always returns a Dashboard object as we can always get the odometer value
340        return Dashboard(
341            self._endpoint_data.get("telemetry", None),
342            self._endpoint_data.get("electric_status", None),
343            self._endpoint_data.get("health_status", None),
344            self._metric,
345        )
346
347    @computed_field  # type: ignore[prop-decorator]
348    @property
349    def climate_settings(self) -> ClimateSettings | None:
350        """Return the vehicle climate settings.
351
352        Returns:
353            Optional[ClimateSettings]: A climate settings
354
355        """
356        return ClimateSettings(self._endpoint_data.get("climate_settings", None))
357
358    @computed_field  # type: ignore[prop-decorator]
359    @property
360    def climate_status(self) -> ClimateStatus | None:
361        """Return the vehicle climate status.
362
363        Returns:
364            Optional[ClimateStatus]: A climate status
365
366        """
367        return ClimateStatus(self._endpoint_data.get("climate_status", None))
368
369    @computed_field  # type: ignore[prop-decorator]
370    @property
371    def electric_status(self) -> ElectricStatus | None:
372        """Returns the Electric Status of the vehicle.
373
374        Returns:
375            Optional[ElectricStatus]: Electric Status
376
377        """
378        return ElectricStatus(self._endpoint_data.get("electric_status", None))
379
380    async def refresh_electric_realtime_status(self) -> StatusModel:
381        """Force update of electric realtime status.
382
383        This will drain the 12V battery of the vehicle if
384        used to often!
385
386        Returns:
387            StatusModel: A status response for the command.
388
389        """
390        return await self._api.refresh_electric_realtime_status(self.vin)
391
392    async def refresh_status(self) -> RefreshStatusResponseModel:
393        """Wake the vehicle and request a fresh /status cache populate.
394
395        Issues POST /v1/global/remote/refresh-status. Use sparingly:
396        each call uses cellular airtime and a small amount of 12V battery.
397        Returns when the gateway has accepted the wake request, NOT when
398        the cache has actually been populated; the caller should poll
399        /status afterwards (and check occurrence_date advancement) to
400        verify the wake succeeded end-to-end.
401
402        Returns:
403            RefreshStatusResponseModel: payload.return_code "000000"
404                = wake accepted, anything else = vehicle does not
405                support refresh-status (caller should disable further
406                attempts for this VIN).
407
408        """
409        return await self._api.refresh_vehicle_status(self.vin)
410
411    @computed_field  # type: ignore[prop-decorator]
412    @property
413    def location(self) -> Location | None:
414        """Return the vehicles latest reported Location.
415
416        Returns:
417            Optional[Location]: The latest location or None. If None vehicle car
418                does not support providing location information.
419                _Note_ an empty location object can be returned when the Vehicle
420                supports location but none is currently available.
421
422        """
423        return Location(self._endpoint_data.get("location", None))
424
425    @computed_field  # type: ignore[prop-decorator]
426    @property
427    def notifications(self) -> list[Notification] | None:
428        r"""Returns a list of notifications for the vehicle.
429
430        Returns:
431            Optional[list[Notification]]: A list of notifications for the vehicle,
432                or None if not supported.
433
434        """
435        if "notifications" in self._endpoint_data:
436            ret: list[Notification] = []
437            for p in self._endpoint_data["notifications"].payload:
438                ret.extend(Notification(n) for n in p.notifications)
439            return ret
440
441        return None
442
443    @computed_field  # type: ignore[prop-decorator]
444    @property
445    def service_history(self) -> list[ServiceHistory] | None:
446        r"""Returns a list of service history entries for the vehicle.
447
448        Returns:
449            Optional[list[ServiceHistory]]: A list of service history entries
450                for the vehicle, or None if not supported.
451
452        """
453        if "service_history" in self._endpoint_data:
454            ret: list[ServiceHistory] = []
455            payload = self._endpoint_data["service_history"].payload
456            if not payload:
457                return None
458            ret.extend(
459                ServiceHistory(service_history)
460                for service_history in payload.service_histories
461            )
462            return ret
463
464        return None
465
466    def get_latest_service_history(self) -> ServiceHistory | None:
467        r"""Return the latest service history entry for the vehicle.
468
469        Returns:
470            Optional[ServiceHistory]: A service history entry for the vehicle,
471                ordered by date and service_category. None if not supported or unknown.
472
473        """
474        if self.service_history is not None:
475            return max(
476                self.service_history, key=lambda x: (x.service_date, x.service_category)
477            )
478        return None
479
480    @computed_field  # type: ignore[prop-decorator]
481    @property
482    def lock_status(self) -> LockStatus | None:
483        """Returns the latest lock status of Doors & Windows.
484
485        Returns:
486            Optional[LockStatus]: The latest lock status of Doors & Windows,
487                or None if not supported.
488
489        """
490        return LockStatus(self._endpoint_data.get("status", None))
491
492    @computed_field  # type: ignore[prop-decorator]
493    @property
494    def last_trip(self) -> Trip | None:
495        """Returns the Vehicle last trip.
496
497        Returns:
498            Optional[Trip]: The last trip
499
500        """
501        ret = None
502        if "trip_history" in self._endpoint_data:
503            ret = next(iter(self._endpoint_data["trip_history"].payload.trips), None)
504
505        return None if ret is None else Trip(ret, self._metric)
506
507    @computed_field  # type: ignore[prop-decorator]
508    @property
509    def trip_history(self) -> list[Trip] | None:
510        """Returns the Vehicle trips.
511
512        Returns:
513            Optional[list[Trip]]: A list of trips
514
515        """
516        if "trip_history" in self._endpoint_data:
517            ret: list[Trip] = []
518            payload = self._endpoint_data["trip_history"].payload
519            ret.extend(Trip(t, self._metric) for t in payload.trips)
520            return ret
521
522        return None
523
524    async def get_summary(
525        self,
526        from_date: date,
527        to_date: date,
528        summary_type: SummaryType = SummaryType.MONTHLY,
529    ) -> list[Summary]:
530        """Return different summarys between the provided dates.
531
532        All but Daily can return a partial time range. For example
533        if the summary_type is weekly and the date ranges selected
534        include partial weeks these partial weeks will be returned.
535        The dates contained in the summary will indicate the range
536        of dates that made up the partial week.
537
538        Note: Weekly and yearly summaries lose a small amount of
539        accuracy due to rounding issues.
540
541        Args:
542            from_date (date, required): The inclusive from date to report summaries.
543            to_date (date, required): The inclusive to date to report summaries.
544            summary_type (SummaryType, optional): Daily, Monthly or Yearly summary.
545                Monthly by default.
546
547        Returns:
548            list[Summary]: A list of summaries or empty list if not supported.
549
550        """
551        to_date = min(to_date, date.today())  # noqa : DTZ011
552
553        # Summary information is always returned in the first response.
554        # No need to check all the following pages
555        resp = await self._api.get_trips(
556            self.vin, from_date, to_date, summary=True, limit=1, offset=0
557        )
558        if resp.payload is None or len(resp.payload.summary) == 0:
559            return []
560
561        # Convert to response
562        if summary_type == SummaryType.DAILY:
563            return self._generate_daily_summaries(resp.payload.summary)
564        if summary_type == SummaryType.WEEKLY:
565            return self._generate_weekly_summaries(resp.payload.summary)
566        if summary_type == SummaryType.MONTHLY:
567            return self._generate_monthly_summaries(
568                resp.payload.summary, from_date, to_date
569            )
570        if summary_type == SummaryType.YEARLY:
571            return self._generate_yearly_summaries(resp.payload.summary, to_date)
572        msg = "No such SummaryType"
573        raise AssertionError(msg)
574
575    async def get_current_day_summary(self) -> Summary | None:
576        """Return a summary for the current day.
577
578        Returns:
579            Optional[Summary]: A summary or None if not supported.
580
581        """
582        summary = await self.get_summary(
583            from_date=Arrow.now().date(),
584            to_date=Arrow.now().date(),
585            summary_type=SummaryType.DAILY,
586        )
587        min_no_of_summaries_required_for_calculation = 2
588        if len(summary) < min_no_of_summaries_required_for_calculation:
589            logger.info("Not enough summaries for calculation.")
590        return summary[0] if len(summary) > 0 else None
591
592    async def get_current_week_summary(self) -> Summary | None:
593        """Return a summary for the current week.
594
595        Returns:
596            Optional[Summary]: A summary or None if not supported.
597
598        """
599        summary = await self.get_summary(
600            from_date=Arrow.now().floor("week").date(),
601            to_date=Arrow.now().date(),
602            summary_type=SummaryType.WEEKLY,
603        )
604        min_no_of_summaries_required_for_calculation = 2
605        if len(summary) < min_no_of_summaries_required_for_calculation:
606            logger.info("Not enough summaries for calculation.")
607        return summary[0] if len(summary) > 0 else None
608
609    async def get_current_month_summary(self) -> Summary | None:
610        """Return a summary for the current month.
611
612        Returns:
613            Optional[Summary]: A summary or None if not supported.
614
615        """
616        summary = await self.get_summary(
617            from_date=Arrow.now().floor("month").date(),
618            to_date=Arrow.now().date(),
619            summary_type=SummaryType.MONTHLY,
620        )
621        min_no_of_summaries_required_for_calculation = 2
622        if len(summary) < min_no_of_summaries_required_for_calculation:
623            logger.info("Not enough summaries for calculation.")
624        return summary[0] if len(summary) > 0 else None
625
626    async def get_current_year_summary(self) -> Summary | None:
627        """Return a summary for the current year.
628
629        Returns:
630            Optional[Summary]: A summary or None if not supported.
631
632        """
633        summary = await self.get_summary(
634            from_date=Arrow.now().floor("year").date(),
635            to_date=Arrow.now().date(),
636            summary_type=SummaryType.YEARLY,
637        )
638        min_no_of_summaries_required_for_calculation = 2
639        if len(summary) < min_no_of_summaries_required_for_calculation:
640            logger.info("Not enough summaries for calculation.")
641        return summary[0] if len(summary) > 0 else None
642
643    async def get_trips(
644        self,
645        from_date: date,
646        to_date: date,
647        full_route: bool = False,  # noqa : FBT001, FBT002
648    ) -> list[Trip] | None:
649        """Return information on all trips made between the provided dates.
650
651        Args:
652            from_date (date, required): The inclusive from date
653            to_date (date, required): The inclusive to date
654            full_route (bool, optional): Provide the full route
655                                         information for each trip.
656
657        Returns:
658            Optional[list[Trip]]: A list of all trips or None if not supported.
659
660        """
661        ret: list[Trip] = []
662        offset = 0
663        while True:
664            resp = await self._api.get_trips(
665                self.vin,
666                from_date,
667                to_date,
668                summary=False,
669                limit=5,
670                offset=offset,
671                route=full_route,
672            )
673            if resp.payload is None:
674                break
675
676            # Convert to response
677            if resp.payload.trips:
678                ret.extend(Trip(t, self._metric) for t in resp.payload.trips)
679
680            offset = resp.payload.metadata.pagination.next_offset
681            if offset is None:
682                break
683
684        return ret
685
686    async def get_last_trip(self) -> Trip | None:
687        """Return information on the last trip.
688
689        Returns:
690            Optional[Trip]: A trip model or None if not supported.
691
692        """
693        resp = await self._api.get_trips(
694            self.vin,
695            date.today() - timedelta(days=90),  # noqa : DTZ011
696            date.today(),  # noqa : DTZ011
697            summary=False,
698            limit=1,
699            offset=0,
700            route=False,
701        )
702
703        if resp.payload is None:
704            return None
705
706        ret = next(iter(resp.payload.trips), None)
707        return None if ret is None else Trip(ret, self._metric)
708
709    async def refresh_climate_status(self) -> StatusModel:
710        """Force update of climate status.
711
712        Returns:
713            StatusModel: A status response for the command.
714
715        """
716        return await self._api.refresh_climate_status(self.vin)
717
718    async def post_command(self, command: CommandType, beeps: int = 0) -> StatusModel:
719        """Send remote command to the vehicle.
720
721        Args:
722            command (CommandType): The remote command model
723            beeps (int): Amount of beeps for commands that support it
724
725        Returns:
726            StatusModel: A status response for the command.
727
728        """
729        return await self._api.send_command(self.vin, command=command, beeps=beeps)
730
731    async def send_next_charging_command(
732        self, command: NextChargeSettings
733    ) -> ElectricCommandResponseModel:
734        """Send the next command to the vehicle.
735
736        Args:
737            command: NextChargeSettings command to send
738
739        Returns:
740            Model containing status of the command request
741
742        """
743        return await self._api.send_next_charging_command(self.vin, command=command)
744
745    #
746    # More get functionality depending on what we find
747    #
748
749    async def set_alias(
750        self,
751        value: bool,  # noqa : FBT001
752    ) -> bool:
753        """Set the alias for the vehicle.
754
755        Args:
756            value: The alias value to set for the vehicle.
757
758        Returns:
759            bool: Indicator if value is set
760
761        """
762        return value
763
764    #
765    # More set functionality depending on what we find
766    #
767
768    def _dump_all(self) -> dict[str, Any]:
769        """Dump data from all endpoints for debugging and further work."""
770        dump: [str, Any] = {
771            "vehicle_info": json.loads(self._vehicle_info.model_dump_json())
772        }
773        for name, data in self._endpoint_data.items():
774            dump[name] = json.loads(data.model_dump_json())
775
776        return censor_all(copy.deepcopy(dump))
777
778    def _generate_daily_summaries(
779        self, summary: list[_SummaryItemModel]
780    ) -> list[Summary]:
781        summary.sort(key=attrgetter("year", "month"))
782        # Skip histograms with summary=None - a hollow Summary crashes
783        # downstream when sensors read its properties (see #278).
784        return [
785            Summary(
786                histogram.summary,
787                self._metric,
788                Arrow(histogram.year, histogram.month, histogram.day).date(),
789                Arrow(histogram.year, histogram.month, histogram.day).date(),
790                histogram.hdc,
791            )
792            for month in summary
793            for histogram in sorted(month.histograms, key=attrgetter("day"))
794            if histogram.summary is not None
795        ]
796
797    def _generate_weekly_summaries(
798        self, summary: list[_SummaryItemModel]
799    ) -> list[Summary]:
800        ret: list[Summary] = []
801        summary.sort(key=attrgetter("year", "month"))
802
803        # Flatten the list of histograms
804        histograms = [histogram for month in summary for histogram in month.histograms]
805        histograms.sort(key=lambda h: date(day=h.day, month=h.month, year=h.year))
806
807        # Group histograms by week
808        for _, week_histograms_iter in groupby(
809            histograms, key=lambda h: Arrow(h.year, h.month, h.day).span("week")[0]
810        ):
811            week_histograms = list(week_histograms_iter)
812            build_hdc = copy.copy(week_histograms[0].hdc)
813            build_summary = copy.copy(week_histograms[0].summary)
814            start_date = Arrow(
815                week_histograms[0].year,
816                week_histograms[0].month,
817                week_histograms[0].day,
818            )
819
820            for histogram in week_histograms[1:]:
821                # ``add_with_none`` returns the sum, so we must capture it;
822                # without the assignment ``build_hdc`` would stay at the
823                # first histogram's hdc (or ``None`` if that was None).
824                build_hdc = add_with_none(build_hdc, histogram.hdc)
825                # histogram.summary (and the seed build_summary) may be None on
826                # days where the Toyota API returned a partial payload. Seed with
827                # the first non-None summary we see, then accumulate.
828                if histogram.summary is None:
829                    continue
830                if build_summary is None:
831                    build_summary = copy.copy(histogram.summary)
832                else:
833                    build_summary += histogram.summary
834
835            end_date = Arrow(
836                week_histograms[-1].year,
837                week_histograms[-1].month,
838                week_histograms[-1].day,
839            )
840            # Skip weeks where every histogram.summary was None - a hollow
841            # Summary crashes downstream when sensors read its properties.
842            if build_summary is None:
843                continue
844            ret.append(
845                Summary(
846                    build_summary,
847                    self._metric,
848                    start_date.date(),
849                    end_date.date(),
850                    build_hdc,
851                )
852            )
853
854        return ret
855
856    def _generate_monthly_summaries(
857        self, summary: list[_SummaryItemModel], from_date: date, to_date: date
858    ) -> list[Summary]:
859        # Convert all the monthly responses from the payload to a summary response
860        ret: list[Summary] = []
861        summary.sort(key=attrgetter("year", "month"))
862        for month in summary:
863            # Skip months with summary=None - a hollow Summary crashes
864            # downstream when sensors read its properties (see #278).
865            if month.summary is None:
866                continue
867            month_start = Arrow(month.year, month.month, 1).date()
868            month_end = (
869                Arrow(month.year, month.month, 1).shift(months=1).shift(days=-1).date()
870            )
871
872            ret.append(
873                Summary(
874                    month.summary,
875                    self._metric,
876                    # The data might not include an entire month
877                    # so update start and end dates.
878                    max(month_start, from_date),
879                    min(month_end, to_date),
880                    month.hdc,
881                )
882            )
883
884        return ret
885
886    def _generate_yearly_summaries(
887        self, summary: list[_SummaryItemModel], to_date: date
888    ) -> list[Summary]:
889        summary.sort(key=attrgetter("year", "month"))
890        ret: list[Summary] = []
891        build_hdc = copy.copy(summary[0].hdc)
892        build_summary = copy.copy(summary[0].summary)
893        start_date = date(day=1, month=summary[0].month, year=summary[0].year)
894
895        if len(summary) == 1:
896            if build_summary is not None:
897                ret.append(
898                    Summary(build_summary, self._metric, start_date, to_date, build_hdc)
899                )
900        else:
901            for month, next_month in zip(
902                summary[1:], [*summary[2:], None], strict=False
903            ):
904                summary_month = date(day=1, month=month.month, year=month.year)
905                # ``add_with_none`` returns the sum; capture it or ``build_hdc``
906                # stays at the year's first month's hdc.
907                build_hdc = add_with_none(build_hdc, month.hdc)
908                # month.summary (and the seed build_summary) may be None when
909                # the Toyota API returned partial data.
910                if month.summary is not None:
911                    if build_summary is None:
912                        build_summary = copy.copy(month.summary)
913                    else:
914                        build_summary += month.summary
915
916                if next_month is None or next_month.year != month.year:
917                    end_date = min(
918                        to_date, date(day=31, month=12, year=summary_month.year)
919                    )
920                    # Skip years where every month.summary was None - a hollow
921                    # Summary crashes downstream when sensors read its properties.
922                    if build_summary is not None:
923                        ret.append(
924                            Summary(
925                                build_summary,
926                                self._metric,
927                                start_date,
928                                end_date,
929                                build_hdc,
930                            )
931                        )
932                    if next_month:
933                        start_date = date(
934                            day=1, month=next_month.month, year=next_month.year
935                        )
936                        build_hdc = copy.copy(next_month.hdc)
937                        build_summary = copy.copy(next_month.summary)
938
939        return ret