Add weekly & yearly summaries

mytoyota/models/endpoints/trips.py

@@ -1,11 +1,14 @@
 """Toyota Connected Services API - Trips Models."""
+from __future__ import annotations
+
 from datetime import date, datetime
 from typing import Any, List, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, Field
 
 from mytoyota.models.endpoints.common import StatusModel
+from mytoyota.utils.helpers import add_with_none
 
 
 class _SummaryBaseModel(BaseModel):
@@ -23,6 +26,30 @@
         alias="fuelConsumption", default=None
     )  # Electric cars might not use fuel. Milliliters.
 
+    def __add__(self, other: _SummaryBaseModel):
+        """Add together two SummaryBaseModel's.
+
+        Handles Min/Max/Average fields correctly.
+
+        Args:
+        ----
+        other: _SummaryBaseModel: to be added
+        """
+        if other is not None:
+            self.length += other.length
+            self.duration += other.duration
+            self.duration_idle += other.duration_idle
+            self.countries.extend(x for x in other.countries if x not in self.countries)
+            self.max_speed = max(self.max_speed, other.max_speed)
+            self.average_speed = (self.average_speed + other.average_speed) / 2.0
+            self.length_overspeed += other.length_overspeed
+            self.duration_overspeed += other.duration_overspeed
+            self.length_highway += other.length_highway
+            self.duration_highway += other.duration_highway
+            self.fuel_consumption = add_with_none(self.fuel_consumption, other.fuel_consumption)
+
+        return self
+
 
 class _SummaryModel(_SummaryBaseModel):
     start_lat: float = Field(alias="startLat")
@@ -66,6 +93,27 @@
     power_time: Optional[int] = Field(alias="powerTime", default=None)
     power_dist: Optional[int] = Field(alias="powerDist", default=None)
 
+    def __add__(self, other: _HDCModel):
+        """Add together two HDCModel's.
+
+        Handles Min/Max/Average fields correctly.
+
+        Args:
+        ----
+        other: _SummaryBaseModel: to be added
+        """
+        if other is not None:
+            self.ev_time = add_with_none(self.ev_time, other.ev_time)
+            self.ev_distance = add_with_none(self.ev_distance, other.ev_distance)
+            self.charge_time = add_with_none(self.charge_time, other.charge_time)
+            self.charge_dist = add_with_none(self.charge_dist, other.charge_dist)
+            self.eco_time = add_with_none(self.eco_time, other.eco_time)
+            self.eco_dist = add_with_none(self.eco_dist, other.eco_dist)
+            self.power_time = add_with_none(self.power_time, other.power_time)
+            self.power_dist = add_with_none(self.power_dist, other.power_dist)
+
+        return self
+
 
 class _RouteModel(BaseModel):
     lat: float = Field(repr=False)

mytoyota/models/vehicle.py

@@ -6,6 +6,7 @@
 import logging
 from datetime import date, timedelta
 from functools import partial
+from operator import attrgetter
 from typing import Any, Dict, List, Optional, Tuple
 
 from mytoyota.api import Api
@@ -16,6 +17,7 @@
 from mytoyota.models.nofication import Notification
 from mytoyota.models.summary import Summary, SummaryType
 from mytoyota.models.trips import Trip
+from mytoyota.utils.helpers import add_with_none
 from mytoyota.utils.logs import censor_all
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -232,13 +234,21 @@
         to_date: date,
         summary_type: SummaryType = SummaryType.MONTHLY,
     ) -> Optional[List[Summary]]:
-        """Return a Daily, Monthly or Yearly summary between the provided dates.
+        """Return a Daily, Weekly, Monthly or Yearly summary between the provided dates.
+
+        All but Daily can return a partial time range. For example if the summary_type is weekly
+        and the date ranges selected include partial weeks these partial weeks will be returned.
+        The dates contained in the summary will indicate the range of dates that made up the
+        partial week.
+
+        Note: Weekly and yearly summaries lose a small amount of accuracy due to rounding issues
 
         Args:
         ----
             from_date (date, required): The inclusive from date to report summaries.
             to_date (date, required): The inclusive to date to report summaries.
-            summary_type (???, optional): Daily, Monthly or Yearly summary. Monthly by default.
+            summary_type (SummaryType, optional): Daily, Monthly or Yearly summary.
+                                                  Monthly by default.
 
         Returns:
         -------
