Chore/fix price next day attributes

.pre-commit-config.yaml

@@ -31,7 +31,6 @@ repos:
           - flake8-bugbear>=20.11.1
           - flake8-builtins>=1.5.3
           - flake8-comprehensions>=3.3.1
-          - flake8-import-order>=0.18.1
           - flake8-mutable>=1.2.0
           - flake8-pie>=0.6.1
           - flake8-quotes>=3.2.0

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## [v2.2.3](https://github.com/azogue/aiopvpc/tree/v2.2.3) - Splitt today / tomorrow price sensor attributes (2021-11-04)
+
+[Full Changelog](https://github.com/azogue/aiopvpc/compare/v2.2.3...v2.2.2)
+
+**Changes:**
+
+* Generate different sets of sensor attributes for hourly prices for current day and for the next day (available at evening), so attrs like `price_position` or `price_ratio` don't change for the current day when next-day prices are received.
+
 ## [v2.2.2](https://github.com/azogue/aiopvpc/tree/v2.2.2) - Migrate CI from travis to gh-actions (2021-11-04)
 
 [Full Changelog](https://github.com/azogue/aiopvpc/compare/v2.2.2...v2.2.1)

aiopvpc/prices.py

@@ -0,0 +1,111 @@
+"""ESIOS API handler for HomeAssistant. Hourly price attributes."""
+import zoneinfo
+from datetime import datetime
+from typing import Any, Dict, Tuple
+
+
+def _is_tomorrow_price(ts: datetime, ref: datetime) -> bool:
+    return any(map(lambda x: x[0] > x[1], zip(ts.isocalendar(), ref.isocalendar())))
+
+
+def _split_today_tomorrow_prices(
+    current_prices: Dict[datetime, float],
+    utc_time: datetime,
+    timezone: zoneinfo.ZoneInfo,
+) -> Tuple[Dict[datetime, float], Dict[datetime, float]]:
+    local_time = utc_time.astimezone(timezone)
+    today, tomorrow = {}, {}
+    for ts_utc, price_h in current_prices.items():
+        ts_local = ts_utc.astimezone(timezone)
+        if _is_tomorrow_price(ts_local, local_time):
+            tomorrow[ts_utc] = price_h
+        else:
+            today[ts_utc] = price_h
+    return today, tomorrow
+
+
+def _make_price_tag_attributes(
+    prices: Dict[datetime, float], timezone: zoneinfo.ZoneInfo, tomorrow: bool
+) -> Dict[str, Any]:
+    prefix = "price_next_day_" if tomorrow else "price_"
+    attributes = {}
+    for ts_utc, price_h in prices.items():
+        ts_local = ts_utc.astimezone(timezone)
+        attr_key = f"{prefix}{ts_local.hour:02d}h"
+        if attr_key in attributes:  # DST change with duplicated hour :)
+            attr_key += "_d"
+        attributes[attr_key] = price_h
+    return attributes
+
+
+def _make_price_stats_attributes(
+    current_price: float,
+    current_prices: Dict[datetime, float],
+    utc_time: datetime,
+    timezone: zoneinfo.ZoneInfo,
+) -> Dict[str, Any]:
+    attributes: Dict[str, Any] = {}
+    better_prices_ahead = [
+        (ts, price)
+        for ts, price in current_prices.items()
+        if ts > utc_time and price < current_price
+    ]
+    if better_prices_ahead:
+        next_better_ts, next_better_price = better_prices_ahead[0]
+        delta_better = next_better_ts - utc_time
+        attributes["next_better_price"] = next_better_price
+        attributes["hours_to_better_price"] = int(delta_better.total_seconds()) // 3600
+        attributes["num_better_prices_ahead"] = len(better_prices_ahead)
+
+    prices_sorted = dict(sorted(current_prices.items(), key=lambda x: x[1]))
+    try:
+        attributes["price_position"] = (
+            list(prices_sorted.values()).index(current_price) + 1
+        )
+    except ValueError:
+        pass
+
+    max_price = max(current_prices.values())
+    min_price = min(current_prices.values())
+    try:
+        attributes["price_ratio"] = round(
+            (current_price - min_price) / (max_price - min_price), 2
+        )
+    except ZeroDivisionError:  # pragma: no cover
+        pass
+    attributes["max_price"] = max_price
+    attributes["max_price_at"] = (
+        next(iter(reversed(prices_sorted))).astimezone(timezone).hour
+    )
+    attributes["min_price"] = min_price
+    attributes["min_price_at"] = next(iter(prices_sorted)).astimezone(timezone).hour
+    attributes["next_best_at"] = list(
+        map(
+            lambda x: x.astimezone(timezone).hour,
+            filter(lambda x: x >= utc_time, prices_sorted.keys()),
+        )
+    )
+    return attributes
+
+
+def make_price_sensor_attributes(
+    current_prices: Dict[datetime, float],
+    utc_time: datetime,
+    timezone: zoneinfo.ZoneInfo,
+) -> Dict[str, Any]:
+    """Generate sensor attributes for hourly prices variables."""
+    current_price = current_prices[utc_time]
+    today, tomorrow = _split_today_tomorrow_prices(current_prices, utc_time, timezone)
+    price_attrs = _make_price_stats_attributes(current_price, today, utc_time, timezone)
+    price_tags = _make_price_tag_attributes(today, timezone, False)
+    if tomorrow:
+        tomorrow_prices = {
+            f"{key} (next day)": value
+            for key, value in _make_price_stats_attributes(
+                current_price, tomorrow, utc_time, timezone
+            ).items()
+        }
+        tomorrow_price_tags = _make_price_tag_attributes(tomorrow, timezone, True)
+        price_attrs = {**price_attrs, **tomorrow_prices}
+        price_tags = {**price_tags, **tomorrow_price_tags}
+    return {**price_attrs, **price_tags}

aiopvpc/pvpc_data.py

@@ -13,7 +13,6 @@
 
 import aiohttp
 import async_timeout
-import holidays
 
 from aiopvpc.const import (
     ATTRIBUTION,
@@ -28,7 +27,9 @@
     UTC_TZ,
     zoneinfo,
 )
+from aiopvpc.prices import make_price_sensor_attributes
 from aiopvpc.pvpc_download import extract_pvpc_data, get_url_for_daily_json
+from aiopvpc.pvpc_tariff import get_current_and_next_tariff_periods
 
 _REQUEST_HEADERS = {
     "User-Agent": "aioPVPC Python library",
@@ -44,102 +45,6 @@ def _ensure_utc_time(ts: datetime):
     return ts
 
 
-def _tariff_period_key(local_ts: datetime, zone_ceuta_melilla: bool) -> str:
-    """Return period key (P1/P2/P3) for current hour."""
-    day = local_ts.date()
-    # TODO review 'festivos nacionales no sustituibles de fecha fija', + 6/1
-    national_holiday = day in holidays.Spain(observed=False, years=day.year).keys()
-    if national_holiday or day.isoweekday() >= 6 or local_ts.hour < 8:
-        return "P3"
-    elif zone_ceuta_melilla and local_ts.hour in (8, 9, 10, 15, 16, 17, 18, 23):
-        return "P2"
-    elif not zone_ceuta_melilla and local_ts.hour in (8, 9, 14, 15, 16, 17, 22, 23):
-        return "P2"
-    return "P1"
-
-
-def _get_current_and_next_tariff_periods(
-    local_ts: datetime, zone_ceuta_melilla: bool
-) -> Tuple[str, str, timedelta]:
-    current_period = _tariff_period_key(local_ts, zone_ceuta_melilla)
-    delta = timedelta(hours=1)
-    while (
-        next_period := _tariff_period_key(local_ts + delta, zone_ceuta_melilla)
-    ) == current_period:
-        delta += timedelta(hours=1)
-    return current_period, next_period, delta
-
-
-def _make_sensor_attributes(
-    current_prices: Dict[datetime, float],
-    utc_time: datetime,
-    timezone: zoneinfo.ZoneInfo,
-    zone_ceuta_melilla: bool,
-    power: int,
-    power_valley: int,
-) -> Dict[str, Any]:
-    attributes: Dict[str, Any] = {}
-    current_price = current_prices[utc_time]
-    local_time = utc_time.astimezone(timezone)
-    current_period, next_period, delta = _get_current_and_next_tariff_periods(
-        local_time, zone_ceuta_melilla
-    )
-    attributes["period"] = current_period
-    attributes["available_power"] = power_valley if current_period == "P3" else power
-    attributes["next_period"] = next_period
-    attributes["hours_to_next_period"] = int(delta.total_seconds()) // 3600
-
-    better_prices_ahead = [
-        (ts, price)
-        for ts, price in current_prices.items()
-        if ts > utc_time and price < current_price
-    ]
-    if better_prices_ahead:
-        next_better_ts, next_better_price = better_prices_ahead[0]
-        delta_better = next_better_ts - utc_time
-        attributes["next_better_price"] = next_better_price
-        attributes["hours_to_better_price"] = int(delta_better.total_seconds()) // 3600
-        attributes["num_better_prices_ahead"] = len(better_prices_ahead)
-
-    prices_sorted = dict(sorted(current_prices.items(), key=lambda x: x[1]))
-    attributes["price_position"] = list(prices_sorted.values()).index(current_price) + 1
-    max_price = max(current_prices.values())
-    min_price = min(current_prices.values())
-    try:
-        attributes["price_ratio"] = round(
-            (current_price - min_price) / (max_price - min_price), 2
-        )
-    except ZeroDivisionError:  # pragma: no cover
-        pass
-    attributes["max_price"] = max_price
-    attributes["max_price_at"] = (
-        next(iter(reversed(prices_sorted))).astimezone(timezone).hour
-    )
-    attributes["min_price"] = min_price
-    attributes["min_price_at"] = next(iter(prices_sorted)).astimezone(timezone).hour
-    attributes["next_best_at"] = list(
-        map(
-            lambda x: x.astimezone(timezone).hour,
-            filter(lambda x: x >= utc_time, prices_sorted.keys()),
-        )
-    )
-
-    def _is_tomorrow_price(ts, ref):
-        return any(map(lambda x: x[0] > x[1], zip(ts.isocalendar(), ref.isocalendar())))
-
-    for ts_utc, price_h in current_prices.items():
-        ts_local = ts_utc.astimezone(timezone)
-        if _is_tomorrow_price(ts_local, local_time):
-            attr_key = f"price_next_day_{ts_local.hour:02d}h"
-        else:
-            attr_key = f"price_{ts_local.hour:02d}h"
-        if attr_key in attributes:  # DST change with duplicated hour :)
-            attr_key += "_d"
-        attributes[attr_key] = price_h
-
-    return attributes
-
-
 class PVPCData:
     """
     Data handler for PVPC hourly prices.
@@ -308,16 +213,22 @@ def process_state_and_attributes(self, utc_now: datetime) -> bool:
             self.attributes = attributes
             return False
 
-        # generate sensor attributes
-        current_hour_attrs = _make_sensor_attributes(
-            self._current_prices,
-            utc_time,
-            self._local_timezone,
-            self._zone_ceuta_melilla,
-            int(1000 * self._power),
-            int(1000 * self._power_valley),
+        # generate PVPC 2.0TD sensor attributes
+        local_time = utc_time.astimezone(self._local_timezone)
+        (current_period, next_period, delta,) = get_current_and_next_tariff_periods(
+            local_time, zone_ceuta_melilla=self._zone_ceuta_melilla
+        )
+        attributes["period"] = current_period
+        power = self._power_valley if current_period == "P3" else self._power
+        attributes["available_power"] = int(1000 * power)
+        attributes["next_period"] = next_period
+        attributes["hours_to_next_period"] = int(delta.total_seconds()) // 3600
+
+        # generate price attributes
+        price_attrs = make_price_sensor_attributes(
+            self._current_prices, utc_time, self._local_timezone
         )
-        self.attributes = {**attributes, **current_hour_attrs}
+        self.attributes = {**attributes, **price_attrs}
         return True
 
     async def _download_worker(self, wk_name: str, queue: asyncio.Queue):

aiopvpc/pvpc_tariff.py

@@ -0,0 +1,36 @@
+"""ESIOS API handler for HomeAssistant. PVPC tariff periods."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+import holidays
+
+_HOURS_P2 = (8, 9, 14, 15, 16, 17, 22, 23)
+_HOURS_P2_CYM = (8, 9, 10, 15, 16, 17, 18, 23)
+
+
+def _tariff_period_key(local_ts: datetime, zone_ceuta_melilla: bool) -> str:
+    """Return period key (P1/P2/P3) for current hour."""
+    day = local_ts.date()
+    # TODO review 'festivos nacionales no sustituibles de fecha fija', + 6/1
+    national_holiday = day in holidays.Spain(observed=False, years=day.year).keys()
+    if national_holiday or day.isoweekday() >= 6 or local_ts.hour < 8:
+        return "P3"
+    elif zone_ceuta_melilla and local_ts.hour in _HOURS_P2_CYM:
+        return "P2"
+    elif not zone_ceuta_melilla and local_ts.hour in _HOURS_P2:
+        return "P2"
+    return "P1"
+
+
+def get_current_and_next_tariff_periods(
+    local_ts: datetime, zone_ceuta_melilla: bool
+) -> tuple[str, str, timedelta]:
+    """Get tariff periods for PVPC 2.0TD."""
+    current_period = _tariff_period_key(local_ts, zone_ceuta_melilla)
+    delta = timedelta(hours=1)
+    while (
+        next_period := _tariff_period_key(local_ts + delta, zone_ceuta_melilla)
+    ) == current_period:
+        delta += timedelta(hours=1)
+    return current_period, next_period, delta