pytoyoda.models.vehicle

  1"""Vehicle model."""
  2
  3import asyncio
  4import copy
  5import json
  6from collections.abc import Callable
  7from dataclasses import dataclass
  8from datetime import date, timedelta
  9from enum import Enum, auto
 10from functools import partial
 11from itertools import groupby
 12from operator import attrgetter
 13from typing import Any, TypeVar
 14
 15from arrow import Arrow
 16from loguru import logger
 17from pydantic import computed_field
 18
 19from pytoyoda.api import Api
 20from pytoyoda.exceptions import ToyotaApiError
 21from pytoyoda.models.climate import ClimateSettings, ClimateStatus
 22from pytoyoda.models.dashboard import Dashboard
 23from pytoyoda.models.electric_status import ElectricStatus
 24from pytoyoda.models.endpoints.command import CommandType
 25from pytoyoda.models.endpoints.common import StatusModel
 26from pytoyoda.models.endpoints.electric import (
 27    ElectricCommandResponseModel,
 28    NextChargeSettings,
 29)
 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(self) -> None:
244        """Update the data for the vehicle.
245
246        This method asynchronously updates the data for the vehicle by
247        calling the endpoint functions in parallel.
248
249        Returns:
250            None
251
252        """
253
254        async def parallel_wrapper(
255            name: str, function: partial
256        ) -> tuple[str, dict[str, Any]]:
257            r = await function()
258            return name, r
259
260        responses = asyncio.gather(
261            *[
262                parallel_wrapper(name, function)
263                for name, function in self._endpoint_collect
264            ]
265        )
266        for name, data in await responses:
267            self._endpoint_data[name] = data
268
269    @computed_field  # type: ignore[prop-decorator]
270    @property
271    def vin(self) -> str | None:
272        """Return the vehicles VIN number.
273
274        Returns:
275            Optional[str]: The vehicles VIN number
276
277        """
278        return self._vehicle_info.vin
279
280    @computed_field  # type: ignore[prop-decorator]
281    @property
282    def alias(self) -> str | None:
283        """Vehicle's alias.
284
285        Returns:
286            Optional[str]: Nickname of vehicle
287
288        """
289        return self._vehicle_info.nickname
290
291    @computed_field  # type: ignore[prop-decorator]
292    @property
293    def type(self) -> str | None:
294        """Returns the "type" of vehicle.
295
296        Returns:
297            Optional[str]: "fuel" if only fuel based
298                "mildhybrid" if hybrid
299                "phev" if plugin hybrid
300                "ev" if full electric vehicle
301
302        """
303        vehicle_type = VehicleType.from_vehicle_info(self._vehicle_info)
304        return vehicle_type.name.lower()
305
306    @computed_field  # type: ignore[prop-decorator]
307    @property
308    def dashboard(self) -> Dashboard | None:
309        """Returns the Vehicle dashboard.
310
311        The dashboard consists of items of information you would expect to
312        find on the dashboard. i.e. Fuel Levels.
313
314        Returns:
315            Optional[Dashboard]: A dashboard
316
317        """
318        # Always returns a Dashboard object as we can always get the odometer value
319        return Dashboard(
320            self._endpoint_data.get("telemetry", None),
321            self._endpoint_data.get("electric_status", None),
322            self._endpoint_data.get("health_status", None),
323            self._metric,
324        )
325
326    @computed_field  # type: ignore[prop-decorator]
327    @property
328    def climate_settings(self) -> ClimateSettings | None:
329        """Return the vehicle climate settings.
330
331        Returns:
332            Optional[ClimateSettings]: A climate settings
333
334        """
335        return ClimateSettings(self._endpoint_data.get("climate_settings", None))
336
337    @computed_field  # type: ignore[prop-decorator]
338    @property
339    def climate_status(self) -> ClimateStatus | None:
340        """Return the vehicle climate status.
341
342        Returns:
343            Optional[ClimateStatus]: A climate status
344
345        """
346        return ClimateStatus(self._endpoint_data.get("climate_status", None))
347
348    @computed_field  # type: ignore[prop-decorator]
349    @property
350    def electric_status(self) -> ElectricStatus | None:
351        """Returns the Electric Status of the vehicle.
352
353        Returns:
354            Optional[ElectricStatus]: Electric Status
355
356        """
357        return ElectricStatus(self._endpoint_data.get("electric_status", None))
358
359    async def refresh_electric_realtime_status(self) -> StatusModel:
360        """Force update of electric realtime status.
361
362        This will drain the 12V battery of the vehicle if
363        used to often!
364
365        Returns:
366            StatusModel: A status response for the command.
367
368        """
369        return await self._api.refresh_electric_realtime_status(self.vin)
370
371    @computed_field  # type: ignore[prop-decorator]
372    @property
373    def location(self) -> Location | None:
374        """Return the vehicles latest reported Location.
375
376        Returns:
377            Optional[Location]: The latest location or None. If None vehicle car
378                does not support providing location information.
379                _Note_ an empty location object can be returned when the Vehicle
380                supports location but none is currently available.
381
382        """
383        return Location(self._endpoint_data.get("location", None))
384
385    @computed_field  # type: ignore[prop-decorator]
386    @property
387    def notifications(self) -> list[Notification] | None:
388        r"""Returns a list of notifications for the vehicle.
389
390        Returns:
391            Optional[list[Notification]]: A list of notifications for the vehicle,
392                or None if not supported.
393
394        """
395        if "notifications" in self._endpoint_data:
396            ret: list[Notification] = []
397            for p in self._endpoint_data["notifications"].payload:
398                ret.extend(Notification(n) for n in p.notifications)
399            return ret
400
401        return None
402
403    @computed_field  # type: ignore[prop-decorator]
404    @property
405    def service_history(self) -> list[ServiceHistory] | None:
406        r"""Returns a list of service history entries for the vehicle.
407
408        Returns:
409            Optional[list[ServiceHistory]]: A list of service history entries
410                for the vehicle, or None if not supported.
411
412        """
413        if "service_history" in self._endpoint_data:
414            ret: list[ServiceHistory] = []
415            payload = self._endpoint_data["service_history"].payload
416            if not payload:
417                return None
418            ret.extend(
419                ServiceHistory(service_history)
420                for service_history in payload.service_histories
421            )
422            return ret
423
424        return None
425
426    def get_latest_service_history(self) -> ServiceHistory | None:
427        r"""Return the latest service history entry for the vehicle.
428
429        Returns:
430            Optional[ServiceHistory]: A service history entry for the vehicle,
431                ordered by date and service_category. None if not supported or unknown.
432
433        """
434        if self.service_history is not None:
435            return max(
436                self.service_history, key=lambda x: (x.service_date, x.service_category)
437            )
438        return None
439
440    @computed_field  # type: ignore[prop-decorator]
441    @property
442    def lock_status(self) -> LockStatus | None:
443        """Returns the latest lock status of Doors & Windows.
444
445        Returns:
446            Optional[LockStatus]: The latest lock status of Doors & Windows,
447                or None if not supported.
448
449        """
450        return LockStatus(self._endpoint_data.get("status", None))
451
452    @computed_field  # type: ignore[prop-decorator]
453    @property
454    def last_trip(self) -> Trip | None:
455        """Returns the Vehicle last trip.
456
457        Returns:
458            Optional[Trip]: The last trip
459
460        """
461        ret = None
462        if "trip_history" in self._endpoint_data:
463            ret = next(iter(self._endpoint_data["trip_history"].payload.trips), None)
464
465        return None if ret is None else Trip(ret, self._metric)
466
467    @computed_field  # type: ignore[prop-decorator]
468    @property
469    def trip_history(self) -> list[Trip] | None:
470        """Returns the Vehicle trips.
471
472        Returns:
473            Optional[list[Trip]]: A list of trips
474
475        """
476        if "trip_history" in self._endpoint_data:
477            ret: list[Trip] = []
478            payload = self._endpoint_data["trip_history"].payload
479            ret.extend(Trip(t, self._metric) for t in payload.trips)
480            return ret
481
482        return None
483
484    async def get_summary(
485        self,
486        from_date: date,
487        to_date: date,
488        summary_type: SummaryType = SummaryType.MONTHLY,
489    ) -> list[Summary]:
490        """Return different summarys between the provided dates.
491
492        All but Daily can return a partial time range. For example
493        if the summary_type is weekly and the date ranges selected
494        include partial weeks these partial weeks will be returned.
495        The dates contained in the summary will indicate the range
496        of dates that made up the partial week.
497
498        Note: Weekly and yearly summaries lose a small amount of
499        accuracy due to rounding issues.
500
501        Args:
502            from_date (date, required): The inclusive from date to report summaries.
503            to_date (date, required): The inclusive to date to report summaries.
504            summary_type (SummaryType, optional): Daily, Monthly or Yearly summary.
505                Monthly by default.
506
507        Returns:
508            list[Summary]: A list of summaries or empty list if not supported.
509
510        """
511        to_date = min(to_date, date.today())  # noqa : DTZ011
512
513        # Summary information is always returned in the first response.
514        # No need to check all the following pages
515        resp = await self._api.get_trips(
516            self.vin, from_date, to_date, summary=True, limit=1, offset=0
517        )
518        if resp.payload is None or len(resp.payload.summary) == 0:
519            return []
520
521        # Convert to response
522        if summary_type == SummaryType.DAILY:
523            return self._generate_daily_summaries(resp.payload.summary)
524        if summary_type == SummaryType.WEEKLY:
525            return self._generate_weekly_summaries(resp.payload.summary)
526        if summary_type == SummaryType.MONTHLY:
527            return self._generate_monthly_summaries(
528                resp.payload.summary, from_date, to_date
529            )
530        if summary_type == SummaryType.YEARLY:
531            return self._generate_yearly_summaries(resp.payload.summary, to_date)
532        msg = "No such SummaryType"
533        raise AssertionError(msg)
534
535    async def get_current_day_summary(self) -> Summary | None:
536        """Return a summary for the current day.
537
538        Returns:
539            Optional[Summary]: A summary or None if not supported.
540
541        """
542        summary = await self.get_summary(
543            from_date=Arrow.now().date(),
544            to_date=Arrow.now().date(),
545            summary_type=SummaryType.DAILY,
546        )
547        min_no_of_summaries_required_for_calculation = 2
548        if len(summary) < min_no_of_summaries_required_for_calculation:
549            logger.info("Not enough summaries for calculation.")
550        return summary[0] if len(summary) > 0 else None
551
552    async def get_current_week_summary(self) -> Summary | None:
553        """Return a summary for the current week.
554
555        Returns:
556            Optional[Summary]: A summary or None if not supported.
557
558        """
559        summary = await self.get_summary(
560            from_date=Arrow.now().floor("week").date(),
561            to_date=Arrow.now().date(),
562            summary_type=SummaryType.WEEKLY,
563        )
564        min_no_of_summaries_required_for_calculation = 2
565        if len(summary) < min_no_of_summaries_required_for_calculation:
566            logger.info("Not enough summaries for calculation.")
567        return summary[0] if len(summary) > 0 else None
568
569    async def get_current_month_summary(self) -> Summary | None:
570        """Return a summary for the current month.
571
572        Returns:
573            Optional[Summary]: A summary or None if not supported.
574
575        """
576        summary = await self.get_summary(
577            from_date=Arrow.now().floor("month").date(),
578            to_date=Arrow.now().date(),
579            summary_type=SummaryType.MONTHLY,
580        )
581        min_no_of_summaries_required_for_calculation = 2
582        if len(summary) < min_no_of_summaries_required_for_calculation:
583            logger.info("Not enough summaries for calculation.")
584        return summary[0] if len(summary) > 0 else None
585
586    async def get_current_year_summary(self) -> Summary | None:
587        """Return a summary for the current year.
588
589        Returns:
590            Optional[Summary]: A summary or None if not supported.
591
592        """
593        summary = await self.get_summary(
594            from_date=Arrow.now().floor("year").date(),
595            to_date=Arrow.now().date(),
596            summary_type=SummaryType.YEARLY,
597        )
598        min_no_of_summaries_required_for_calculation = 2
599        if len(summary) < min_no_of_summaries_required_for_calculation:
600            logger.info("Not enough summaries for calculation.")
601        return summary[0] if len(summary) > 0 else None
602
603    async def get_trips(
604        self,
605        from_date: date,
606        to_date: date,
607        full_route: bool = False,  # noqa : FBT001, FBT002
608    ) -> list[Trip] | None:
609        """Return information on all trips made between the provided dates.
610
611        Args:
612            from_date (date, required): The inclusive from date
613            to_date (date, required): The inclusive to date
614            full_route (bool, optional): Provide the full route
615                                         information for each trip.
616
617        Returns:
618            Optional[list[Trip]]: A list of all trips or None if not supported.
619
620        """
621        ret: list[Trip] = []
622        offset = 0
623        while True:
624            resp = await self._api.get_trips(
625                self.vin,
626                from_date,
627                to_date,
628                summary=False,
629                limit=5,
630                offset=offset,
631                route=full_route,
632            )
633            if resp.payload is None:
634                break
635
636            # Convert to response
637            if resp.payload.trips:
638                ret.extend(Trip(t, self._metric) for t in resp.payload.trips)
639
640            offset = resp.payload.metadata.pagination.next_offset
641            if offset is None:
642                break
643
644        return ret
645
646    async def get_last_trip(self) -> Trip | None:
647        """Return information on the last trip.
648
649        Returns:
650            Optional[Trip]: A trip model or None if not supported.
651
652        """
653        resp = await self._api.get_trips(
654            self.vin,
655            date.today() - timedelta(days=90),  # noqa : DTZ011
656            date.today(),  # noqa : DTZ011
657            summary=False,
658            limit=1,
659            offset=0,
660            route=False,
661        )
662
663        if resp.payload is None:
664            return None
665
666        ret = next(iter(resp.payload.trips), None)
667        return None if ret is None else Trip(ret, self._metric)
668
669    async def refresh_climate_status(self) -> StatusModel:
670        """Force update of climate status.
671
672        Returns:
673            StatusModel: A status response for the command.
674
675        """
676        return await self._api.refresh_climate_status(self.vin)
677
678    async def post_command(self, command: CommandType, beeps: int = 0) -> StatusModel:
679        """Send remote command to the vehicle.
680
681        Args:
682            command (CommandType): The remote command model
683            beeps (int): Amount of beeps for commands that support it
684
685        Returns:
686            StatusModel: A status response for the command.
687
688        """
689        return await self._api.send_command(self.vin, command=command, beeps=beeps)
690
691    async def send_next_charging_command(
692        self, command: NextChargeSettings
693    ) -> ElectricCommandResponseModel:
694        """Send the next command to the vehicle.
695
696        Args:
697            command: NextChargeSettings command to send
698
699        Returns:
700            Model containing status of the command request
701
702        """
703        return await self._api.send_next_charging_command(self.vin, command=command)
704
705    #
706    # More get functionality depending on what we find
707    #
708
709    async def set_alias(
710        self,
711        value: bool,  # noqa : FBT001
712    ) -> bool:
713        """Set the alias for the vehicle.
714
715        Args:
716            value: The alias value to set for the vehicle.
717
718        Returns:
719            bool: Indicator if value is set
720
721        """
722        return value
723
724    #
725    # More set functionality depending on what we find
726    #
727
728    def _dump_all(self) -> dict[str, Any]:
729        """Dump data from all endpoints for debugging and further work."""
730        dump: [str, Any] = {
731            "vehicle_info": json.loads(self._vehicle_info.model_dump_json())
732        }
733        for name, data in self._endpoint_data.items():
734            dump[name] = json.loads(data.model_dump_json())
735
736        return censor_all(copy.deepcopy(dump))
737
738    def _generate_daily_summaries(
739        self, summary: list[_SummaryItemModel]
740    ) -> list[Summary]:
741        summary.sort(key=attrgetter("year", "month"))
742        return [
743            Summary(
744                histogram.summary,
745                self._metric,
746                Arrow(histogram.year, histogram.month, histogram.day).date(),
747                Arrow(histogram.year, histogram.month, histogram.day).date(),
748                histogram.hdc,
749            )
750            for month in summary
751            for histogram in sorted(month.histograms, key=attrgetter("day"))
752        ]
753
754    def _generate_weekly_summaries(
755        self, summary: list[_SummaryItemModel]
756    ) -> list[Summary]:
757        ret: list[Summary] = []
758        summary.sort(key=attrgetter("year", "month"))
759
760        # Flatten the list of histograms
761        histograms = [histogram for month in summary for histogram in month.histograms]
762        histograms.sort(key=lambda h: date(day=h.day, month=h.month, year=h.year))
763
764        # Group histograms by week
765        for _, week_histograms_iter in groupby(
766            histograms, key=lambda h: Arrow(h.year, h.month, h.day).span("week")[0]
767        ):
768            week_histograms = list(week_histograms_iter)
769            build_hdc = copy.copy(week_histograms[0].hdc)
770            build_summary = copy.copy(week_histograms[0].summary)
771            start_date = Arrow(
772                week_histograms[0].year,
773                week_histograms[0].month,
774                week_histograms[0].day,
775            )
776
777            for histogram in week_histograms[1:]:
778                add_with_none(build_hdc, histogram.hdc)
779                build_summary += histogram.summary
780
781            end_date = Arrow(
782                week_histograms[-1].year,
783                week_histograms[-1].month,
784                week_histograms[-1].day,
785            )
786            ret.append(
787                Summary(
788                    build_summary,
789                    self._metric,
790                    start_date.date(),
791                    end_date.date(),
792                    build_hdc,
793                )
794            )
795
796        return ret
797
798    def _generate_monthly_summaries(
799        self, summary: list[_SummaryItemModel], from_date: date, to_date: date
800    ) -> list[Summary]:
801        # Convert all the monthly responses from the payload to a summary response
802        ret: list[Summary] = []
803        summary.sort(key=attrgetter("year", "month"))
804        for month in summary:
805            month_start = Arrow(month.year, month.month, 1).date()
806            month_end = (
807                Arrow(month.year, month.month, 1).shift(months=1).shift(days=-1).date()
808            )
809
810            ret.append(
811                Summary(
812                    month.summary,
813                    self._metric,
814                    # The data might not include an entire month
815                    # so update start and end dates.
816                    max(month_start, from_date),
817                    min(month_end, to_date),
818                    month.hdc,
819                )
820            )
821
822        return ret
823
824    def _generate_yearly_summaries(
825        self, summary: list[_SummaryItemModel], to_date: date
826    ) -> list[Summary]:
827        summary.sort(key=attrgetter("year", "month"))
828        ret: list[Summary] = []
829        build_hdc = copy.copy(summary[0].hdc)
830        build_summary = copy.copy(summary[0].summary)
831        start_date = date(day=1, month=summary[0].month, year=summary[0].year)
832
833        if len(summary) == 1:
834            ret.append(
835                Summary(build_summary, self._metric, start_date, to_date, build_hdc)
836            )
837        else:
838            for month, next_month in zip(
839                summary[1:], [*summary[2:], None], strict=False
840            ):
841                summary_month = date(day=1, month=month.month, year=month.year)
842                add_with_none(build_hdc, month.hdc)
843                build_summary += month.summary
844
845                if next_month is None or next_month.year != month.year:
846                    end_date = min(
847                        to_date, date(day=31, month=12, year=summary_month.year)
848                    )
849                    ret.append(
850                        Summary(
851                            build_summary, self._metric, start_date, end_date, build_hdc
852                        )
853                    )
854                    if next_month:
855                        start_date = date(
856                            day=1, month=next_month.month, year=next_month.year
857                        )
858                        build_hdc = copy.copy(next_month.hdc)
859                        build_summary = copy.copy(next_month.summary)
860
861        return ret