diff --git a/CODEOWNERS b/CODEOWNERS
index c2198123382..bd76f51b4d4 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -107,6 +107,7 @@ homeassistant/components/dunehd/* @bieniu
 homeassistant/components/dweet/* @fabaff
 homeassistant/components/dynalite/* @ziv1234
 homeassistant/components/dyson/* @etheralm
+homeassistant/components/eafm/* @Jc2k
 homeassistant/components/ecobee/* @marthoc
 homeassistant/components/ecovacs/* @OverloadUT
 homeassistant/components/edl21/* @mtdcr
diff --git a/homeassistant/components/eafm/__init__.py b/homeassistant/components/eafm/__init__.py
new file mode 100644
index 00000000000..f0ce5128624
--- /dev/null
+++ b/homeassistant/components/eafm/__init__.py
@@ -0,0 +1,23 @@
+"""UK Environment Agency Flood Monitoring Integration."""
+
+from .const import DOMAIN
+
+
+async def async_setup(hass, config):
+    """Set up devices."""
+    hass.data[DOMAIN] = {}
+    return True
+
+
+async def async_setup_entry(hass, entry):
+    """Set up flood monitoring sensors for this config entry."""
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, "sensor")
+    )
+
+    return True
+
+
+async def async_unload_entry(hass, config_entry):
+    """Unload flood monitoring sensors."""
+    return await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")
diff --git a/homeassistant/components/eafm/config_flow.py b/homeassistant/components/eafm/config_flow.py
new file mode 100644
index 00000000000..0f640951f9f
--- /dev/null
+++ b/homeassistant/components/eafm/config_flow.py
@@ -0,0 +1,61 @@
+"""Config flow to configure flood monitoring gauges."""
+import logging
+
+from aioeafm import get_stations
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+# pylint: disable=unused-import
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class UKFloodsFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a UK Environment Agency flood monitoring config flow."""
+
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
+
+    def __init__(self):
+        """Handle a UK Floods config flow."""
+        self.stations = {}
+
+    async def async_step_user(self, user_input=None):
+        """Handle a flow start."""
+        errors = {}
+
+        if user_input is not None:
+            station = self.stations[user_input["station"]]
+            await self.async_set_unique_id(station, raise_on_progress=False)
+            self._abort_if_unique_id_configured()
+            return self.async_create_entry(
+                title=user_input["station"], data={"station": station},
+            )
+
+        session = async_get_clientsession(hass=self.hass)
+        stations = await get_stations(session)
+
+        self.stations = {}
+        for station in stations:
+            label = station["label"]
+
+            # API annoyingly sometimes returns a list and some times returns a string
+            # E.g. L3121 has a label of ['Scurf Dyke', 'Scurf Dyke Dyke Level']
+            if isinstance(label, list):
+                label = label[-1]
+
+            self.stations[label] = station["stationReference"]
+
+        if not self.stations:
+            return self.async_abort(reason="no_stations")
+
+        return self.async_show_form(
+            step_id="user",
+            errors=errors,
+            data_schema=vol.Schema(
+                {vol.Required("station"): vol.In(sorted(self.stations))}
+            ),
+        )
diff --git a/homeassistant/components/eafm/const.py b/homeassistant/components/eafm/const.py
new file mode 100644
index 00000000000..c87cf263a9d
--- /dev/null
+++ b/homeassistant/components/eafm/const.py
@@ -0,0 +1,3 @@
+"""Constants for the eafm component."""
+
+DOMAIN = "eafm"
diff --git a/homeassistant/components/eafm/manifest.json b/homeassistant/components/eafm/manifest.json
new file mode 100644
index 00000000000..66813d33036
--- /dev/null
+++ b/homeassistant/components/eafm/manifest.json
@@ -0,0 +1,8 @@
+{
+  "domain": "eafm",
+  "name": "Environment Agency Flood Gauges",
+  "documentation": "https://www.home-assistant.io/integrations/eafm",
+  "config_flow": true,
+  "codeowners": ["@Jc2k"],
+  "requirements": ["aioeafm==0.1.2"]
+}
diff --git a/homeassistant/components/eafm/sensor.py b/homeassistant/components/eafm/sensor.py
new file mode 100644
index 00000000000..ae4968bcf31
--- /dev/null
+++ b/homeassistant/components/eafm/sensor.py
@@ -0,0 +1,183 @@
+"""Support for guages from flood monitoring API."""
+from datetime import timedelta
+import logging
+
+from aioeafm import get_station
+import async_timeout
+
+from homeassistant.const import ATTR_ATTRIBUTION, LENGTH_METERS
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+UNIT_MAPPING = {
+    "http://qudt.org/1.1/vocab/unit#Meter": LENGTH_METERS,
+}
+
+
+def get_measures(station_data):
+    """Force measure key to always be a list."""
+    if "measures" not in station_data:
+        return []
+    if isinstance(station_data["measures"], dict):
+        return [station_data["measures"]]
+    return station_data["measures"]
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up UK Flood Monitoring Sensors."""
+    station_key = config_entry.data["station"]
+    session = async_get_clientsession(hass=hass)
+
+    measurements = set()
+
+    async def async_update_data():
+        # DataUpdateCoordinator will handle aiohttp ClientErrors and timouts
+        async with async_timeout.timeout(30):
+            data = await get_station(session, station_key)
+
+        measures = get_measures(data)
+        entities = []
+
+        # Look to see if payload contains new measures
+        for measure in measures:
+            if measure["@id"] in measurements:
+                continue
+
+            if "latestReading" not in measure:
+                # Don't create a sensor entity for a gauge that isn't available
+                continue
+
+            entities.append(Measurement(hass.data[DOMAIN][station_key], measure["@id"]))
+            measurements.add(measure["@id"])
+
+        async_add_entities(entities)
+
+        # Turn data.measures into a dict rather than a list so easier for entities to
+        # find themselves.
+        data["measures"] = {measure["@id"]: measure for measure in measures}
+
+        return data
+
+    hass.data[DOMAIN][station_key] = coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name="sensor",
+        update_method=async_update_data,
+        update_interval=timedelta(seconds=15 * 60),
+    )
+
+    # Fetch initial data so we have data when entities subscribe
+    await coordinator.async_refresh()
+
+
+class Measurement(Entity):
+    """A gauge at a flood monitoring station."""
+
+    attribution = "This uses Environment Agency flood and river level data from the real-time data API"
+
+    def __init__(self, coordinator, key):
+        """Initialise the gauge with a data instance and station."""
+        self.coordinator = coordinator
+        self.key = key
+
+    @property
+    def station_name(self):
+        """Return the station name for the measure."""
+        return self.coordinator.data["label"]
+
+    @property
+    def station_id(self):
+        """Return the station id for the measure."""
+        return self.coordinator.data["measures"][self.key]["stationReference"]
+
+    @property
+    def qualifier(self):
+        """Return the qualifier for the station."""
+        return self.coordinator.data["measures"][self.key]["qualifier"]
+
+    @property
+    def parameter_name(self):
+        """Return the parameter name for the station."""
+        return self.coordinator.data["measures"][self.key]["parameterName"]
+
+    @property
+    def name(self):
+        """Return the name of the gauge."""
+        return f"{self.station_name} {self.parameter_name} {self.qualifier}"
+
+    @property
+    def should_poll(self) -> bool:
+        """Stations are polled as a group - the entity shouldn't poll by itself."""
+        return False
+
+    @property
+    def unique_id(self):
+        """Return the unique id of the gauge."""
+        return self.key
+
+    @property
+    def device_info(self):
+        """Return the device info."""
+        return {
+            "identifiers": {(DOMAIN, "measure-id", self.station_id)},
+            "name": self.name,
+            "manufacturer": "https://environment.data.gov.uk/",
+            "model": self.parameter_name,
+            "entry_type": "service",
+        }
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        if not self.coordinator.last_update_success:
+            return False
+
+        # If sensor goes offline it will no longer contain a reading
+        if "latestReading" not in self.coordinator.data["measures"][self.key]:
+            return False
+
+        # Sometimes lastestReading key is present but actually a URL rather than a piece of data
+        # This is usually because the sensor has been archived
+        if not isinstance(
+            self.coordinator.data["measures"][self.key]["latestReading"], dict
+        ):
+            return False
+
+        return True
+
+    async def async_added_to_hass(self):
+        """When entity is added to hass."""
+        self.async_on_remove(
+            self.coordinator.async_add_listener(self.async_write_ha_state)
+        )
+
+    @property
+    def unit_of_measurement(self):
+        """Return units for the sensor."""
+        measure = self.coordinator.data["measures"][self.key]
+        if "unit" not in measure:
+            return None
+        return UNIT_MAPPING.get(measure["unit"], measure["unitName"])
+
+    @property
+    def device_state_attributes(self):
+        """Return the sensor specific state attributes."""
+        return {ATTR_ATTRIBUTION: self.attribution}
+
+    @property
+    def state(self):
+        """Return the current sensor value."""
+        return self.coordinator.data["measures"][self.key]["latestReading"]["value"]
+
+    async def async_update(self):
+        """
+        Update the entity.
+
+        Only used by the generic entity update service.
+        """
+        await self.coordinator.async_request_refresh()
diff --git a/homeassistant/components/eafm/strings.json b/homeassistant/components/eafm/strings.json
new file mode 100644
index 00000000000..9c829abcd1a
--- /dev/null
+++ b/homeassistant/components/eafm/strings.json
@@ -0,0 +1,17 @@
+{
+    "config": {
+        "step": {
+            "user": {
+                "title": "Track a flood monitoring station",
+                "description": "Select the station you want to monitor",
+                "data": {
+                    "station": "Station"
+                }
+            }
+        },
+        "abort": {
+            "no_stations": "No flood monitoring stations found.",
+            "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+        }
+    }
+}
diff --git a/homeassistant/components/eafm/translations/en.json b/homeassistant/components/eafm/translations/en.json
new file mode 100644
index 00000000000..3a3ddae390b
--- /dev/null
+++ b/homeassistant/components/eafm/translations/en.json
@@ -0,0 +1,17 @@
+{
+    "config": {
+        "step": {
+            "user": {
+                "title": "Track a flood monitoring station",
+                "description": "Select the station you want to monitor",
+                "data": {
+                    "station": "Station"
+                }
+            }
+        },
+        "abort": {
+            "no_stations": "No flood monitoring stations found.",
+            "already_configured": "This station is already configured."
+        }
+    }
+}
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index f6171eb9f47..3698fad422c 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -44,6 +44,7 @@ FLOWS = [
     "doorbird",
     "dunehd",
     "dynalite",
+    "eafm",
     "ecobee",
     "elgato",
     "elkm1",
diff --git a/requirements_all.txt b/requirements_all.txt
index bcd39c115c7..4c4ce6de1ba 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -157,6 +157,9 @@ aiobotocore==0.11.1
 # homeassistant.components.minecraft_server
 aiodns==2.0.0
 
+# homeassistant.components.eafm
+aioeafm==0.1.2
+
 # homeassistant.components.esphome
 aioesphomeapi==2.6.1
 
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 3c1a97e6bfd..da524c6423d 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -85,6 +85,9 @@ aiobotocore==0.11.1
 # homeassistant.components.minecraft_server
 aiodns==2.0.0
 
+# homeassistant.components.eafm
+aioeafm==0.1.2
+
 # homeassistant.components.esphome
 aioesphomeapi==2.6.1
 
diff --git a/tests/components/eafm/__init__.py b/tests/components/eafm/__init__.py
new file mode 100644
index 00000000000..d94537dd966
--- /dev/null
+++ b/tests/components/eafm/__init__.py
@@ -0,0 +1 @@
+"""Tests for eafm."""
diff --git a/tests/components/eafm/conftest.py b/tests/components/eafm/conftest.py
new file mode 100644
index 00000000000..b25c0f4cdba
--- /dev/null
+++ b/tests/components/eafm/conftest.py
@@ -0,0 +1,18 @@
+"""eafm fixtures."""
+
+from asynctest import patch
+import pytest
+
+
+@pytest.fixture()
+def mock_get_stations():
+    """Mock aioeafm.get_stations."""
+    with patch("homeassistant.components.eafm.config_flow.get_stations") as patched:
+        yield patched
+
+
+@pytest.fixture()
+def mock_get_station():
+    """Mock aioeafm.get_station."""
+    with patch("homeassistant.components.eafm.sensor.get_station") as patched:
+        yield patched
diff --git a/tests/components/eafm/test_config_flow.py b/tests/components/eafm/test_config_flow.py
new file mode 100644
index 00000000000..4656e34a34c
--- /dev/null
+++ b/tests/components/eafm/test_config_flow.py
@@ -0,0 +1,59 @@
+"""Tests for eafm config flow."""
+from asynctest import patch
+import pytest
+from voluptuous.error import MultipleInvalid
+
+from homeassistant.components.eafm import const
+
+
+async def test_flow_no_discovered_stations(hass, mock_get_stations):
+    """Test config flow discovers no station."""
+    mock_get_stations.return_value = []
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": "user"}
+    )
+    assert result["type"] == "abort"
+    assert result["reason"] == "no_stations"
+
+
+async def test_flow_invalid_station(hass, mock_get_stations):
+    """Test config flow errors on invalid station."""
+    mock_get_stations.return_value = [
+        {"label": "My station", "stationReference": "L12345"}
+    ]
+
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": "user"}
+    )
+    assert result["type"] == "form"
+
+    with pytest.raises(MultipleInvalid):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={"station": "My other station"}
+        )
+
+
+async def test_flow_works(hass, mock_get_stations, mock_get_station):
+    """Test config flow discovers no station."""
+    mock_get_stations.return_value = [
+        {"label": "My station", "stationReference": "L12345"}
+    ]
+    mock_get_station.return_value = [
+        {"label": "My station", "stationReference": "L12345"}
+    ]
+
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": "user"}
+    )
+    assert result["type"] == "form"
+
+    with patch("homeassistant.components.eafm.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={"station": "My station"}
+        )
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == "My station"
+    assert result["data"] == {
+        "station": "L12345",
+    }
diff --git a/tests/components/eafm/test_sensor.py b/tests/components/eafm/test_sensor.py
new file mode 100644
index 00000000000..6cce7a2bc4b
--- /dev/null
+++ b/tests/components/eafm/test_sensor.py
@@ -0,0 +1,431 @@
+"""Tests for polling measures."""
+import datetime
+
+import aiohttp
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.setup import async_setup_component
+import homeassistant.util.dt as dt_util
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+DUMMY_REQUEST_INFO = aiohttp.client.RequestInfo(
+    url="http://example.com", method="GET", headers={}, real_url="http://example.com"
+)
+
+CONNECTION_EXCEPTIONS = [
+    aiohttp.ClientConnectionError("Mock connection error"),
+    aiohttp.ClientResponseError(DUMMY_REQUEST_INFO, [], message="Mock response error"),
+]
+
+
+async def async_setup_test_fixture(hass, mock_get_station, initial_value):
+    """Create a dummy config entry for testing polling."""
+    mock_get_station.return_value = initial_value
+
+    entry = MockConfigEntry(
+        version=1,
+        domain="eafm",
+        entry_id="VikingRecorder1234",
+        data={"station": "L1234"},
+        title="Viking Recorder",
+        connection_class=config_entries.CONN_CLASS_CLOUD_PUSH,
+    )
+    entry.add_to_hass(hass)
+
+    assert await async_setup_component(hass, "eafm", {})
+    assert entry.state == config_entries.ENTRY_STATE_LOADED
+    await hass.async_block_till_done()
+
+    async def poll(value):
+        mock_get_station.reset_mock(return_value=True, side_effect=True)
+
+        if isinstance(value, Exception):
+            mock_get_station.side_effect = value
+        else:
+            mock_get_station.return_value = value
+
+        next_update = dt_util.utcnow() + datetime.timedelta(60 * 15)
+        async_fire_time_changed(hass, next_update)
+        await hass.async_block_till_done()
+
+    return entry, poll
+
+
+async def test_reading_measures_not_list(hass, mock_get_station):
+    """
+    Test that a measure can be a dict not a list.
+
+    E.g. https://environment.data.gov.uk/flood-monitoring/id/stations/751110
+    """
+    _ = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": {
+                "@id": "really-long-unique-id",
+                "label": "York Viking Recorder - level-stage-i-15_min----",
+                "qualifier": "Stage",
+                "parameterName": "Water Level",
+                "latestReading": {"value": 5},
+                "stationReference": "L1234",
+            },
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+
+async def test_reading_no_unit(hass, mock_get_station):
+    """
+    Test that a sensor functions even if its unit is not known.
+
+    E.g. https://environment.data.gov.uk/flood-monitoring/id/stations/L0410
+    """
+    _ = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                }
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+
+async def test_ignore_invalid_latest_reading(hass, mock_get_station):
+    """
+    Test that a sensor functions even if its unit is not known.
+
+    E.g. https://environment.data.gov.uk/flood-monitoring/id/stations/L0410
+    """
+    _ = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": "http://environment.data.gov.uk/flood-monitoring/data/readings/L0410-level-stage-i-15_min----/2017-02-22T10-30-00Z",
+                    "stationReference": "L0410",
+                },
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Other",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L0411",
+                },
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state is None
+
+    state = hass.states.get("sensor.my_station_other_stage")
+    assert state.state == "5"
+
+
+@pytest.mark.parametrize("exception", CONNECTION_EXCEPTIONS)
+async def test_reading_unavailable(hass, mock_get_station, exception):
+    """Test that a sensor is marked as unavailable if there is a connection error."""
+    _, poll = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+    await poll(exception)
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "unavailable"
+
+
+@pytest.mark.parametrize("exception", CONNECTION_EXCEPTIONS)
+async def test_recover_from_failure(hass, mock_get_station, exception):
+    """Test that a sensor recovers from failures."""
+    _, poll = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+    await poll(exception)
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "unavailable"
+
+    await poll(
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 56},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "56"
+
+
+async def test_reading_is_sampled(hass, mock_get_station):
+    """Test that a sensor is added and polled."""
+    await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+    assert state.attributes["unit_of_measurement"] == "m"
+
+
+async def test_multiple_readings_are_sampled(hass, mock_get_station):
+    """Test that multiple sensors are added and polled."""
+    await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                },
+                {
+                    "@id": "really-long-unique-id-2",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Second Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 4},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                },
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+    assert state.attributes["unit_of_measurement"] == "m"
+
+    state = hass.states.get("sensor.my_station_water_level_second_stage")
+    assert state.state == "4"
+    assert state.attributes["unit_of_measurement"] == "m"
+
+
+async def test_ignore_no_latest_reading(hass, mock_get_station):
+    """Test that a measure is ignored if it has no latest reading."""
+    await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                },
+                {
+                    "@id": "really-long-unique-id-2",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Second Stage",
+                    "parameterName": "Water Level",
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                },
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+    assert state.attributes["unit_of_measurement"] == "m"
+
+    state = hass.states.get("sensor.my_station_water_level_second_stage")
+    assert state is None
+
+
+async def test_mark_existing_as_unavailable_if_no_latest(hass, mock_get_station):
+    """Test that a measure is marked as unavailable if it has no latest reading."""
+    _, poll = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+    assert state.attributes["unit_of_measurement"] == "m"
+
+    await poll(
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        }
+    )
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "unavailable"
+
+    await poll(
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        }
+    )
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+
+async def test_unload_entry(hass, mock_get_station):
+    """Test being able to unload an entry."""
+    entry, _ = await async_setup_test_fixture(
+        hass,
+        mock_get_station,
+        {
+            "label": "My station",
+            "measures": [
+                {
+                    "@id": "really-long-unique-id",
+                    "label": "York Viking Recorder - level-stage-i-15_min----",
+                    "qualifier": "Stage",
+                    "parameterName": "Water Level",
+                    "latestReading": {"value": 5},
+                    "stationReference": "L1234",
+                    "unit": "http://qudt.org/1.1/vocab/unit#Meter",
+                    "unitName": "m",
+                }
+            ],
+        },
+    )
+
+    # And there should be an entity
+    state = hass.states.get("sensor.my_station_water_level_stage")
+    assert state.state == "5"
+
+    assert await entry.async_unload(hass)
+
+    # And the entity should be gone
+    assert not hass.states.get("sensor.my_station_water_level_stage")