diff --git a/mytoyota/models/endpoints/trips.py b/mytoyota/models/endpoints/trips.py index f8d929d4..0468eab2 100644 --- a/mytoyota/models/endpoints/trips.py +++ b/mytoyota/models/endpoints/trips.py @@ -1,4 +1,6 @@ """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 @@ -6,6 +8,7 @@ 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 @@ class _SummaryBaseModel(BaseModel): 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 @@ class _HDCModel(BaseModel): 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) diff --git a/mytoyota/models/vehicle.py b/mytoyota/models/vehicle.py index e32fbcf7..51885244 100644 --- a/mytoyota/models/vehicle.py +++ b/mytoyota/models/vehicle.py @@ -1,13 +1,16 @@ """Vehicle model.""" import asyncio -import calendar import copy import json import logging from datetime import date, timedelta from functools import partial +from itertools import groupby +from operator import attrgetter from typing import Any, Dict, List, Optional, Tuple +from arrow import Arrow + from mytoyota.api import Api from mytoyota.models.dashboard import Dashboard from mytoyota.models.endpoints.vehicle_guid import VehicleGuidModel @@ -16,6 +19,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 +236,21 @@ async def get_summary( 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: ------- @@ -256,46 +268,16 @@ async def get_summary( return None # 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._generate_daily_summaries(resp.payload.summary) elif summary_type == SummaryType.WEEKLY: - raise NotImplementedError + return self._generate_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._generate_monthly_summaries(resp.payload.summary, from_date, to_date) elif summary_type == SummaryType.YEARLY: - raise NotImplementedError - - return ret + return self._generate_yearly_summaries(resp.payload.summary, 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,94 @@ def _dump_all(self) -> Dict[str, Any]: dump[name] = json.loads(data.model_dump_json()) return censor_all(copy.deepcopy(dump)) + + def _generate_daily_summaries(self, summary) -> List[Summary]: + summary.sort(key=attrgetter("year", "month")) + return [ + Summary( + histogram.summary, + self._metric, + Arrow(histogram.year, histogram.month, histogram.day).date(), + Arrow(histogram.year, histogram.month, histogram.day).date(), + histogram.hdc, + ) + for month in summary + for histogram in sorted(month.histograms, key=attrgetter("day")) + ] + + def _generate_weekly_summaries(self, summary) -> List[Summary]: + ret: List[Summary] = [] + summary.sort(key=attrgetter("year", "month")) + + # Flatten the list of histograms + histograms = [histogram for month in summary for histogram in month.histograms] + histograms.sort(key=lambda h: date(day=h.day, month=h.month, year=h.year)) + + # Group histograms by week + for _, week_histograms_iter in groupby( + histograms, key=lambda h: Arrow(h.year, h.month, h.day).span("week")[0] + ): + week_histograms = list(week_histograms_iter) + build_hdc = copy.copy(week_histograms[0].hdc) + build_summary = copy.copy(week_histograms[0].summary) + start_date = Arrow( + week_histograms[0].year, week_histograms[0].month, week_histograms[0].day + ) + + for histogram in week_histograms[1:]: + add_with_none(build_hdc, histogram.hdc) + build_summary += histogram.summary + + end_date = Arrow( + week_histograms[-1].year, week_histograms[-1].month, week_histograms[-1].day + ) + ret.append( + Summary(build_summary, self._metric, start_date.date(), end_date.date(), build_hdc) + ) + + return ret + + def _generate_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_start = Arrow(month.year, month.month, 1).date() + month_end = Arrow(month.year, month.month, 1).shift(months=1).shift(days=-1).date() + + ret.append( + Summary( + month.summary, + self._metric, + # The data might not include an entire month so update start and end dates + max(month_start, from_date), + min(month_end, to_date), + month.hdc, + ) + ) + + return ret + + def _generate_yearly_summaries(self, summary, to_date: date) -> List[Summary]: + summary.sort(key=attrgetter("year", "month")) + ret: List[Summary] = [] + build_hdc = copy.copy(summary[0].hdc) + build_summary = copy.copy(summary[0].summary) + start_date = date(day=1, month=summary[0].month, year=summary[0].year) + + for month, next_month in zip(summary, summary[1:] + [None]): + summary_month = date(day=1, month=month.month, year=month.year) + add_with_none(build_hdc, month.hdc) + build_summary += month.summary + + if next_month is None or next_month.year != month.year: + 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)) + if next_month: + start_date = date(day=1, month=next_month.month, year=next_month.year) + build_hdc = copy.copy(next_month.hdc) + build_summary = copy.copy(next_month.summary) + + return ret diff --git a/mytoyota/utils/helpers.py b/mytoyota/utils/helpers.py new file mode 100644 index 00000000..3d3ac990 --- /dev/null +++ b/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 diff --git a/simple_client_example.py b/simple_client_example.py index 80389da6..3f924af4 100644 --- a/simple_client_example.py +++ b/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: diff --git a/tests/test_utils/test_helpers.py b/tests/test_utils/test_helpers.py new file mode 100644 index 00000000..781e003f --- /dev/null +++ b/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)