diff --git a/.coveragerc b/.coveragerc
index 6a2a0db3ea4e7..75402c713252d 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -859,6 +859,9 @@ omit =
     homeassistant/components/openweathermap/sensor.py
     homeassistant/components/openweathermap/weather_update_coordinator.py
     homeassistant/components/opnsense/__init__.py
+    homeassistant/components/opower/__init__.py
+    homeassistant/components/opower/coordinator.py
+    homeassistant/components/opower/sensor.py
     homeassistant/components/opnsense/device_tracker.py
     homeassistant/components/opple/light.py
     homeassistant/components/oru/*
diff --git a/CODEOWNERS b/CODEOWNERS
index 3f8f27187f3f1..7e09c3c814715 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -897,6 +897,8 @@ build.json @home-assistant/supervisor
 /tests/components/openweathermap/ @fabaff @freekode @nzapponi
 /homeassistant/components/opnsense/ @mtreinish
 /tests/components/opnsense/ @mtreinish
+/homeassistant/components/opower/ @tronikos
+/tests/components/opower/ @tronikos
 /homeassistant/components/oralb/ @bdraco @Lash-L
 /tests/components/oralb/ @bdraco @Lash-L
 /homeassistant/components/oru/ @bvlaicu
diff --git a/homeassistant/components/opower/__init__.py b/homeassistant/components/opower/__init__.py
new file mode 100644
index 0000000000000..f4fca22c9b442
--- /dev/null
+++ b/homeassistant/components/opower/__init__.py
@@ -0,0 +1,31 @@
+"""The Opower integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import OpowerCoordinator
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Opower from a config entry."""
+
+    coordinator = OpowerCoordinator(hass, entry.data)
+    await coordinator.async_config_entry_first_refresh()
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
diff --git a/homeassistant/components/opower/config_flow.py b/homeassistant/components/opower/config_flow.py
new file mode 100644
index 0000000000000..fdf007c3b681a
--- /dev/null
+++ b/homeassistant/components/opower/config_flow.py
@@ -0,0 +1,112 @@
+"""Config flow for Opower integration."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from opower import CannotConnect, InvalidAuth, Opower, get_supported_utility_names
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+
+from .const import CONF_UTILITY, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_UTILITY): vol.In(get_supported_utility_names()),
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+async def _validate_login(
+    hass: HomeAssistant, login_data: dict[str, str]
+) -> dict[str, str]:
+    """Validate login data and return any errors."""
+    api = Opower(
+        async_create_clientsession(hass),
+        login_data[CONF_UTILITY],
+        login_data[CONF_USERNAME],
+        login_data[CONF_PASSWORD],
+    )
+    errors: dict[str, str] = {}
+    try:
+        await api.async_login()
+    except InvalidAuth:
+        errors["base"] = "invalid_auth"
+    except CannotConnect:
+        errors["base"] = "cannot_connect"
+    return errors
+
+
+class OpowerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Opower."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize a new OpowerConfigFlow."""
+        self.reauth_entry: config_entries.ConfigEntry | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match(
+                {
+                    CONF_UTILITY: user_input[CONF_UTILITY],
+                    CONF_USERNAME: user_input[CONF_USERNAME],
+                }
+            )
+            errors = await _validate_login(self.hass, user_input)
+            if not errors:
+                return self.async_create_entry(
+                    title=f"{user_input[CONF_UTILITY]} ({user_input[CONF_USERNAME]})",
+                    data=user_input,
+                )
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
+        """Handle configuration by re-auth."""
+        self.reauth_entry = self.hass.config_entries.async_get_entry(
+            self.context["entry_id"]
+        )
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Dialog that informs the user that reauth is required."""
+        assert self.reauth_entry
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            data = {**self.reauth_entry.data, **user_input}
+            errors = await _validate_login(self.hass, data)
+            if not errors:
+                self.hass.config_entries.async_update_entry(
+                    self.reauth_entry, data=data
+                )
+                await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
+                return self.async_abort(reason="reauth_successful")
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_USERNAME): self.reauth_entry.data[CONF_USERNAME],
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            errors=errors,
+        )
diff --git a/homeassistant/components/opower/const.py b/homeassistant/components/opower/const.py
new file mode 100644
index 0000000000000..b996a214a0591
--- /dev/null
+++ b/homeassistant/components/opower/const.py
@@ -0,0 +1,5 @@
+"""Constants for the Opower integration."""
+
+DOMAIN = "opower"
+
+CONF_UTILITY = "utility"
diff --git a/homeassistant/components/opower/coordinator.py b/homeassistant/components/opower/coordinator.py
new file mode 100644
index 0000000000000..4d40bb3356bfe
--- /dev/null
+++ b/homeassistant/components/opower/coordinator.py
@@ -0,0 +1,220 @@
+"""Coordinator to handle Opower connections."""
+from datetime import datetime, timedelta
+import logging
+from types import MappingProxyType
+from typing import Any, cast
+
+from opower import (
+    Account,
+    AggregateType,
+    CostRead,
+    Forecast,
+    InvalidAuth,
+    MeterType,
+    Opower,
+)
+
+from homeassistant.components.recorder import get_instance
+from homeassistant.components.recorder.models import StatisticData, StatisticMetaData
+from homeassistant.components.recorder.statistics import (
+    async_add_external_statistics,
+    get_last_statistics,
+    statistics_during_period,
+)
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, UnitOfEnergy, UnitOfVolume
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import CONF_UTILITY, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OpowerCoordinator(DataUpdateCoordinator):
+    """Handle fetching Opower data, updating sensors and inserting statistics."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry_data: MappingProxyType[str, Any],
+    ) -> None:
+        """Initialize the data handler."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Opower",
+            # Data is updated daily on Opower.
+            # Refresh every 12h to be at most 12h behind.
+            update_interval=timedelta(hours=12),
+        )
+        self.api = Opower(
+            aiohttp_client.async_get_clientsession(hass),
+            entry_data[CONF_UTILITY],
+            entry_data[CONF_USERNAME],
+            entry_data[CONF_PASSWORD],
+        )
+
+    async def _async_update_data(
+        self,
+    ) -> dict[str, Forecast]:
+        """Fetch data from API endpoint."""
+        try:
+            # Login expires after a few minutes.
+            # Given the infrequent updating (every 12h)
+            # assume previous session has expired and re-login.
+            await self.api.async_login()
+        except InvalidAuth as err:
+            raise ConfigEntryAuthFailed from err
+        forecasts: list[Forecast] = await self.api.async_get_forecast()
+        _LOGGER.debug("Updating sensor data with: %s", forecasts)
+        await self._insert_statistics([forecast.account for forecast in forecasts])
+        return {forecast.account.utility_account_id: forecast for forecast in forecasts}
+
+    async def _insert_statistics(self, accounts: list[Account]) -> None:
+        """Insert Opower statistics."""
+        for account in accounts:
+            id_prefix = "_".join(
+                (
+                    self.api.utility.subdomain(),
+                    account.meter_type.name.lower(),
+                    account.utility_account_id,
+                )
+            )
+            cost_statistic_id = f"{DOMAIN}:{id_prefix}_energy_cost"
+            consumption_statistic_id = f"{DOMAIN}:{id_prefix}_energy_consumption"
+            _LOGGER.debug(
+                "Updating Statistics for %s and %s",
+                cost_statistic_id,
+                consumption_statistic_id,
+            )
+
+            last_stat = await get_instance(self.hass).async_add_executor_job(
+                get_last_statistics, self.hass, 1, consumption_statistic_id, True, set()
+            )
+            if not last_stat:
+                _LOGGER.debug("Updating statistic for the first time")
+                cost_reads = await self._async_get_all_cost_reads(account)
+                cost_sum = 0.0
+                consumption_sum = 0.0
+                last_stats_time = None
+            else:
+                cost_reads = await self._async_get_recent_cost_reads(
+                    account, last_stat[consumption_statistic_id][0]["start"]
+                )
+                if not cost_reads:
+                    _LOGGER.debug("No recent usage/cost data. Skipping update")
+                    continue
+                stats = await get_instance(self.hass).async_add_executor_job(
+                    statistics_during_period,
+                    self.hass,
+                    cost_reads[0].start_time,
+                    None,
+                    {cost_statistic_id, consumption_statistic_id},
+                    "hour" if account.meter_type == MeterType.ELEC else "day",
+                    None,
+                    {"sum"},
+                )
+                cost_sum = cast(float, stats[cost_statistic_id][0]["sum"])
+                consumption_sum = cast(float, stats[consumption_statistic_id][0]["sum"])
+                last_stats_time = stats[cost_statistic_id][0]["start"]
+
+            cost_statistics = []
+            consumption_statistics = []
+
+            for cost_read in cost_reads:
+                start = cost_read.start_time
+                if last_stats_time is not None and start.timestamp() <= last_stats_time:
+                    continue
+                cost_sum += cost_read.provided_cost
+                consumption_sum += cost_read.consumption
+
+                cost_statistics.append(
+                    StatisticData(
+                        start=start, state=cost_read.provided_cost, sum=cost_sum
+                    )
+                )
+                consumption_statistics.append(
+                    StatisticData(
+                        start=start, state=cost_read.consumption, sum=consumption_sum
+                    )
+                )
+
+            name_prefix = " ".join(
+                (
+                    "Opower",
+                    self.api.utility.subdomain(),
+                    account.meter_type.name.lower(),
+                    account.utility_account_id,
+                )
+            )
+            cost_metadata = StatisticMetaData(
+                has_mean=False,
+                has_sum=True,
+                name=f"{name_prefix} cost",
+                source=DOMAIN,
+                statistic_id=cost_statistic_id,
+                unit_of_measurement=None,
+            )
+            consumption_metadata = StatisticMetaData(
+                has_mean=False,
+                has_sum=True,
+                name=f"{name_prefix} consumption",
+                source=DOMAIN,
+                statistic_id=consumption_statistic_id,
+                unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR
+                if account.meter_type == MeterType.ELEC
+                else UnitOfVolume.CENTUM_CUBIC_FEET,
+            )
+
+            async_add_external_statistics(self.hass, cost_metadata, cost_statistics)
+            async_add_external_statistics(
+                self.hass, consumption_metadata, consumption_statistics
+            )
+
+    async def _async_get_all_cost_reads(self, account: Account) -> list[CostRead]:
+        """Get all cost reads since account activation but at different resolutions depending on age.
+
+        - month resolution for all years (since account activation)
+        - day resolution for past 3 years
+        - hour resolution for past 2 months, only for electricity, not gas
+        """
+        cost_reads = []
+        start = None
+        end = datetime.now() - timedelta(days=3 * 365)
+        cost_reads += await self.api.async_get_cost_reads(
+            account, AggregateType.BILL, start, end
+        )
+        start = end if not cost_reads else cost_reads[-1].end_time
+        end = (
+            datetime.now() - timedelta(days=2 * 30)
+            if account.meter_type == MeterType.ELEC
+            else datetime.now()
+        )
+        cost_reads += await self.api.async_get_cost_reads(
+            account, AggregateType.DAY, start, end
+        )
+        if account.meter_type == MeterType.ELEC:
+            start = end if not cost_reads else cost_reads[-1].end_time
+            end = datetime.now()
+            cost_reads += await self.api.async_get_cost_reads(
+                account, AggregateType.HOUR, start, end
+            )
+        return cost_reads
+
+    async def _async_get_recent_cost_reads(
+        self, account: Account, last_stat_time: float
+    ) -> list[CostRead]:
+        """Get cost reads within the past 30 days to allow corrections in data from utilities.
+
+        Hourly for electricity, daily for gas.
+        """
+        return await self.api.async_get_cost_reads(
+            account,
+            AggregateType.HOUR
+            if account.meter_type == MeterType.ELEC
+            else AggregateType.DAY,
+            datetime.fromtimestamp(last_stat_time) - timedelta(days=30),
+            datetime.now(),
+        )
diff --git a/homeassistant/components/opower/manifest.json b/homeassistant/components/opower/manifest.json
new file mode 100644
index 0000000000000..969583f050a0a
--- /dev/null
+++ b/homeassistant/components/opower/manifest.json
@@ -0,0 +1,10 @@
+{
+  "domain": "opower",
+  "name": "Opower",
+  "codeowners": ["@tronikos"],
+  "config_flow": true,
+  "dependencies": ["recorder"],
+  "documentation": "https://www.home-assistant.io/integrations/opower",
+  "iot_class": "cloud_polling",
+  "requirements": ["opower==0.0.11"]
+}
diff --git a/homeassistant/components/opower/sensor.py b/homeassistant/components/opower/sensor.py
new file mode 100644
index 0000000000000..e28dcbd0661a7
--- /dev/null
+++ b/homeassistant/components/opower/sensor.py
@@ -0,0 +1,219 @@
+"""Support for Opower sensors."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from opower import Forecast, MeterType, UnitOfMeasure
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import UnitOfEnergy, UnitOfVolume
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import OpowerCoordinator
+
+
+@dataclass
+class OpowerEntityDescriptionMixin:
+    """Mixin values for required keys."""
+
+    value_fn: Callable[[Forecast], str | float]
+
+
+@dataclass
+class OpowerEntityDescription(SensorEntityDescription, OpowerEntityDescriptionMixin):
+    """Class describing Opower sensors entities."""
+
+
+# suggested_display_precision=0 for all sensors since
+# Opower provides 0 decimal points for all these.
+# (for the statistics in the energy dashboard Opower does provide decimal points)
+ELEC_SENSORS: tuple[OpowerEntityDescription, ...] = (
+    OpowerEntityDescription(
+        key="elec_usage_to_date",
+        name="Current bill electric usage to date",
+        device_class=SensorDeviceClass.ENERGY,
+        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.usage_to_date,
+    ),
+    OpowerEntityDescription(
+        key="elec_forecasted_usage",
+        name="Current bill electric forecasted usage",
+        device_class=SensorDeviceClass.ENERGY,
+        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.forecasted_usage,
+    ),
+    OpowerEntityDescription(
+        key="elec_typical_usage",
+        name="Typical monthly electric usage",
+        device_class=SensorDeviceClass.ENERGY,
+        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.typical_usage,
+    ),
+    OpowerEntityDescription(
+        key="elec_cost_to_date",
+        name="Current bill electric cost to date",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.cost_to_date,
+    ),
+    OpowerEntityDescription(
+        key="elec_forecasted_cost",
+        name="Current bill electric forecasted cost",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.forecasted_cost,
+    ),
+    OpowerEntityDescription(
+        key="elec_typical_cost",
+        name="Typical monthly electric cost",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.typical_cost,
+    ),
+)
+GAS_SENSORS: tuple[OpowerEntityDescription, ...] = (
+    OpowerEntityDescription(
+        key="gas_usage_to_date",
+        name="Current bill gas usage to date",
+        device_class=SensorDeviceClass.GAS,
+        native_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        suggested_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.usage_to_date,
+    ),
+    OpowerEntityDescription(
+        key="gas_forecasted_usage",
+        name="Current bill gas forecasted usage",
+        device_class=SensorDeviceClass.GAS,
+        native_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        suggested_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.forecasted_usage,
+    ),
+    OpowerEntityDescription(
+        key="gas_typical_usage",
+        name="Typical monthly gas usage",
+        device_class=SensorDeviceClass.GAS,
+        native_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        suggested_unit_of_measurement=UnitOfVolume.CENTUM_CUBIC_FEET,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.typical_usage,
+    ),
+    OpowerEntityDescription(
+        key="gas_cost_to_date",
+        name="Current bill gas cost to date",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.cost_to_date,
+    ),
+    OpowerEntityDescription(
+        key="gas_forecasted_cost",
+        name="Current bill gas forecasted cost",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.forecasted_cost,
+    ),
+    OpowerEntityDescription(
+        key="gas_typical_cost",
+        name="Typical monthly gas cost",
+        device_class=SensorDeviceClass.MONETARY,
+        state_class=SensorStateClass.TOTAL,
+        suggested_display_precision=0,
+        value_fn=lambda data: data.typical_cost,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the Opower sensor."""
+
+    coordinator: OpowerCoordinator = hass.data[DOMAIN][entry.entry_id]
+    entities: list[OpowerSensor] = []
+    forecasts: list[Forecast] = coordinator.data.values()
+    for forecast in forecasts:
+        device_id = f"{coordinator.api.utility.subdomain()}_{forecast.account.utility_account_id}"
+        device = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            name=f"{forecast.account.meter_type.name} account {forecast.account.utility_account_id}",
+            manufacturer="Opower",
+            model=coordinator.api.utility.name(),
+        )
+        sensors: tuple[OpowerEntityDescription, ...] = ()
+        if (
+            forecast.account.meter_type == MeterType.ELEC
+            and forecast.unit_of_measure == UnitOfMeasure.KWH
+        ):
+            sensors = ELEC_SENSORS
+        elif (
+            forecast.account.meter_type == MeterType.GAS
+            and forecast.unit_of_measure == UnitOfMeasure.THERM
+        ):
+            sensors = GAS_SENSORS
+        for sensor in sensors:
+            entities.append(
+                OpowerSensor(
+                    coordinator,
+                    sensor,
+                    forecast.account.utility_account_id,
+                    device,
+                    device_id,
+                )
+            )
+
+    async_add_entities(entities)
+
+
+class OpowerSensor(SensorEntity, CoordinatorEntity[OpowerCoordinator]):
+    """Representation of an Opower sensor."""
+
+    def __init__(
+        self,
+        coordinator: OpowerCoordinator,
+        description: OpowerEntityDescription,
+        utility_account_id: str,
+        device: DeviceInfo,
+        device_id: str,
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator)
+        self.entity_description: OpowerEntityDescription = description
+        self._attr_unique_id = f"{device_id}_{description.key}"
+        self._attr_device_info = device
+        self.utility_account_id = utility_account_id
+
+    @property
+    def native_value(self) -> StateType:
+        """Return the state."""
+        if self.coordinator.data is not None:
+            return self.entity_description.value_fn(
+                self.coordinator.data[self.utility_account_id]
+            )
+        return None
diff --git a/homeassistant/components/opower/strings.json b/homeassistant/components/opower/strings.json
new file mode 100644
index 0000000000000..79d8bf80feeb3
--- /dev/null
+++ b/homeassistant/components/opower/strings.json
@@ -0,0 +1,28 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "utility": "Utility name",
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      },
+      "reauth_confirm": {
+        "title": "[%key:common::config_flow::title::reauth%]",
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
+    }
+  }
+}
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index 2925ea3425cc5..d8ffa25b765da 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -325,6 +325,7 @@
         "opentherm_gw",
         "openuv",
         "openweathermap",
+        "opower",
         "oralb",
         "otbr",
         "overkiz",
diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json
index 98571b6905e48..314e8ffa092f0 100644
--- a/homeassistant/generated/integrations.json
+++ b/homeassistant/generated/integrations.json
@@ -4016,6 +4016,12 @@
       "config_flow": false,
       "iot_class": "local_polling"
     },
+    "opower": {
+      "name": "Opower",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "cloud_polling"
+    },
     "opple": {
       "name": "Opple",
       "integration_type": "hub",
diff --git a/requirements_all.txt b/requirements_all.txt
index 87f74d66523cc..74c0b80ad5702 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -1362,6 +1362,9 @@ openwrt-luci-rpc==1.1.16
 # homeassistant.components.ubus
 openwrt-ubus-rpc==0.0.2
 
+# homeassistant.components.opower
+opower==0.0.11
+
 # homeassistant.components.oralb
 oralb-ble==0.17.6
 
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 86e22c3644c3f..09706afca237d 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -1028,6 +1028,9 @@ openerz-api==0.2.0
 # homeassistant.components.openhome
 openhomedevice==2.2.0
 
+# homeassistant.components.opower
+opower==0.0.11
+
 # homeassistant.components.oralb
 oralb-ble==0.17.6
 
diff --git a/tests/components/opower/__init__.py b/tests/components/opower/__init__.py
new file mode 100644
index 0000000000000..71aea27a69835
--- /dev/null
+++ b/tests/components/opower/__init__.py
@@ -0,0 +1 @@
+"""Tests for the Opower integration."""
diff --git a/tests/components/opower/conftest.py b/tests/components/opower/conftest.py
new file mode 100644
index 0000000000000..17c6896593b03
--- /dev/null
+++ b/tests/components/opower/conftest.py
@@ -0,0 +1,25 @@
+"""Fixtures for the Opower integration tests."""
+import pytest
+
+from homeassistant.components.opower.const import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture
+def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
+    """Return the default mocked config entry."""
+    config_entry = MockConfigEntry(
+        title="Pacific Gas & Electric (test-username)",
+        domain=DOMAIN,
+        data={
+            "utility": "Pacific Gas and Electric Company (PG&E)",
+            "username": "test-username",
+            "password": "test-password",
+        },
+        state=ConfigEntryState.LOADED,
+    )
+    config_entry.add_to_hass(hass)
+    return config_entry
diff --git a/tests/components/opower/test_config_flow.py b/tests/components/opower/test_config_flow.py
new file mode 100644
index 0000000000000..7f6a847f52e3b
--- /dev/null
+++ b/tests/components/opower/test_config_flow.py
@@ -0,0 +1,204 @@
+"""Test the Opower config flow."""
+from collections.abc import Generator
+from unittest.mock import AsyncMock, patch
+
+from opower import CannotConnect, InvalidAuth
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.opower.const import DOMAIN
+from homeassistant.components.recorder import Recorder
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture(autouse=True, name="mock_setup_entry")
+def override_async_setup_entry() -> Generator[AsyncMock, None, None]:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.opower.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+@pytest.fixture
+def mock_unload_entry() -> Generator[AsyncMock, None, None]:
+    """Mock unloading a config entry."""
+    with patch(
+        "homeassistant.components.opower.async_unload_entry",
+        return_value=True,
+    ) as mock_unload_entry:
+        yield mock_unload_entry
+
+
+async def test_form(
+    recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
+) -> None:
+    """Test we get the form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert not result["errors"]
+
+    with patch(
+        "homeassistant.components.opower.config_flow.Opower.async_login",
+    ) as mock_login:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "utility": "Pacific Gas and Electric Company (PG&E)",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == "Pacific Gas and Electric Company (PG&E) (test-username)"
+    assert result2["data"] == {
+        "utility": "Pacific Gas and Electric Company (PG&E)",
+        "username": "test-username",
+        "password": "test-password",
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert mock_login.call_count == 1
+
+
+@pytest.mark.parametrize(
+    ("api_exception", "expected_error"),
+    [
+        (InvalidAuth(), "invalid_auth"),
+        (CannotConnect(), "cannot_connect"),
+    ],
+)
+async def test_form_exceptions(
+    recorder_mock: Recorder, hass: HomeAssistant, api_exception, expected_error
+) -> None:
+    """Test we handle exceptions."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.opower.config_flow.Opower.async_login",
+        side_effect=api_exception,
+    ) as mock_login:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "utility": "Pacific Gas and Electric Company (PG&E)",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": expected_error}
+    assert mock_login.call_count == 1
+
+
+async def test_form_already_configured(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test user input for config_entry that already exists."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.opower.config_flow.Opower.async_login",
+    ) as mock_login:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "utility": "Pacific Gas and Electric Company (PG&E)",
+                "username": "test-username",
+                "password": "test-password",
+            },
+        )
+
+    assert result2["type"] == FlowResultType.ABORT
+    assert result2["reason"] == "already_configured"
+    assert mock_login.call_count == 0
+
+
+async def test_form_not_already_configured(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test user input for config_entry different than the existing one."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.opower.config_flow.Opower.async_login",
+    ) as mock_login:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "utility": "Pacific Gas and Electric Company (PG&E)",
+                "username": "test-username2",
+                "password": "test-password",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert (
+        result2["title"] == "Pacific Gas and Electric Company (PG&E) (test-username2)"
+    )
+    assert result2["data"] == {
+        "utility": "Pacific Gas and Electric Company (PG&E)",
+        "username": "test-username2",
+        "password": "test-password",
+    }
+    assert len(mock_setup_entry.mock_calls) == 2
+    assert mock_login.call_count == 1
+
+
+async def test_form_valid_reauth(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_unload_entry: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test that we can handle a valid reauth."""
+    mock_config_entry.async_start_reauth(hass)
+    await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    result = flows[0]
+    assert result["step_id"] == "reauth_confirm"
+    assert result["context"]["source"] == "reauth"
+    assert result["context"]["title_placeholders"] == {"name": mock_config_entry.title}
+
+    with patch(
+        "homeassistant.components.opower.config_flow.Opower.async_login",
+    ) as mock_login:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"username": "test-username", "password": "test-password2"},
+        )
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+
+    await hass.async_block_till_done()
+    assert hass.config_entries.async_entries(DOMAIN)[0].data == {
+        "utility": "Pacific Gas and Electric Company (PG&E)",
+        "username": "test-username",
+        "password": "test-password2",
+    }
+    assert len(mock_unload_entry.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert mock_login.call_count == 1