Add tariff sensors to Teslemetry

homeassistant/components/teslemetry/coordinator.py

@@ -77,7 +77,10 @@ async def _async_update_data(self) -> dict[str, Any]:
         except TeslaFleetError as e:
             raise UpdateFailed(e.message) from e
 
-        return flatten(data)
+        return flatten(
+            data,
+            exceptions=["daily_charges", "demand_charges", "energy_charges", "seasons"],
+        )
 
 
 class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@@ -160,7 +163,10 @@ async def _async_update_data(self) -> dict[str, Any]:
         except TeslaFleetError as e:
             raise UpdateFailed(e.message) from e
 
-        return flatten(data)
+        return flatten(
+            data,
+            exceptions=["daily_charges", "demand_charges", "energy_charges", "seasons"],
+        )
 
 
 class TeslemetryEnergyHistoryCoordinator(DataUpdateCoordinator[dict[str, Any]]):

homeassistant/components/teslemetry/helpers.py

@@ -9,14 +9,17 @@
 from .const import DOMAIN, LOGGER
 
 
-def flatten(data: dict[str, Any], parent: str | None = None) -> dict[str, Any]:
+def flatten(
+    data: dict[str, Any], parent: str | None = None, exceptions: list[str] | None = None
+) -> dict[str, Any]:
     """Flatten the data structure."""
     result = {}
     for key, value in data.items():
+        exception = exceptions and key in exceptions
         if parent:
             key = f"{parent}_{key}"
-        if isinstance(value, dict):
-            result.update(flatten(value, key))
+        if isinstance(value, dict) and not exception:
+            result.update(flatten(value, key, exceptions))
         else:
             result[key] = value
     return result

homeassistant/components/teslemetry/sensor.py

@@ -3,8 +3,9 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+from typing import Any
 
 from teslemetry_stream import TeslemetryStream, TeslemetryStreamVehicle
 
@@ -1623,9 +1624,189 @@
         )
     )
 
+    # Add Tariff Sensors
+    for energy_site in entry.runtime_data.energysites:
+        # Buy Tariff Sensor
+        if energy_site.info_coordinator.data.get("tariff_content_v2_seasons"):
+            entities.append(TeslemetryTariffSensor(energy_site, "tariff_content_v2"))
+        # Sell Tariff Sensor
+        if energy_site.info_coordinator.data.get(
+            "tariff_content_v2_sell_tariff_seasons"
+        ):
+            entities.append(
+                TeslemetryTariffSensor(energy_site, "tariff_content_v2_sell_tariff")
+            )
+
     async_add_entities(entities)
 
 
