diff --git a/homeassistant/components/modbus/__init__.py b/homeassistant/components/modbus/__init__.py
index df1fc2f6995..830dadfcdb8 100644
--- a/homeassistant/components/modbus/__init__.py
+++ b/homeassistant/components/modbus/__init__.py
@@ -369,6 +369,11 @@ SERVICE_WRITE_COIL_SCHEMA = vol.Schema(
         ),
     }
 )
+SERVICE_STOP_START_SCHEMA = vol.Schema(
+    {
+        vol.Required(ATTR_HUB): cv.string,
+    }
+)
 
 
 def get_hub(hass: HomeAssistant, name: str) -> ModbusHub:
@@ -379,5 +384,9 @@ def get_hub(hass: HomeAssistant, name: str) -> ModbusHub:
 async def async_setup(hass, config):
     """Set up Modbus component."""
     return await async_modbus_setup(
-        hass, config, SERVICE_WRITE_REGISTER_SCHEMA, SERVICE_WRITE_COIL_SCHEMA
+        hass,
+        config,
+        SERVICE_WRITE_REGISTER_SCHEMA,
+        SERVICE_WRITE_COIL_SCHEMA,
+        SERVICE_STOP_START_SCHEMA,
     )
diff --git a/homeassistant/components/modbus/base_platform.py b/homeassistant/components/modbus/base_platform.py
index efcb70b5b16..3167313fae8 100644
--- a/homeassistant/components/modbus/base_platform.py
+++ b/homeassistant/components/modbus/base_platform.py
@@ -5,7 +5,7 @@ from abc import abstractmethod
 from datetime import timedelta
 import logging
 import struct
