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