pytoyoda.models.vehicle

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