-from typing import Any
+from typing import Any, Callable
 
 from homeassistant.const import (
     CONF_ADDRESS,
@@ -21,6 +21,8 @@ from homeassistant.const import (
     CONF_STRUCTURE,
     STATE_ON,
 )
+from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity import Entity, ToggleEntity
 from homeassistant.helpers.event import async_call_later, async_track_time_interval
 from homeassistant.helpers.restore_state import RestoreEntity
@@ -50,6 +52,8 @@ from .const import (
     CONF_VERIFY,
     CONF_WRITE_TYPE,
     DATA_TYPE_STRING,
+    SIGNAL_START_ENTITY,
+    SIGNAL_STOP_ENTITY,
 )
 from .modbus import ModbusHub
 
@@ -73,6 +77,7 @@ class BasePlatform(Entity):
         self._value = None
         self._scan_interval = int(entry[CONF_SCAN_INTERVAL])
         self._call_active = False
+        self._cancel_timer: Callable[[], None] | None = None
 
         self._attr_name = entry[CONF_NAME]
         self._attr_should_poll = False
@@ -86,13 +91,35 @@ class BasePlatform(Entity):
     async def async_update(self, now=None):
         """Virtual function to be overwritten."""
 
-    async def async_base_added_to_hass(self):
-        """Handle entity which will be added."""
+    @callback
+    def async_remote_start(self) -> None:
+        """Remote start entity."""
+        if self._cancel_timer:
+            self._cancel_timer()
+            self._cancel_timer = None
         if self._scan_interval > 0:
-            cancel_func = async_track_time_interval(
+            self._cancel_timer = async_track_time_interval(
                 self.hass, self.async_update, timedelta(seconds=self._scan_interval)
             )
-            self._hub.entity_timers.append(cancel_func)
+        self._attr_available = True
+        self.async_write_ha_state()
+
+    @callback
+    def async_remote_stop(self) -> None:
+        """Remote stop entity."""
+        if self._cancel_timer:
+            self._cancel_timer()
+            self._cancel_timer = None
+        self._attr_available = False
+        self.async_write_ha_state()
+
+    async def async_base_added_to_hass(self):
+        """Handle entity which will be added."""
+        self.async_remote_start()
+        async_dispatcher_connect(self.hass, SIGNAL_STOP_ENTITY, self.async_remote_stop)
+        async_dispatcher_connect(
+            self.hass, SIGNAL_START_ENTITY, self.async_remote_start
+        )
 
 
 class BaseStructPlatform(BasePlatform, RestoreEntity):
diff --git a/homeassistant/components/modbus/const.py b/homeassistant/components/modbus/const.py
index 3bcd85053d2..f2c3b7dd19c 100644
--- a/homeassistant/components/modbus/const.py
+++ b/homeassistant/components/modbus/const.py
@@ -106,6 +106,12 @@ CALL_TYPE_X_REGISTER_HOLDINGS = "holdings"
 # service calls
 SERVICE_WRITE_COIL = "write_coil"
 SERVICE_WRITE_REGISTER = "write_register"
+SERVICE_STOP = "stop"
+SERVICE_RESTART = "restart"
+
+# dispatcher signals
+SIGNAL_STOP_ENTITY = "modbus.stop"
+SIGNAL_START_ENTITY = "modbus.start"
 
 # integration names
 DEFAULT_HUB = "modbus_hub"
diff --git a/homeassistant/components/modbus/cover.py b/homeassistant/components/modbus/cover.py
index 5fa77eb1cb8..ca4f25ca1cc 100644
--- a/homeassistant/components/modbus/cover.py
+++ b/homeassistant/components/modbus/cover.py
@@ -74,6 +74,7 @@ class ModbusCover(BasePlatform, CoverEntity, RestoreEntity):
         self._status_register_type = config[CONF_STATUS_REGISTER_TYPE]
 
         self._attr_supported_features = SUPPORT_OPEN | SUPPORT_CLOSE
+        self._attr_is_closed = False
 
         # If we read cover status from coil, and not from optional status register,
         # we interpret boolean value False as closed cover, and value True as open cover.
diff --git a/homeassistant/components/modbus/modbus.py b/homeassistant/components/modbus/modbus.py
index 42505215622..fb9af241048 100644
--- a/homeassistant/components/modbus/modbus.py
+++ b/homeassistant/components/modbus/modbus.py
@@ -20,8 +20,9 @@ from homeassistant.const import (
     CONF_TYPE,
     EVENT_HOMEASSISTANT_STOP,
 )
-from homeassistant.core import CALLBACK_TYPE, callback
+from homeassistant.core import callback
 from homeassistant.helpers.discovery import async_load_platform
+from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.event import async_call_later
 
 from .const import (
@@ -51,8 +52,12 @@ from .const import (
     PLATFORMS,
     RTUOVERTCP,
     SERIAL,
+    SERVICE_RESTART,
+    SERVICE_STOP,
     SERVICE_WRITE_COIL,
     SERVICE_WRITE_REGISTER,
+    SIGNAL_START_ENTITY,
+    SIGNAL_STOP_ENTITY,
     TCP,
     UDP,
 )
@@ -107,7 +112,11 @@ PYMODBUS_CALL = [
 
 
 async def async_modbus_setup(
-    hass, config, service_write_register_schema, service_write_coil_schema
+    hass,
+    config,
+    service_write_register_schema,
+    service_write_coil_schema,
+    service_stop_start_schema,
 ):
     """Set up Modbus component."""
 
@@ -131,9 +140,9 @@ async def async_modbus_setup(
     async def async_stop_modbus(event):
         """Stop Modbus service."""
 
+        async_dispatcher_send(hass, SIGNAL_STOP_ENTITY)
         for client in hub_collect.values():
             await client.async_close()
-            del client
 
     hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_stop_modbus)
 
@@ -142,15 +151,15 @@ async def async_modbus_setup(
         unit = int(float(service.data[ATTR_UNIT]))
         address = int(float(service.data[ATTR_ADDRESS]))
         value = service.data[ATTR_VALUE]
-        client_name = (
+        hub = hub_collect[
             service.data[ATTR_HUB] if ATTR_HUB in service.data else DEFAULT_HUB
-        )
+        ]
         if isinstance(value, list):
-            await hub_collect[client_name].async_pymodbus_call(
+            await hub.async_pymodbus_call(
                 unit, address, [int(float(i)) for i in value], CALL_TYPE_WRITE_REGISTERS
             )
         else:
-            await hub_collect[client_name].async_pymodbus_call(
+            await hub.async_pymodbus_call(
                 unit, address, int(float(value)), CALL_TYPE_WRITE_REGISTER
             )
 
@@ -166,35 +175,48 @@ async def async_modbus_setup(
         unit = service.data[ATTR_UNIT]
         address = service.data[ATTR_ADDRESS]
         state = service.data[ATTR_STATE]
-        client_name = (
+        hub = hub_collect[
             service.data[ATTR_HUB] if ATTR_HUB in service.data else DEFAULT_HUB
-        )
+        ]
         if isinstance(state, list):
-            await hub_collect[client_name].async_pymodbus_call(
-                unit, address, state, CALL_TYPE_WRITE_COILS
-            )
+            await hub.async_pymodbus_call(unit, address, state, CALL_TYPE_WRITE_COILS)
         else:
-            await hub_collect[client_name].async_pymodbus_call(
-                unit, address, state, CALL_TYPE_WRITE_COIL
-            )
+            await hub.async_pymodbus_call(unit, address, state, CALL_TYPE_WRITE_COIL)
 
     hass.services.async_register(
         DOMAIN, SERVICE_WRITE_COIL, async_write_coil, schema=service_write_coil_schema
     )
+
+    async def async_stop_hub(service):
+        """Stop Modbus hub."""
+        async_dispatcher_send(hass, SIGNAL_STOP_ENTITY)
+        hub = hub_collect[service.data[ATTR_HUB]]
+        await hub.async_close()
+
+    hass.services.async_register(
+        DOMAIN, SERVICE_STOP, async_stop_hub, schema=service_stop_start_schema
+    )
+
+    async def async_restart_hub(service):
+        """Restart Modbus hub."""
+        async_dispatcher_send(hass, SIGNAL_START_ENTITY)
+        hub = hub_collect[service.data[ATTR_HUB]]
+        await hub.async_restart()
+
+    hass.services.async_register(
+        DOMAIN, SERVICE_RESTART, async_restart_hub, schema=service_stop_start_schema
+    )
     return True
 
 
 class ModbusHub:
     """Thread safe wrapper class for pymodbus."""
 
-    name: str
-
     def __init__(self, hass, client_config):
         """Initialize the Modbus hub."""
 
         # generic configuration
         self._client = None
-        self.entity_timers: list[CALLBACK_TYPE] = []
         self._async_cancel_listener = None
         self._in_error = False
         self._lock = asyncio.Lock()
@@ -284,29 +306,40 @@ class ModbusHub:
         self._async_cancel_listener = None
         self._config_delay = 0
 
+    async def async_restart(self):
+        """Reconnect client."""
+        if self._client:
+            await self.async_close()
+
+        await self.async_setup()
+
     async def async_close(self):
         """Disconnect client."""
         if self._async_cancel_listener:
             self._async_cancel_listener()
             self._async_cancel_listener = None
-        for call in self.entity_timers:
-            call()
-        self.entity_timers = []
         async with self._lock:
             if self._client:
                 try:
                     self._client.close()
                 except ModbusException as exception_error:
                     self._log_error(str(exception_error))
+                del self._client
                 self._client = None
+                message = f"modbus {self.name} communication closed"
+                _LOGGER.warning(message)
 
     def _pymodbus_connect(self):
         """Connect client."""
         try:
-            return self._client.connect()
+            self._client.connect()
         except ModbusException as exception_error:
             self._log_error(str(exception_error), error_state=False)
             return False
+        else:
+            message = f"modbus {self.name} communication open"
+            _LOGGER.warning(message)
+            return True
 
     def _pymodbus_call(self, unit, address, value, use_call):
         """Call sync. pymodbus."""
diff --git a/homeassistant/components/modbus/services.yaml b/homeassistant/components/modbus/services.yaml
index 855303aef07..835927e4627 100644
--- a/homeassistant/components/modbus/services.yaml
+++ b/homeassistant/components/modbus/services.yaml
@@ -66,3 +66,25 @@ write_register:
       default: "modbus_hub"
       selector:
         text:
+stop:
+  name: Stop
+  description: Stop modbus hub.
+  fields:
+    hub:
+      name: Hub
+      description: Modbus hub name.
+      example: "hub1"
+      default: "modbus_hub"
+      selector:
+        text:
+restart:
+  name: Restart
+  description: Restart modbus hub (if running stop then start).
+  fields:
+    hub:
+      name: Hub
+      description: Modbus hub name.
+      example: "hub1"
+      default: "modbus_hub"
+      selector:
+        text:
diff --git a/tests/components/modbus/test_init.py b/tests/components/modbus/test_init.py
index 3ae271467ca..ba99df19b4d 100644
--- a/tests/components/modbus/test_init.py
+++ b/tests/components/modbus/test_init.py
@@ -53,6 +53,8 @@ from homeassistant.components.modbus.const import (
     MODBUS_DOMAIN as DOMAIN,
     RTUOVERTCP,
     SERIAL,
+    SERVICE_RESTART,
+    SERVICE_STOP,
     SERVICE_WRITE_COIL,
     SERVICE_WRITE_REGISTER,
     TCP,
@@ -82,6 +84,7 @@ from homeassistant.const import (
     EVENT_HOMEASSISTANT_STOP,
     STATE_ON,
     STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
 )
 from homeassistant.setup import async_setup_component
 import homeassistant.util.dt as dt_util
@@ -715,3 +718,54 @@ async def test_shutdown(hass, caplog, mock_pymodbus, mock_modbus_with_pymodbus):
     await hass.async_block_till_done()
     assert mock_pymodbus.close.called
     assert caplog.text == ""
+
+
+@pytest.mark.parametrize(
+    "do_config",
+    [
+        {
+            CONF_SENSORS: [
+                {
+                    CONF_NAME: TEST_ENTITY_NAME,
+                    CONF_ADDRESS: 51,
+                }
+            ]
+        },
+    ],
+)
+async def test_stop_restart(hass, caplog, mock_modbus):
+    """Run test for service stop."""
+
+    entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}"
+    assert hass.states.get(entity_id).state == STATE_UNKNOWN
+    hass.states.async_set(entity_id, 17)
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == "17"
+
+    mock_modbus.reset_mock()
+    caplog.clear()
+    data = {
+        ATTR_HUB: TEST_MODBUS_NAME,
+    }
+    await hass.services.async_call(DOMAIN, SERVICE_STOP, data, blocking=True)
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
+    assert mock_modbus.close.called
+    assert f"modbus {TEST_MODBUS_NAME} communication closed" in caplog.text
+
+    mock_modbus.reset_mock()
+    caplog.clear()
+    await hass.services.async_call(DOMAIN, SERVICE_RESTART, data, blocking=True)
+    await hass.async_block_till_done()
+    assert not mock_modbus.close.called
+    assert mock_modbus.connect.called
+    assert f"modbus {TEST_MODBUS_NAME} communication open" in caplog.text
+
+    mock_modbus.reset_mock()
+    caplog.clear()
+    await hass.services.async_call(DOMAIN, SERVICE_RESTART, data, blocking=True)
+    await hass.async_block_till_done()
+    assert mock_modbus.close.called
+    assert mock_modbus.connect.called
+    assert f"modbus {TEST_MODBUS_NAME} communication closed" in caplog.text
+    assert f"modbus {TEST_MODBUS_NAME} communication open" in caplog.text