diff --git a/.coveragerc b/.coveragerc
index c58601b2245..8b3067313d3 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -1062,6 +1062,7 @@ omit =
     homeassistant/components/rabbitair/fan.py
     homeassistant/components/rachio/__init__.py
     homeassistant/components/rachio/binary_sensor.py
+    homeassistant/components/rachio/coordinator.py
     homeassistant/components/rachio/device.py
     homeassistant/components/rachio/entity.py
     homeassistant/components/rachio/switch.py
diff --git a/homeassistant/components/rachio/__init__.py b/homeassistant/components/rachio/__init__.py
index e5a64d25fc8..f91a7b4fa75 100644
--- a/homeassistant/components/rachio/__init__.py
+++ b/homeassistant/components/rachio/__init__.py
@@ -83,7 +83,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         raise ConfigEntryNotReady from error
 
     # Check for Rachio controller devices
-    if not person.controllers:
+    if not person.controllers and not person.base_stations:
         _LOGGER.error("No Rachio devices found in account %s", person.username)
         return False
     _LOGGER.info(
@@ -91,10 +91,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             "%d Rachio device(s) found; The url %s must be accessible from the internet"
             " in order to receive updates"
         ),
-        len(person.controllers),
+        len(person.controllers) + len(person.base_stations),
         webhook_url,
     )
 
+    for base in person.base_stations:
+        await base.coordinator.async_config_entry_first_refresh()
+
     # Enable platform
     hass.data.setdefault(DOMAIN, {})[entry.entry_id] = person
     async_register_webhook(hass, entry)
diff --git a/homeassistant/components/rachio/const.py b/homeassistant/components/rachio/const.py
index dad044e5049..22c92be2b74 100644
--- a/homeassistant/components/rachio/const.py
+++ b/homeassistant/components/rachio/const.py
@@ -26,6 +26,7 @@ KEY_NAME = "name"
 KEY_MODEL = "model"
 KEY_ON = "on"
 KEY_DURATION = "totalDuration"
+KEY_DURATION_MINUTES = "duration"
 KEY_RAIN_DELAY = "rainDelayExpirationDate"
 KEY_RAIN_DELAY_END = "endTime"
 KEY_RAIN_SENSOR_TRIPPED = "rainSensorTripped"
@@ -47,6 +48,21 @@ KEY_CUSTOM_SHADE = "customShade"
 KEY_CUSTOM_CROP = "customCrop"
 KEY_CUSTOM_SLOPE = "customSlope"
 
+# Smart Hose timer
+KEY_BASE_STATIONS = "baseStations"
+KEY_VALVES = "valves"
+KEY_REPORTED_STATE = "reportedState"
+KEY_STATE = "state"
+KEY_CONNECTED = "connected"
+KEY_CURRENT_STATUS = "lastWateringAction"
+KEY_DETECT_FLOW = "detectFlow"
+KEY_BATTERY_STATUS = "batteryStatus"
+KEY_REASON = "reason"
+KEY_DEFAULT_RUNTIME = "defaultRuntimeSeconds"
+KEY_DURATION_SECONDS = "durationSeconds"
+KEY_FLOW_DETECTED = "flowDetected"
+KEY_START_TIME = "start"
+
 STATUS_ONLINE = "ONLINE"
 
 MODEL_GENERATION_1 = "GENERATION1"
@@ -56,6 +72,7 @@ SERVICE_PAUSE_WATERING = "pause_watering"
 SERVICE_RESUME_WATERING = "resume_watering"
 SERVICE_STOP_WATERING = "stop_watering"
 SERVICE_SET_ZONE_MOISTURE = "set_zone_moisture_percent"
+SERVICE_START_WATERING = "start_watering"
 SERVICE_START_MULTIPLE_ZONES = "start_multiple_zone_schedule"
 
 SIGNAL_RACHIO_UPDATE = f"{DOMAIN}_update"