+class TeslemetryTariffSensor(TeslemetryEnergyInfoEntity, SensorEntity):
+    """Energy Site Tariff Price Sensor."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+
+    seasons: dict[str, dict[str, Any]] = field(default_factory=dict)
+    charges: dict[str, dict[str, Any]] = field(default_factory=dict)
+    key: str
+
+    def __init__(
+        self,
+        data: TeslemetryEnergyData,
+        key: str,
+    ) -> None:
+        """Initialize the tariff price sensor."""
+        self.key = key
+        super().__init__(data, key)
+
+    @property
+    def native_value(self) -> float | None:
+        """Return the current tariff price."""
+        price_info = self._get_current_tariff_info()
+        return price_info["price"] if price_info else None
+
+    @property
+    def native_unit_of_measurement(self) -> str | None:
+        """Return the unit of measurement."""
+        return f"{self.hass.config.currency}/kWh" if self.hass.config.currency else None
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        """Return other tariff details as attributes."""
+        price_info = self._get_current_tariff_info()
+        if not price_info:
+            return None
+        return {
+            "season": price_info["season_name"],
+            "period_name": price_info["period_name"],
+            "start_time": price_info["start_time"],
+            "end_time": price_info["end_time"],
+        }
+
+    def _get_current_tariff_info(self) -> dict[str, Any] | None:
+        """Calculate the current tariff price and details."""
+        now = dt_util.now()
+        current_season_name = self._get_current_season(now)
+
+        if not current_season_name or not self.seasons.get(current_season_name):
+            return None
+
+        season_data = self.seasons[current_season_name]
+        tou_periods = season_data.get("tou_periods", {})
+
+        for period_name_key, period_group in tou_periods.items():
+            for period_def in period_group.get("periods", []):
+                day_of_week = now.weekday()
+                from_day = period_def.get("fromDayOfWeek", 0)
+                to_day = period_def.get("toDayOfWeek", 6)
+                if not (from_day <= day_of_week <= to_day):
+                    continue
+
+                from_hour, from_minute = (
+                    period_def.get("fromHour", 0) % 24,
+                    period_def.get("fromMinute", 0) % 60,
+                )
+                to_hour, to_minute = (
+                    period_def.get("toHour", 0) % 24,
+                    period_def.get("toMinute", 0) % 60,
+                )
+
+                start_time = now.replace(
+                    hour=from_hour, minute=from_minute, second=0, microsecond=0
+                )
+                end_time = now.replace(
+                    hour=to_hour, minute=to_minute, second=0, microsecond=0
+                )
+
+                if end_time <= start_time:  # Period crosses midnight
+                    potential_end_time = end_time + timedelta(days=1)
+                    if start_time <= now < potential_end_time:
+                        end_time = potential_end_time
+                    elif (start_time - timedelta(days=1)) <= now < end_time:
+                        start_time -= timedelta(days=1)
+                    else:
+                        continue
+                elif not (start_time <= now < end_time):
+                    continue
+
+                price = self._get_price_for_period(current_season_name, period_name_key)
+                return {
+                    "price": price,
+                    "season_name": current_season_name.capitalize(),
+                    "period_name": period_name_key.capitalize().replace("_", " "),
+                    "start_time": start_time,
+                    "end_time": end_time,
+                }
+        return None
+
+    def _get_current_season(self, date_to_check: datetime) -> str | None:
+        """Determine the active season for a given date."""
+        local_date = dt_util.as_local(date_to_check)
+        year = local_date.year
+        for season_name, season_data in self.seasons.items():
+            if not season_data:
+                continue
+            try:
+                from_month, from_day = season_data["fromMonth"], season_data["fromDay"]
+                to_month, to_day = season_data["toMonth"], season_data["toDay"]
+                start_year, end_year = year, year
+                if from_month > to_month or (
+                    from_month == to_month and from_day > to_day
+                ):
+                    if local_date.month > from_month or (
+                        local_date.month == from_month and local_date.day >= from_day
+                    ):
+                        end_year = year + 1
+                    else:
+                        start_year = year - 1
+                season_start = local_date.replace(
+                    year=start_year,
+                    month=from_month,
+                    day=from_day,
+                    hour=0,
+                    minute=0,
+                    second=0,
+                    microsecond=0,
+                )
+                season_end = local_date.replace(
+                    year=end_year,
+                    month=to_month,
+                    day=to_day,
+                    hour=0,
+                    minute=0,
+                    second=0,
+                    microsecond=0,
+                ) + timedelta(days=1)
+                if season_start <= local_date < season_end:
+                    return season_name
+            except (KeyError, ValueError):
+                continue
+        return None
+
+    def _get_price_for_period(self, season_name: str, period_name: str) -> float | None:
+        """Get the price for a specific season and period name."""
+        try:
+            season_charges = self.charges.get(season_name, self.charges.get("ALL", {}))
+            rates = season_charges.get("rates", {})
+            price = rates.get(period_name, rates.get("ALL"))
+            return float(price) if price is not None else None
+        except (KeyError, ValueError, TypeError):
+            return None
+
+    def _async_update_attrs(self) -> None:
+        """Update the Sensor attributes from coordinator data."""
+        self.seasons = self.coordinator.data.get(f"{self.key}_seasons", {})
+        self.charges = self.coordinator.data.get(f"{self.key}_energy_charges", {})
+
+        # native_value is determined by property, but availability depends on data
+        current_tariff_info = self._get_current_tariff_info()
+        if current_tariff_info and current_tariff_info["price"] is not None:
+            self._attr_available = True
+            # native_value and native_unit_of_measurement are handled by properties
+        else:
+            self._attr_available = False
+        # extra_state_attributes is also handled by a property
+
+
 class TeslemetryStreamSensorEntity(TeslemetryVehicleStreamEntity, RestoreSensor):
     """Base class for Teslemetry vehicle streaming sensors."""
 

homeassistant/components/teslemetry/strings.json

@@ -426,6 +426,12 @@
       }
     },
     "sensor": {
+      "tariff_content_v2": {
+        "name": "Buy tariff"
+      },
+      "tariff_content_v2_sell_tariff": {
+        "name": "Sell tariff"
+      },
       "battery_power": {
         "name": "Battery power"
       },

tests/components/teslemetry/fixtures/site_info.json

@@ -122,6 +122,130 @@
     "installation_time_zone": "",
     "max_site_meter_power_ac": 1000000000,
     "min_site_meter_power_ac": -1000000000,
+    "tariff_content_v2": {
+      "code": "Test",
+      "name": "Battery Maximiser",
+      "utility": "Origin",
+      "daily_charges": [
+        {
+          "name": "Charge"
+        }
+      ],
+      "demand_charges": {
+        "ALL": {
+          "rates": {
+            "ALL": 0
+          }
+        },
+        "Summer": {},
+        "Winter": {}
+      },
+      "energy_charges": {
+        "ALL": {
+          "rates": {
+            "ALL": 0
+          }
+        },
+        "Summer": {
+          "rates": {
+            "OFF_PEAK": 0.198,
+            "ON_PEAK": 0.22
+          }
+        },
+        "Winter": {}
+      },
+      "seasons": {
+        "Summer": {
+          "fromDay": 1,
+          "toDay": 31,
+          "fromMonth": 1,
+          "toMonth": 12,
+          "tou_periods": {
+            "OFF_PEAK": {
+              "periods": [
+                {
+                  "toDayOfWeek": 6,
+                  "fromHour": 21,
+                  "toHour": 16
+                }
+              ]
+            },
+            "ON_PEAK": {
+              "periods": [
+                {
+                  "toDayOfWeek": 6,
+                  "fromHour": 16,
+                  "toHour": 21
+                }
+              ]
+            }
+          }
+        },
+        "Winter": {}
+      },
+      "sell_tariff": {
+        "name": "Battery Maximiser",
+        "utility": "Origin",
+        "daily_charges": [
+          {
+            "name": "Charge"
+          }
+        ],
+        "demand_charges": {
+          "ALL": {
+            "rates": {
+              "ALL": 0
+            }
+          },
+          "Summer": {},
+          "Winter": {}
+        },
+        "energy_charges": {
+          "ALL": {
+            "rates": {
+              "ALL": 0
+            }
+          },
+          "Summer": {
+            "rates": {
+              "OFF_PEAK": 0.08,
+              "ON_PEAK": 0.16
+            }
+          },
+          "Winter": {}
+        },
+        "seasons": {
+          "Summer": {
+            "fromDay": 1,
+            "toDay": 31,
+            "fromMonth": 1,
+            "toMonth": 12,
+            "tou_periods": {
+              "OFF_PEAK": {
+                "periods": [
+                  {
+                    "toDayOfWeek": 6,
+                    "fromHour": 21,
+                    "toHour": 16
+                  }
+                ]
+              },
+              "ON_PEAK": {
+                "periods": [
+                  {
+                    "toDayOfWeek": 6,
+                    "fromHour": 16,
+                    "toHour": 21
+                  }
+                ]
+              }
+            }
+          },
+          "Winter": {}
+        }
+      },
+      "version": 1
+    },
     "vpp_backup_reserve_percent": 0
   }
 }