Stream Teslemetry energy live status, site info and tariff over SSE (#178735)

homeassistant/components/teslemetry/__init__.py

@@ -14,8 +14,9 @@ from tesla_fleet_api.exceptions import (
     SubscriptionRequired,
     TeslaFleetError,
 )
-from tesla_fleet_api.teslemetry import Teslemetry
+from tesla_fleet_api.teslemetry import EnergySite, Teslemetry
 from teslemetry_stream import TeslemetryStream
+from teslemetry_stream.const import SseTopic
 
 from homeassistant.components.application_credentials import (
     ClientCredential,
@@ -43,6 +44,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
 )
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.typing import ConfigType
+from homeassistant.helpers.update_coordinator import UpdateFailed
 
 from .const import CLIENT_ID, DOMAIN, LOGGER, VEHICLE_ISSUE_LEARN_MORE
 from .coordinator import (
@@ -76,6 +78,19 @@ type TeslemetryConfigEntry = ConfigEntry[TeslemetryData]
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
+# Exact SSE topics the integration consumes. An explicit allowlist keeps a
+# new server topic from silently adding traffic or data exposure to HA.
+STREAM_TOPICS: Final = (
+    SseTopic.STATE,
+    SseTopic.VEHICLE_DATA,
+    SseTopic.DATA,
+    SseTopic.CONNECTIVITY,
+    SseTopic.CREDITS,
+    SseTopic.LIVE_STATUS,
+    SseTopic.SITE_INFO,
+    SseTopic.TARIFF_CONTENT_V2,
+)
+
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the Telemetry integration."""
@@ -314,9 +329,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
     vehicles: list[TeslemetryVehicleData] = []
     energysites: list[TeslemetryEnergyData] = []
 
-    # Create the stream (created lazily when first vehicle is found)
+    # Create the stream (created lazily for the first eligible vehicle or
+    # energy site, so energy-only accounts still open the account stream)
     stream: TeslemetryStream | None = None
 
+    def create_stream() -> TeslemetryStream:
+        return TeslemetryStream(
+            session,
+            access_token,
+            server=f"{region.lower()}.teslemetry.com",
+            parse_timestamp=True,
+            manual=True,
+            topics=STREAM_TOPICS,
+        )
+
     # Remember each device identifier we create
     current_devices: set[tuple[str, str]] = set()
 
@@ -334,13 +360,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
 
             # Create stream if required (for first vehicle)
             if not stream:
-                stream = TeslemetryStream(
-                    session,
-                    access_token,
-                    server=f"{region.lower()}.teslemetry.com",
-                    parse_timestamp=True,
-                    manual=True,
-                )
+                stream = create_stream()
 
             # Remove the protobuff 'cached_data' that we do not use to save memory
             product.pop("cached_data", None)
@@ -404,6 +424,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
                 )
                 continue
 
+            # Create stream if required (for first energy site)
+            if not stream:
+                stream = create_stream()
+
             current_devices.add((DOMAIN, str(site_id)))
             if wall_connector:
                 current_devices |= {
@@ -419,53 +443,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
                 serial_number=str(site_id),
             )
 
-            # For initial setup, raise auth errors properly
-            try:
-                live_status = (await energy_site.live_status())["response"]
-            except InvalidToken as e:
-                raise ConfigEntryAuthFailed(
-                    translation_domain=DOMAIN,
-                    translation_key="auth_failed_invalid_token",
-                ) from e
-            except LoginRequired as e:
-                raise ConfigEntryAuthFailed(
-                    translation_domain=DOMAIN,
-                    translation_key="auth_failed_login_required",
-                ) from e
-            except SubscriptionRequired as e:
-                raise ConfigEntryAuthFailed(
-                    translation_domain=DOMAIN,
-                    translation_key="auth_failed_subscription_required",
-                ) from e
-            except Forbidden as e:
-                raise ConfigEntryAuthFailed(
-                    translation_domain=DOMAIN,
-                    translation_key="auth_failed_invalid_token",
-                ) from e
-            except TeslaFleetError as e:
-                raise ConfigEntryNotReady(
-                    translation_domain=DOMAIN,
-                    translation_key="not_ready_api_error",
-                ) from e
-
+            (
+                live_coordinator,
+                info_coordinator,
+                history_coordinator,
+            ) = await _async_setup_energy_site(
+                hass,
+                entry,
+                stream,
+                energy_site,
+                product,
+                site_id,
+                powerwall,
+            )
             energysites.append(
                 TeslemetryEnergyData(
                     api=energy_site,
-                    live_coordinator=(
-                        TeslemetryEnergySiteLiveCoordinator(
-                            hass, entry, energy_site, live_status
-                        )
-                        if isinstance(live_status, dict)
-                        else None
-                    ),
-                    info_coordinator=TeslemetryEnergySiteInfoCoordinator(
-                        hass, entry, energy_site, product
-                    ),
-                    history_coordinator=(
-                        TeslemetryEnergyHistoryCoordinator(hass, entry, energy_site)
-                        if powerwall
-                        else None
-                    ),
+                    live_coordinator=live_coordinator,
+                    info_coordinator=info_coordinator,
+                    history_coordinator=history_coordinator,
                     id=site_id,
                     device=device,
                 )
@@ -526,11 +522,123 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
 
     if stream:
         entry.async_on_unload(stream.close)
+        # The stream is the only freshness signal for the energy coordinators, so
+        # a dropped connection must mark their entities unavailable rather than
+        # leaving stale live/info/tariff data available indefinitely.
+        if energysites:
+            entry.async_on_unload(
+                stream.async_add_connection_listener(
+                    create_handle_energy_stream_connection(energysites)
+                )
+            )
         entry.async_create_background_task(hass, stream.listen(), "Teslemetry Stream")
 
     return True
 
 
+def create_handle_energy_stream_connection(
+    energysites: list[TeslemetryEnergyData],
+) -> Callable[[bool], None]:
+    """Create a stream connection listener for the energy coordinators."""
+
+    @callback
+    def handle_connection(connected: bool) -> None:
+        """Fail stream-driven energy coordinators while the stream is down.
+
+        Each subsequent streamed document restores its coordinator via
+        async_set_updated_data, so no reload is required on reconnect.
+        """
+        if connected:
+            return
+        error = UpdateFailed(
+            translation_domain=DOMAIN,
+            translation_key="stream_disconnected",
+        )
+        for energysite in energysites:
+            if energysite.live_coordinator is not None:
+                energysite.live_coordinator.async_set_update_error(error)
+            energysite.info_coordinator.async_set_update_error(error)
+
+    return handle_connection
+
+
+async def _async_setup_energy_site(
+    hass: HomeAssistant,
+    entry: TeslemetryConfigEntry,
+    stream: TeslemetryStream,
+    energy_site: EnergySite,
+    product: dict[str, Any],
+    site_id: int,
+    powerwall: Any,
+) -> tuple[
+    TeslemetryEnergySiteLiveCoordinator | None,
+    TeslemetryEnergySiteInfoCoordinator,
+    TeslemetryEnergyHistoryCoordinator | None,
+]:
+    """Cold-read live status, build the energy coordinators, and register listeners."""
+    # The stream has no ready boundary, so keep a deterministic REST cold read
+    # for setup auth/error handling before switching to listener-driven updates.
+    try:
+        live_status = (await energy_site.live_status())["response"]
+    except InvalidToken as e:
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="auth_failed_invalid_token",
+        ) from e
+    except LoginRequired as e:
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="auth_failed_login_required",
+        ) from e
+    except SubscriptionRequired as e:
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="auth_failed_subscription_required",
+        ) from e
+    except Forbidden as e:
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="auth_failed_invalid_token",
+        ) from e
+    except TeslaFleetError as e:
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="not_ready_api_error",
+        ) from e
+
+    live_coordinator = (
+        TeslemetryEnergySiteLiveCoordinator(hass, entry, energy_site, live_status)
+        if isinstance(live_status, dict)
+        else None
+    )
+    info_coordinator = TeslemetryEnergySiteInfoCoordinator(
+        hass, entry, energy_site, product
+    )
+
+    # Register before stream.listen() so the opening snapshot cannot be missed.
+    stream_energysite = stream.get_energysite(site_id)
+    if live_coordinator is not None:
+        entry.async_on_unload(
+            stream_energysite.listen_LiveStatus(live_coordinator.handle_stream_update)
+        )
+    entry.async_on_unload(
+        stream_energysite.listen_SiteInfo(info_coordinator.handle_site_info)
+    )
+    entry.async_on_unload(
+        stream_energysite.listen_TariffContentV2(
+            info_coordinator.handle_tariff_content_v2
+        )
+    )
+
+    history_coordinator = (
+        TeslemetryEnergyHistoryCoordinator(hass, entry, energy_site)
+        if powerwall
+        else None
+    )
+
+    return live_coordinator, info_coordinator, history_coordinator
+
+
 async def async_unload_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool:
     """Unload Teslemetry Config."""
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/teslemetry/coordinator.py

@@ -45,11 +45,13 @@ def _get_retry_after(e: TeslaFleetError) -> float:
 
 VEHICLE_INTERVAL = timedelta(seconds=60)
 VEHICLE_WAIT = timedelta(minutes=15)
-ENERGY_LIVE_INTERVAL = timedelta(seconds=30)
-ENERGY_INFO_INTERVAL = timedelta(seconds=30)
 ENERGY_HISTORY_INTERVAL = timedelta(seconds=60)
 METADATA_INTERVAL = timedelta(hours=1)
 
+# Keys within tariff_content_v2 kept as nested dicts rather than flattened,
+# since entities and calendars read them as whole structures.
+TARIFF_SKIP_KEYS = ["daily_charges", "demand_charges", "energy_charges", "seasons"]
+
 # Insufficient credits will not resolve themselves quickly, so back off polling
 # instead of hammering the API at the coordinator's normal interval.
 INSUFFICIENT_CREDITS_RETRY_AFTER = timedelta(hours=1).total_seconds()
@@ -175,8 +177,21 @@ class TeslemetryVehicleDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         return data
 
 
+def _index_wall_connectors(data: dict[str, Any]) -> dict[str, Any]:
+    """Convert the live_status wall_connectors list into a DIN-keyed dict."""
+    data["wall_connectors"] = {
+        wc["din"]: wc for wc in (data.get("wall_connectors") or [])
+    }
+    return data
+
+
 class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]):
-    """Class to manage fetching energy site live status from the Teslemetry API."""
+    """Class to manage energy site live status from the Teslemetry stream.
+
+    Updates are driven by ``live_status`` stream events; the REST update
+    method is retained for the deterministic setup cold read and manual
+    recovery only.
+    """
 
     config_entry: TeslemetryConfigEntry
     updated_once: bool
@@ -194,15 +209,13 @@ class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]])
             LOGGER,
             config_entry=config_entry,
             name="Teslemetry Energy Site Live",
-            update_interval=ENERGY_LIVE_INTERVAL,
         )
         self.api = api
+        self.data = _index_wall_connectors(data)
 
-        # Convert Wall Connectors from array to dict
-        data["wall_connectors"] = {
-            wc["din"]: wc for wc in (data.get("wall_connectors") or [])
-        }
-        self.data = data
+    def handle_stream_update(self, data: dict[str, Any]) -> None:
+        """Handle a live_status document from the stream."""
+        self.async_set_updated_data(_index_wall_connectors(data))
 
     @override
     async def _async_update_data(self) -> dict[str, Any]:
@@ -224,15 +237,18 @@ class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]])
                 translation_key="update_failed",
                 translation_placeholders={"message": e.message},
             ) from e
-        # Convert Wall Connectors from array to dict
-        data["wall_connectors"] = {
-            wc["din"]: wc for wc in (data.get("wall_connectors") or [])
-        }
-        return data
+        return _index_wall_connectors(data)
 
 
 class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]):
-    """Class to manage fetching energy site info from the Teslemetry API."""
+    """Class to manage energy site info from the Teslemetry stream.
+
+    Site info and the V2 tariff are two independently replaceable partitions.
+    The flattened coordinator view is recomposed from both whenever either
+    changes, so a removed field or a cleared tariff never lingers. Stream
+    events drive updates; the REST update method is retained for the
+    deterministic setup cold read and manual recovery only.
+    """
 
     config_entry: TeslemetryConfigEntry
 
@@ -249,11 +265,46 @@ class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]])
             LOGGER,
             config_entry=config_entry,
             name="Teslemetry Energy Site Info",
-            update_interval=ENERGY_INFO_INTERVAL,
         )
         self.api = api
+        self._site_info: dict[str, Any] = product
+        self._tariff_content_v2: dict[str, Any] | None = None
         self.data = product
 
+    def _compose(self) -> dict[str, Any]:
+        """Flatten the two partitions into the coordinator view."""
+        result = flatten(self._site_info, skip_keys=TARIFF_SKIP_KEYS)
+        if self._tariff_content_v2 is not None:
+            result.update(
+                flatten(
+                    {"tariff_content_v2": self._tariff_content_v2},
+                    skip_keys=TARIFF_SKIP_KEYS,
+                )
+            )
+        return result
+
+    def _ingest_site_info(self, site_info: dict[str, Any]) -> dict[str, Any]:
+        """Split a full REST site_info response into both partitions.
+
+        The REST document carries both tariff versions inline; the V2 tariff
+        moves to its own partition (matching the slim stream event shape) so
+        removal semantics stay identical across both delivery paths.
+        """
+        site_info = dict(site_info)
+        self._tariff_content_v2 = site_info.pop("tariff_content_v2", None)
+        self._site_info = site_info
+        return self._compose()
+
+    def handle_site_info(self, site_info: dict[str, Any]) -> None:
+        """Handle a slim site_info document from the stream."""
+        self._site_info = site_info
+        self.async_set_updated_data(self._compose())
+
+    def handle_tariff_content_v2(self, tariff: dict[str, Any] | None) -> None:
+        """Handle a V2 tariff document (or removal) from the stream."""
+        self._tariff_content_v2 = tariff
+        self.async_set_updated_data(self._compose())
+
     @override
     async def _async_update_data(self) -> dict[str, Any]:
         """Update energy site data using Teslemetry API."""
@@ -275,10 +326,7 @@ class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]])
                 translation_placeholders={"message": e.message},
             ) from e
 
-        return flatten(
-            data,
-            skip_keys=["daily_charges", "demand_charges", "energy_charges", "seasons"],
-        )
+        return self._ingest_site_info(data)
 
 
 class TeslemetryEnergyHistoryCoordinator(DataUpdateCoordinator[dict[str, Any]]):

homeassistant/components/teslemetry/strings.json

@@ -1203,6 +1203,9 @@
     "set_scheduled_departure_preconditioning": {
       "message": "Preconditioning departure time is required when enabling"
     },
+    "stream_disconnected": {
+      "message": "Disconnected from the Teslemetry stream"
+    },
     "token_data_malformed": {
       "message": "Token data malformed, try reauthenticating"
     },

tests/components/teslemetry/conftest.py

@@ -3,7 +3,7 @@
 from collections.abc import Generator
 from copy import deepcopy
 from typing import Any
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from teslemetry_stream.stream import recursive_match
@@ -15,8 +15,10 @@ from .const import (
     ENERGY_HISTORY,
     LIVE_STATUS,
     METADATA,
+    METADATA_ENERGY,
     METADATA_LEGACY,
     PRODUCTS,
+    PRODUCTS_ENERGY,
     SITE_INFO,
     VEHICLE_DATA,
     WAKE_UP_ONLINE,
@@ -184,6 +186,100 @@ def mock_add_listener():
         yield mock_add_listener
 
 
+@pytest.fixture
+def mock_add_connection_listener():
+    """Mock Teslemetry Stream add connection listener method."""
+    with patch(
+        "teslemetry_stream.TeslemetryStream.async_add_connection_listener",
+    ) as mock_add_connection_listener:
+        mock_add_connection_listener.listeners = []
+
+        def unsubscribe() -> None:
+            return
+
+        def side_effect(callback):
+            mock_add_connection_listener.listeners.append(callback)
+            return unsubscribe
+
+        def send(connected: bool) -> None:
+            for listener in mock_add_connection_listener.listeners:
+                listener(connected)
+
+        mock_add_connection_listener.send = send
+        mock_add_connection_listener.side_effect = side_effect
+        yield mock_add_connection_listener
+
+
+@pytest.fixture
+def mock_energy_live_stream() -> Generator[MagicMock]:
+    """Capture the callback the integration registers for live_status events."""
+    with patch(
+        "teslemetry_stream.TeslemetryStreamEnergySite.listen_LiveStatus",
+    ) as mock_listen:
+        callbacks: list = []
+
+        def side_effect(callback):
+            callbacks.append(callback)
+            return MagicMock()
+
+        def send(live_status: dict) -> None:
+            for callback in callbacks:
+                callback(live_status)
+
+        mock_listen.side_effect = side_effect
+        mock_listen.send = send
+        yield mock_listen
+
+
+@pytest.fixture
+def mock_energy_info_stream() -> Generator[MagicMock]:
+    """Capture the callback the integration registers for site_info events."""
+    with patch(
+        "teslemetry_stream.TeslemetryStreamEnergySite.listen_SiteInfo",
+    ) as mock_listen:
+        callbacks: list = []
+
+        def side_effect(callback):
+            callbacks.append(callback)
+            return MagicMock()
+
+        def send(site_info: dict) -> None:
+            for callback in callbacks:
+                callback(site_info)
+
+        mock_listen.side_effect = side_effect
+        mock_listen.send = send
+        yield mock_listen
+
+
+@pytest.fixture
+def mock_energy_tariff_stream() -> Generator[MagicMock]:
+    """Capture the callback the integration registers for tariff events."""
+    with patch(
+        "teslemetry_stream.TeslemetryStreamEnergySite.listen_TariffContentV2",
+    ) as mock_listen:
+        callbacks: list = []
+
+        def side_effect(callback):
+            callbacks.append(callback)
+            return MagicMock()
+
+        def send(tariff: dict | None) -> None:
+            for callback in callbacks:
+                callback(tariff)
+
+        mock_listen.side_effect = side_effect
+        mock_listen.send = send
+        yield mock_listen
+
+
+@pytest.fixture
+def mock_energy_only(mock_products: AsyncMock, mock_metadata: MagicMock) -> None:
+    """Patch products and metadata to an energy-only account."""
+    mock_products.return_value = PRODUCTS_ENERGY
+    mock_metadata.return_value = METADATA_ENERGY
+
+
 @pytest.fixture(autouse=True)
 def mock_stream_get_config():
     """Mock Teslemetry Stream listen method."""

tests/components/teslemetry/const.py

@@ -145,3 +145,27 @@ METADATA_NOSCOPE = {
         }
     },
 }
+
+# Energy-only account: no accessible vehicle, one accessible energy site.
+METADATA_ENERGY = {
+    "uid": UNIQUE_ID,
+    "region": "NA",
+    "scopes": [
+        "openid",
+        "offline_access",
+        "user_data",
+        "energy_device_data",
+        "energy_cmds",
+    ],
+    "vehicles": {},
+    "energy_sites": {
+        "123456": {
+            "access": True,
+            "name": "Energy Site",
+        }
+    },
+}
+PRODUCTS_ENERGY = load_json_object_fixture("products.json", DOMAIN)
+PRODUCTS_ENERGY["response"] = [
+    product for product in PRODUCTS_ENERGY["response"] if "energy_site_id" in product
+]

tests/components/teslemetry/test_calendar.py

@@ -3,7 +3,7 @@
 from collections.abc import Generator
 from copy import deepcopy
 from datetime import datetime
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from freezegun.api import FrozenDateTimeFactory
 import pytest
@@ -15,7 +15,12 @@ from homeassistant.components.calendar import (
     EVENT_START_DATETIME,
     SERVICE_GET_EVENTS,
 )
-from homeassistant.const import ATTR_ENTITY_ID, Platform
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+    Platform,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 from homeassistant.util import dt as dt_util
@@ -25,6 +30,14 @@ from .const import SITE_INFO, SITE_INFO_MULTI_SEASON, SITE_INFO_WEEK_CROSSING
 
 ENTITY_BUY = "calendar.energy_site_buy_tariff"
 ENTITY_SELL = "calendar.energy_site_sell_tariff"
+ENTITY_OPERATION_MODE = "select.energy_site_operation_mode"
+
+TARIFF_V2 = SITE_INFO["response"]["tariff_content_v2"]
+SLIM_SITE_INFO = {
+    key: value
+    for key, value in SITE_INFO["response"].items()
+    if key != "tariff_content_v2"
+}
 
 
 @pytest.fixture
@@ -469,3 +482,100 @@ async def test_calendar_invalid_price(
     assert state
     assert state.state == "on"
     assert "Unknown Price" in state.attributes["message"]
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+@pytest.mark.parametrize(
+    "order",
+    [
+        pytest.param(("site_info", "tariff"), id="site_info_first"),
+        pytest.param(("tariff", "site_info"), id="tariff_first"),
+    ],
+)
+async def test_energy_stream_site_info_and_tariff_compose(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    mock_legacy: AsyncMock,
+    mock_energy_info_stream: MagicMock,
+    mock_energy_tariff_stream: MagicMock,
+    order: tuple[str, str],
+) -> None:
+    """Slim site_info and tariff events are independently replaceable partitions."""
+    tz = dt_util.get_default_time_zone()
+    freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz))
+
+    await setup_platform(hass, [Platform.CALENDAR, Platform.SELECT])
+
+    # The REST cold read populated both partitions.
+    assert hass.states.get(ENTITY_OPERATION_MODE).state == "self_consumption"
+    buy = hass.states.get(ENTITY_BUY)
+    assert buy.state == "on"
+    assert "0.20/kWh" in buy.attributes["message"]
+
+    slim = deepcopy(SLIM_SITE_INFO)
+    slim["default_real_mode"] = "autonomous"
+    # A distinct OFF_PEAK price proves the streamed tariff replaced the cold read.
+    streamed_tariff = deepcopy(TARIFF_V2)
+    streamed_tariff["energy_charges"]["Summer"]["rates"]["OFF_PEAK"] = 0.99
+    events = {
+        "site_info": lambda: mock_energy_info_stream.send(slim),
+        "tariff": lambda: mock_energy_tariff_stream.send(streamed_tariff),
+    }
+
+    for name in order:
+        events[name]()
+        await hass.async_block_till_done()
+
+    # Both partitions survive regardless of arrival order.
+    assert hass.states.get(ENTITY_OPERATION_MODE).state == "autonomous"
+    buy = hass.states.get(ENTITY_BUY)
+    assert buy.state == "on"
+    assert "0.99/kWh" in buy.attributes["message"]
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_energy_stream_tariff_removal_clears_calendar(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    mock_legacy: AsyncMock,
+    mock_energy_tariff_stream: MagicMock,
+) -> None:
+    """A tariff_content_v2 removal marks the tariff calendar unavailable."""
+    tz = dt_util.get_default_time_zone()
+    freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz))
+
+    await setup_platform(hass, [Platform.CALENDAR])
+
+    assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE
+
+    mock_energy_tariff_stream.send(None)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(ENTITY_BUY).state == STATE_UNAVAILABLE
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_energy_stream_slim_site_info_drops_removed_field(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    mock_legacy: AsyncMock,
+    mock_energy_info_stream: MagicMock,
+) -> None:
+    """A replacement site_info event drops a removed field without touching tariff."""
+    tz = dt_util.get_default_time_zone()
+    freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz))
+
+    await setup_platform(hass, [Platform.CALENDAR, Platform.SELECT])
+
+    assert hass.states.get(ENTITY_OPERATION_MODE).state == "self_consumption"
+    assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE
+
+    slim = deepcopy(SLIM_SITE_INFO)
+    slim.pop("default_real_mode")
+    mock_energy_info_stream.send(slim)
+    await hass.async_block_till_done()
+
+    # The removed field no longer lingers in the site_info-derived entity.
+    assert hass.states.get(ENTITY_OPERATION_MODE).state == STATE_UNKNOWN
+    # The tariff partition is untouched by a site_info replacement.
+    assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE

tests/components/teslemetry/test_init.py

@@ -19,14 +19,12 @@ from tesla_fleet_api.exceptions import (
     TeslaFleetError,
 )
 
-from homeassistant.components.teslemetry import _get_access_token
+from homeassistant.components.teslemetry import STREAM_TOPICS, _get_access_token
 from homeassistant.components.teslemetry.const import CLIENT_ID, DOMAIN
 
 # Coordinator constants
 from homeassistant.components.teslemetry.coordinator import (
     ENERGY_HISTORY_INTERVAL,
-    ENERGY_INFO_INTERVAL,
-    ENERGY_LIVE_INTERVAL,
     INSUFFICIENT_CREDITS_RETRY_AFTER,
     METADATA_INTERVAL,
     VEHICLE_INTERVAL,
@@ -165,7 +163,7 @@ async def test_energy_site_refresh_error(
 @pytest.mark.usefixtures("entity_registry_enabled_by_default")
 async def test_vehicle_stream(
     hass: HomeAssistant,
-    mock_add_listener: AsyncMock,
+    mock_add_listener: MagicMock,
     snapshot: SnapshotAssertion,
 ) -> None:
     """Test vehicle stream events."""
@@ -593,7 +591,6 @@ async def test_vehicle_data_retry_exceptions(
 @pytest.mark.parametrize(("exception", "expected_retry_after"), RETRY_EXCEPTIONS)
 async def test_live_status_coordinator_retry_exceptions(
     hass: HomeAssistant,
-    freezer: FrozenDateTimeFactory,
     mock_live_status: AsyncMock,
     exception: TeslaFleetError,
     expected_retry_after: float,
@@ -616,9 +613,8 @@ async def test_live_status_coordinator_retry_exceptions(
     assert entry.state is ConfigEntryState.LOADED
     assert call_count == 1
 
-    # Trigger coordinator refresh - this will raise the exception
-    freezer.tick(ENERGY_LIVE_INTERVAL)
-    async_fire_time_changed(hass)
+    # The recovery/manual REST path still raises the exception
+    await entry.runtime_data.energysites[0].live_coordinator.async_refresh()
     await hass.async_block_till_done()
 
     # API was called exactly once for this refresh (no manual retry loop)
@@ -665,7 +661,6 @@ async def test_energy_history_coordinator_retry_exceptions(
 
 async def test_live_status_auth_error(
     hass: HomeAssistant,
-    freezer: FrozenDateTimeFactory,
 ) -> None:
     """Test live status coordinator handles auth errors."""
     call_count = 0
@@ -684,9 +679,8 @@ async def test_live_status_auth_error(
         entry = await setup_platform(hass)
         assert entry.state is ConfigEntryState.LOADED
 
-        # Trigger a coordinator refresh by advancing time
-        freezer.tick(ENERGY_LIVE_INTERVAL)
-        async_fire_time_changed(hass)
+        # The recovery/manual REST path surfaces the auth error
+        await entry.runtime_data.energysites[0].live_coordinator.async_refresh()
         await hass.async_block_till_done()
 
         # Auth error triggers reauth flow
@@ -695,7 +689,6 @@ async def test_live_status_auth_error(
 
 async def test_live_status_generic_error(
     hass: HomeAssistant,
-    freezer: FrozenDateTimeFactory,
 ) -> None:
     """Test live status coordinator handles generic TeslaFleetError."""
     call_count = 0
@@ -714,9 +707,8 @@ async def test_live_status_generic_error(
         entry = await setup_platform(hass)
         assert entry.state is ConfigEntryState.LOADED
 
-        # Trigger a coordinator refresh by advancing time
-        freezer.tick(ENERGY_LIVE_INTERVAL)
-        async_fire_time_changed(hass)
+        # The recovery/manual REST path surfaces the error
+        await entry.runtime_data.energysites[0].live_coordinator.async_refresh()
         await hass.async_block_till_done()
 
         # Entry stays loaded but coordinator will have failed
@@ -911,10 +903,9 @@ async def test_vehicle_polling_stops_when_all_entities_disabled(
 async def test_energy_site_version_update(
     hass: HomeAssistant,
     device_registry: dr.DeviceRegistry,
-    mock_site_info: AsyncMock,
-    freezer: FrozenDateTimeFactory,
+    mock_add_listener: MagicMock,
 ) -> None:
-    """Test energy site sw_version updates when info coordinator refreshes."""
+    """Test energy site sw_version updates from a site_info stream event."""
     entry = await setup_platform(hass)
     assert entry.state is ConfigEntryState.LOADED
 
@@ -925,14 +916,11 @@ async def test_energy_site_version_update(
     assert device is not None
     assert device.sw_version == "23.44.0 eb113390"
 
-    # Update mock to return new version on next poll
-    updated_site_info = deepcopy(SITE_INFO)
-    updated_site_info["response"]["version"] = "24.1.0 abc123"
-    mock_site_info.side_effect = lambda: updated_site_info
-
-    # Trigger coordinator refresh
-    freezer.tick(ENERGY_INFO_INTERVAL)
-    async_fire_time_changed(hass)
+    # A slim site_info stream event carries the new version
+    updated_site_info = deepcopy(SITE_INFO["response"])
+    updated_site_info.pop("tariff_content_v2", None)
+    updated_site_info["version"] = "24.1.0 abc123"
+    mock_add_listener.send({"site_id": site_id, "site_info": updated_site_info})
     await hass.async_block_till_done()
 
     # Check device sw_version was updated
@@ -962,7 +950,6 @@ async def test_live_status_auth_failed_forbidden(
 )
 async def test_live_status_coordinator_refresh_error(
     hass: HomeAssistant,
-    freezer: FrozenDateTimeFactory,
     mock_live_status: AsyncMock,
     side_effect: list,
 ) -> None:
@@ -972,8 +959,7 @@ async def test_live_status_coordinator_refresh_error(
     entry = await setup_platform(hass)
     assert entry.state is ConfigEntryState.LOADED
 
-    freezer.tick(ENERGY_LIVE_INTERVAL)
-    async_fire_time_changed(hass)
+    await entry.runtime_data.energysites[0].live_coordinator.async_refresh()
     await hass.async_block_till_done()
 
     assert entry.state is ConfigEntryState.LOADED
@@ -1229,4 +1215,115 @@ async def test_get_access_token_rate_limited_after_setup_is_not_fatal(
         await _get_access_token(session)
     await hass.async_block_till_done()
 
-    assert not hass.config_entries.flow.async_progress()
+
+def test_stream_topic_allowlist() -> None:
+    """The stream subscribes to exactly the topics the integration consumes."""
+    assert [topic.value for topic in STREAM_TOPICS] == [
+        "state",
+        "vehicle_data",
+        "data",
+        "connectivity",
+        "credits",
+        "live_status",
+        "site_info",
+        "tariff_content_v2",
+    ]
+
+
+async def test_energy_stream_no_recurring_rest_polling(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    mock_live_status: AsyncMock,
+    mock_site_info: AsyncMock,
+) -> None:
+    """The live/info REST cold reads happen once and do not recur."""
+    await setup_platform(hass, [Platform.SENSOR])
+    assert mock_live_status.call_count == 1
+    assert mock_site_info.call_count == 1
+
+    # Advancing well past the old 30-second poll intervals triggers no REST reads.
+    freezer.tick(ENERGY_HISTORY_INTERVAL * 2)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    assert mock_live_status.call_count == 1
+    assert mock_site_info.call_count == 1
+
+
+async def test_energy_stream_unload_unsubscribes_and_closes_stream(
+    hass: HomeAssistant,
+) -> None:
+    """Unload runs each listener unsubscribe and closes the shared stream."""
+    live_unsub = MagicMock()
+    info_unsub = MagicMock()
+    tariff_unsub = MagicMock()
+
+    with (
+        patch(
+            "teslemetry_stream.TeslemetryStreamEnergySite.listen_LiveStatus",
+            return_value=live_unsub,
+        ),
+        patch(
+            "teslemetry_stream.TeslemetryStreamEnergySite.listen_SiteInfo",
+            return_value=info_unsub,
+        ),
+        patch(
+            "teslemetry_stream.TeslemetryStreamEnergySite.listen_TariffContentV2",
+            return_value=tariff_unsub,
+        ),
+        patch("teslemetry_stream.TeslemetryStream.close") as mock_close,
+    ):
+        entry = await setup_platform(hass, [Platform.SENSOR])
+        assert entry.state is ConfigEntryState.LOADED
+
+        assert await hass.config_entries.async_unload(entry.entry_id)
+        await hass.async_block_till_done()
+
+    live_unsub.assert_called_once()
+    info_unsub.assert_called_once()
+    tariff_unsub.assert_called_once()
+    mock_close.assert_called_once()
+
+
+async def test_energy_stream_disconnect_marks_unavailable_and_recovers(
+    hass: HomeAssistant,
+    mock_add_connection_listener: MagicMock,
+    mock_energy_live_stream: MagicMock,
+    mock_energy_info_stream: MagicMock,
+) -> None:
+    """A dropped stream marks energy entities unavailable until documents resume."""
+    await setup_platform(hass, [Platform.SENSOR, Platform.CALENDAR])
+
+    # Both stream-driven coordinators start available from the setup cold read.
+    assert hass.states.get("sensor.energy_site_solar_power").state == "1.185"
+    assert hass.states.get("calendar.energy_site_buy_tariff").state != STATE_UNAVAILABLE
+
+    # A stream disconnect fails the live and info/tariff coordinators.
+    mock_add_connection_listener.send(False)
+    await hass.async_block_till_done()
+
+    assert hass.states.get("sensor.energy_site_solar_power").state == STATE_UNAVAILABLE
+    assert hass.states.get("calendar.energy_site_buy_tariff").state == STATE_UNAVAILABLE
+
+    # A streamed live_status document restores the live coordinator on reconnect.
+    live_status = deepcopy(LIVE_STATUS["response"])
+    live_status["solar_power"] = 456
+    mock_energy_live_stream.send(live_status)
+    await hass.async_block_till_done()
+    assert hass.states.get("sensor.energy_site_solar_power").state == "0.456"
+
+    # A streamed site_info document restores the info/tariff coordinator.
+    slim_site_info = {
+        key: value
+        for key, value in deepcopy(SITE_INFO["response"]).items()
+        if key != "tariff_content_v2"
+    }
+    mock_energy_info_stream.send(slim_site_info)
+    await hass.async_block_till_done()
+    assert hass.states.get("calendar.energy_site_buy_tariff").state != STATE_UNAVAILABLE
+
+    assert not [
+        flow
+        for flow in hass.config_entries.flow.async_progress()
+        if flow["handler"] == DOMAIN
+    ]

tests/components/teslemetry/test_select.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Awaitable, Callable
 from copy import deepcopy
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from freezegun.api import FrozenDateTimeFactory
 import pytest
@@ -16,10 +16,7 @@ from homeassistant.components.select import (
     DOMAIN as SELECT_DOMAIN,
     SERVICE_SELECT_OPTION,
 )
-from homeassistant.components.teslemetry.coordinator import (
-    ENERGY_INFO_INTERVAL,
-    VEHICLE_INTERVAL,
-)
+from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
 from homeassistant.components.teslemetry.select import HIGH, LEVEL, LOW, MEDIUM, OFF
 from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
 from homeassistant.core import HomeAssistant
@@ -534,8 +531,8 @@ async def test_export_rule_restore(
 )
 async def test_export_rule_update_attrs_logic(
     hass: HomeAssistant,
-    freezer: FrozenDateTimeFactory,
     mock_site_info: AsyncMock,
+    mock_energy_info_stream: MagicMock,
     previous_data: dict,
     new_data: str | None,
     expected_state: str,
@@ -549,14 +546,12 @@ async def test_export_rule_update_attrs_logic(
     # Set up platform
     await setup_platform(hass, [Platform.SELECT])
 
-    # Change the state
-    test_site_info = deepcopy(SITE_INFO)
-    test_site_info["response"]["components"].update(new_data)
-    mock_site_info.side_effect = lambda: test_site_info
-
-    # Coordinator refresh
-    freezer.tick(ENERGY_INFO_INTERVAL)
-    async_fire_time_changed(hass)
+    # Change the state via a streamed site_info event, driven through the
+    # callback the integration registered with the library.
+    streamed_site_info = deepcopy(SITE_INFO["response"])
+    streamed_site_info.pop("tariff_content_v2", None)
+    streamed_site_info["components"].update(new_data)
+    mock_energy_info_stream.send(streamed_site_info)
     await hass.async_block_till_done()
 
     # Check the final state matches expected

tests/components/teslemetry/test_sensor.py

@@ -1,7 +1,7 @@
 """Test the Teslemetry sensor platform."""
 
 from copy import deepcopy
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 from freezegun.api import FrozenDateTimeFactory
 import pytest
@@ -23,7 +23,13 @@ from homeassistant.helpers import entity_registry as er
 from homeassistant.util.unit_conversion import PressureConverter
 
 from . import assert_entities, assert_entities_alt, setup_platform
-from .const import ENERGY_HISTORY_EMPTY, METADATA, PRODUCTS, VEHICLE_DATA_ALT
+from .const import (
+    ENERGY_HISTORY_EMPTY,
+    LIVE_STATUS,
+    METADATA,
+    PRODUCTS,
+    VEHICLE_DATA_ALT,
+)
 
 from tests.common import async_fire_time_changed
 
@@ -38,6 +44,59 @@ def _products_with_driver_assist(driver_assist: str) -> dict:
     return products
 
 
+def _live_status(**overrides: object) -> dict:
+    """Return a copy of the live_status document with overrides applied."""
+    data = deepcopy(LIVE_STATUS["response"])
+    data.update(overrides)
+    return data
+
+
+async def test_energy_live_status_stream_updates(
+    hass: HomeAssistant,
+    mock_energy_live_stream: MagicMock,
+) -> None:
+    """A streamed live_status document drives the energy sensor states."""
+    await setup_platform(hass, [Platform.SENSOR])
+
+    # The REST cold read populated the fixture values.
+    assert hass.states.get("sensor.energy_site_solar_power").state == "1.185"
+    assert hass.states.get("sensor.wall_connector_power").state == "0.0"
+
+    live_status = _live_status(solar_power=456)
+    live_status["wall_connectors"][0]["wall_connector_power"] = 789
+    mock_energy_live_stream.send(live_status)
+    await hass.async_block_till_done()
+
+    assert hass.states.get("sensor.energy_site_solar_power").state == "0.456"
+    assert hass.states.get("sensor.wall_connector_power").state == "0.789"
+
+
+@pytest.mark.usefixtures("mock_energy_only")
+async def test_energy_only_account_streams_live_status(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_stream_listen: AsyncMock,
+    mock_energy_live_stream: MagicMock,
+) -> None:
+    """An energy-only account starts one stream and streams live_status to sensors."""
+    entry = await setup_platform(hass, [Platform.SENSOR])
+    assert entry.state is ConfigEntryState.LOADED
+
+    # The account-wide stream is started and the live_status listener registered.
+    mock_stream_listen.assert_called_once()
+    mock_energy_live_stream.assert_called_once()
+
+    mock_energy_live_stream.send(_live_status(solar_power=999))
+    await hass.async_block_till_done()
+    assert hass.states.get("sensor.energy_site_solar_power").state == "0.999"
+
+    # Credit sensors are still created for an energy-only account.
+    assert entry.unique_id is not None
+    assert entity_registry.async_get_entity_id(
+        Platform.SENSOR, "teslemetry", f"{entry.unique_id}_credit_quota"
+    )
+
+
 @pytest.mark.usefixtures("entity_registry_enabled_by_default")
 async def test_sensors(
     hass: HomeAssistant,