diff --git a/homeassistant/components/kmtronic/__init__.py b/homeassistant/components/kmtronic/__init__.py
index edec0b32af2..b49efebc35e 100644
--- a/homeassistant/components/kmtronic/__init__.py
+++ b/homeassistant/components/kmtronic/__init__.py
@@ -1,10 +1,5 @@
 """The kmtronic integration."""
 
-import asyncio
-from datetime import timedelta
-import logging
-
-import aiohttp
 from pykmtronic.auth import Auth
 from pykmtronic.hub import KMTronicHubAPI
 
@@ -12,14 +7,12 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import aiohttp_client
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from .const import DATA_COORDINATOR, DATA_HUB, DOMAIN, MANUFACTURER, UPDATE_LISTENER
+from .const import DATA_COORDINATOR, DATA_HUB, DOMAIN, UPDATE_LISTENER
+from .coordinator import KMtronicCoordinator
 
 PLATFORMS = [Platform.SWITCH]
 
-_LOGGER = logging.getLogger(__name__)
-
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up kmtronic from a config entry."""
@@ -31,24 +24,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         entry.data[CONF_PASSWORD],
     )
     hub = KMTronicHubAPI(auth)
-
-    async def async_update_data():
-        try:
-            async with asyncio.timeout(10):
-                await hub.async_update_relays()
-        except aiohttp.client_exceptions.ClientResponseError as err:
-            raise UpdateFailed(f"Wrong credentials: {err}") from err
-        except aiohttp.client_exceptions.ClientConnectorError as err:
-            raise UpdateFailed(f"Error communicating with API: {err}") from err
-
-    coordinator = DataUpdateCoordinator(
-        hass,
-        _LOGGER,
-        config_entry=entry,
-        name=f"{MANUFACTURER} {hub.name}",
-        update_method=async_update_data,
-        update_interval=timedelta(seconds=30),
-    )
+    coordinator = KMtronicCoordinator(hass, entry, hub)
     await coordinator.async_config_entry_first_refresh()
 
     hass.data.setdefault(DOMAIN, {})
diff --git a/homeassistant/components/kmtronic/coordinator.py b/homeassistant/components/kmtronic/coordinator.py
new file mode 100644
index 00000000000..8a94949dea6
--- /dev/null
+++ b/homeassistant/components/kmtronic/coordinator.py
@@ -0,0 +1,46 @@
+"""The kmtronic integration."""
+
+import asyncio
+from datetime import timedelta
+import logging
+
+from aiohttp.client_exceptions import ClientConnectorError, ClientResponseError
+from pykmtronic.hub import KMTronicHubAPI
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import MANUFACTURER
+
+PLATFORMS = [Platform.SWITCH]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class KMtronicCoordinator(DataUpdateCoordinator[None]):
+    """Coordinator for KMTronic."""
+
+    def __init__(
+        self, hass: HomeAssistant, entry: ConfigEntry, hub: KMTronicHubAPI
+    ) -> None:
+        """Initialize the KMTronic coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=entry,
+            name=f"{MANUFACTURER} {hub.name}",
+            update_interval=timedelta(seconds=30),
+        )
+        self.hub = hub
+
+    async def _async_update_data(self) -> None:
+        """Fetch the latest data from the source."""
+        try:
+            async with asyncio.timeout(10):
+                await self.hub.async_update_relays()
+        except ClientResponseError as err:
+            raise UpdateFailed(f"Wrong credentials: {err}") from err
+        except ClientConnectorError as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err