diff --git a/homeassistant/components/rachio/coordinator.py b/homeassistant/components/rachio/coordinator.py
new file mode 100644
index 00000000000..4f8cc87daef
--- /dev/null
+++ b/homeassistant/components/rachio/coordinator.py
@@ -0,0 +1,56 @@
+"""Coordinator object for the Rachio integration."""
+
+from datetime import timedelta
+import logging
+from typing import Any
+
+from rachiopy import Rachio
+from requests.exceptions import Timeout
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.debounce import Debouncer
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, KEY_ID, KEY_VALVES
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_DELAY_TIME = 8
+
+
+class RachioUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Coordinator Class for Rachio Hose Timers."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        rachio: Rachio,
+        base_station,
+        base_count: int,
+    ) -> None:
+        """Initialize the Rachio Update Coordinator."""
+        self.hass = hass
+        self.rachio = rachio
+        self.base_station = base_station
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"{DOMAIN} update coordinator",
+            # To avoid exceeding the rate limit, increase polling interval for
+            # each additional base station on the account
+            update_interval=timedelta(minutes=(base_count + 1)),
+            # Debouncer used because the API takes a bit to update state changes
+            request_refresh_debouncer=Debouncer(
+                hass, _LOGGER, cooldown=UPDATE_DELAY_TIME, immediate=False
+            ),
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Update smart hose timer data."""
+        try:
+            data = await self.hass.async_add_executor_job(
+                self.rachio.valve.list_valves, self.base_station[KEY_ID]
+            )
+        except Timeout as err:
+            raise UpdateFailed(f"Could not connect to the Rachio API: {err}") from err
+        return {valve[KEY_ID]: valve for valve in data[1][KEY_VALVES]}
diff --git a/homeassistant/components/rachio/device.py b/homeassistant/components/rachio/device.py
index dea91fcc6fd..c018d7e6f86 100644
--- a/homeassistant/components/rachio/device.py
+++ b/homeassistant/components/rachio/device.py
@@ -17,6 +17,7 @@ from homeassistant.helpers import config_validation as cv
 
 from .const import (
     DOMAIN,
+    KEY_BASE_STATIONS,
     KEY_DEVICES,
     KEY_ENABLED,
     KEY_EXTERNAL_ID,
@@ -37,6 +38,7 @@ from .const import (
     SERVICE_STOP_WATERING,
     WEBHOOK_CONST_ID,
 )
+from .coordinator import RachioUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -67,6 +69,7 @@ class RachioPerson:
         self.username = None
         self._id: str | None = None
         self._controllers: list[RachioIro] = []
+        self._base_stations: list[RachioBaseStation] = []
 
     async def async_setup(self, hass: HomeAssistant) -> None:
         """Create rachio devices and services."""
@@ -78,30 +81,34 @@ class RachioPerson:
                 can_pause = True
                 break
 
-        all_devices = [rachio_iro.name for rachio_iro in self._controllers]
+        all_controllers = [rachio_iro.name for rachio_iro in self._controllers]
 
         def pause_water(service: ServiceCall) -> None:
             """Service to pause watering on all or specific controllers."""
             duration = service.data[ATTR_DURATION]
-            devices = service.data.get(ATTR_DEVICES, all_devices)
+            devices = service.data.get(ATTR_DEVICES, all_controllers)
             for iro in self._controllers:
                 if iro.name in devices:
                     iro.pause_watering(duration)
 
         def resume_water(service: ServiceCall) -> None:
             """Service to resume watering on all or specific controllers."""
-            devices = service.data.get(ATTR_DEVICES, all_devices)
+            devices = service.data.get(ATTR_DEVICES, all_controllers)
             for iro in self._controllers:
                 if iro.name in devices:
                     iro.resume_watering()
 
         def stop_water(service: ServiceCall) -> None:
             """Service to stop watering on all or specific controllers."""
-            devices = service.data.get(ATTR_DEVICES, all_devices)
+            devices = service.data.get(ATTR_DEVICES, all_controllers)
             for iro in self._controllers:
                 if iro.name in devices:
                     iro.stop_watering()
 
+        # If only hose timers on account, none of these services apply
+        if not all_controllers:
+            return
+
         hass.services.async_register(
             DOMAIN,
             SERVICE_STOP_WATERING,
@@ -145,6 +152,9 @@ class RachioPerson:
             raise ConfigEntryNotReady(f"API Error: {data}")
         self.username = data[1][KEY_USERNAME]
         devices: list[dict[str, Any]] = data[1][KEY_DEVICES]
+        base_station_data = rachio.valve.list_base_stations(self._id)
+        base_stations: list[dict[str, Any]] = base_station_data[1][KEY_BASE_STATIONS]
+
         for controller in devices:
             webhooks = rachio.notification.get_device_webhook(controller[KEY_ID])[1]
             # The API does not provide a way to tell if a controller is shared
@@ -174,6 +184,14 @@ class RachioPerson:
             rachio_iro.setup()
             self._controllers.append(rachio_iro)
 
+        base_count = len(base_stations)
+        self._base_stations.extend(
+            RachioBaseStation(
+                rachio, base, RachioUpdateCoordinator(hass, rachio, base, base_count)
+            )
+            for base in base_stations
+        )
+
         _LOGGER.info('Using Rachio API as user "%s"', self.username)
 
     @property
@@ -186,6 +204,11 @@ class RachioPerson:
         """Get a list of controllers managed by this account."""
         return self._controllers
 
+    @property
+    def base_stations(self) -> list[RachioBaseStation]:
+        """List of smart hose timer base stations."""
+        return self._base_stations
+
     def start_multiple_zones(self, zones) -> None:
         """Start multiple zones."""
         self.rachio.zone.start_multiple(zones)
@@ -321,6 +344,28 @@ class RachioIro:
         _LOGGER.debug("Resuming watering on %s", self)
 
 
+class RachioBaseStation:
+    """Represent a smart hose timer base station."""
+
+    def __init__(
+        self, rachio: Rachio, data: dict[str, Any], coordinator: RachioUpdateCoordinator
+    ) -> None:
+        """Initialize a hose time base station."""
+        self.rachio = rachio
+        self._id = data[KEY_ID]
+        self.serial_number = data[KEY_SERIAL_NUMBER]
+        self.mac_address = data[KEY_MAC_ADDRESS]
+        self.coordinator = coordinator
+
+    def start_watering(self, valve_id: str, duration: int) -> None:
+        """Start watering on this valve."""
+        self.rachio.valve.start_watering(valve_id, duration)
+
+    def stop_watering(self, valve_id: str) -> None:
+        """Stop watering on this valve."""
+        self.rachio.valve.stop_watering(valve_id)
+
+
 def is_invalid_auth_code(http_status_code: int) -> bool:
     """HTTP status codes that mean invalid auth."""
     return http_status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN)
diff --git a/homeassistant/components/rachio/icons.json b/homeassistant/components/rachio/icons.json
index 3b3ec860514..dfab8788fc8 100644
--- a/homeassistant/components/rachio/icons.json
+++ b/homeassistant/components/rachio/icons.json
@@ -14,6 +14,7 @@
     "start_multiple_zone_schedule": "mdi:play",
     "pause_watering": "mdi:pause",
     "resume_watering": "mdi:play",
-    "stop_watering": "mdi:stop"
+    "stop_watering": "mdi:stop",
+    "start_watering": "mdi:water"
   }
 }
diff --git a/homeassistant/components/rachio/services.yaml b/homeassistant/components/rachio/services.yaml
index 6a6a8bf5cf6..72582de870a 100644
--- a/homeassistant/components/rachio/services.yaml
+++ b/homeassistant/components/rachio/services.yaml
@@ -11,6 +11,17 @@ set_zone_moisture_percent:
           min: 0
           max: 100
           unit_of_measurement: "%"
+start_watering:
+  target:
+    entity:
+      integration: rachio
+      domain: switch
+  fields:
+    duration:
+      example: 15
+      required: false
+      selector:
+        object:
 start_multiple_zone_schedule:
   target:
     entity:
diff --git a/homeassistant/components/rachio/strings.json b/homeassistant/components/rachio/strings.json
index 560c300db17..2e4de262d21 100644
--- a/homeassistant/components/rachio/strings.json
+++ b/homeassistant/components/rachio/strings.json
@@ -63,6 +63,16 @@
         }
       }
     },
+    "start_watering": {
+      "name": "Start watering",
+      "description": "Start a single zone, a schedule or any number of smart hose timers.",
+      "fields": {
+        "duration": {
+          "name": "Duration",
+          "description": "Number of minutes to run. For sprinkler zones the maximum duration is 3 hours, or 24 hours for smart hose timers. Leave empty for schedules."
+        }
+      }
+    },
     "pause_watering": {
       "name": "Pause watering",
       "description": "Pause any currently running zones or schedules.",
diff --git a/homeassistant/components/rachio/switch.py b/homeassistant/components/rachio/switch.py
index 8b8d10248e0..fe3d455df3c 100644
--- a/homeassistant/components/rachio/switch.py
+++ b/homeassistant/components/rachio/switch.py
@@ -11,19 +11,28 @@ import voluptuous as vol
 from homeassistant.components.switch import SwitchEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_ENTITY_ID, ATTR_ID
-from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, ServiceCall, callback
 from homeassistant.exceptions import HomeAssistantError
-from homeassistant.helpers import config_validation as cv, entity_platform
+from homeassistant.helpers import (
+    config_validation as cv,
+    device_registry as dr,
+    entity_platform,
+)
+from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from homeassistant.util.dt import as_timestamp, now, parse_datetime, utc_from_timestamp
 
 from .const import (
     CONF_MANUAL_RUN_MINS,
     DEFAULT_MANUAL_RUN_MINS,
+    DEFAULT_NAME,
     DOMAIN as DOMAIN_RACHIO,
+    KEY_CONNECTED,
+    KEY_CURRENT_STATUS,
     KEY_CUSTOM_CROP,
     KEY_CUSTOM_SHADE,
     KEY_CUSTOM_SLOPE,
@@ -36,7 +45,9 @@ from .const import (
     KEY_ON,
     KEY_RAIN_DELAY,
     KEY_RAIN_DELAY_END,
+    KEY_REPORTED_STATE,
     KEY_SCHEDULE_ID,
+    KEY_STATE,
     KEY_SUBTYPE,
     KEY_SUMMARY,
     KEY_TYPE,
@@ -46,6 +57,7 @@ from .const import (
     SCHEDULE_TYPE_FLEX,
     SERVICE_SET_ZONE_MOISTURE,
     SERVICE_START_MULTIPLE_ZONES,
+    SERVICE_START_WATERING,
     SIGNAL_RACHIO_CONTROLLER_UPDATE,
     SIGNAL_RACHIO_RAIN_DELAY_UPDATE,
     SIGNAL_RACHIO_SCHEDULE_UPDATE,
@@ -55,6 +67,7 @@ from .const import (
     SLOPE_SLIGHT,
     SLOPE_STEEP,
 )
+from .coordinator import RachioUpdateCoordinator
 from .device import RachioPerson
 from .entity import RachioDevice
 from .webhooks import (
@@ -80,6 +93,7 @@ ATTR_SCHEDULE_ENABLED = "Enabled"
 ATTR_SCHEDULE_DURATION = "Duration"
 ATTR_SCHEDULE_TYPE = "Type"
 ATTR_SORT_ORDER = "sortOrder"
+ATTR_WATERING_DURATION = "Watering Duration seconds"
 ATTR_ZONE_NUMBER = "Zone number"
 ATTR_ZONE_SHADE = "Shade"
 ATTR_ZONE_SLOPE = "Slope"
@@ -141,6 +155,19 @@ async def async_setup_entry(
         else:
             raise HomeAssistantError("No matching zones found in given entity_ids")
 
+    platform = entity_platform.async_get_current_platform()
+    platform.async_register_entity_service(
+        SERVICE_START_WATERING,
+        {
+            vol.Optional(ATTR_DURATION): cv.positive_int,
+        },
+        "turn_on",
+    )
+
+    # If only hose timers on account, none of these services apply
+    if not zone_entities:
+        return
+
     hass.services.async_register(
         DOMAIN_RACHIO,
         SERVICE_START_MULTIPLE_ZONES,
@@ -176,6 +203,11 @@ def _create_entities(hass: HomeAssistant, config_entry: ConfigEntry) -> list[Ent
             RachioSchedule(person, controller, schedule, current_schedule)
             for schedule in schedules + flex_schedules
         )
+    entities.extend(
+        RachioValve(person, base_station, valve, base_station.coordinator)
+        for base_station in person.base_stations
+        for valve in base_station.coordinator.data.values()
+    )
     return entities
 
 
@@ -246,9 +278,9 @@ class RachioRainDelay(RachioSwitch):
     _attr_has_entity_name = True
     _attr_translation_key = "rain_delay"
 
-    def __init__(self, controller):
+    def __init__(self, controller) -> None:
         """Set up a Rachio rain delay switch."""
-        self._cancel_update = None
+        self._cancel_update: CALLBACK_TYPE | None = None
         super().__init__(controller)
 
     @property
@@ -324,7 +356,7 @@ class RachioZone(RachioSwitch):
 
     _attr_icon = "mdi:water"
 
-    def __init__(self, person, controller, data, current_schedule):
+    def __init__(self, person, controller, data, current_schedule) -> None:
         """Initialize a new Rachio Zone."""
         self.id = data[KEY_ID]
         self._attr_name = data[KEY_NAME]
@@ -379,11 +411,14 @@ class RachioZone(RachioSwitch):
         self.turn_off()
 
         # Start this zone
-        manual_run_time = timedelta(
-            minutes=self._person.config_entry.options.get(
-                CONF_MANUAL_RUN_MINS, DEFAULT_MANUAL_RUN_MINS
+        if ATTR_DURATION in kwargs:
+            manual_run_time = timedelta(minutes=kwargs[ATTR_DURATION])
+        else:
+            manual_run_time = timedelta(
+                minutes=self._person.config_entry.options.get(
+                    CONF_MANUAL_RUN_MINS, DEFAULT_MANUAL_RUN_MINS
+                )
             )
-        )
         # The API limit is 3 hours, and requires an int be passed
         self._controller.rachio.zone.start(self.zone_id, manual_run_time.seconds)
         _LOGGER.debug(
@@ -435,7 +470,7 @@ class RachioZone(RachioSwitch):
 class RachioSchedule(RachioSwitch):
     """Representation of one fixed schedule on the Rachio Iro."""
 
-    def __init__(self, person, controller, data, current_schedule):
+    def __init__(self, person, controller, data, current_schedule) -> None:
         """Initialize a new Rachio Schedule."""
         self._schedule_id = data[KEY_ID]
         self._duration = data[KEY_DURATION]
@@ -509,3 +544,70 @@ class RachioSchedule(RachioSwitch):
                 self.hass, SIGNAL_RACHIO_SCHEDULE_UPDATE, self._async_handle_update
             )
         )
+
+
+class RachioValve(CoordinatorEntity[RachioUpdateCoordinator], SwitchEntity):
+    """Representation of one smart hose timer valve."""
+
+    def __init__(
+        self, person, base, data, coordinator: RachioUpdateCoordinator
+    ) -> None:
+        """Initialize a new smart hose valve."""
+        super().__init__(coordinator)
+        self._person = person
+        self._base = base
+        self.id = data[KEY_ID]
+        self._attr_name = data[KEY_NAME]
+        self._attr_unique_id = f"{self.id}-valve"
+        self._static_attrs = data[KEY_STATE][KEY_REPORTED_STATE]
+        self._attr_is_on = KEY_CURRENT_STATUS in self._static_attrs
+        self._attr_device_info = DeviceInfo(
+            identifiers={
+                (
+                    DOMAIN_RACHIO,
+                    self.id,
+                )
+            },
+            connections={(dr.CONNECTION_NETWORK_MAC, self._base.mac_address)},
+            manufacturer=DEFAULT_NAME,
+            model="Smart Hose Timer",
+            name=self._attr_name,
+            configuration_url="https://app.rach.io",
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return if the valve is available."""
+        return super().available and self._static_attrs[KEY_CONNECTED]
+
+    def turn_on(self, **kwargs: Any) -> None:
+        """Turn on this valve."""
+        if ATTR_DURATION in kwargs:
+            manual_run_time = timedelta(minutes=kwargs[ATTR_DURATION])
+        else:
+            manual_run_time = timedelta(
+                minutes=self._person.config_entry.options.get(
+                    CONF_MANUAL_RUN_MINS, DEFAULT_MANUAL_RUN_MINS
+                )
+            )
+
+        self._base.start_watering(self.id, manual_run_time.seconds)
+        self._attr_is_on = True
+        self.schedule_update_ha_state(force_refresh=True)
+        _LOGGER.debug("Starting valve %s for %s", self.name, str(manual_run_time))
+
+    def turn_off(self, **kwargs: Any) -> None:
+        """Turn off this valve."""
+        self._base.stop_watering(self.id)
+        self._attr_is_on = False
+        self.schedule_update_ha_state(force_refresh=True)
+        _LOGGER.debug("Stopping watering on valve %s", self.name)
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated coordinator data."""
+        data = self.coordinator.data[self.id]
+
+        self._static_attrs = data[KEY_STATE][KEY_REPORTED_STATE]
+        self._attr_is_on = KEY_CURRENT_STATUS in self._static_attrs
+        super()._handle_coordinator_update()