diff --git a/.coveragerc b/.coveragerc
index a63e1ae33a5..9182edfd756 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -21,6 +21,7 @@ omit =
     homeassistant/components/airly/sensor.py
     homeassistant/components/airly/const.py
     homeassistant/components/airvisual/__init__.py
+    homeassistant/components/airvisual/air_quality.py
     homeassistant/components/airvisual/sensor.py
     homeassistant/components/aladdin_connect/cover.py
     homeassistant/components/alarmdecoder/*
diff --git a/homeassistant/components/airvisual/__init__.py b/homeassistant/components/airvisual/__init__.py
index 4352b15b8a5..7f81b906237 100644
--- a/homeassistant/components/airvisual/__init__.py
+++ b/homeassistant/components/airvisual/__init__.py
@@ -1,22 +1,29 @@
 """The airvisual component."""
-import logging
+import asyncio
+from datetime import timedelta
 
 from pyairvisual import Client
-from pyairvisual.errors import AirVisualError, InvalidKeyError
+from pyairvisual.errors import AirVisualError, NodeProError
 import voluptuous as vol
 
 from homeassistant.config_entries import SOURCE_IMPORT
 from homeassistant.const import (
+    ATTR_ATTRIBUTION,
     CONF_API_KEY,
+    CONF_IP_ADDRESS,
     CONF_LATITUDE,
     CONF_LONGITUDE,
+    CONF_PASSWORD,
     CONF_SHOW_ON_MAP,
     CONF_STATE,
 )
 from homeassistant.core import callback
-from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import aiohttp_client, config_validation as cv
-from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.dispatcher import (
+    async_dispatcher_connect,
+    async_dispatcher_send,
+)
+from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.event import async_track_time_interval
 
 from .const import (
@@ -24,15 +31,20 @@ from .const import (
     CONF_COUNTRY,
     CONF_GEOGRAPHIES,
     DATA_CLIENT,
-    DEFAULT_SCAN_INTERVAL,
     DOMAIN,
+    INTEGRATION_TYPE_GEOGRAPHY,
+    INTEGRATION_TYPE_NODE_PRO,
+    LOGGER,
     TOPIC_UPDATE,
 )
 
-_LOGGER = logging.getLogger(__name__)
+PLATFORMS = ["air_quality", "sensor"]
 
 DATA_LISTENER = "listener"
 
+DEFAULT_ATTRIBUTION = "Data provided by AirVisual"
+DEFAULT_GEOGRAPHY_SCAN_INTERVAL = timedelta(minutes=10)
+DEFAULT_NODE_PRO_SCAN_INTERVAL = timedelta(minutes=1)
 DEFAULT_OPTIONS = {CONF_SHOW_ON_MAP: True}
 
 GEOGRAPHY_COORDINATES_SCHEMA = vol.Schema(
@@ -66,6 +78,9 @@ CONFIG_SCHEMA = vol.Schema({DOMAIN: CLOUD_API_SCHEMA}, extra=vol.ALLOW_EXTRA)
 @callback
 def async_get_geography_id(geography_dict):
     """Generate a unique ID from a geography dict."""
+    if not geography_dict:
+        return
+
     if CONF_CITY in geography_dict:
         return ", ".join(
             (
@@ -103,45 +118,58 @@ async def async_setup(hass, config):
     return True
 
 
-async def async_setup_entry(hass, config_entry):
-    """Set up AirVisual as config entry."""
+@callback
+def _standardize_geography_config_entry(hass, config_entry):
+    """Ensure that geography observables have appropriate properties."""
     entry_updates = {}
+
     if not config_entry.unique_id:
         # If the config entry doesn't already have a unique ID, set one:
         entry_updates["unique_id"] = config_entry.data[CONF_API_KEY]
     if not config_entry.options:
         # If the config entry doesn't already have any options set, set defaults:
-        entry_updates["options"] = DEFAULT_OPTIONS
+        entry_updates["options"] = {CONF_SHOW_ON_MAP: True}
 
-    if entry_updates:
-        hass.config_entries.async_update_entry(config_entry, **entry_updates)
+    if not entry_updates:
+        return
 
+    hass.config_entries.async_update_entry(config_entry, **entry_updates)
+
+
+async def async_setup_entry(hass, config_entry):
+    """Set up AirVisual as config entry."""
     websession = aiohttp_client.async_get_clientsession(hass)
 
-    hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = AirVisualData(
-        hass, Client(websession, api_key=config_entry.data[CONF_API_KEY]), config_entry
-    )
+    if CONF_API_KEY in config_entry.data:
+        _standardize_geography_config_entry(hass, config_entry)
+        airvisual = AirVisualGeographyData(
+            hass,
+            Client(websession, api_key=config_entry.data[CONF_API_KEY]),
+            config_entry,
+        )
 
-    try:
-        await hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id].async_update()
-    except InvalidKeyError:
-        _LOGGER.error("Invalid API key provided")
-        raise ConfigEntryNotReady
+        # Only geography-based entries have options:
+        config_entry.add_update_listener(async_update_options)
+    else:
+        airvisual = AirVisualNodeProData(hass, Client(websession), config_entry)
 
-    hass.async_create_task(
-        hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
-    )
+    await airvisual.async_update()
+
+    hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = airvisual
+
+    for component in PLATFORMS:
+        hass.async_create_task(
+            hass.config_entries.async_forward_entry_setup(config_entry, component)
+        )
 
     async def refresh(event_time):
         """Refresh data from AirVisual."""
-        await hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id].async_update()
+        await airvisual.async_update()
 
     hass.data[DOMAIN][DATA_LISTENER][config_entry.entry_id] = async_track_time_interval(
-        hass, refresh, DEFAULT_SCAN_INTERVAL
+        hass, refresh, airvisual.scan_interval
     )
 
-    config_entry.add_update_listener(async_update_options)
-
     return True
 
 
@@ -149,7 +177,7 @@ async def async_migrate_entry(hass, config_entry):
     """Migrate an old config entry."""
     version = config_entry.version
 
-    _LOGGER.debug("Migrating from version %s", version)
+    LOGGER.debug("Migrating from version %s", version)
 
     # 1 -> 2: One geography per config entry
     if version == 1:
@@ -178,21 +206,27 @@ async def async_migrate_entry(hass, config_entry):
                 )
             )
 
-    _LOGGER.info("Migration to version %s successful", version)
+    LOGGER.info("Migration to version %s successful", version)
 
     return True
 
 
 async def async_unload_entry(hass, config_entry):
     """Unload an AirVisual config entry."""
-    hass.data[DOMAIN][DATA_CLIENT].pop(config_entry.entry_id)
+    unload_ok = all(
+        await asyncio.gather(
+            *[
+                hass.config_entries.async_forward_entry_unload(config_entry, component)
+                for component in PLATFORMS
+            ]
+        )
+    )
+    if unload_ok:
+        hass.data[DOMAIN][DATA_CLIENT].pop(config_entry.entry_id)
+        remove_listener = hass.data[DOMAIN][DATA_LISTENER].pop(config_entry.entry_id)
+        remove_listener()
 
-    remove_listener = hass.data[DOMAIN][DATA_LISTENER].pop(config_entry.entry_id)
-    remove_listener()
-
-    await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")
-
-    return True
+    return unload_ok
 
 
 async def async_update_options(hass, config_entry):
@@ -201,7 +235,53 @@ async def async_update_options(hass, config_entry):
     airvisual.async_update_options(config_entry.options)
 
 
-class AirVisualData:
+class AirVisualEntity(Entity):
+    """Define a generic AirVisual entity."""
+
+    def __init__(self, airvisual):
+        """Initialize."""
+        self._airvisual = airvisual
+        self._attrs = {ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION}
+        self._icon = None
+        self._unit = None
+
+    @property
+    def device_state_attributes(self):
+        """Return the device state attributes."""
+        return self._attrs
+
+    @property
+    def icon(self):
+        """Return the icon."""
+        return self._icon
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return self._unit
+
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+
+        @callback
+        def update():
+            """Update the state."""
+            self.update_from_latest_data()
+            self.async_write_ha_state()
+
+        self.async_on_remove(
+            async_dispatcher_connect(self.hass, self._airvisual.topic_update, update)
+        )
+
+        self.update_from_latest_data()
+
+    @callback
+    def update_from_latest_data(self):
+        """Update the entity from the latest data."""
+        raise NotImplementedError
+
+
+class AirVisualGeographyData:
     """Define a class to manage data from the AirVisual cloud API."""
 
     def __init__(self, hass, client, config_entry):
@@ -211,7 +291,10 @@ class AirVisualData:
         self.data = {}
         self.geography_data = config_entry.data
         self.geography_id = config_entry.unique_id
+        self.integration_type = INTEGRATION_TYPE_GEOGRAPHY
         self.options = config_entry.options
+        self.scan_interval = DEFAULT_GEOGRAPHY_SCAN_INTERVAL
+        self.topic_update = TOPIC_UPDATE.format(config_entry.unique_id)
 
     async def async_update(self):
         """Get new data for all locations from the AirVisual cloud API."""
@@ -229,14 +312,43 @@ class AirVisualData:
         try:
             self.data[self.geography_id] = await api_coro
         except AirVisualError as err:
-            _LOGGER.error("Error while retrieving data: %s", err)
+            LOGGER.error("Error while retrieving data: %s", err)
             self.data[self.geography_id] = {}
 
-        _LOGGER.debug("Received new data")
-        async_dispatcher_send(self._hass, TOPIC_UPDATE)
+        LOGGER.debug("Received new geography data")
+        async_dispatcher_send(self._hass, self.topic_update)
 
     @callback
     def async_update_options(self, options):
         """Update the data manager's options."""
         self.options = options
-        async_dispatcher_send(self._hass, TOPIC_UPDATE)
+        async_dispatcher_send(self._hass, self.topic_update)
+
+
+class AirVisualNodeProData:
+    """Define a class to manage data from an AirVisual Node/Pro."""
+
+    def __init__(self, hass, client, config_entry):
+        """Initialize."""
+        self._client = client
+        self._hass = hass
+        self._password = config_entry.data[CONF_PASSWORD]
+        self.data = {}
+        self.integration_type = INTEGRATION_TYPE_NODE_PRO
+        self.ip_address = config_entry.data[CONF_IP_ADDRESS]
+        self.scan_interval = DEFAULT_NODE_PRO_SCAN_INTERVAL
+        self.topic_update = TOPIC_UPDATE.format(config_entry.data[CONF_IP_ADDRESS])
+
+    async def async_update(self):
+        """Get new data from the Node/Pro."""
+        try:
+            self.data = await self._client.node.from_samba(
+                self.ip_address, self._password, include_history=False
+            )
+        except NodeProError as err:
+            LOGGER.error("Error while retrieving Node/Pro data: %s", err)
+            self.data = {}
+            return
+
+        LOGGER.debug("Received new Node/Pro data")
+        async_dispatcher_send(self._hass, self.topic_update)
diff --git a/homeassistant/components/airvisual/air_quality.py b/homeassistant/components/airvisual/air_quality.py
new file mode 100644
index 00000000000..9da5b83d79f
--- /dev/null
+++ b/homeassistant/components/airvisual/air_quality.py
@@ -0,0 +1,117 @@
+"""Support for AirVisual Node/Pro units."""
+from homeassistant.components.air_quality import AirQualityEntity
+from homeassistant.const import CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
+from homeassistant.core import callback
+from homeassistant.util import slugify
+
+from . import AirVisualEntity
+from .const import DATA_CLIENT, DOMAIN, INTEGRATION_TYPE_GEOGRAPHY
+
+ATTR_HUMIDITY = "humidity"
+ATTR_SENSOR_LIFE = "{0}_sensor_life"
+ATTR_TREND = "{0}_trend"
+ATTR_VOC = "voc"
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up AirVisual air quality entities based on a config entry."""
+    airvisual = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
+
+    # Geography-based AirVisual integrations don't utilize this platform:
+    if airvisual.integration_type == INTEGRATION_TYPE_GEOGRAPHY:
+        return
+
+    async_add_entities([AirVisualNodeProSensor(airvisual)], True)
+
+
+class AirVisualNodeProSensor(AirVisualEntity, AirQualityEntity):
+    """Define a sensor for a AirVisual Node/Pro."""
+
+    def __init__(self, airvisual):
+        """Initialize."""
+        super().__init__(airvisual)
+
+        self._icon = "mdi:chemical-weapon"
+        self._unit = CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
+
+    @property
+    def air_quality_index(self):
+        """Return the Air Quality Index (AQI)."""
+        if self._airvisual.data["current"]["settings"]["is_aqi_usa"]:
+            return self._airvisual.data["current"]["measurements"]["aqi_us"]
+        return self._airvisual.data["current"]["measurements"]["aqi_cn"]
+
+    @property
+    def available(self):
+        """Return True if entity is available."""
+        return bool(self._airvisual.data)
+
+    @property
+    def carbon_dioxide(self):
+        """Return the CO2 (carbon dioxide) level."""
+        return self._airvisual.data["current"]["measurements"].get("co2_ppm")
+
+    @property
+    def device_info(self):
+        """Return device registry information for this entity."""
+        return {
+            "identifiers": {(DOMAIN, self._airvisual.data["current"]["serial_number"])},
+            "name": self._airvisual.data["current"]["settings"]["node_name"],
+            "manufacturer": "AirVisual",
+            "model": f'{self._airvisual.data["current"]["status"]["model"]}',
+            "sw_version": (
+                f'Version {self._airvisual.data["current"]["status"]["system_version"]}'
+                f'{self._airvisual.data["current"]["status"]["app_version"]}'
+            ),
+        }
+
+    @property
+    def name(self):
+        """Return the name."""
+        node_name = self._airvisual.data["current"]["settings"]["node_name"]
+        return f"{node_name} Node/Pro: Air Quality"
+
+    @property
+    def particulate_matter_2_5(self):
+        """Return the particulate matter 2.5 level."""
+        return self._airvisual.data["current"]["measurements"].get("pm2_5")
+
+    @property
+    def particulate_matter_10(self):
+        """Return the particulate matter 10 level."""
+        return self._airvisual.data["current"]["measurements"].get("pm1_0")
+
+    @property
+    def particulate_matter_0_1(self):
+        """Return the particulate matter 0.1 level."""
+        return self._airvisual.data["current"]["measurements"].get("pm0_1")
+
+    @property
+    def unique_id(self):
+        """Return a unique, Home Assistant friendly identifier for this entity."""
+        return self._airvisual.data["current"]["serial_number"]
+
+    @callback
+    def update_from_latest_data(self):
+        """Update from the Node/Pro's data."""
+        trends = {
+            ATTR_TREND.format(slugify(pollutant)): trend
+            for pollutant, trend in self._airvisual.data["trends"].items()
+        }
+        if self._airvisual.data["current"]["settings"]["is_aqi_usa"]:
+            trends.pop(ATTR_TREND.format("aqi_cn"))
+        else:
+            trends.pop(ATTR_TREND.format("aqi_us"))
+
+        self._attrs.update(
+            {
+                ATTR_VOC: self._airvisual.data["current"]["measurements"].get("voc"),
+                **{
+                    ATTR_SENSOR_LIFE.format(pollutant): lifespan
+                    for pollutant, lifespan in self._airvisual.data["current"][
+                        "status"
+                    ]["sensor_life"].items()
+                },
+                **trends,
+            }
+        )
diff --git a/homeassistant/components/airvisual/config_flow.py b/homeassistant/components/airvisual/config_flow.py
index 0c9c0e65ff1..ef15f8dcc99 100644
--- a/homeassistant/components/airvisual/config_flow.py
+++ b/homeassistant/components/airvisual/config_flow.py
@@ -2,21 +2,29 @@
 import asyncio
 
 from pyairvisual import Client
-from pyairvisual.errors import InvalidKeyError
+from pyairvisual.errors import InvalidKeyError, NodeProError
 import voluptuous as vol
 
 from homeassistant import config_entries
 from homeassistant.const import (
     CONF_API_KEY,
+    CONF_IP_ADDRESS,
     CONF_LATITUDE,
     CONF_LONGITUDE,
+    CONF_PASSWORD,
     CONF_SHOW_ON_MAP,
 )
 from homeassistant.core import callback
 from homeassistant.helpers import aiohttp_client, config_validation as cv
 
 from . import async_get_geography_id
-from .const import CONF_GEOGRAPHIES, DOMAIN  # pylint: disable=unused-import
+from .const import (  # pylint: disable=unused-import
+    CONF_GEOGRAPHIES,
+    DOMAIN,
+    INTEGRATION_TYPE_GEOGRAPHY,
+    INTEGRATION_TYPE_NODE_PRO,
+    LOGGER,
+)
 
 
 class AirVisualFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@@ -26,7 +34,7 @@ class AirVisualFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
     CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
 
     @property
-    def cloud_api_schema(self):
+    def geography_schema(self):
         """Return the data schema for the cloud API."""
         return vol.Schema(
             {
@@ -40,38 +48,47 @@ class AirVisualFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
             }
         )
 
+    @property
+    def pick_integration_type_schema(self):
+        """Return the data schema for picking the integration type."""
+        return vol.Schema(
+            {
+                vol.Required("type"): vol.In(
+                    [INTEGRATION_TYPE_GEOGRAPHY, INTEGRATION_TYPE_NODE_PRO]
+                )
+            }
+        )
+
+    @property
+    def node_pro_schema(self):
+        """Return the data schema for a Node/Pro."""
+        return vol.Schema(
+            {vol.Required(CONF_IP_ADDRESS): str, vol.Required(CONF_PASSWORD): str}
+        )
+
     async def _async_set_unique_id(self, unique_id):
         """Set the unique ID of the config flow and abort if it already exists."""
         await self.async_set_unique_id(unique_id)
         self._abort_if_unique_id_configured()
 
-    @callback
-    async def _show_form(self, errors=None):
-        """Show the form to the user."""
-        return self.async_show_form(
-            step_id="user", data_schema=self.cloud_api_schema, errors=errors or {},
-        )
-
     @staticmethod
     @callback
     def async_get_options_flow(config_entry):
         """Define the config flow to handle options."""
         return AirVisualOptionsFlowHandler(config_entry)
 
-    async def async_step_import(self, import_config):
-        """Import a config entry from configuration.yaml."""
-        return await self.async_step_user(import_config)
-
-    async def async_step_user(self, user_input=None):
-        """Handle the start of the config flow."""
+    async def async_step_geography(self, user_input=None):
+        """Handle the initialization of the integration via the cloud API."""
         if not user_input:
-            return await self._show_form()
+            return self.async_show_form(
+                step_id="geography", data_schema=self.geography_schema
+            )
 
         geo_id = async_get_geography_id(user_input)
         await self._async_set_unique_id(geo_id)
         self._abort_if_unique_id_configured()
 
-        # Find older config entries without unique ID
+        # Find older config entries without unique ID:
         for entry in self._async_current_entries():
             if entry.version != 1:
                 continue
@@ -97,8 +114,10 @@ class AirVisualFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
                 try:
                     await client.api.nearest_city()
                 except InvalidKeyError:
-                    return await self._show_form(
-                        errors={CONF_API_KEY: "invalid_api_key"}
+                    return self.async_show_form(
+                        step_id="geography",
+                        data_schema=self.geography_schema,
+                        errors={CONF_API_KEY: "invalid_api_key"},
                     )
 
                 checked_keys.add(user_input[CONF_API_KEY])
@@ -107,6 +126,49 @@ class AirVisualFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
                 title=f"Cloud API ({geo_id})", data=user_input
             )
 
+    async def async_step_import(self, import_config):
+        """Import a config entry from configuration.yaml."""
+        return await self.async_step_geography(import_config)
+
+    async def async_step_node_pro(self, user_input=None):
+        """Handle the initialization of the integration with a Node/Pro."""
+        if not user_input:
+            return self.async_show_form(
+                step_id="node_pro", data_schema=self.node_pro_schema
+            )
+
+        await self._async_set_unique_id(user_input[CONF_IP_ADDRESS])
+
+        websession = aiohttp_client.async_get_clientsession(self.hass)
+        client = Client(websession)
+
+        try:
+            await client.node.from_samba(
+                user_input[CONF_IP_ADDRESS], user_input[CONF_PASSWORD]
+            )
+        except NodeProError as err:
+            LOGGER.error("Error connecting to Node/Pro unit: %s", err)
+            return self.async_show_form(
+                step_id="node_pro",
+                data_schema=self.node_pro_schema,
+                errors={CONF_IP_ADDRESS: "unable_to_connect"},
+            )
+
+        return self.async_create_entry(
+            title=f"Node/Pro ({user_input[CONF_IP_ADDRESS]})", data=user_input
+        )
+
+    async def async_step_user(self, user_input=None):
+        """Handle the start of the config flow."""
+        if not user_input:
+            return self.async_show_form(
+                step_id="user", data_schema=self.pick_integration_type_schema
+            )
+
+        if user_input["type"] == INTEGRATION_TYPE_GEOGRAPHY:
+            return await self.async_step_geography()
+        return await self.async_step_node_pro()
+
 
 class AirVisualOptionsFlowHandler(config_entries.OptionsFlow):
     """Handle an AirVisual options flow."""
diff --git a/homeassistant/components/airvisual/const.py b/homeassistant/components/airvisual/const.py
index ab54e191116..482c4191480 100644
--- a/homeassistant/components/airvisual/const.py
+++ b/homeassistant/components/airvisual/const.py
@@ -1,7 +1,11 @@
 """Define AirVisual constants."""
-from datetime import timedelta
+import logging
 
 DOMAIN = "airvisual"
+LOGGER = logging.getLogger(__package__)
+
+INTEGRATION_TYPE_GEOGRAPHY = "Geographical Location"
+INTEGRATION_TYPE_NODE_PRO = "AirVisual Node/Pro"
 
 CONF_CITY = "city"
 CONF_COUNTRY = "country"
@@ -9,6 +13,4 @@ CONF_GEOGRAPHIES = "geographies"
 
 DATA_CLIENT = "client"
 
-DEFAULT_SCAN_INTERVAL = timedelta(minutes=10)
-
-TOPIC_UPDATE = f"{DOMAIN}_update"
+TOPIC_UPDATE = f"airvisual_update_{0}"
diff --git a/homeassistant/components/airvisual/manifest.json b/homeassistant/components/airvisual/manifest.json
index d5c7dc6853d..d97fcfb78ef 100644
--- a/homeassistant/components/airvisual/manifest.json
+++ b/homeassistant/components/airvisual/manifest.json
@@ -3,6 +3,6 @@
   "name": "AirVisual",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/airvisual",
-  "requirements": ["pyairvisual==3.0.1"],
+  "requirements": ["pyairvisual==4.3.0"],
   "codeowners": ["@bachya"]
 }
diff --git a/homeassistant/components/airvisual/sensor.py b/homeassistant/components/airvisual/sensor.py
index 20e76bf86b6..5009788e6fa 100644
--- a/homeassistant/components/airvisual/sensor.py
+++ b/homeassistant/components/airvisual/sensor.py
@@ -2,7 +2,6 @@
 from logging import getLogger
 
 from homeassistant.const import (
-    ATTR_ATTRIBUTION,
     ATTR_LATITUDE,
     ATTR_LONGITUDE,
     ATTR_STATE,
@@ -13,12 +12,22 @@ from homeassistant.const import (
     CONF_LONGITUDE,
     CONF_SHOW_ON_MAP,
     CONF_STATE,
+    DEVICE_CLASS_BATTERY,
+    DEVICE_CLASS_HUMIDITY,
+    DEVICE_CLASS_TEMPERATURE,
+    TEMP_CELSIUS,
+    UNIT_PERCENTAGE,
 )
 from homeassistant.core import callback
-from homeassistant.helpers.dispatcher import async_dispatcher_connect
-from homeassistant.helpers.entity import Entity
 
-from .const import CONF_CITY, CONF_COUNTRY, DATA_CLIENT, DOMAIN, TOPIC_UPDATE
+from . import AirVisualEntity
+from .const import (
+    CONF_CITY,
+    CONF_COUNTRY,
+    DATA_CLIENT,
+    DOMAIN,
+    INTEGRATION_TYPE_GEOGRAPHY,
+)
 
 _LOGGER = getLogger(__name__)
 
@@ -28,8 +37,6 @@ ATTR_POLLUTANT_SYMBOL = "pollutant_symbol"
 ATTR_POLLUTANT_UNIT = "pollutant_unit"
 ATTR_REGION = "region"
 
-DEFAULT_ATTRIBUTION = "Data provided by AirVisual"
-
 MASS_PARTS_PER_MILLION = "ppm"
 MASS_PARTS_PER_BILLION = "ppb"
 VOLUME_MICROGRAMS_PER_CUBIC_METER = "µg/m3"
@@ -37,11 +44,22 @@ VOLUME_MICROGRAMS_PER_CUBIC_METER = "µg/m3"
 SENSOR_KIND_LEVEL = "air_pollution_level"
 SENSOR_KIND_AQI = "air_quality_index"
 SENSOR_KIND_POLLUTANT = "main_pollutant"
-SENSORS = [
+SENSOR_KIND_BATTERY_LEVEL = "battery_level"
+SENSOR_KIND_HUMIDITY = "humidity"
+SENSOR_KIND_TEMPERATURE = "temperature"
+
+GEOGRAPHY_SENSORS = [
     (SENSOR_KIND_LEVEL, "Air Pollution Level", "mdi:gauge", None),
     (SENSOR_KIND_AQI, "Air Quality Index", "mdi:chart-line", "AQI"),
     (SENSOR_KIND_POLLUTANT, "Main Pollutant", "mdi:chemical-weapon", None),
 ]
+GEOGRAPHY_SENSOR_LOCALES = {"cn": "Chinese", "us": "U.S."}
+
+NODE_PRO_SENSORS = [
+    (SENSOR_KIND_BATTERY_LEVEL, "Battery", DEVICE_CLASS_BATTERY, UNIT_PERCENTAGE),
+    (SENSOR_KIND_HUMIDITY, "Humidity", DEVICE_CLASS_HUMIDITY, UNIT_PERCENTAGE),
+    (SENSOR_KIND_TEMPERATURE, "Temperature", DEVICE_CLASS_TEMPERATURE, TEMP_CELSIUS),
+]
 
 POLLUTANT_LEVEL_MAPPING = [
     {"label": "Good", "icon": "mdi:emoticon-excited", "minimum": 0, "maximum": 50},
@@ -71,44 +89,64 @@ POLLUTANT_MAPPING = {
     "s2": {"label": "Sulfur Dioxide", "unit": CONCENTRATION_PARTS_PER_BILLION},
 }
 
-SENSOR_LOCALES = {"cn": "Chinese", "us": "U.S."}
 
-
-async def async_setup_entry(hass, entry, async_add_entities):
+async def async_setup_entry(hass, config_entry, async_add_entities):
     """Set up AirVisual sensors based on a config entry."""
-    airvisual = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
+    airvisual = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
 
-    async_add_entities(
-        [
-            AirVisualSensor(airvisual, kind, name, icon, unit, locale, geography_id)
+    if airvisual.integration_type == INTEGRATION_TYPE_GEOGRAPHY:
+        sensors = [
+            AirVisualGeographySensor(
+                airvisual, kind, name, icon, unit, locale, geography_id,
+            )
             for geography_id in airvisual.data
-            for locale in SENSOR_LOCALES
-            for kind, name, icon, unit in SENSORS
-        ],
-        True,
-    )
+            for locale in GEOGRAPHY_SENSOR_LOCALES
+            for kind, name, icon, unit in GEOGRAPHY_SENSORS
+        ]
+    else:
+        sensors = [
+            AirVisualNodeProSensor(airvisual, kind, name, device_class, unit)
+            for kind, name, device_class, unit in NODE_PRO_SENSORS
+        ]
+
+    async_add_entities(sensors, True)
 
 
-class AirVisualSensor(Entity):
-    """Define an AirVisual sensor."""
+class AirVisualSensor(AirVisualEntity):
+    """Define a generic AirVisual sensor."""
 
-    def __init__(self, airvisual, kind, name, icon, unit, locale, geography_id):
+    def __init__(self, airvisual, kind, name, unit):
         """Initialize."""
-        self._airvisual = airvisual
-        self._geography_id = geography_id
-        self._icon = icon
+        super().__init__(airvisual)
+
         self._kind = kind
-        self._locale = locale
         self._name = name
         self._state = None
         self._unit = unit
 
-        self._attrs = {
-            ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION,
-            ATTR_CITY: airvisual.data[geography_id].get(CONF_CITY),
-            ATTR_STATE: airvisual.data[geography_id].get(CONF_STATE),
-            ATTR_COUNTRY: airvisual.data[geography_id].get(CONF_COUNTRY),
-        }
+    @property
+    def state(self):
+        """Return the state."""
+        return self._state
+
+
+class AirVisualGeographySensor(AirVisualSensor):
+    """Define an AirVisual sensor related to geography data via the Cloud API."""
+
+    def __init__(self, airvisual, kind, name, icon, unit, locale, geography_id):
+        """Initialize."""
+        super().__init__(airvisual, kind, name, unit)
+
+        self._attrs.update(
+            {
+                ATTR_CITY: airvisual.data[geography_id].get(CONF_CITY),
+                ATTR_STATE: airvisual.data[geography_id].get(CONF_STATE),
+                ATTR_COUNTRY: airvisual.data[geography_id].get(CONF_COUNTRY),
+            }
+        )
+        self._geography_id = geography_id
+        self._icon = icon
+        self._locale = locale
 
     @property
     def available(self):
@@ -120,47 +158,18 @@ class AirVisualSensor(Entity):
         except KeyError:
             return False
 
-    @property
-    def device_state_attributes(self):
-        """Return the device state attributes."""
-        return self._attrs
-
-    @property
-    def icon(self):
-        """Return the icon."""
-        return self._icon
-
     @property
     def name(self):
         """Return the name."""
-        return f"{SENSOR_LOCALES[self._locale]} {self._name}"
-
-    @property
-    def state(self):
-        """Return the state."""
-        return self._state
+        return f"{GEOGRAPHY_SENSOR_LOCALES[self._locale]} {self._name}"
 
     @property
     def unique_id(self):
         """Return a unique, Home Assistant friendly identifier for this entity."""
         return f"{self._geography_id}_{self._locale}_{self._kind}"
 
-    @property
-    def unit_of_measurement(self):
-        """Return the unit the value is expressed in."""
-        return self._unit
-
-    async def async_added_to_hass(self):
-        """Register callbacks."""
-
-        @callback
-        def update():
-            """Update the state."""
-            self.async_schedule_update_ha_state(True)
-
-        self.async_on_remove(async_dispatcher_connect(self.hass, TOPIC_UPDATE, update))
-
-    async def async_update(self):
+    @callback
+    def update_from_latest_data(self):
         """Update the sensor."""
         try:
             data = self._airvisual.data[self._geography_id]["current"]["pollution"]
@@ -203,3 +212,62 @@ class AirVisualSensor(Entity):
                 self._attrs["long"] = self._airvisual.geography_data[CONF_LONGITUDE]
                 self._attrs.pop(ATTR_LATITUDE, None)
                 self._attrs.pop(ATTR_LONGITUDE, None)
+
+
+class AirVisualNodeProSensor(AirVisualSensor):
+    """Define an AirVisual sensor related to a Node/Pro unit."""
+
+    def __init__(self, airvisual, kind, name, device_class, unit):
+        """Initialize."""
+        super().__init__(airvisual, kind, name, unit)
+
+        self._device_class = device_class
+
+    @property
+    def available(self):
+        """Return True if entity is available."""
+        return bool(self._airvisual.data)
+
+    @property
+    def device_class(self):
+        """Return the device class."""
+        return self._device_class
+
+    @property
+    def device_info(self):
+        """Return device registry information for this entity."""
+        return {
+            "identifiers": {(DOMAIN, self._airvisual.data["current"]["serial_number"])},
+            "name": self._airvisual.data["current"]["settings"]["node_name"],
+            "manufacturer": "AirVisual",
+            "model": f'{self._airvisual.data["current"]["status"]["model"]}',
+            "sw_version": (
+                f'Version {self._airvisual.data["current"]["status"]["system_version"]}'
+                f'{self._airvisual.data["current"]["status"]["app_version"]}'
+            ),
+        }
+
+    @property
+    def name(self):
+        """Return the name."""
+        node_name = self._airvisual.data["current"]["settings"]["node_name"]
+        return f"{node_name} Node/Pro: {self._name}"
+
+    @property
+    def unique_id(self):
+        """Return a unique, Home Assistant friendly identifier for this entity."""
+        return f"{self._airvisual.data['current']['serial_number']}_{self._kind}"
+
+    @callback
+    def update_from_latest_data(self):
+        """Update from the Node/Pro's data."""
+        if self._kind == SENSOR_KIND_BATTERY_LEVEL:
+            self._state = self._airvisual.data["current"]["status"]["battery"]
+        elif self._kind == SENSOR_KIND_HUMIDITY:
+            self._state = self._airvisual.data["current"]["measurements"].get(
+                "humidity"
+            )
+        elif self._kind == SENSOR_KIND_TEMPERATURE:
+            self._state = self._airvisual.data["current"]["measurements"].get(
+                "temperature_C"
+            )
diff --git a/homeassistant/components/airvisual/strings.json b/homeassistant/components/airvisual/strings.json
index cd81d1862dd..8b9978b611f 100644
--- a/homeassistant/components/airvisual/strings.json
+++ b/homeassistant/components/airvisual/strings.json
@@ -1,27 +1,49 @@
 {
   "config": {
     "step": {
-      "user": {
-        "title": "Configure AirVisual",
-        "description": "Monitor air quality in a geographical location.",
+      "geography": {
+        "title": "Configure a Geography",
+        "description": "Use the AirVisual cloud API to monitor a geographical location.",
         "data": {
           "api_key": "API Key",
           "latitude": "Latitude",
           "longitude": "Longitude"
         }
+      },
+      "node_pro": {
+        "title": "Configure an AirVisual Node/Pro",
+        "description": "Monitor a personal AirVisual unit. The password can be retrieved from the unit's UI.",
+        "data": {
+          "ip_address": "Unit IP Address/Hostname",
+          "password": "Unit Password"
+        }
+      },
+      "user": {
+        "title": "Configure AirVisual",
+        "description": "Pick what type of AirVisual data you want to monitor.",
+        "data": {
+          "cloud_api": "Geographical Location",
+          "node_pro": "AirVisual Node Pro",
+          "type": "Integration Type"
+        }
       }
     },
-    "error": { "invalid_api_key": "Invalid API key" },
+    "error": {
+      "general_error": "There was an unknown error.",
+      "invalid_api_key": "Invalid API key provided.",
+      "unable_to_connect": "Unable to connect to Node/Pro unit."
+    },
     "abort": {
-      "already_configured": "These coordinates have already been registered."
+      "already_configured": "These coordinates or Node/Pro ID are already registered."
     }
   },
   "options": {
     "step": {
       "init": {
         "title": "Configure AirVisual",
-        "description": "Set various options for the AirVisual integration.",
-        "data": { "show_on_map": "Show monitored geography on the map" }
+        "data": {
+          "show_on_map": "Show monitored geography on the map"
+        }
       }
     }
   }
diff --git a/homeassistant/components/airvisual/translations/en.json b/homeassistant/components/airvisual/translations/en.json
index c7f2c32b4e3..b32ba9ec1ab 100644
--- a/homeassistant/components/airvisual/translations/en.json
+++ b/homeassistant/components/airvisual/translations/en.json
@@ -1,19 +1,38 @@
 {
     "config": {
         "abort": {
-            "already_configured": "These coordinates have already been registered."
+            "already_configured": "These coordinates or Node/Pro ID are already registered."
         },
         "error": {
-            "invalid_api_key": "Invalid API key"
+            "general_error": "There was an unknown error.",
+            "invalid_api_key": "Invalid API key provided.",
+            "unable_to_connect": "Unable to connect to Node/Pro unit."
         },
         "step": {
-            "user": {
+            "geography": {
                 "data": {
                     "api_key": "API Key",
                     "latitude": "Latitude",
                     "longitude": "Longitude"
                 },
-                "description": "Monitor air quality in a geographical location.",
+                "description": "Use the AirVisual cloud API to monitor a geographical location.",
+                "title": "Configure a Geography"
+            },
+            "node_pro": {
+                "data": {
+                    "ip_address": "Unit IP Address/Hostname",
+                    "password": "Unit Password"
+                },
+                "description": "Monitor a personal AirVisual unit. The password can be retrieved from the unit's UI.",
+                "title": "Configure an AirVisual Node/Pro"
+            },
+            "user": {
+                "data": {
+                    "cloud_api": "Geographical Location",
+                    "node_pro": "AirVisual Node Pro",
+                    "type": "Integration Type"
+                },
+                "description": "Pick what type of AirVisual data you want to monitor.",
                 "title": "Configure AirVisual"
             }
         }
@@ -24,7 +43,6 @@
                 "data": {
                     "show_on_map": "Show monitored geography on the map"
                 },
-                "description": "Set various options for the AirVisual integration.",
                 "title": "Configure AirVisual"
             }
         }
diff --git a/requirements_all.txt b/requirements_all.txt
index 8c89391d060..faeaa3e0036 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -1173,7 +1173,7 @@ pyaehw4a1==0.3.4
 pyaftership==0.1.2
 
 # homeassistant.components.airvisual
-pyairvisual==3.0.1
+pyairvisual==4.3.0
 
 # homeassistant.components.almond
 pyalmond==0.0.2
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 77c12ead2c7..2509a3d3935 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -470,7 +470,7 @@ py_nextbusnext==0.1.4
 pyaehw4a1==0.3.4
 
 # homeassistant.components.airvisual
-pyairvisual==3.0.1
+pyairvisual==4.3.0
 
 # homeassistant.components.almond
 pyalmond==0.0.2
diff --git a/tests/components/airvisual/test_config_flow.py b/tests/components/airvisual/test_config_flow.py
index d21aec14fa0..57852969d71 100644
--- a/tests/components/airvisual/test_config_flow.py
+++ b/tests/components/airvisual/test_config_flow.py
@@ -1,14 +1,21 @@
 """Define tests for the AirVisual config flow."""
 from asynctest import patch
-from pyairvisual.errors import InvalidKeyError
+from pyairvisual.errors import InvalidKeyError, NodeProError
 
 from homeassistant import data_entry_flow
-from homeassistant.components.airvisual import CONF_GEOGRAPHIES, DOMAIN
+from homeassistant.components.airvisual import (
+    CONF_GEOGRAPHIES,
+    DOMAIN,
+    INTEGRATION_TYPE_GEOGRAPHY,
+    INTEGRATION_TYPE_NODE_PRO,
+)
 from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
 from homeassistant.const import (
     CONF_API_KEY,
+    CONF_IP_ADDRESS,
     CONF_LATITUDE,
     CONF_LONGITUDE,
+    CONF_PASSWORD,
     CONF_SHOW_ON_MAP,
 )
 from homeassistant.setup import async_setup_component
@@ -17,28 +24,43 @@ from tests.common import MockConfigEntry
 
 
 async def test_duplicate_error(hass):
-    """Test that errors are shown when duplicates are added."""
-    conf = {
+    """Test that errors are shown when duplicate entries are added."""
+    geography_conf = {
         CONF_API_KEY: "abcde12345",
         CONF_LATITUDE: 51.528308,
         CONF_LONGITUDE: -0.3817765,
     }
+    node_pro_conf = {CONF_IP_ADDRESS: "192.168.1.100", CONF_PASSWORD: "12345"}
 
     MockConfigEntry(
-        domain=DOMAIN, unique_id="51.528308, -0.3817765", data=conf
+        domain=DOMAIN, unique_id="51.528308, -0.3817765", data=geography_conf
     ).add_to_hass(hass)
 
     result = await hass.config_entries.flow.async_init(
-        DOMAIN, context={"source": SOURCE_USER}, data=conf
+        DOMAIN, context={"source": SOURCE_IMPORT}, data=geography_conf
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"
+
+    MockConfigEntry(
+        domain=DOMAIN, unique_id="192.168.1.100", data=node_pro_conf
+    ).add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}, data={"type": "AirVisual Node/Pro"}
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input=node_pro_conf
     )
 
     assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
     assert result["reason"] == "already_configured"
 
 
-async def test_invalid_api_key(hass):
-    """Test that invalid credentials throws an error."""
-    conf = {
+async def test_invalid_identifier(hass):
+    """Test that an invalid API key or Node/Pro ID throws an error."""
+    geography_conf = {
         CONF_API_KEY: "abcde12345",
         CONF_LATITUDE: 51.528308,
         CONF_LONGITUDE: -0.3817765,
@@ -48,11 +70,29 @@ async def test_invalid_api_key(hass):
         "pyairvisual.api.API.nearest_city", side_effect=InvalidKeyError,
     ):
         result = await hass.config_entries.flow.async_init(
-            DOMAIN, context={"source": SOURCE_USER}, data=conf
+            DOMAIN, context={"source": SOURCE_IMPORT}, data=geography_conf
         )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
         assert result["errors"] == {CONF_API_KEY: "invalid_api_key"}
 
 
+async def test_node_pro_error(hass):
+    """Test that an invalid Node/Pro ID shows an error."""
+    node_pro_conf = {CONF_IP_ADDRESS: "192.168.1.100", CONF_PASSWORD: "my_password"}
+
+    with patch(
+        "pyairvisual.node.Node.from_samba", side_effect=NodeProError,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}, data={"type": "AirVisual Node/Pro"}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input=node_pro_conf
+        )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+        assert result["errors"] == {CONF_IP_ADDRESS: "unable_to_connect"}
+
+
 async def test_migration_1_2(hass):
     """Test migrating from version 1 to version 2."""
     conf = {
@@ -96,12 +136,16 @@ async def test_migration_1_2(hass):
 
 async def test_options_flow(hass):
     """Test config flow options."""
-    conf = {CONF_API_KEY: "abcde12345"}
+    geography_conf = {
+        CONF_API_KEY: "abcde12345",
+        CONF_LATITUDE: 51.528308,
+        CONF_LONGITUDE: -0.3817765,
+    }
 
     config_entry = MockConfigEntry(
         domain=DOMAIN,
-        unique_id="abcde12345",
-        data=conf,
+        unique_id="51.528308, -0.3817765",
+        data=geography_conf,
         options={CONF_SHOW_ON_MAP: True},
     )
     config_entry.add_to_hass(hass)
@@ -122,18 +166,8 @@ async def test_options_flow(hass):
         assert config_entry.options == {CONF_SHOW_ON_MAP: False}
 
 
-async def test_show_form(hass):
-    """Test that the form is served with no input."""
-    result = await hass.config_entries.flow.async_init(
-        DOMAIN, context={"source": SOURCE_USER}
-    )
-
-    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
-    assert result["step_id"] == "user"
-
-
-async def test_step_import(hass):
-    """Test that the import step works."""
+async def test_step_geography(hass):
+    """Test the geograph (cloud API) step."""
     conf = {
         CONF_API_KEY: "abcde12345",
         CONF_LATITUDE: 51.528308,
@@ -146,6 +180,50 @@ async def test_step_import(hass):
         result = await hass.config_entries.flow.async_init(
             DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
         )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+        assert result["title"] == "Cloud API (51.528308, -0.3817765)"
+        assert result["data"] == {
+            CONF_API_KEY: "abcde12345",
+            CONF_LATITUDE: 51.528308,
+            CONF_LONGITUDE: -0.3817765,
+        }
+
+
+async def test_step_node_pro(hass):
+    """Test the Node/Pro step."""
+    conf = {CONF_IP_ADDRESS: "192.168.1.100", CONF_PASSWORD: "my_password"}
+
+    with patch(
+        "homeassistant.components.airvisual.async_setup_entry", return_value=True
+    ), patch("pyairvisual.node.Node.from_samba"):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}, data={"type": "AirVisual Node/Pro"}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input=conf
+        )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+        assert result["title"] == "Node/Pro (192.168.1.100)"
+        assert result["data"] == {
+            CONF_IP_ADDRESS: "192.168.1.100",
+            CONF_PASSWORD: "my_password",
+        }
+
+
+async def test_step_import(hass):
+    """Test the import step for both types of configuration."""
+    geography_conf = {
+        CONF_API_KEY: "abcde12345",
+        CONF_LATITUDE: 51.528308,
+        CONF_LONGITUDE: -0.3817765,
+    }
+
+    with patch(
+        "homeassistant.components.airvisual.async_setup_entry", return_value=True
+    ), patch("pyairvisual.api.API.nearest_city"):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_IMPORT}, data=geography_conf
+        )
 
         assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
         assert result["title"] == "Cloud API (51.528308, -0.3817765)"
@@ -157,23 +235,28 @@ async def test_step_import(hass):
 
 
 async def test_step_user(hass):
-    """Test that the user step works."""
-    conf = {
-        CONF_API_KEY: "abcde12345",
-        CONF_LATITUDE: 32.87336,
-        CONF_LONGITUDE: -117.22743,
-    }
+    """Test the user ("pick the integration type") step."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
 
-    with patch(
-        "homeassistant.components.airvisual.async_setup_entry", return_value=True
-    ), patch("pyairvisual.api.API.nearest_city"):
-        result = await hass.config_entries.flow.async_init(
-            DOMAIN, context={"source": SOURCE_USER}, data=conf
-        )
-        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
-        assert result["title"] == "Cloud API (32.87336, -117.22743)"
-        assert result["data"] == {
-            CONF_API_KEY: "abcde12345",
-            CONF_LATITUDE: 32.87336,
-            CONF_LONGITUDE: -117.22743,
-        }
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={"type": INTEGRATION_TYPE_GEOGRAPHY},
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "geography"
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={"type": INTEGRATION_TYPE_NODE_PRO},
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "node_pro"