Move solarlog coordinator to own file (#100402)

.coveragerc

@@ -1157,6 +1157,7 @@ omit =
     homeassistant/components/solaredge_local/sensor.py
     homeassistant/components/solarlog/__init__.py
     homeassistant/components/solarlog/sensor.py
+    homeassistant/components/solarlog/coordinator.py
     homeassistant/components/solax/__init__.py
     homeassistant/components/solax/sensor.py
     homeassistant/components/soma/__init__.py

homeassistant/components/solarlog/__init__.py

@@ -1,19 +1,10 @@
 """Solar-Log integration."""
-from datetime import timedelta
-import logging
-from urllib.parse import ParseResult, urlparse
-
-from requests.exceptions import HTTPError, Timeout
-from sunwatcher.solarlog.solarlog import SolarLog
-
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_HOST, Platform
+from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import update_coordinator
 
 from .const import DOMAIN
-
-_LOGGER = logging.getLogger(__name__)
+from .coordinator import SolarlogData
 
 PLATFORMS = [Platform.SENSOR]
 
@@ -30,45 +21,3 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
-
-
-class SolarlogData(update_coordinator.DataUpdateCoordinator):
-    """Get and update the latest data."""
-
-    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
-        """Initialize the data object."""
-        super().__init__(
-            hass, _LOGGER, name="SolarLog", update_interval=timedelta(seconds=60)
-        )
-
-        host_entry = entry.data[CONF_HOST]
-
-        url = urlparse(host_entry, "http")
-        netloc = url.netloc or url.path
-        path = url.path if url.netloc else ""
-        url = ParseResult("http", netloc, path, *url[3:])
-        self.unique_id = entry.entry_id
-        self.name = entry.title
-        self.host = url.geturl()
-
-    async def _async_update_data(self):
-        """Update the data from the SolarLog device."""
-        try:
-            data = await self.hass.async_add_executor_job(SolarLog, self.host)
-        except (OSError, Timeout, HTTPError) as err:
-            raise update_coordinator.UpdateFailed(err) from err
-
-        if data.time.year == 1999:
-            raise update_coordinator.UpdateFailed(
-                "Invalid data returned (can happen after Solarlog restart)."
-            )
-
-        self.logger.debug(
-            (
-                "Connection to Solarlog successful. Retrieving latest Solarlog update"
-                " of %s"
-            ),
-            data.time,
-        )
-
-        return data

homeassistant/components/solarlog/coordinator.py

@@ -0,0 +1,56 @@
+"""DataUpdateCoordinator for solarlog integration."""
+from datetime import timedelta
+import logging
+from urllib.parse import ParseResult, urlparse
+
+from requests.exceptions import HTTPError, Timeout
+from sunwatcher.solarlog.solarlog import SolarLog
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import update_coordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SolarlogData(update_coordinator.DataUpdateCoordinator):
+    """Get and update the latest data."""
+
+    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
+        """Initialize the data object."""
+        super().__init__(
+            hass, _LOGGER, name="SolarLog", update_interval=timedelta(seconds=60)
+        )
+
+        host_entry = entry.data[CONF_HOST]
+
+        url = urlparse(host_entry, "http")
+        netloc = url.netloc or url.path
+        path = url.path if url.netloc else ""
+        url = ParseResult("http", netloc, path, *url[3:])
+        self.unique_id = entry.entry_id
+        self.name = entry.title
+        self.host = url.geturl()
+
+    async def _async_update_data(self):
+        """Update the data from the SolarLog device."""
+        try:
+            data = await self.hass.async_add_executor_job(SolarLog, self.host)
+        except (OSError, Timeout, HTTPError) as err:
+            raise update_coordinator.UpdateFailed(err) from err
+
+        if data.time.year == 1999:
+            raise update_coordinator.UpdateFailed(
+                "Invalid data returned (can happen after Solarlog restart)."
+            )
+
+        self.logger.debug(
+            (
+                "Connection to Solarlog successful. Retrieving latest Solarlog update"
+                " of %s"
+            ),
+            data.time,
+        )
+
+        return data