Add Monarch Money Integration

CODEOWNERS

@@ -922,6 +922,8 @@ build.json @home-assistant/supervisor
 /tests/components/modern_forms/ @wonderslug
 /homeassistant/components/moehlenhoff_alpha2/ @j-a-n
 /tests/components/moehlenhoff_alpha2/ @j-a-n
+/homeassistant/components/monarch_money/ @jeeftor
+/tests/components/monarch_money/ @jeeftor
 /homeassistant/components/monoprice/ @etsinko @OnFreund
 /tests/components/monoprice/ @etsinko @OnFreund
 /homeassistant/components/monzo/ @jakemartin-icl

homeassistant/components/monarch_money/__init__.py

@@ -0,0 +1,35 @@
+"""The Monarch Money integration."""
+
+from __future__ import annotations
+
+from typedmonarchmoney import TypedMonarchMoney
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_TOKEN, Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import MonarchMoneyDataUpdateCoordinator
+
+type MonarchMoneyConfigEntry = ConfigEntry[MonarchMoneyDataUpdateCoordinator]
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: MonarchMoneyConfigEntry
+) -> bool:
+    """Set up Monarch Money from a config entry."""
+    monarch_client = TypedMonarchMoney(token=entry.data.get(CONF_TOKEN))
+
+    mm_coordinator = MonarchMoneyDataUpdateCoordinator(hass, monarch_client)
+    await mm_coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = mm_coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: MonarchMoneyConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/monarch_money/config_flow.py

@@ -0,0 +1,157 @@
+"""Config flow for Monarch Money integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from monarchmoney import LoginFailedException, RequireMFAException
+from monarchmoney.monarchmoney import SESSION_FILE
+from typedmonarchmoney import TypedMonarchMoney
+from typedmonarchmoney.models import MonarchSubscription
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_EMAIL, CONF_ID, CONF_PASSWORD, CONF_TOKEN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.selector import (
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import CONF_MFA_CODE, DOMAIN, LOGGER
+
+_LOGGER = logging.getLogger(__name__)
+
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_EMAIL): TextSelector(
+            TextSelectorConfig(
+                type=TextSelectorType.EMAIL,
+            ),
+        ),
+        vol.Required(CONF_PASSWORD): TextSelector(
+            TextSelectorConfig(
+                type=TextSelectorType.PASSWORD,
+            ),
+        ),
+    }
+)
+
+STEP_MFA_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_MFA_CODE): str,
+    }
+)
+
+
+async def validate_login(
+    hass: HomeAssistant,
+    data: dict[str, Any],
+    email: str | None = None,
+    password: str | None = None,
+) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. Upon success a session will be saved
+    """
+
+    if not email:
+        email = data[CONF_EMAIL]
+    if not password:
+        password = data[CONF_PASSWORD]
+    monarch_client = TypedMonarchMoney()
+    if CONF_MFA_CODE in data:
+        mfa_code = data[CONF_MFA_CODE]
+        LOGGER.debug("Attempting to authenticate with MFA code")
+        try:
+            await monarch_client.multi_factor_authenticate(email, password, mfa_code)
+        except KeyError as err:
+            # A bug in the backing lib that I don't control throws a KeyError if the MFA code is wrong
+            LOGGER.debug("Bad MFA Code")
+            raise BadMFA from err
+    else:
+        LOGGER.debug("Attempting to authenticate")
+        try:
+            await monarch_client.login(
+                email=email,
+                password=password,
+                save_session=False,
+                use_saved_session=False,
+            )
+        except RequireMFAException:
+            raise
+        except LoginFailedException as err:
+            raise InvalidAuth from err
+
+    LOGGER.debug(f"Connection successful - saving session to file {SESSION_FILE}")
+    LOGGER.debug("Obtaining subscription id")
+    subs: MonarchSubscription = await monarch_client.get_subscription_details()
+    assert subs is not None
+    subscription_id = subs.id
+    return {
+        CONF_TOKEN: monarch_client.token,
+        CONF_ID: subscription_id,
+    }
+
+
+class MonarchMoneyConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Monarch Money."""
+
+    VERSION = 1
+
+    def __init__(self):
+        """Initialize config flow."""
+        self.email: str | None = None
+        self.password: str | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            try:
+                info = await validate_login(
+                    self.hass, user_input, email=self.email, password=self.password
+                )
+            except RequireMFAException:
+                self.email = user_input[CONF_EMAIL]
+                self.password = user_input[CONF_PASSWORD]
+
+                return self.async_show_form(
+                    step_id="user",
+                    data_schema=STEP_MFA_DATA_SCHEMA,
+                    errors={"base": "mfa_required"},
+                )
+            except BadMFA:
+                return self.async_show_form(
+                    step_id="user",
+                    data_schema=STEP_MFA_DATA_SCHEMA,
+                    errors={"base": "bad_mfa"},
+                )
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            else:
+                await self.async_set_unique_id(info[CONF_ID])
+                self._abort_if_unique_id_configured()
+
+                return self.async_create_entry(
+                    title="Monarch Money",
+                    data={CONF_TOKEN: info[CONF_TOKEN]},
+                )
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class BadMFA(HomeAssistantError):
+    """Error to indicate the MFA code was bad."""

homeassistant/components/monarch_money/const.py

@@ -0,0 +1,10 @@
+"""Constants for the Monarch Money integration."""
+
+import logging
+
+DOMAIN = "monarch_money"
+
+LOGGER = logging.getLogger(__package__)
+
+CONF_MFA_SECRET = "mfa_secret"
+CONF_MFA_CODE = "mfa_code"

homeassistant/components/monarch_money/coordinator.py

@@ -0,0 +1,91 @@
+"""Data coordinator for monarch money."""
+
+import asyncio
+from dataclasses import dataclass
+from datetime import timedelta
+
+from aiohttp import ClientResponseError
+from gql.transport.exceptions import TransportServerError
+from monarchmoney import LoginFailedException
+from typedmonarchmoney import TypedMonarchMoney
+from typedmonarchmoney.models import (
+    MonarchAccount,
+    MonarchCashflowSummary,
+    MonarchSubscription,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryError
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import LOGGER
+
+
+@dataclass
+class MonarchData:
+    """Data class to hold monarch data."""
+
+    account_data: dict[str, MonarchAccount]
+    cashflow_summary: MonarchCashflowSummary
+
+
+class MonarchMoneyDataUpdateCoordinator(DataUpdateCoordinator[MonarchData]):
+    """Data update coordinator for Monarch Money."""
+
+    config_entry: ConfigEntry
+    subscription_id: str
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        client: TypedMonarchMoney,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass=hass,
+            logger=LOGGER,
+            name="monarchmoney",
+            update_interval=timedelta(hours=4),
+        )
+        self.client = client
+
+    async def _async_setup(self) -> None:
+        """Obtain subscription ID in setup phase."""
+        try:
+            sub_details: MonarchSubscription = (
+                await self.client.get_subscription_details()
+            )
+        except (TransportServerError, LoginFailedException, ClientResponseError) as err:
+            raise ConfigEntryError("Authentication failed") from err
+        self.subscription_id = sub_details.id
+
+    async def _async_update_data(self) -> MonarchData:
+        """Fetch data for all accounts."""
+
+        account_data, cashflow_summary = await asyncio.gather(
+            self.client.get_accounts_as_dict_with_id_key(),
+            self.client.get_cashflow_summary(),
+        )
+
+        return MonarchData(account_data=account_data, cashflow_summary=cashflow_summary)
+
+    @property
+    def cashflow_summary(self) -> MonarchCashflowSummary:
+        """Return cashflow summary."""
+        return self.data.cashflow_summary
+
+    @property
+    def accounts(self) -> list[MonarchAccount]:
+        """Return accounts."""
+        return list(self.data.account_data.values())
+
+    @property
+    def value_accounts(self) -> list[MonarchAccount]:
+        """Return value accounts."""
+        return [x for x in self.accounts if x.is_value_account]
+
+    @property
+    def balance_accounts(self) -> list[MonarchAccount]:
+        """Return accounts that aren't assets."""
+        return [x for x in self.accounts if x.is_balance_account]