@@ -255,47 +265,19 @@
         if resp.payload is None:
             return None
 
+        # Sort payload as code requires data in date order. Sorts the monthly summaries only
+
         # Convert to response
-        ret: List[Summary] = []
         if summary_type == SummaryType.DAILY:
-            for summary in resp.payload.summary:
-                for histogram in summary.histograms:
-                    summary_date = date(
-                        day=histogram.day, month=histogram.month, year=histogram.year
-                    )
-                    ret.append(
-                        Summary(
-                            histogram.summary,
-                            self._metric,
-                            summary_date,
-                            summary_date,
-                            summary.hdc,
-                        )
-                    )
+            return self._daily_summaries(resp.payload.summary)
         elif summary_type == SummaryType.WEEKLY:
-            raise NotImplementedError
+            return self._weekly_summaries(resp.payload.summary)
         elif summary_type == SummaryType.MONTHLY:
-            for summary in resp.payload.summary:
-                summary_from_date = date(day=1, month=summary.month, year=summary.year)
-                summary_to_date = date(
-                    day=calendar.monthrange(summary.year, summary.month)[1],
-                    month=summary.month,
-                    year=summary.year,
-                )
-
-                ret.append(
-                    Summary(
-                        summary.summary,
-                        self._metric,
-                        summary_from_date if summary_from_date > from_date else from_date,
-                        summary_to_date if summary_to_date < to_date else to_date,
-                        summary.hdc,
-                    )
-                )
+            return self._monthly_summaries(resp.payload.summary, from_date, to_date)
         elif summary_type == SummaryType.YEARLY:
-            raise NotImplementedError
-
-        return ret
+            return self._yearly_summaries(resp.payload.summary, from_date, to_date)
+        else:
+            raise AssertionError("No such SummaryType")
 
     async def get_trips(
         self, from_date: date, to_date: date, full_route: bool = False
@@ -365,3 +347,118 @@
             dump[name] = json.loads(data.model_dump_json())
 
         return censor_all(copy.deepcopy(dump))
+
+    def _daily_summaries(self, summary) -> List[Summary]:
+        summary.sort(key=attrgetter("year", "month"))
+
+        ret: List[Summary] = []
+        for month in summary:  # Loop the months
+            # Ensure daily data is in date order
+            for histogram in sorted(month.histograms, key=attrgetter("day")):  # Loop the days
+                summary_date = date(day=histogram.day, month=histogram.month, year=histogram.year)
+                ret.append(
+                    Summary(
+                        histogram.summary,
+                        self._metric,
+                        summary_date,
+                        summary_date,
+                        histogram.hdc,
+                    )
+                )
+
+        return ret
+
+    def _weekly_summaries(self, summary) -> List[Summary]:
+        # To build a weekly summary we need to add together all the daily summaries for the
+        # week. A week starts on a Monday so data is added into build_hdc & build_summary
+        # until the current day is Sunday(i.e. last day of the week). At this point the
+        # build_hdc & build_summary are added to the response and we start again.
+        build_hdc = None
+        build_summary = None
+        start_date = None
+
+        ret: List[Summary] = []
+        summary.sort(key=attrgetter("year", "month"))
+        for month in summary:
+            month.histograms.sort(key=attrgetter("day"))  # Code required days are in date order
+            for histogram in month.histograms:
+                histogram_day = date(day=histogram.day, month=histogram.month, year=histogram.year)
+                if start_date is None:
+                    start_date = histogram_day
+                    build_hdc = copy.copy(histogram.hdc)
+                    build_summary = copy.copy(histogram.summary)
+                else:
+                    add_with_none(build_hdc, histogram.hdc)
+                    build_summary += histogram.summary
+
+                # If current histogram is a Sunday then we have a full week of data so add it
+                # to our response, also need to add to response if ts the last item in the
+                # loops(otherwise partial week is lost)
+                if histogram_day.weekday() == 6 or (
+                    month == summary[-1] and histogram == month.histograms[-1]
+                ):
+                    ret.append(
+                        Summary(
+                            build_summary,
+                            self._metric,
+                            start_date,
+                            histogram_day,
+                            build_hdc,
+                        )
+                    )
+                    start_date = None  # Reset start date so we can start a new week
+
+        return ret
+
+    def _monthly_summaries(self, summary, from_date: date, to_date: date) -> List[Summary]:
+        # Convert all the monthly responses from the payload to a summary response
+
+        ret: List[Summary] = []
+        summary.sort(key=attrgetter("year", "month"))
+        for month in summary:
+            month_from_date = date(day=1, month=month.month, year=month.year)
+            month_to_date = date(
+                day=calendar.monthrange(month.year, month.month)[1],
+                month=month.month,
+                year=month.year,
+            )
+
+            ret.append(
+                Summary(
+                    month.summary,
+                    self._metric,
+                    # The data might not include an entire month so update start and end dates
+                    max(month_from_date, from_date),
+                    min(month_to_date, to_date),
+                    month.hdc,
+                )
+            )
+
+        return ret
+
+    def _yearly_summaries(self, summar, from_date: date, to_date: date) -> List[Summary]:
+        build_hdc = None
+        build_summary = None
+        start_date = None
+
+        ret: List[Summary] = []
+        summar.sort(key=attrgetter("year", "month"))
+        for month in summar:
+            summary_month = date(day=1, month=month.month, year=month.year)
+            if start_date is None or summary_month < from_date:
+                start_date = summary_month
+                build_hdc = copy.copy(month.hdc)
+                build_summary = copy.copy(month.summary)
+            else:
+                add_with_none(build_hdc, month.hdc)
+                build_summary += month.summary
+
+            # If "today" is december then we have reached the end of the year so add it the
+            # response also need to add to response if last item in list to add the last
+            # partial year
+            if summary_month.month == 12 or month == summar[-1]:
+                end_date = min(to_date, date(day=31, month=12, year=summary_month.year))
+                ret.append(Summary(build_summary, self._metric, start_date, end_date, build_hdc))
+                start_date = None
+
+        return ret

mytoyota/utils/helpers.py

@@ -0,0 +1,14 @@
+"""Helper functions used in multiple places."""
+
+
+def add_with_none(this, that):
+    """Add two items.
+
+    First checking if either item is None.
+    """
+    if this is None:
+        return that
+    if that is None:
+        return this
+
+    return this + that

simple_client_example.py

@@ -57,12 +57,22 @@ async def get_information():
         # Notifications
         pp.pprint(f"Notifications: {[[x] for x in car.notifications]}")
         # Summary
+        # pp.pprint(
+        #    f"Summary: {[[x] for x in await car.get_summary(date.today() - timedelta(days=7), date.today(), summary_type=SummaryType.DAILY)]}"  # noqa: E501 # pylint: disable=C0301
+        # )
+        # pp.pprint(
+        #    f"Summary: {[[x] for x in await car.get_summary(date.today() - timedelta(days=7 * 4), date.today(), summary_type=SummaryType.WEEKLY)]}"  # noqa: E501 # pylint: disable=C0301
+        # )
         pp.pprint(
             f"Summary: {[[x] for x in await car.get_summary(date.today() - timedelta(days=6 * 30), date.today(), summary_type=SummaryType.MONTHLY)]}"  # noqa: E501
         )
+        # pp.pprint(
+        #    f"Summary: {[[x] for x in await car.get_summary(date.today() - timedelta(days=365), date.today(), summary_type=SummaryType.YEARLY)]}"  # noqa: E501 # pylint: disable=C0301
+        # )
+
         # Trips
         pp.pprint(
-            f"Trips: f{await car.get_trips(date.today() - timedelta(days=2), date.today(), full_route=True)}"  # noqa: E501
+            f"Trips: f{await car.get_trips(date.today() - timedelta(days=7), date.today(), full_route=True)}"  # noqa: E501
         )
 
         # Dump all the information collected so far:

tests/test_utils/test_helpers.py

@@ -0,0 +1,18 @@
+"""Test Helper Utils."""
+import pytest
+
+from mytoyota.utils.helpers import add_with_none
+
+
+# Parametrized test for happy path with various realistic test values
+@pytest.mark.parametrize(
+    "this, that, result",
+    [
+        pytest.param(1, None, 1),
+        pytest.param(None, 1, 1),
+        pytest.param(1, 1, 2),
+        pytest.param(None, None, None),
+    ],
+)
+def test_is_valid_locale_happy_path(this, that, result):  # noqa: D103
+    assert result == add_with_none(this, that)