homeassistant/components/monarch_money/entity.py

@@ -0,0 +1,83 @@
+"""Monarch money entity definition."""
+
+from typedmonarchmoney.models import MonarchAccount, MonarchCashflowSummary
+
+from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
+from homeassistant.helpers.entity import EntityDescription
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import MonarchMoneyDataUpdateCoordinator
+
+
+class MonarchMoneyEntityBase(CoordinatorEntity[MonarchMoneyDataUpdateCoordinator]):
+    """Base entity for Monarch Money with entity name attribute."""
+
+    _attr_has_entity_name = True
+
+
+class MonarchMoneyCashFlowEntity(MonarchMoneyEntityBase):
+    """Entity for Cashflow sensors."""
+
+    def __init__(
+        self,
+        coordinator: MonarchMoneyDataUpdateCoordinator,
+        description: EntityDescription,
+    ) -> None:
+        """Initialize the Monarch Money Entity."""
+        super().__init__(coordinator)
+        self._attr_unique_id = (
+            f"{coordinator.subscription_id}_cashflow_{description.key}"
+        )
+        self.entity_description = description
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, str(coordinator.subscription_id))},
+            name="Cashflow",
+        )
+
+    @property
+    def summary_data(self) -> MonarchCashflowSummary:
+        """Return cashflow summary data."""
+        return self.coordinator.cashflow_summary
+
+
+class MonarchMoneyAccountEntity(MonarchMoneyEntityBase):
+    """Entity for Account Sensors."""
+
+    def __init__(
+        self,
+        coordinator: MonarchMoneyDataUpdateCoordinator,
+        description: EntityDescription,
+        account: MonarchAccount,
+    ) -> None:
+        """Initialize the Monarch Money Entity."""
+        super().__init__(coordinator)
+
+        self.entity_description = description
+        self._account_id = account.id
+        self._attr_attribution = (
+            f"Data provided by Monarch Money API via {account.data_provider}"
+        )
+        self._attr_unique_id = (
+            f"{coordinator.subscription_id}_{account.id}_{description.translation_key}"
+        )
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, str(account.id))},
+            name=f"{account.institution_name} {account.name}",
+            entry_type=DeviceEntryType.SERVICE,
+            manufacturer=account.data_provider,
+            model=f"{account.institution_name} - {account.type_name} - {account.subtype_name}",
+            configuration_url=account.institution_url,
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return super().available and (
+            self._account_id in self.coordinator.data.account_data
+        )
+
+    @property
+    def account_data(self) -> MonarchAccount:
+        """Return the account data."""
+        return self.coordinator.data.account_data[self._account_id]