diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..342b2f8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +.storage/ +.DS_Store +.vscode/ diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..09ac841 --- /dev/null +++ b/__init__.py @@ -0,0 +1,160 @@ +"""The YNAB Custom integration.""" + +import logging +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.typing import ConfigType +from homeassistant.helpers import entity_registry as er +import re + +from .const import DOMAIN, CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL, CONF_INCLUDE_CLOSED_ACCOUNTS, CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_HIDDEN_CATEGORIES, CONF_SELECTED_ACCOUNTS, CONF_SELECTED_CATEGORIES +from .coordinator import YNABDataUpdateCoordinator + +_LOGGER = logging.getLogger(__name__) + +def sanitize_budget_name(budget_name: str) -> str: + """Sanitize the budget name to create a valid Home Assistant entity ID.""" + # Replace spaces with underscores and remove any special characters + sanitized_name = re.sub(r'[^a-zA-Z0-9_]', '', budget_name.replace(" ", "_")) + return sanitized_name + +async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Migrate config entry from version 1 to version 2.""" + _LOGGER.info(f"Migrating YNAB config entry {entry.entry_id} from version {entry.version} to version 2") + _LOGGER.debug(f"Entry data before migration: {entry.data}") + _LOGGER.debug(f"Entry options before migration: {entry.options}") + + try: + # Create new data and options dictionaries + new_data = dict(entry.data) + new_options = dict(entry.options) if entry.options else {} + + # Migrate update_interval from data to options if it exists in data + if CONF_UPDATE_INTERVAL in entry.data: + new_options.setdefault(CONF_UPDATE_INTERVAL, entry.data[CONF_UPDATE_INTERVAL]) + # Remove from data since it should be in options + new_data.pop(CONF_UPDATE_INTERVAL, None) + else: + # Ensure update_interval exists in options with default + new_options.setdefault(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) + + # Add new checkbox options with safe defaults + new_options.setdefault(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + new_options.setdefault(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + + # Update the config entry + hass.config_entries.async_update_entry( + entry, + data=new_data, + options=new_options, + version=2 + ) + + _LOGGER.info(f"Successfully migrated YNAB config entry {entry.entry_id} to version 2") + _LOGGER.debug(f"Entry data after migration: {new_data}") + _LOGGER.debug(f"Entry options after migration: {new_options}") + return True + + except Exception as e: + _LOGGER.exception(f"Failed to migrate YNAB config entry {entry.entry_id}: {e}") + return False + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up YNAB Custom from a config entry.""" + _LOGGER.debug(f"Setting up YNAB integration {entry.entry_id}") + + budget_id = entry.data.get("budget_id") + budget_name = entry.data.get("budget_name") + + if not budget_id or not budget_name: + _LOGGER.error("Missing budget_id or budget_name in config entry.") + return False + + # Sanitize the budget name to avoid issues with special characters or spaces + sanitized_budget_name = sanitize_budget_name(budget_name) + coordinator = YNABDataUpdateCoordinator(hass, entry, budget_id, sanitized_budget_name) + + # Load persistent data before first refresh + await coordinator.async_load_persistent_data() + + await coordinator.async_config_entry_first_refresh() # Ensure the first update occurs + hass.async_create_task(coordinator.async_refresh()) # 🔹 Manually force an update on startup + + if DOMAIN not in hass.data: + hass.data[DOMAIN] = {} + + hass.data[DOMAIN][entry.entry_id] = coordinator + await hass.config_entries.async_forward_entry_setups(entry, ["sensor", "number"]) + + + + # Set up options update listener + entry.async_on_unload(entry.add_update_listener(async_update_options)) + + _LOGGER.info(f"YNAB Custom integration for {sanitized_budget_name} successfully loaded.") + return True + + +async def async_update_options(hass: HomeAssistant, entry: ConfigEntry) -> None: + """Update options.""" + + # Get the coordinator to check what changed + coordinator = hass.data.get(DOMAIN, {}).get(entry.entry_id) + if coordinator: + old_accounts = set(coordinator.selected_accounts) + old_categories = set(coordinator.selected_categories) + new_accounts = set(entry.data.get(CONF_SELECTED_ACCOUNTS, [])) + new_categories = set(entry.data.get(CONF_SELECTED_CATEGORIES, [])) + + _LOGGER.debug(f"Options update - OLD accounts: {old_accounts}, NEW accounts: {new_accounts}") + _LOGGER.debug(f"Options update - OLD categories: {old_categories}, NEW categories: {new_categories}") + + if old_accounts != new_accounts or old_categories != new_categories: + _LOGGER.info(f"Account/category selections changed - doing full unload/reload cycle") + + # Force a complete unload/reload cycle to ensure clean state + unload_result = await hass.config_entries.async_unload(entry.entry_id) + setup_result = await hass.config_entries.async_setup(entry.entry_id) + + _LOGGER.info(f"Unload/reload cycle complete - Unload: {unload_result}, Setup: {setup_result}") + return + else: + _LOGGER.debug(f"Only settings changed, doing simple reload") + else: + _LOGGER.warning(f"No coordinator found for {entry.entry_id}") + + # Simple reload for non-selection changes + await hass.config_entries.async_reload(entry.entry_id) + _LOGGER.debug(f"Simple reload complete") + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + + # Clean up entity registry entries for this config entry + entity_registry = er.async_get(hass) + entities_to_remove = [] + + for entity_entry in entity_registry.entities.values(): + if entity_entry.config_entry_id == entry.entry_id: + entities_to_remove.append(entity_entry.entity_id) + + # Remove entities from registry + removed_entities = [] + for entity_id in entities_to_remove: + entity_registry.async_remove(entity_id) + removed_entities.append(entity_id) + + coordinator = hass.data[DOMAIN].pop(entry.entry_id, None) + + if coordinator: + await coordinator.async_shutdown() + + result = await hass.config_entries.async_unload_platforms(entry, ["sensor"]) + + if removed_entities: + _LOGGER.debug(f"Unload complete for {entry.entry_id} - removed {len(removed_entities)} entities") + else: + _LOGGER.debug(f"Unload complete for {entry.entry_id}") + + return result diff --git a/api.py b/api.py new file mode 100644 index 0000000..3ba5b0e --- /dev/null +++ b/api.py @@ -0,0 +1,166 @@ +"""YNAB API Handler.""" + +import aiohttp +import logging +from datetime import datetime, timedelta +from collections import deque + +_LOGGER = logging.getLogger(__name__) + +class YNABApi: + """YNAB API class for handling requests.""" + + BASE_URL = "https://api.youneedabudget.com/v1" + + def __init__(self, access_token: str, shared_tracking: dict = None): + """Initialize the API client.""" + self.access_token = access_token + self.headers = {"Authorization": f"Bearer {access_token}"} + + # Use shared request tracking if provided, otherwise create local tracking + if shared_tracking: + self.request_timestamps = shared_tracking["request_timestamps"] + self._shared_tracking = shared_tracking # Keep reference for shared counter + else: + # Fallback to local tracking (for backwards compatibility) + self.request_timestamps = deque() + self._shared_tracking = None + + def _track_request(self): + """Track a new API request with timestamp.""" + now = datetime.now() + self.request_timestamps.append(now) + + # Update shared total counter (cumulative across all instances) + if self._shared_tracking: + self._shared_tracking["total_requests"] += 1 # Increment shared counter + else: + # Fallback for local tracking + if not hasattr(self, 'total_requests'): + self.total_requests = 0 + self.total_requests += 1 + + # Clean old requests (older than 1 hour) + cutoff_time = now - timedelta(hours=1) + cleaned_count = 0 + while self.request_timestamps and self.request_timestamps[0] < cutoff_time: + self.request_timestamps.popleft() + cleaned_count += 1 + + + def get_rate_limit_info(self): + """Get current rate limit status and statistics.""" + now = datetime.now() + + # Clean old requests first + cutoff_time = now - timedelta(hours=1) + while self.request_timestamps and self.request_timestamps[0] < cutoff_time: + self.request_timestamps.popleft() + + requests_this_hour = len(self.request_timestamps) + estimated_remaining = max(0, 200 - requests_this_hour) + + # Calculate when the rate limit resets (1 hour from oldest request) + if self.request_timestamps: + oldest_request = self.request_timestamps[0] + rate_limit_resets_at = oldest_request + timedelta(hours=1) + else: + rate_limit_resets_at = now + timedelta(hours=1) + + return { + "requests_made_total": self._shared_tracking["total_requests"] if self._shared_tracking else getattr(self, 'total_requests', 0), + "requests_this_hour": requests_this_hour, + "estimated_remaining": estimated_remaining, + "rate_limit_resets_at": rate_limit_resets_at.strftime("%I:%M %p"), + } + + async def get_budgets(self): + """Fetch available budgets.""" + url = f"{self.BASE_URL}/budgets" + _LOGGER.debug("Fetching budgets from URL: %s", url) + return await self._get(url) + + async def get_budget(self, budget_id: str): + """Fetch full details for a specific budget.""" + if not budget_id or budget_id == "budgets": + _LOGGER.error("Invalid budget_id before API call: %s", budget_id) + return {} + + url = f"{self.BASE_URL}/budgets/{budget_id}" + _LOGGER.debug("Fetching budget from URL: %s", url) + return await self._get(url) + + async def get_accounts(self, budget_id: str): + """Fetch accounts for a specific budget.""" + if not budget_id or budget_id == "budgets": + _LOGGER.error("Invalid budget_id before accounts API call: %s", budget_id) + return {} + + url = f"{self.BASE_URL}/budgets/{budget_id}/accounts" + _LOGGER.debug("Fetching accounts from URL: %s", url) + return await self._get(url) + + async def get_categories(self, budget_id: str): + """Fetch categories for a specific budget.""" + if not budget_id or budget_id == "budgets": + _LOGGER.error("Invalid budget_id before categories API call: %s", budget_id) + return {} + + url = f"{self.BASE_URL}/budgets/{budget_id}/categories" + _LOGGER.debug("Fetching categories from URL: %s", url) + return await self._get(url) + + async def get_monthly_summary(self, budget_id: str, current_month: str): + """Fetch the latest monthly summary for a specific budget.""" + if not budget_id or budget_id == "budgets": + _LOGGER.error("Invalid budget_id before months API call: %s", budget_id) + return {} + + # Form the URL to query the specific month + url = f"{self.BASE_URL}/budgets/{budget_id}/months/{current_month}" # Include the full date with day + _LOGGER.debug(f"Requesting URL: {url}") # Log the full URL being requested + + # Fetch the data - let 429 errors propagate to coordinator + response = await self._get(url) + + # Log the response data + _LOGGER.debug(f"Response for {url}: {response}") + + # If response contains data, return it + if response: + return response + else: + _LOGGER.warning(f"No data found for {current_month} in budget: {budget_id}") + return {} + + async def get_transactions(self, budget_id: str): + """Fetch recent transactions for a specific budget.""" + if not budget_id or budget_id == "budgets": + _LOGGER.error("Invalid budget_id before transactions API call: %s", budget_id) + return {} + + url = f"{self.BASE_URL}/budgets/{budget_id}/transactions" + _LOGGER.debug("Fetching transactions from URL: %s", url) + return await self._get(url) + + async def _get(self, url: str): + """Generic GET request handler.""" + # Track this request + self._track_request() + + + async with aiohttp.ClientSession() as session: + async with session.get(url, headers=self.headers) as response: + if response.status == 200: + # If the response is successful, return the data + data = await response.json() + return data.get("data", {}) + else: + # Log if the request fails + _LOGGER.error("YNAB API error: %s - URL: %s", response.status, url) + + # Raise exception for 429 errors so coordinator can catch them + if response.status == 429: + raise Exception(f"429 - Too Many Requests: {url}") + + return {} diff --git a/config_flow.py b/config_flow.py new file mode 100644 index 0000000..01d5460 --- /dev/null +++ b/config_flow.py @@ -0,0 +1,611 @@ +from __future__ import annotations + +import logging +import re +import voluptuous as vol +from typing import Any, Dict + +from homeassistant import config_entries, exceptions +from homeassistant.const import CONF_ACCESS_TOKEN +from homeassistant.helpers import config_validation as cv +from homeassistant.data_entry_flow import FlowResult +from homeassistant.core import callback + +from .const import DOMAIN, CONF_SELECTED_ACCOUNTS, CONF_SELECTED_CATEGORIES, CONF_CURRENCY, CONF_SELECTED_BUDGET, CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL, CONF_INCLUDE_CLOSED_ACCOUNTS, CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_HIDDEN_CATEGORIES +from .api import YNABApi + +_LOGGER = logging.getLogger(__name__) + +SELECT_ALL_OPTION = "Select All" + +# Supported currency options +CURRENCY_OPTIONS = { + "USD": "$ (US Dollar)", + "EUR": "€ (Euro)", + "GBP": "£ (British Pound)", + "AUD": "A$ (Australian Dollar)", + "CAD": "C$ (Canadian Dollar)", + "JPY": "¥ (Japanese Yen)", + "CHF": "CHF (Swiss Franc)", + "SEK": "kr (Swedish Krona)", + "NZD": "NZ$ (New Zealand Dollar)", +} + +# Polling interval options (5-60 minutes) +POLLING_INTERVAL_OPTIONS = {i: f"{i} minute{'s' if i > 1 else ''}" for i in range(5, 61)} + +# Function to sanitize the budget name +def sanitize_budget_name(budget_name: str) -> str: + """Sanitize the budget name to create a valid Home Assistant entity ID.""" + # Replace spaces with underscores and remove any special characters + sanitized_name = re.sub(r'[^a-zA-Z0-9_]', '', budget_name.replace(" ", "_")) + return sanitized_name + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for YNAB Custom integration.""" + + VERSION = 2 + + @staticmethod + @callback + def async_get_options_flow(config_entry): + """Return the options flow.""" + return OptionsFlow(config_entry) + + async def async_step_user(self, user_input: Dict[str, Any] | None = None) -> FlowResult: + """Handle the initial step to enter an access token.""" + errors = {} + + + if user_input is None: + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({ + vol.Required(CONF_ACCESS_TOKEN): str, + vol.Required("Accept Terms", default=False): bool # Checkbox for accepting the terms + }), + errors=errors + ) + + # Check if terms are accepted, otherwise prompt again + if not user_input.get("Accept Terms"): + errors["base"] = "Terms not accepted, please accept terms to continue" + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({ + vol.Required(CONF_ACCESS_TOKEN): str, + vol.Required("Accept Terms", default=False): bool # Custom label again for consistency + }), + errors=errors + ) + + # If terms are accepted, continue the flow + try: + self.access_token = user_input[CONF_ACCESS_TOKEN] + + # Initialize shared request tracking for config flow + if DOMAIN not in self.hass.data: + self.hass.data[DOMAIN] = {} + + api_token_key = f"api_tracking_{self.access_token[-8:]}" + if api_token_key not in self.hass.data[DOMAIN]: + from collections import deque + self.hass.data[DOMAIN][api_token_key] = { + "request_timestamps": deque(), + "total_requests": 0 + } + + self.api = YNABApi(self.access_token, self.hass.data[DOMAIN][api_token_key]) + + budgets_response = await self.api.get_budgets() + if not budgets_response or "budgets" not in budgets_response: + raise CannotConnect + + self.budgets = {b["id"]: b["name"] for b in budgets_response["budgets"]} + _LOGGER.debug("Available budgets: %s", self.budgets) + + return await self.async_step_budget_selection() + + except CannotConnect: + errors["base"] = "cannot_connect" + except InvalidAuth: + errors["base"] = "invalid_auth" + except Exception as e: + error_str = str(e) + if "429" in error_str or "rate limit" in error_str.lower(): + errors["base"] = "rate_limited" + else: + _LOGGER.exception("Unexpected exception") + errors["base"] = "unknown" + + return self.async_show_form(step_id="user", errors=errors) + + async def async_step_budget_selection(self, user_input: Dict[str, Any] | None = None) -> FlowResult: + """Handle budget selection.""" + errors = {} + + if user_input is not None: + selected_budget_id = user_input[CONF_SELECTED_BUDGET] + + if not selected_budget_id or selected_budget_id not in self.budgets: + _LOGGER.error("Invalid budget selected: %s", selected_budget_id) + return self.async_abort(reason="invalid_budget") + + self.budget_id = selected_budget_id + self.budget_name = self.budgets[self.budget_id] + + try: + # Fetch accounts and categories now that we know the budget + accounts_response = await self.api.get_accounts(self.budget_id) + categories_response = await self.api.get_categories(self.budget_id) + + # Store all accounts and categories for later filtering based on user preferences + self.all_accounts = {a["id"]: a["name"] for a in accounts_response.get("accounts", [])} + self.all_categories = { + c["id"]: c["name"] + for group in categories_response.get("category_groups", []) + for c in group.get("categories", []) + } + + # Store additional info for filtering + self.accounts_info = {a["id"]: a for a in accounts_response.get("accounts", [])} + self.categories_info = { + c["id"]: c + for group in categories_response.get("category_groups", []) + for c in group.get("categories", []) + } + + _LOGGER.debug("Fetched %d accounts and %d categories", len(self.all_accounts), len(self.all_categories)) + + return await self.async_step_config_page() + + except Exception as e: + error_str = str(e) + if "429" in error_str or "rate limit" in error_str.lower(): + errors["base"] = "rate_limited" + else: + _LOGGER.exception("Error fetching accounts/categories") + errors["base"] = "unknown" + + schema = vol.Schema({ + vol.Required(CONF_SELECTED_BUDGET): vol.In(self.budgets) + }) + + return self.async_show_form( + step_id="budget_selection", + data_schema=schema, + errors=errors + ) + + async def async_step_config_page(self, user_input: Dict[str, Any] | None = None) -> FlowResult: + """Prompt for instance name, currency, update interval, accounts, and categories on a single page.""" + errors = {} + + # Default values for the prompt fields + if user_input is None: + user_input = {} + + selected_accounts = user_input.get(CONF_SELECTED_ACCOUNTS, [SELECT_ALL_OPTION]) # Default to "Select All" + selected_categories = user_input.get(CONF_SELECTED_CATEGORIES, [SELECT_ALL_OPTION]) # Default to "Select All" + + # Ensure that if a currency was previously selected, it remains selected in UI + selected_currency = user_input.get(CONF_CURRENCY, getattr(self, "selected_currency", "USD")) + + update_interval = user_input.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) # Default to 10 minutes + + # Filter options - default to excluding closed accounts and hidden categories + include_closed_accounts = user_input.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + include_hidden_categories = user_input.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + + # Apply filtering based on user preferences + self.accounts = self._filter_accounts(include_closed_accounts) + self.categories = self._filter_categories(include_hidden_categories) + + if user_input: + self.instance_name = user_input.get("instance_name", self.budget_name) # Default to raw budget name + self.selected_currency = selected_currency + self.update_interval = update_interval + + # If "Select All" is selected, use the entire list of accounts/categories + if SELECT_ALL_OPTION in selected_accounts: + selected_accounts = list(self.accounts.keys()) + if SELECT_ALL_OPTION in selected_categories: + selected_categories = list(self.categories.keys()) + + self.selected_accounts = selected_accounts + self.selected_categories = selected_categories + + return await self.async_step_create_entry() + + # Add "Select All" as an option to the dropdown + account_options = {**self.accounts, SELECT_ALL_OPTION: "Select All Accounts"} + category_options = {**self.categories, SELECT_ALL_OPTION: "Select All Categories"} + + schema = vol.Schema({ + vol.Optional("instance_name", default=self.budget_name): str, # Default to raw budget name + vol.Required(CONF_CURRENCY, default=self.selected_currency if hasattr(self, "selected_currency") else "USD"): vol.In(CURRENCY_OPTIONS), # Default currency + vol.Required(CONF_UPDATE_INTERVAL, default=DEFAULT_UPDATE_INTERVAL): vol.In(POLLING_INTERVAL_OPTIONS), + vol.Optional(CONF_INCLUDE_CLOSED_ACCOUNTS, default=DEFAULT_INCLUDE_CLOSED_ACCOUNTS): bool, # Checkbox for including closed accounts + vol.Optional(CONF_INCLUDE_HIDDEN_CATEGORIES, default=DEFAULT_INCLUDE_HIDDEN_CATEGORIES): bool, # Checkbox for including hidden categories + vol.Required(CONF_SELECTED_ACCOUNTS, default=[SELECT_ALL_OPTION]): cv.multi_select(account_options), + vol.Required(CONF_SELECTED_CATEGORIES, default=[SELECT_ALL_OPTION]): cv.multi_select(category_options), + }) + + return self.async_show_form( + step_id="config_page", + data_schema=schema, + errors=errors + ) + + async def async_step_create_entry(self) -> FlowResult: + """Create the entry with the user data.""" + currency_to_store = self.selected_currency + _LOGGER.error(f"🔴 DEBUG: Storing selected currency in config entry: {currency_to_store}") + + return self.async_create_entry( + title=self.instance_name, # Use the raw instance name (display name) + data={ + CONF_ACCESS_TOKEN: self.access_token, + "budget_id": self.budget_id, + "budget_name": self.budget_name, # Store the raw budget name + CONF_CURRENCY: currency_to_store, # Store selected currency + CONF_SELECTED_ACCOUNTS: self.selected_accounts, + CONF_SELECTED_CATEGORIES: self.selected_categories, + "instance_name": self.instance_name, # Make sure instance_name is added here + }, + options={ # Store CONF_UPDATE_INTERVAL in options instead of data + CONF_UPDATE_INTERVAL: self.update_interval + } + ) + + def _filter_accounts(self, include_closed: bool) -> Dict[str, str]: + """Filter accounts based on closed status.""" + filtered = {} + + for account_id, name in self.all_accounts.items(): + is_closed = self.accounts_info.get(account_id, {}).get("closed", False) + + # Skip closed accounts if not including them + if is_closed and not include_closed: + continue + + # Add (Closed) prefix for closed accounts + display_name = f"(Closed) {name}" if is_closed else name + filtered[account_id] = display_name + + return filtered + + def _filter_categories(self, include_hidden: bool) -> Dict[str, str]: + """Filter categories based on hidden status.""" + filtered = {} + + for category_id, name in self.all_categories.items(): + is_hidden = self.categories_info.get(category_id, {}).get("hidden", False) + + # Skip hidden categories if not including them + if is_hidden and not include_hidden: + continue + + # Add (Hidden) prefix for hidden categories + display_name = f"(Hidden) {name}" if is_hidden else name + filtered[category_id] = display_name + + return filtered + + def _build_config_schema(self, defaults: Dict[str, Any]) -> vol.Schema: + """Build the configuration schema with provided defaults.""" + # Apply filtering based on user preferences + include_closed = defaults.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + include_hidden = defaults.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + + filtered_accounts = self._filter_accounts(include_closed) + filtered_categories = self._filter_categories(include_hidden) + + # Add "Select All" as an option to the dropdown + account_options = {**filtered_accounts, SELECT_ALL_OPTION: "Select All Accounts"} + category_options = {**filtered_categories, SELECT_ALL_OPTION: "Select All Categories"} + + return vol.Schema({ + vol.Optional("instance_name", default=defaults.get("instance_name", "")): str, + vol.Required(CONF_CURRENCY, default=defaults.get(CONF_CURRENCY, "USD")): vol.In(CURRENCY_OPTIONS), + vol.Required(CONF_UPDATE_INTERVAL, default=defaults.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL)): vol.In(POLLING_INTERVAL_OPTIONS), + vol.Optional(CONF_INCLUDE_CLOSED_ACCOUNTS, default=defaults.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS)): bool, + vol.Optional(CONF_INCLUDE_HIDDEN_CATEGORIES, default=defaults.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES)): bool, + vol.Required(CONF_SELECTED_ACCOUNTS, default=defaults.get(CONF_SELECTED_ACCOUNTS, [SELECT_ALL_OPTION])): cv.multi_select(account_options), + vol.Required(CONF_SELECTED_CATEGORIES, default=defaults.get(CONF_SELECTED_CATEGORIES, [SELECT_ALL_OPTION])): cv.multi_select(category_options), + }) + +class OptionsFlow(config_entries.OptionsFlow): + """Handle options flow for YNAB integration.""" + + def __init__(self, config_entry: config_entries.ConfigEntry) -> None: + """Initialize options flow.""" + super().__init__() + try: + # Store config entry data we need, not the entry itself + self.entry_data = config_entry.data + self.entry_options = config_entry.options + self.entry_id = config_entry.entry_id + self.access_token = config_entry.data[CONF_ACCESS_TOKEN] # Store token directly + self.budget_id = config_entry.data["budget_id"] + self.budget_name = config_entry.data["budget_name"] + + # We'll initialize the API with shared tracking in async_step_init + # since we need access to self.hass which isn't available yet + self.api = None + + _LOGGER.debug(f"OptionsFlow initialized for budget: {self.budget_name}") + except Exception as e: + _LOGGER.error(f"Error initializing OptionsFlow: {e}") + raise + + async def async_step_init(self, user_input: Dict[str, Any] | None = None) -> FlowResult: + """Handle the initial step of options flow.""" + errors = {} + _LOGGER.debug(f"OptionsFlow async_step_init called with user_input: {user_input is not None}") + + # Initialize API with shared tracking if not already done + if self.api is None: + # Initialize shared request tracking for options flow + if DOMAIN not in self.hass.data: + self.hass.data[DOMAIN] = {} + + api_token_key = f"api_tracking_{self.access_token[-8:]}" + if api_token_key not in self.hass.data[DOMAIN]: + from collections import deque + self.hass.data[DOMAIN][api_token_key] = { + "request_timestamps": deque(), + "total_requests": 0 + } + + self.api = YNABApi(self.access_token, self.hass.data[DOMAIN][api_token_key]) + _LOGGER.debug("OptionsFlow API initialized with shared tracking") + + if user_input is not None: + # Process the form submission + instance_name = user_input.get("instance_name", self.entry_data.get("instance_name", self.budget_name)) + selected_currency = user_input.get(CONF_CURRENCY) + update_interval = user_input.get(CONF_UPDATE_INTERVAL) + include_closed_accounts = user_input.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + include_hidden_categories = user_input.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + + _LOGGER.debug(f"Form submitted with include_closed: {include_closed_accounts}, include_hidden: {include_hidden_categories}") + selected_accounts = user_input.get(CONF_SELECTED_ACCOUNTS, []) + selected_categories = user_input.get(CONF_SELECTED_CATEGORIES, []) + + # Handle filtering changes and "Select All" logic + filtered_accounts = self._filter_accounts(include_closed_accounts) + filtered_categories = self._filter_categories(include_hidden_categories) + + if SELECT_ALL_OPTION in selected_accounts: + # Select all filtered accounts + selected_accounts = list(filtered_accounts.keys()) + else: + # Use the form selection, but add newly available items when checkboxes are enabled + current_filtered = set(filtered_accounts.keys()) + + # If include_closed was just enabled, add newly available closed accounts + old_include_closed = self.entry_options.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + if include_closed_accounts and not old_include_closed: + # Closed accounts were just enabled, add them to selection + newly_available = current_filtered - set(self._filter_accounts(False).keys()) + selected_accounts = list(set(selected_accounts) | newly_available) + _LOGGER.debug(f"Added newly available closed accounts: {newly_available}") + + # Filter out accounts that are no longer available + selected_accounts = [acc for acc in selected_accounts if acc in current_filtered] + + if SELECT_ALL_OPTION in selected_categories: + # Select all filtered categories + selected_categories = list(filtered_categories.keys()) + else: + # Use the form selection, but add newly available items when checkboxes are enabled + current_filtered = set(filtered_categories.keys()) + + # If include_hidden was just enabled, add newly available hidden categories + old_include_hidden = self.entry_options.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + if include_hidden_categories and not old_include_hidden: + # Hidden categories were just enabled, add them to selection + newly_available = current_filtered - set(self._filter_categories(False).keys()) + selected_categories = list(set(selected_categories) | newly_available) + _LOGGER.debug(f"Added newly available hidden categories: {newly_available}") + + # Filter out categories that are no longer available + selected_categories = [cat for cat in selected_categories if cat in current_filtered] + + # Update the config entry + new_data = dict(self.entry_data) + new_data.update({ + "instance_name": instance_name, + CONF_CURRENCY: selected_currency, + CONF_SELECTED_ACCOUNTS: selected_accounts, + CONF_SELECTED_CATEGORIES: selected_categories, + }) + + new_options = dict(self.entry_options) + new_options.update({ + CONF_UPDATE_INTERVAL: update_interval, + CONF_INCLUDE_CLOSED_ACCOUNTS: include_closed_accounts, + CONF_INCLUDE_HIDDEN_CATEGORIES: include_hidden_categories, + }) + + # Get config entry from hass + config_entry = self.hass.config_entries.async_get_entry(self.entry_id) + if config_entry: + self.hass.config_entries.async_update_entry( + config_entry, + data=new_data, + options=new_options, + title=instance_name, + ) + + return self.async_create_entry(title="", data={}) + + # Fetch current accounts and categories + try: + accounts_response = await self.api.get_accounts(self.budget_id) + categories_response = await self.api.get_categories(self.budget_id) + + self.all_accounts = {a["id"]: a["name"] for a in accounts_response.get("accounts", [])} + self.all_categories = { + c["id"]: c["name"] + for group in categories_response.get("category_groups", []) + for c in group.get("categories", []) + } + + # Store additional info for filtering + self.accounts_info = {a["id"]: a for a in accounts_response.get("accounts", [])} + self.categories_info = { + c["id"]: c + for group in categories_response.get("category_groups", []) + for c in group.get("categories", []) + } + + except Exception as e: + _LOGGER.error(f"Error fetching YNAB data for options: {e}") + errors["base"] = "cannot_connect" + + if not errors: + # Get current filter settings - need to check if they were explicitly set + # If not in options, check if we can infer from current selections + current_include_closed = self.entry_options.get(CONF_INCLUDE_CLOSED_ACCOUNTS) + current_include_hidden = self.entry_options.get(CONF_INCLUDE_HIDDEN_CATEGORIES) + + # If not explicitly set, try to infer from current account/category selections + if current_include_closed is None: + # Check if any currently selected accounts are closed + selected_accounts = self.entry_data.get(CONF_SELECTED_ACCOUNTS, []) + closed_account_ids = { + acc_id for acc_id, acc_info in self.accounts_info.items() + if acc_info.get("closed", False) + } + current_include_closed = bool(set(selected_accounts) & closed_account_ids) + _LOGGER.debug(f"Inferred include_closed: {current_include_closed} (selected: {len(selected_accounts)}, closed: {len(closed_account_ids)}, intersection: {len(set(selected_accounts) & closed_account_ids)})") + + if current_include_hidden is None: + # Check if any currently selected categories are hidden + selected_categories = self.entry_data.get(CONF_SELECTED_CATEGORIES, []) + hidden_category_ids = { + cat_id for cat_id, cat_info in self.categories_info.items() + if cat_info.get("hidden", False) + } + current_include_hidden = bool(set(selected_categories) & hidden_category_ids) + _LOGGER.debug(f"Inferred include_hidden: {current_include_hidden} (selected: {len(selected_categories)}, hidden: {len(hidden_category_ids)}, intersection: {len(set(selected_categories) & hidden_category_ids)})") + + # Filter current accounts/categories based on current filter settings + filtered_accounts = self._filter_accounts(current_include_closed) + filtered_categories = self._filter_categories(current_include_hidden) + + # Get all currently selected accounts/categories + all_current_accounts = self.entry_data.get(CONF_SELECTED_ACCOUNTS, []) + all_current_categories = self.entry_data.get(CONF_SELECTED_CATEGORIES, []) + + # Filter current selections to only include available ones + current_selected_accounts = [ + acc_id for acc_id in all_current_accounts + if acc_id in filtered_accounts + ] + current_selected_categories = [ + cat_id for cat_id in all_current_categories + if cat_id in filtered_categories + ] + + # Check if all available items are selected (show "Select All") + if set(current_selected_accounts) == set(filtered_accounts.keys()) and current_selected_accounts: + current_selected_accounts = [SELECT_ALL_OPTION] + # Don't default to "Select All" if no items are selected - let user choose + + if set(current_selected_categories) == set(filtered_categories.keys()) and current_selected_categories: + current_selected_categories = [SELECT_ALL_OPTION] + # Don't default to "Select All" if no items are selected - let user choose + + # Prepare current values for the form + current_values = { + "instance_name": self.entry_data.get("instance_name", self.budget_name), + CONF_CURRENCY: self.entry_data.get(CONF_CURRENCY, "USD"), + CONF_UPDATE_INTERVAL: self.entry_options.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL), + CONF_INCLUDE_CLOSED_ACCOUNTS: current_include_closed, + CONF_INCLUDE_HIDDEN_CATEGORIES: current_include_hidden, + CONF_SELECTED_ACCOUNTS: current_selected_accounts, + CONF_SELECTED_CATEGORIES: current_selected_categories, + } + + schema = self._build_config_schema(current_values) + + return self.async_show_form( + step_id="init", + data_schema=schema, + errors=errors, + ) + + return self.async_show_form( + step_id="init", + data_schema=vol.Schema({}), + errors=errors, + ) + + def _filter_accounts(self, include_closed: bool) -> Dict[str, str]: + """Filter accounts based on closed status.""" + filtered = {} + + for account_id, name in self.all_accounts.items(): + is_closed = self.accounts_info.get(account_id, {}).get("closed", False) + + # Skip closed accounts if not including them + if is_closed and not include_closed: + continue + + # Add (Closed) prefix for closed accounts + display_name = f"(Closed) {name}" if is_closed else name + filtered[account_id] = display_name + + return filtered + + def _filter_categories(self, include_hidden: bool) -> Dict[str, str]: + """Filter categories based on hidden status.""" + filtered = {} + + for category_id, name in self.all_categories.items(): + is_hidden = self.categories_info.get(category_id, {}).get("hidden", False) + + # Skip hidden categories if not including them + if is_hidden and not include_hidden: + continue + + # Add (Hidden) prefix for hidden categories + display_name = f"(Hidden) {name}" if is_hidden else name + filtered[category_id] = display_name + + return filtered + + def _build_config_schema(self, defaults: Dict[str, Any]) -> vol.Schema: + """Build the configuration schema with provided defaults.""" + # Apply filtering based on user preferences + include_closed = defaults.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS) + include_hidden = defaults.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES) + + filtered_accounts = self._filter_accounts(include_closed) + filtered_categories = self._filter_categories(include_hidden) + + # Add "Select All" as an option to the dropdown + account_options = {**filtered_accounts, SELECT_ALL_OPTION: "Select All Accounts"} + category_options = {**filtered_categories, SELECT_ALL_OPTION: "Select All Categories"} + + return vol.Schema({ + vol.Optional("instance_name", default=defaults.get("instance_name", "")): str, + vol.Required(CONF_CURRENCY, default=defaults.get(CONF_CURRENCY, "USD")): vol.In(CURRENCY_OPTIONS), + vol.Required(CONF_UPDATE_INTERVAL, default=defaults.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL)): vol.In(POLLING_INTERVAL_OPTIONS), + vol.Optional(CONF_INCLUDE_CLOSED_ACCOUNTS, default=defaults.get(CONF_INCLUDE_CLOSED_ACCOUNTS, DEFAULT_INCLUDE_CLOSED_ACCOUNTS)): bool, + vol.Optional(CONF_INCLUDE_HIDDEN_CATEGORIES, default=defaults.get(CONF_INCLUDE_HIDDEN_CATEGORIES, DEFAULT_INCLUDE_HIDDEN_CATEGORIES)): bool, + vol.Required(CONF_SELECTED_ACCOUNTS, default=defaults.get(CONF_SELECTED_ACCOUNTS, [SELECT_ALL_OPTION])): cv.multi_select(account_options), + vol.Required(CONF_SELECTED_CATEGORIES, default=defaults.get(CONF_SELECTED_CATEGORIES, [SELECT_ALL_OPTION])): cv.multi_select(category_options), + }) + + +# Define CannotConnect exception +class CannotConnect(exceptions.HomeAssistantError): + """Error to indicate we cannot connect.""" + +# Define InvalidAuth exception +class InvalidAuth(exceptions.HomeAssistantError): + """Error to indicate invalid authentication.""" diff --git a/const.py b/const.py new file mode 100644 index 0000000..36dbca4 --- /dev/null +++ b/const.py @@ -0,0 +1,33 @@ +"""Constants for YNAB Custom integration.""" + +DOMAIN = "ynab_custom" + +CONF_SELECTED_ACCOUNTS = "Select Accounts to Import" +CONF_SELECTED_CATEGORIES = "Select Categories to Import" +CONF_CURRENCY = "Select Currency" +CONF_SELECTED_BUDGET = "Select Budget to Import" +CONF_UPDATE_INTERVAL = "Update Interval" +CONF_BUDGET_NAME = "budget_name" +CONF_INCLUDE_CLOSED_ACCOUNTS = "Include Closed Accounts" +CONF_INCLUDE_HIDDEN_CATEGORIES = "Include Hidden Categories" +DEFAULT_UPDATE_INTERVAL = 10 # Default to 10 minutes +DEFAULT_INCLUDE_CLOSED_ACCOUNTS = False # Default to excluding closed accounts +DEFAULT_INCLUDE_HIDDEN_CATEGORIES = False # Default to excluding hidden categories + + +PLATFORMS = ["sensor"] + +def get_currency_symbol(currency_code): + """Convert currency code to symbol.""" + currency_map = { + "USD": "$", + "EUR": "€", + "GBP": "£", + "AUD": "A$", + "CAD": "C$", + "JPY": "¥", + "CHF": "CHF", + "SEK": "kr", + "NZD": "NZ$", + } + return currency_map.get(currency_code, "$") # Default to USD if not found diff --git a/coordinator.py b/coordinator.py new file mode 100644 index 0000000..b7e3b48 --- /dev/null +++ b/coordinator.py @@ -0,0 +1,471 @@ +"""YNAB Data Update Coordinator.""" + +import logging +from datetime import datetime, timedelta +from collections import deque + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator +from homeassistant.helpers.storage import Store +from homeassistant.config_entries import ConfigEntry + +from .api import YNABApi +from .const import ( + DOMAIN, + CONF_SELECTED_ACCOUNTS, + CONF_SELECTED_CATEGORIES, + CONF_CURRENCY, + CONF_UPDATE_INTERVAL, + DEFAULT_UPDATE_INTERVAL, + get_currency_symbol, +) + +_LOGGER = logging.getLogger(__name__) + + +class YNABDataUpdateCoordinator(DataUpdateCoordinator): + """Class to manage fetching YNAB data from API + user values.""" + + def __init__( + self, + hass: HomeAssistant, + entry: ConfigEntry, + budget_id: str, + budget_name: str, + due_days: dict[str, int] | None = None, + ) -> None: + """Initialize the coordinator.""" + self.hass = hass + self.entry = entry + self.budget_id = budget_id + self.budget_name = budget_name + + # ================================ + # 🔵 SHARED API REQUEST TRACKING + # ================================ + if DOMAIN not in hass.data: + hass.data[DOMAIN] = {} + + api_token_key = f"api_tracking_{entry.data['access_token'][-8:]}" + if api_token_key not in hass.data[DOMAIN]: + hass.data[DOMAIN][api_token_key] = { + "request_timestamps": deque(), + "total_requests": 0, + } + + self.api = YNABApi( + entry.data["access_token"], + hass.data[DOMAIN][api_token_key], + ) + self.selected_accounts = entry.data.get(CONF_SELECTED_ACCOUNTS, []) + self.selected_categories = entry.data.get(CONF_SELECTED_CATEGORIES, []) + + # ================================ + # 🔵 API STATUS STRUCTURE + # ================================ + self.api_status: dict[str, object] = { + "status": "Unknown", + "last_error": "None", + "last_error_time": "Never", + "consecutive_failures": 0, + "last_successful_request": "Never", + "requests_made_total": 0, + "requests_this_hour": 0, + "estimated_remaining": 200, + "rate_limit_resets_at": "Unknown", + "is_at_limit": False, + } + + # Keep your persistent key for API data + self.persistent_data_key = f"ynab_data_{entry.entry_id}" + + # ================================ + # 🔵 USER-EDITABLE VALUE STORAGE + # ================================ + # Separate store JUST for credit limits / APRs so config_entry + # changes don’t reload the whole integration. + self._user_store = Store( + hass, + version=1, + key=f"{DOMAIN}_userdata_{entry.entry_id}", + ) + + # In-memory copies used by entities + self.credit_limits: dict[str, float] = {} + self.aprs: dict[str, float] = {} + self.due_days: dict[str, int] = {} + + # ================================ + # 🔵 CURRENCY + UPDATE INTERVAL + # ================================ + update_interval = ( + self.entry.options.get(CONF_UPDATE_INTERVAL) + or self.entry.data.get(CONF_UPDATE_INTERVAL) + or DEFAULT_UPDATE_INTERVAL + ) + + self.currency_symbol = get_currency_symbol( + self.entry.data.get(CONF_CURRENCY, "USD") + ) + + super().__init__( + hass, + _LOGGER, + name=f"YNAB Coordinator - {budget_name}", + update_interval=timedelta(minutes=update_interval), + ) + + # ================================================================ + # 🔥 STARTUP LOADING: USER VALUES + API CACHE + # ================================================================ + async def async_load_persistent_data(self) -> None: + """Load user values and cached API data at startup.""" + await self._load_user_values() + await self._load_persistent_data() + + # ---------------------------------------------------------------- + # USER VALUES: CREDIT LIMITS / APRS + # ---------------------------------------------------------------- + async def _load_user_values(self) -> None: + """Load credit limits, APRs, and due days from HA storage (with migration).""" + stored = await self._user_store.async_load() or {} + + self.credit_limits = { + k: float(v) for k, v in stored.get("credit_limits", {}).items() + } + self.aprs = { + k: float(v) for k, v in stored.get("aprs", {}).items() + } + self.due_days = { + k: int(v) for k, v in stored.get("due_days", {}).items() + } + + # Migration from old config-entry options + migrated = False + opts = self.entry.options + + if not self.credit_limits and "credit_limits" in opts: + self.credit_limits = { + k: float(v) for k, v in opts["credit_limits"].items() + } + migrated = True + + if not self.aprs and "aprs" in opts: + self.aprs = { + k: float(v) for k, v in opts["aprs"].items() + } + migrated = True + + if not self.due_days and "due_days" in opts: + self.due_days = { + k: int(v) for k, v in opts["due_days"].items() + } + migrated = True + + if migrated: + new_opts = dict(opts) + new_opts.pop("credit_limits", None) + new_opts.pop("aprs", None) + new_opts.pop("due_days", None) + + self.hass.config_entries.async_update_entry( + self.entry, + options=new_opts, + ) + + _LOGGER.warning( + "YNAB: migrated credit_limits/aprs/due_days from config entry → storage" + ) + + # Ensure store is written (covers first-run case) + await self._user_store.async_save( + { + "credit_limits": self.credit_limits, + "aprs": self.aprs, + "due_days": self.due_days, + } + ) + + async def async_save_user_values(self) -> None: + """Persist user-editable values to storage.""" + await self._user_store.async_save( + { + "credit_limits": self.credit_limits, + "aprs": self.aprs, + "due_days": self.due_days, + } + ) + _LOGGER.debug("Saved YNAB user limits/APRs/due_days to HA storage") + + # Convenient helpers for the Number entities + def _get_credit_card_accounts(self): + accounts = self.coordinator.data.get("accounts", []) + return [a for a in accounts if a.get("type") == "creditCard"] + + def _notify_dependents(self) -> None: + self.async_set_updated_data(self.data) + + def get_credit_limit(self, account_id: str) -> float: + return float(self.credit_limits.get(account_id, 0.0)) + + def get_apr(self, account_id: str) -> float: + return float(self.aprs.get(account_id, 0.0)) + + def get_due_day(self, account_id: str) -> int | None: + return self.due_days.get(account_id) + + async def async_set_credit_limit(self, account_id: str, value: float) -> None: + self.credit_limits[account_id] = float(value) + await self.async_save_user_values() + self._notify_dependents() + + async def async_set_due_day(self, account_id: str, value: int) -> None: + self.due_days[account_id] = int(value) + await self.async_save_user_values() + self._notify_dependents() + + async def async_set_apr(self, account_id: str, value: float) -> None: + self.aprs[account_id] = float(value) + await self.async_save_user_values() + self._notify_dependents() + + # ================================================================ + # 🔵 API DATA PERSISTENCE (UNCHANGED CONCEPTUALLY) + # ================================================================ + async def _load_persistent_data(self) -> None: + """Load cached API data from HA storage.""" + try: + store = Store( + self.hass, + version=1, + key=f"ynab_data_{self.entry.entry_id}", + private=True, + ) + persistent_data = await store.async_load() + if persistent_data: + self.data = persistent_data + if "api_status" in persistent_data: + self.api_status.update(persistent_data["api_status"]) + except Exception as exc: # pylint: disable=broad-except + _LOGGER.warning( + "Failed to load persistent API data for %s: %s", + self.budget_name, + exc, + ) + + def _save_persistent_data(self, data: dict) -> None: + """Store API data snapshot (budget summary, categories, etc.).""" + try: + store = Store( + self.hass, + version=1, + key=f"ynab_data_{self.entry.entry_id}", + private=True, + ) + self.hass.async_create_task(store.async_save(data)) + _LOGGER.debug( + "Saved persistent YNAB API dataset for %s", + self.budget_name, + ) + except Exception as exc: # pylint: disable=broad-except + _LOGGER.warning( + "Failed to save persistent API data for %s: %s", + self.budget_name, + exc, + ) + + # ================================================================ + # 🔵 ORIGINAL UPDATE LOGIC + # ================================================================ + @staticmethod + def get_current_month() -> str: + """Return current month in YYYY-MM-01 format.""" + return datetime.now().strftime("%Y-%m-01") + + async def _async_update_data(self) -> dict: + """Fetch budget details from the API.""" + try: + _LOGGER.debug("Fetching latest YNAB data...") + + if self.api_status["consecutive_failures"] > 0: + _LOGGER.debug("Attempting API calls after previous failures...") + + current_month = self.get_current_month() + _LOGGER.debug( + "Fetching data for budget_id: %s and month: %s", + self.budget_id, + current_month, + ) + + budget_data = await self.api.get_budget(self.budget_id) + accounts = await self.api.get_accounts(self.budget_id) + categories = await self.api.get_categories(self.budget_id) + monthly_summary = await self.api.get_monthly_summary( + self.budget_id, + current_month, + ) + transactions = await self.api.get_transactions(self.budget_id) + + rate_limit_info = self.api.get_rate_limit_info() + + self.api_status.update( + { + "status": "Connected", + "last_error": "None", + "consecutive_failures": 0, + } + ) + self.api_status.update(rate_limit_info) + self.api_status["is_at_limit"] = ( + self.api_status["status"] == "Rate Limited" + ) + + last_successful_poll = datetime.now().strftime( + "%B %d, %Y - %I:%M %p" + ) + self.api_status["last_successful_request"] = last_successful_poll + + budget_data["accounts"] = [ + a + for a in accounts.get("accounts", []) + if a["id"] in self.selected_accounts + ] + _LOGGER.debug("Filtered accounts: %s", budget_data["accounts"]) + for account in budget_data["accounts"]: + account_id = account["id"] + + if account_id in self.credit_limits: + account["credit_limit"] = self.credit_limits[account_id] + + if account_id in self.aprs: + account["apr"] = self.aprs[account_id] + + budget_data["categories"] = [ + c + for c_group in categories.get("category_groups", []) + for c in c_group.get("categories", []) + if c["id"] in self.selected_categories + ] + + budget_data["monthly_summary"] = monthly_summary + budget_data["transactions"] = transactions.get("transactions", []) + budget_data["last_successful_poll"] = last_successful_poll + budget_data["api_status"] = self.api_status.copy() + + all_transactions = budget_data["transactions"] + unapproved_transactions = len( + [t for t in all_transactions if not t.get("approved", True)] + ) + + selected_active_account_ids = { + a["id"] + for a in accounts.get("accounts", []) + if not a.get("closed", False) + and not a.get("deleted", False) + and a["id"] + in [acc["id"] for acc in budget_data["accounts"]] + } + + uncleared_transactions = len( + [ + t + for t in all_transactions + if t.get("cleared") == "uncleared" + and t.get("account_id") in selected_active_account_ids + and not t.get("scheduled_transaction_id") + ] + ) + + overspent_categories = len( + [ + c + for c in monthly_summary.get("month", {}).get( + "categories", [] + ) + if c.get("balance", 0) < 0 + ] + ) + + needs_attention_count = sum( + [ + unapproved_transactions > 0, + uncleared_transactions > 0, + overspent_categories > 0, + ] + ) + + budget_data["unapproved_transactions"] = unapproved_transactions + budget_data["uncleared_transactions"] = uncleared_transactions + budget_data["overspent_categories"] = overspent_categories + budget_data["needs_attention_count"] = needs_attention_count + + self._save_persistent_data(budget_data) + return budget_data + + except Exception as exc: # pylint: disable=broad-except + error_time = datetime.now().strftime("%B %d, %Y - %I:%M %p") + self.api_status["consecutive_failures"] += 1 + self.api_status["last_error_time"] = error_time + + rate_limit_info = self.api.get_rate_limit_info() + error_str = str(exc) + + if "429" in error_str or "rate limit" in error_str.lower(): + self.api_status["status"] = "Rate Limited" + self.api_status["last_error"] = "429 - Too Many Requests" + _LOGGER.warning( + "YNAB API rate limited. Consecutive failures: %s", + self.api_status["consecutive_failures"], + ) + elif "401" in error_str or "unauthorized" in error_str.lower(): + self.api_status["status"] = "Unauthorized" + self.api_status["last_error"] = "401 - Invalid API Token" + elif "503" in error_str or "service unavailable" in error_str.lower(): + self.api_status["status"] = "Service Unavailable" + self.api_status["last_error"] = "503 - YNAB Service Down" + else: + self.api_status["status"] = "API Error" + self.api_status["last_error"] = f"Error: {error_str[:100]}" + + self.api_status.update(rate_limit_info) + self.api_status["is_at_limit"] = ( + self.api_status["status"] == "Rate Limited" + ) + + _LOGGER.error("Error fetching YNAB data: %s", exc) + + if not getattr(self, "data", None): + await self._load_persistent_data() + + if getattr(self, "data", None): + updated_data = self.data.copy() + updated_data["api_status"] = self.api_status.copy() + if "last_successful_poll" in updated_data: + updated_data["api_status"][ + "last_successful_request" + ] = updated_data["last_successful_poll"] + _LOGGER.info( + "Rate limited/API error – preserving existing sensor data" + ) + return updated_data + + _LOGGER.warning( + "No previous data available during API error – sensors may be empty" + ) + return { + "api_status": self.api_status.copy(), + "accounts": [], + "categories": [], + "transactions": [], + "monthly_summary": {}, + "unapproved_transactions": 0, + "uncleared_transactions": 0, + "overspent_categories": 0, + "needs_attention_count": 0, + "last_successful_poll": "Never", + } + + async def manual_refresh(self, call) -> None: + """Manually refresh YNAB data when the service is called.""" + _LOGGER.info("Manual refresh triggered for YNAB") + await self.async_refresh() diff --git a/icons.py b/icons.py new file mode 100644 index 0000000..9c159d7 --- /dev/null +++ b/icons.py @@ -0,0 +1,127 @@ +# icons.py + +CATEGORY_ICONS = { + "groceries": "mdi:food", + "rent": "mdi:home", + "mortgage": "mdi:home", + "utilities": "mdi:lightning-bolt", + "electricity": "mdi:lightning-bolt", + "gas": "mdi:gas-station", + "fuel": "mdi:gas-station", + "transport": "mdi:car", + "car": "mdi:car", + "public transport": "mdi:bus", + "entertainment": "mdi:television", + "fun": "mdi:television", + "savings": "mdi:piggy-bank", + "investment": "mdi:piggy-bank", + "insurance": "mdi:shield-check", + "phone": "mdi:cellphone", + "internet": "mdi:cellphone", + "medical": "mdi:heart-pulse", + "health": "mdi:heart-pulse", + "subscriptions": "mdi:credit-card-multiple", + "streaming": "mdi:credit-card-multiple", + "education": "mdi:book-open-variant", + "school": "mdi:book-open-variant", + "travel": "mdi:airplane", + "vacation": "mdi:airplane", + "dining": "mdi:silverware-fork-knife", + "restaurant": "mdi:silverware-fork-knife", + "gifts": "mdi:gift", + "donations": "mdi:gift", + "clothing": "mdi:tshirt-crew", + "apparel": "mdi:tshirt-crew", + "electronics": "mdi:laptop", + "pets": "mdi:paw", + "home improvement": "mdi:tools", + "beauty": "mdi:lipstick", + "sports": "mdi:soccer", + "books": "mdi:book", + "transportation": "mdi:car", + "childcare": "mdi:baby-bottle", + "vacations": "mdi:airplane", + "furniture": "mdi:sofa", + "toys": "mdi:toys", + "alcohol": "mdi:beer", + "fitness": "mdi:run", + "charity": "mdi:heart", + "grocery": "mdi:cart", + "bills": "mdi:cash-multiple", + "personal care": "mdi:shower", + "photography": "mdi:camera", + "music": "mdi:music", + "shopping": "mdi:shopping", + "household": "mdi:home-modern", + "tech": "mdi:smartphone", + "diy": "mdi:hammer-wrench", + "garden": "mdi:flower", + "taxes": "mdi:calculator", + "loan": "mdi:loan", + "membership": "mdi:account-group", + "electronics": "mdi:television", + "fashion": "mdi:tshirt-crew", + "office": "mdi:desktop-tower-monitor", + "food": "mdi:food-apple", + "transportation": "mdi:train", + "outdoor": "mdi:snowflake", + "party": "mdi:balloon", + "hobbies": "mdi:palette", + "gifts": "mdi:gift", + "funds": "mdi:coin", + "beauty": "mdi:makeup", + "pets": "mdi:dog", + "gift shop": "mdi:gift", + "dental": "mdi:tooth", + "banking": "mdi:bank", + "shopping": "mdi:shopping-outline", + "television": "mdi:television-box", + "decor": "mdi:sofa", + "skincare": "mdi:heart-pulse", + "construction": "mdi:home-city", + "bookshelf": "mdi:book-open", + "traveling": "mdi:plane", + "home repair": "mdi:hammer-wrench", + "vacation": "mdi:airplane", + "home office": "mdi:office-building", + "pets supplies": "mdi:paw", + "medical expenses": "mdi:heart-pulse", + "mobile phone": "mdi:cellphone", + "electricity bill": "mdi:lightning-bolt", + "cable": "mdi:cable", + "personal loan": "mdi:loan", + "savings account": "mdi:bank", + "student loan": "mdi:school", + "loan payments": "mdi:loan", + "gym membership": "mdi:run", + "insurance payment": "mdi:shield", + "credit card payment": "mdi:credit-card-check", + "internet bill": "mdi:cellphone", + "house maintenance": "mdi:tools", + "cleaning supplies": "mdi:brush", + "grocery shopping": "mdi:cart", + "vehicle expenses": "mdi:car", + "subscriptions service": "mdi:credit-card-multiple", + "property tax": "mdi:calculator", + "home mortgage": "mdi:home", + "child support": "mdi:baby-bottle", + "education fees": "mdi:book-open-variant", + "study materials": "mdi:book" +} + +ACCOUNT_ICONS = { + "checking": "mdi:bank", + "savings": "mdi:cash-multiple", + "cash": "mdi:currency-usd", + "creditcard": "mdi:credit-card", + "lineofcredit": "mdi:credit-card-plus", + "otherasset": "mdi:home", + "otherliability": "mdi:shield-alert", + "mortgage": "mdi:home-lock", + "autoloan": "mdi:car", + "studentloan": "mdi:account-school", + "personalloan": "mdi:handshake", + "medicaldebt": "mdi:heart-pulse", + "otherdebt": "mdi:currency-usd-off", + "default": "mdi:wallet", +} \ No newline at end of file diff --git a/manifest.json b/manifest.json new file mode 100644 index 0000000..3e1edc7 --- /dev/null +++ b/manifest.json @@ -0,0 +1,14 @@ +{ + "domain": "ynab_custom", + "name": "YNAB Integration for Home Assistant", + + "codeowners": ["@DeLuca21"], + "config_flow": true, + "dependencies": [], + "documentation": "https://github.com/DeLuca21/ynab-ha", + "iot_class": "cloud_polling", + "issue_tracker": "https://github.com/DeLuca21/ynab-ha/issues", + "requirements": ["aiohttp"], + "platforms": ["sensor", "number"], + "version": "1.5.1" +} diff --git a/number.py b/number.py new file mode 100644 index 0000000..ac09a8e --- /dev/null +++ b/number.py @@ -0,0 +1,164 @@ +import logging +from homeassistant.components.number import NumberEntity +from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.const import EntityCategory +from .coordinator import YNABDataUpdateCoordinator +from .const import DOMAIN +_LOGGER = logging.getLogger(__name__) + +class YNABCreditLimitNumber(CoordinatorEntity, NumberEntity): + """Editable credit limit for credit card accounts.""" + + _attr_icon = "mdi:account-credit-card" + _attr_mode = "box" + + # Sensible defaults + _attr_native_min_value = 0 + _attr_native_max_value = 20000 + _attr_native_step = 1 + + + def __init__(self, coordinator, account, entry): + super().__init__(coordinator) + + self.coordinator = coordinator + self.account = account + self.entry = entry + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + self._attr_name = f"{account['name']} Credit Limit" + self._attr_unique_id = f"{budget_id}_{account_id}_credit_limit" + self._attr_entity_category = EntityCategory.CONFIG + self._attr_native_unit_of_measurement = self.coordinator.hass.config.currency + + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": account["name"], + "manufacturer": "YNAB", + "model": "Credit Card", + } + + @property + def native_value(self): + return self.coordinator.get_credit_limit(self.account["id"]) + + async def async_set_native_value(self, value: float) -> None: + account_id = self.account["id"] + + # 1️⃣ Update in-memory account object (used by sensors) + self.account["credit_limit"] = value + + # 2️⃣ Persist via coordinator Store + await self.coordinator.async_set_credit_limit(account_id, value) + + # 3️⃣ Tell HA the entity state changed + self.async_write_ha_state() + +class YNABDueDayNumber(CoordinatorEntity, NumberEntity): + """Monthly credit card payment due day (1–31).""" + + _attr_has_entity_name = True + _attr_entity_category = EntityCategory.CONFIG + + _attr_native_min_value = 1 + _attr_native_max_value = 28 + _attr_native_step = 1 + _attr_mode = "box" + + _attr_icon = "mdi:calendar-month" + + def __init__(self, coordinator, account: dict, entry): + super().__init__(coordinator) + + self.coordinator = coordinator + self.account = account + self.entry = entry + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + self._attr_name = f"{account['name']} Due Day" + self._attr_unique_id = f"{budget_id}_{account_id}_due_day" + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": account["name"], + "manufacturer": "YNAB", + "model": "Credit Card", + } + + @property + def native_value(self) -> int | None: + return self.coordinator.get_due_day(self.account["id"]) + + async def async_set_native_value(self, value: float) -> None: + await self.coordinator.async_set_due_day(self.account["id"],int(value)) + self.async_write_ha_state() + +class YNABAPRNumber(CoordinatorEntity, NumberEntity): + """APR number entity for credit card accounts.""" + + _attr_native_unit_of_measurement = "%" + _attr_icon = "mdi:percent" + _attr_mode = "box" + + # Sensible defaults + _attr_native_min_value = 0 + _attr_native_max_value = 40 + _attr_native_step = 0.01 + + def __init__(self, coordinator, account, entry): + super().__init__(coordinator) + + self.coordinator = coordinator + self.account = account + self.entry = entry + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + self._attr_name = f"{account['name']} APR" + self._attr_unique_id = f"{budget_id}_{account_id}_apr" + self._attr_entity_category = EntityCategory.CONFIG + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": account["name"], + "manufacturer": "YNAB", + "model": "Credit Card", + } + + @property + def native_value(self): + return self.coordinator.get_apr(self.account["id"]) + + async def async_set_native_value(self, value: float) -> None: + account_id = self.account["id"] + + self.account["apr"] = value + await self.coordinator.async_set_apr(account_id, value) + self.async_write_ha_state() + +async def async_setup_entry(hass, entry, async_add_entities): + coordinator = hass.data[DOMAIN][entry.entry_id] + + entities = [] + + accounts = coordinator.data.get("accounts", []) + for account in accounts: + account_type = account.get("type") + + # Credit Limit → credit cards and line of credit + if account_type in ("creditCard", "lineOfCredit"): + entities.append(YNABCreditLimitNumber(coordinator, account, entry)) + + + # APR → credit cards AND personal loans + if account_type in ("creditCard", "personalLoan"): + entities.append(YNABAPRNumber(coordinator, account, entry)) + entities.append(YNABDueDayNumber(coordinator, account, entry)) + + async_add_entities(entities) diff --git a/options_flow.py b/options_flow.py new file mode 100644 index 0000000..05db7d3 --- /dev/null +++ b/options_flow.py @@ -0,0 +1,64 @@ +"""Options flow for YNAB Custom integration.""" + +import logging +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.core import HomeAssistant +from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig, SelectSelectorMode + +from .const import DOMAIN, CONF_CURRENCY + +_LOGGER = logging.getLogger(__name__) + +# Predefined options for the update interval (in minutes) +POLLING_INTERVAL_OPTIONS = {i: f"{i} minute{'s' if i > 1 else ''}" for i in range(5, 61)} + +class YNABOptionsFlowHandler(config_entries.OptionsFlow): + """Handles the options flow for YNAB Custom integration.""" + + def __init__(self, config_entry): + """Initialize the options flow.""" + self.config_entry = config_entry + + async def async_step_init(self, user_input=None): + """Manage the integration options.""" + hass: HomeAssistant = self.hass + coordinator = hass.data[DOMAIN].get(self.config_entry.entry_id) + + if not coordinator: + _LOGGER.error("Coordinator not found. Cannot load options.") + return self.async_abort(reason="unknown_error") + + # Fetch available accounts and categories + account_options = {acc["id"]: acc["name"] for acc in coordinator.accounts} + category_options = {cat["id"]: cat["name"] for cat in coordinator.categories} + + # Get user-configured options (default values if not set) + current_accounts = self.config_entry.options.get("selected_accounts", list(account_options.keys())) + current_categories = self.config_entry.options.get("selected_categories", list(category_options.keys())) + current_interval = self.config_entry.options.get("update_interval", 5) + current_currency = self.config_entry.options.get(CONF_CURRENCY, "USD") + + # Supported currency options + currency_options = { + "USD": "$ (US Dollar)", + "EUR": "€ (Euro)", + "GBP": "£ (British Pound)", + "AUD": "A$ (Australian Dollar)", + "CAD": "C$ (Canadian Dollar)", + "JPY": "¥ (Japanese Yen)", + "CHF": "CHF (Swiss Franc)", + "SEK": "kr (Swedish Krona)", + "NZD": "NZ$ (New Zealand Dollar)", + } + + # Allow the user to change the update interval via a dropdown + schema = vol.Schema({ + vol.Optional("selected_accounts", default=current_accounts): vol.In(account_options), + vol.Optional("selected_categories", default=current_categories): vol.In(category_options), + vol.Optional("update_interval", default=current_interval): vol.In(POLLING_INTERVAL_OPTIONS), # Dropdown for interval + vol.Optional(CONF_CURRENCY, default=current_currency): vol.In(currency_options), # Currency selection + }) + + return self.async_show_form(step_id="init", data_schema=schema) diff --git a/sensor.py b/sensor.py new file mode 100644 index 0000000..e22a644 --- /dev/null +++ b/sensor.py @@ -0,0 +1,669 @@ +import logging +import re +from datetime import datetime +from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.components.sensor import SensorEntity, SensorDeviceClass, SensorStateClass +from homeassistant.components.number import NumberEntity +from homeassistant.const import CONF_CURRENCY +from homeassistant.core import callback +from .coordinator import YNABDataUpdateCoordinator +from .const import DOMAIN +from .icons import CATEGORY_ICONS, ACCOUNT_ICONS +from homeassistant.helpers.entity import EntityCategory +from .const import DOMAIN + + + +_LOGGER = logging.getLogger(__name__) + +def sanitize_budget_name(budget_name: str) -> str: + """Sanitize the budget name to create a valid Home Assistant entity ID.""" + # Sanitize for the entity ID (replace spaces with underscores and remove special characters) + sanitized_name = re.sub(r'[^a-zA-Z0-9_]', '', budget_name.replace(" ", "_")) + return sanitized_name + +def get_currency_symbol(currency_code): + """Convert currency code to symbol.""" + currency_map = { + "USD": "$", + "EUR": "€", + "GBP": "£", + "AUD": "A$", + "CAD": "C$", + "JPY": "¥", + "CHF": "CHF", + "SEK": "kr", + "NZD": "NZ$", + } + return currency_map.get(currency_code, "$") + +async def async_setup_entry(hass, entry, async_add_entities): + """Set up YNAB sensors.""" + coordinator: YNABDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + _LOGGER.debug("🔹 Setting up YNAB sensors...") + + # Fetch the currency symbol directly from coordinator + currency_symbol = coordinator.currency_symbol + + entities = [] + raw_budget_name = entry.data["budget_name"] + sanitized_budget_name = sanitize_budget_name(raw_budget_name) + + # Create monthly summary sensor - it will use data from coordinator.data + # No need to fetch data here, the coordinator handles all API calls + entities.append(YNABExtrasSensor(coordinator, currency_symbol, raw_budget_name)) + + # Ensure diagnostics sensors are always added + entities.append(YNABAPIStatusSensor(coordinator, raw_budget_name)) + entities.append(YNABTotalCreditLimitSensor(coordinator, currency_symbol, raw_budget_name)) + entities.append(YNABTotalAvailableCreditSensor(coordinator, currency_symbol, raw_budget_name)) + entities.append(YNABTotalCreditUtilizationSensor(coordinator, raw_budget_name)) + + + _LOGGER.debug(f"🔹 Coordinator Accounts Data: {coordinator.data.get('accounts', [])}") + + # Create account sensors + for account in coordinator.data.get("accounts", []): + if account["id"] in coordinator.selected_accounts: + _LOGGER.debug(f"🔹 Adding Account Sensor: {account}") + entities.append(YNABAccountSensor(coordinator, account, entry, currency_symbol, raw_budget_name)) + + # Create category sensors + for category in coordinator.data.get("categories", []): + if category["id"] in coordinator.selected_categories: + _LOGGER.debug(f"🔹 Adding Category Sensor: {category}") + entities.append(YNABCategorySensor(coordinator, category, entry, currency_symbol, raw_budget_name)) + + # Create credit utilization sensors + for account in coordinator.data.get("accounts", []): + if account["id"] in coordinator.selected_accounts: + if account.get("type") == "creditCard": + _LOGGER.debug(f"🔹 Adding Utilization and Available Credit Sensors for Account: {account}") + entities.append(YNABUtilizationSensor(coordinator, account, entry)) + entities.append(YNABAvailableCreditSensor(coordinator, account, currency_symbol, entry)) + + # Add the entity list for the sensors + async_add_entities(entities) + +class YNABTotalCreditLimitSensor(CoordinatorEntity, SensorEntity): + """Total credit limit across all credit card accounts.""" + + _attr_icon = "mdi:credit-card-multiple" + _attr_has_entity_name = True + + def __init__(self, coordinator, currency_symbol, instance_name): + super().__init__(coordinator) + self.coordinator = coordinator + self.currency_symbol = currency_symbol + + self._attr_name = f"Total Credit Limit" + self._attr_unique_id = f"{coordinator.entry.entry_id}_total_credit_limit" + self._attr_native_unit_of_measurement = currency_symbol + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{coordinator.entry.entry_id}_extras")}, + "name": f"YNAB {instance_name} - Extras", + "manufacturer": "YNAB", + "model": "YNAB Extras", + "entry_type": "service", + } + + @property + def native_value(self): + total = 0.0 + for account in self.coordinator.data.get("accounts", []): + if account.get("type") != "creditCard": + continue + account_id = account["id"] + total += self.coordinator.get_credit_limit(account_id) + return round(total, 2) + +class YNABTotalCreditUtilizationSensor(CoordinatorEntity, SensorEntity): + """Overall credit utilization across all credit card accounts.""" + + _attr_icon = "mdi:percent" + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "%" + + def __init__(self, coordinator, instance_name): + super().__init__(coordinator) + self.coordinator = coordinator + + self._attr_name = "Total Credit Utilization" + self._attr_unique_id = f"{coordinator.entry.entry_id}_total_credit_utilization" + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{coordinator.entry.entry_id}_extras")}, + "name": f"YNAB {instance_name} - Extras", + "manufacturer": "YNAB", + "model": "YNAB Extras", + "entry_type": "service", + } + + @property + def native_value(self): + total_balance = 0.0 + total_limit = 0.0 + + for account in self.coordinator.data.get("accounts", []): + if account.get("type") != "creditCard": + continue + + account_id = account["id"] + total_balance -= (account.get("balance", 0) / 1000) + total_limit += self.coordinator.get_credit_limit(account_id) + + if total_limit <= 0: + return None + + return round((total_balance / total_limit) * 100, 1) + +class YNABTotalAvailableCreditSensor(CoordinatorEntity, SensorEntity): + """Total available credit across all credit card accounts.""" + + _attr_icon = "mdi:credit-card-check" + _attr_has_entity_name = True + + def __init__(self, coordinator, currency_symbol, instance_name): + super().__init__(coordinator) + self.coordinator = coordinator + self.currency_symbol = currency_symbol + + self._attr_name = "Total Available Credit" + self._attr_unique_id = f"{coordinator.entry.entry_id}_total_available_credit" + self._attr_native_unit_of_measurement = currency_symbol + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{coordinator.entry.entry_id}_extras")}, + "name": f"YNAB {instance_name} - Extras", + "manufacturer": "YNAB", + "model": "YNAB Extras", + "entry_type": "service", + } + + @property + def native_value(self): + total_available = 0.0 + + for account in self.coordinator.data.get("accounts", []): + if account.get("type") != "creditCard": + continue + + account_id = account["id"] + + # Credit limit (already in currency units) + limit = self.coordinator.get_credit_limit(account_id) + if not limit: + continue + + # Balance from YNAB is in milliunits + balance = account.get("balance") + if balance is None: + continue + + balance = balance / 1000 # normalize + + # Same logic as per-account sensor + total_available += limit + balance + + return round(total_available, 2) + + +class YNABExtrasSensor(CoordinatorEntity, SensorEntity): + """Representation of the YNAB Extras sensor (combining Monthly Budget & Diagnostics as separate sensors).""" + + def __init__(self, coordinator: YNABDataUpdateCoordinator, currency_symbol, instance_name): + """Initialize the sensor.""" + super().__init__(coordinator) + self.coordinator = coordinator + self.currency_symbol = currency_symbol + self.instance_name = instance_name + self._state = None + # Updated name format + self._name = f"Latest Month Summary YNAB {self.instance_name}" + self._unique_id = f"latest_month_summary_ynab_{self.coordinator.entry.entry_id}" + self._attr_extra_state_attributes = {} + + # Set default icon for the sensor + self._attr_icon = "mdi:calendar-month" + + # Define as an Extras device + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{self.coordinator.entry.entry_id}_extras")}, + "name": f"YNAB {self.instance_name} - Extras", + "manufacturer": "YNAB", + "model": "YNAB Extras", + "entry_type": "service", + } + self._attr_native_unit_of_measurement = self.currency_symbol + _LOGGER.debug(f"Initialized YNABExtrasSensor with ID: {self._unique_id}") + + @property + def device_class(self): + """Return the device class for statistics support.""" + return SensorDeviceClass.MONETARY + + @property + def state_class(self): + """Return the state class for statistics support.""" + return SensorStateClass.TOTAL + + async def async_added_to_hass(self): + """Call when entity is added to hass.""" + await super().async_added_to_hass() + # Fetch and update attributes when the sensor is added to HA + self.update_attributes() + + def update_attributes(self): + """Update attributes for Sensors - works like other sensors, using coordinator.data.""" + # Get monthly summary data from coordinator (same as other sensors) + monthly_summary = self.coordinator.data.get("monthly_summary", {}) + + if monthly_summary and "month" in monthly_summary: + month_data = monthly_summary.get("month", {}) + if month_data: + # Set the state as the activity (default to activity if not found) + self._state = month_data.get("activity", 0) / 1000 # Default to 0 if activity is missing + + # Fix attribute formatting (keep proper case) + self._attr_extra_state_attributes = { + "Budgeted": month_data.get("budgeted", 0) / 1000, + "Activity": month_data.get("activity", 0) / 1000, + "To Be Budgeted": month_data.get("to_be_budgeted", 0) / 1000, + "Age of Money": month_data.get("age_of_money", 0), + } + + # Add new attention/transaction-related metrics + unapproved = self.coordinator.data.get("unapproved_transactions", 0) + uncleared = self.coordinator.data.get("uncleared_transactions", 0) + overspent = self.coordinator.data.get("overspent_categories", 0) + needs_attention = self.coordinator.data.get("needs_attention_count", 0) + + self._attr_extra_state_attributes.update({ + "Unapproved Transactions": unapproved, + "Uncleared Transactions": uncleared, + "Overspent Categories": overspent, + "Needs Attention Count": needs_attention + }) + + else: + _LOGGER.error(f"Failed to retrieve valid month data for {self.instance_name}") + self._state = None + else: + _LOGGER.warning(f"No monthly summary data available for {self.instance_name}") + self._state = None + + @property + def name(self): + """Return the name of the sensor.""" + return self._name + + @property + def unique_id(self): + """Return the unique id of the sensor.""" + return self._unique_id + + @property + def native_value(self): + """Return the state of the sensor (activity).""" + return self._state + + @property + def extra_state_attributes(self): + """Return structured attributes with proper formatting.""" + return self._attr_extra_state_attributes + + @property + def native_unit_of_measurement(self): + """Return the unit of measurement.""" + return self.currency_symbol # Use currency symbol here + + @property + def icon(self): + """Return the icon for the sensor.""" + return self._attr_icon # Return the default icon for the monthly summary sensor + +class YNABAPIStatusSensor(CoordinatorEntity, SensorEntity): + """Sensor to track YNAB API connection status and health.""" + + def __init__(self, coordinator, instance_name): + """Initialize the sensor.""" + super().__init__(coordinator) + self.instance_name = instance_name + self._attr_name = f"YNAB API Status YNAB {self.instance_name}" + self._attr_unique_id = f"ynab_api_status_ynab_{self.coordinator.entry.entry_id}" + self._attr_icon = "mdi:api" + + # Assign to the "Extras" device + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{coordinator.entry.entry_id}_extras")}, + "name": f"YNAB {self.instance_name} - Extras", + "manufacturer": "YNAB", + "model": "YNAB Extras", + "entry_type": "service", + } + + # Assign to the Diagnostic Category + self._attr_entity_category = EntityCategory.DIAGNOSTIC + + @property + def native_value(self): + """Return the current API status.""" + api_status = self.coordinator.data.get("api_status", {}) + base_status = api_status.get("status", "Unknown") + + # Only show "Rate Limited" if we're actually getting 429 errors + # Don't override based on our estimated 200 limit since YNAB's real limit appears higher + return base_status + + @property + def extra_state_attributes(self): + """Return additional API status attributes.""" + api_status = self.coordinator.data.get("api_status", {}) + return { + "last_error": api_status.get("last_error", "None"), + "last_error_time": api_status.get("last_error_time", "Never"), + "consecutive_failures": api_status.get("consecutive_failures", 0), + "requests_made_total_all_integrations": api_status.get("requests_made_total", 0), + "requests_this_hour_all_integrations": api_status.get("requests_this_hour", 0), + "estimated_remaining": api_status.get("estimated_remaining", 200), + "rate_limit_resets_at": api_status.get("rate_limit_resets_at", "Unknown"), + "is_at_limit": api_status.get("is_at_limit", False), + "last_successful_request": api_status.get("last_successful_request", "Never"), + "note": "Counts include all YNAB integrations using the same API token", + } + + +class YNABUtilizationSensor(CoordinatorEntity, SensorEntity): + """Credit utilization sensor for credit card accounts only.""" + + def __init__(self, coordinator, account, entry): + super().__init__(coordinator) + + self.coordinator = coordinator + self.account = account + self.entry = entry + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + self._attr_name = f"{account['name']} Utilization" + self._attr_unique_id = f"{budget_id}_{account_id}_utilization" + self._attr_native_unit_of_measurement = "%" + self._attr_icon = "mdi:percent" + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": account["name"], + "manufacturer": "YNAB", + "model": "Credit Card", + } + + @property + def native_value(self): + """Return utilization percentage based on balance vs credit limit.""" + balance = self.account.get("balance") or self.account.get("cleared_balance") + balance = abs(balance) / 1000 + credit_limit = self.coordinator.get_credit_limit(self.account["id"]) + + if balance is None or credit_limit in (None, 0): + return None + + return round((abs(balance) / credit_limit) * 100, 1) + +class YNABAvailableCreditSensor(CoordinatorEntity, SensorEntity): + def __init__(self, coordinator, account, currency_symbol, entry): + super().__init__(coordinator) + + self.coordinator = coordinator + self.account = account + self.entry = entry + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + self._attr_name = f"{account['name']} Available Credit" + self._attr_unique_id = f"{budget_id}_{account_id}_available_credit" + self.currency_symbol = currency_symbol + self._attr_native_unit_of_measurement = currency_symbol + self._attr_icon = "mdi:account-credit-card-outline" + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": account["name"], + "manufacturer": "YNAB", + "model": "Credit Card", + } + + @property + def native_value(self): + balance = self.account.get("balance") or self.account.get("cleared_balance") + balance = balance / 1000 + credit_limit = self.coordinator.get_credit_limit(self.account["id"]) + + if balance is None or credit_limit in (None, 0): + return None + + return round(credit_limit + balance, 2) + +class YNABAccountSensor(CoordinatorEntity, SensorEntity): + """YNAB Account Sensor.""" + + def __init__(self, coordinator, account, entry, currency_symbol, instance_name): + """Initialize the sensor.""" + super().__init__(coordinator) + self.coordinator = coordinator + self.account = account + self.currency_symbol = currency_symbol + self.instance_name = instance_name + + account_id = account["id"] + budget_id = entry.data["budget_id"] + + + # Preserve your naming structure + self._attr_unique_id = f"ynab_{instance_name}_{account['id']}" + + # Add (Closed) suffix for closed accounts, matching category pattern + account_name = account["name"] + if account.get("closed"): + account_name += " (Closed)" + self._attr_name = f"{account_name} YNAB {instance_name}" # Friendly name for sensor + + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_{account_id}")}, + "name": f"{account['name']}", + "manufacturer": "YNAB", + "model": "YNAB Account", + "entry_type": None, + "configuration_url": ( + f"https://app.ynab.com/{budget_id}/accounts/{account_id}" + ), + } + + # Set the appropriate icon based on the account type + account_type = account.get("type", "").lower() + self._attr_icon = self.get_account_icon(account_type) # Get the icon based on type + + def get_account_icon(self, account_type: str): + """Return the appropriate icon based on the account type.""" + # Loop through the keys in ACCOUNT_ICONS and check if any part of the account_type matches + for key, icon in ACCOUNT_ICONS.items(): + if key in account_type: # Check if the account type contains any key in ACCOUNT_ICONS + return icon + return ACCOUNT_ICONS["default"] # Default icon if no match is found + + @property + def device_class(self): + """Return the device class for statistics support.""" + return SensorDeviceClass.MONETARY + + @property + def state_class(self): + """Return the state class for statistics support.""" + return SensorStateClass.TOTAL + + @property + def native_unit_of_measurement(self): + """Return the correct currency symbol explicitly for HA to recognize it.""" + return self.currency_symbol # Ensure HA correctly applies the currency + + @property + def native_value(self): + """Return the state of the sensor (cleared balance).""" + return self.account.get("cleared_balance", 0) / 1000 # Convert from milliunits + + @property + def extra_state_attributes(self): + """Return additional state attributes.""" + return { + "balance": self.account.get("balance", 0) / 1000, + "cleared_balance": self.account.get("cleared_balance", 0) / 1000, + "uncleared_balance": self.account.get("uncleared_balance", 0) / 1000, + "on_budget": self.account.get("on_budget", False), + "type": self.account.get("type", "Unknown"), + } + + async def async_added_to_hass(self): + """When added to Home Assistant, subscribe to updates.""" + await super().async_added_to_hass() + self.async_on_remove(self.coordinator.async_add_listener(self._handle_coordinator_update)) + _LOGGER.debug(f"🔹 Account Sensor Added: {self.name} (ID: {self._attr_unique_id})") + + @callback + def _handle_coordinator_update(self): + """Handle an update from the coordinator.""" + _LOGGER.debug(f"🔹 Updating Account Sensor: {self.name}") + + # Find the updated account in the coordinator data + self.account = next( + (a for a in self.coordinator.data.get("accounts", []) if a["id"] == self.account["id"]), + self.account # Keep the old data if not found + ) + + + # Update the icon if the account type changed + account_type = self.account.get("type", "").lower() + self._attr_icon = self.get_account_icon(account_type) + + # Write the new state to Home Assistant + self.async_write_ha_state() + +class YNABCategorySensor(CoordinatorEntity, SensorEntity): + """YNAB Category Sensor.""" + + def __init__(self, coordinator, category, entry, currency_symbol, instance_name): + """Initialize the sensor.""" + super().__init__(coordinator) + self.coordinator = coordinator + self.category = category + self.currency_symbol = currency_symbol + self.instance_name = instance_name + + budget_id = entry.data["budget_id"] + self._attr_unique_id = f"ynab_{instance_name}_{category['id']}" # Use instance_name for unique_id + self._attr_device_info = { + "identifiers": {(DOMAIN, f"{budget_id}_categories")}, + "name": f"YNAB {instance_name} - Categories", # Friendly name for device + "manufacturer": "YNAB", + "model": "YNAB Category", + "entry_type": "service", + } + name = category["name"] # Friendly name for sensor + if category.get("hidden"): + name += " (Hidden)" + self._attr_name = f"{name} YNAB {instance_name}" + + self._attr_native_unit_of_measurement = self.currency_symbol + + # Set the appropriate icon based on the category name + category_name = category.get("name", "").lower().replace(" ", "_") # Normalise the category name to match keys + self._attr_icon = self.get_category_icon(category_name) + + def get_category_icon(self, category_name: str): + """Return the appropriate icon based on the category name.""" + # Look for a matching word in the category name that matches the keys in CATEGORY_ICONS + for key in CATEGORY_ICONS: + if category_name.startswith(key): # Check if the category name starts with any key in CATEGORY_ICONS + return CATEGORY_ICONS[key] + return "mdi:currency-usd" # Default icon if no match is found + + @property + def device_class(self): + """Return the device class for statistics support.""" + return SensorDeviceClass.MONETARY + + @property + def state_class(self): + """Return the state class for statistics support.""" + return SensorStateClass.TOTAL + + @property + def native_unit_of_measurement(self): + """Return the correct currency symbol explicitly for HA to recognize it.""" + return self.currency_symbol # Ensure HA correctly applies the currency + + @property + def native_value(self): + """Return the state of the sensor (category balance).""" + return (self.category.get("balance") or 0) / 1000 # Convert from milliunits + + @property + def extra_state_attributes(self): + """Return additional state attributes.""" + return { + "budgeted": (self.category.get("budgeted") or 0) / 1000, + "activity": (self.category.get("activity") or 0) / 1000, + "balance": (self.category.get("balance") or 0) / 1000, + "category_group": self.category.get("category_group_name") or self.category.get("group_name", "Unknown"), + "goal_type": self.category.get("goal_type", None), + "goal_target": (self.category.get("goal_target") or 0) / 1000, + "goal_percentage_complete": (self.category.get("goal_percentage_complete") or 0), + "goal_overall_left": (self.category.get("goal_overall_left") or 0) / 1000, + "percentage_spent": ( + round( + abs(self.category.get("activity") or 0) / + abs(self.category.get("budgeted") or 1) * 100, + 2 + ) + if (self.category.get("budgeted") or 0) else 0.0 + ), + "needs_attention": ( + (self.category.get("balance") or 0) < 0 or + ( + (self.category.get("goal_target") or 0) > 0 and + (self.category.get("goal_overall_left") or 0) > 0 + ) + ), + "attention_reason": ( + "Overspent" if (self.category.get("balance") or 0) < 0 else + "Underfunded" if ( + (self.category.get("goal_target") or 0) > 0 and + (self.category.get("goal_overall_left") or 0) > 0 + ) else + "Ok" + ), + "hidden": self.category.get("hidden", False), + } + + async def async_added_to_hass(self): + """When added to Home Assistant, subscribe to updates.""" + await super().async_added_to_hass() + self.async_on_remove(self.coordinator.async_add_listener(self._handle_coordinator_update)) + + @callback + def _handle_coordinator_update(self): + """Handle an update from the coordinator.""" + # Find the updated category in the coordinator data + self.category = next( + (c for c in self.coordinator.data.get("categories", []) if c["id"] == self.category["id"]), + self.category # Keep the old data if not found + ) + + + # Update the icon in case the category name changed + category_name = self.category.get("name", "").lower().replace(" ", "_") + self._attr_icon = self.get_category_icon(category_name) + + # Write the new state to Home Assistant + self.async_write_ha_state() diff --git a/translations/en.json b/translations/en.json new file mode 100644 index 0000000..1d2eb42 --- /dev/null +++ b/translations/en.json @@ -0,0 +1,37 @@ +{ + "config": { + "step": { + "user": { + "title": "Connect to YNAB", + "description": "Enter your YNAB access token to connect your budget.\n\n🔹 **How to Get Your Access Token:**\nGo to https://app.ynab.com/settings/developer and generate a new API key.\n\n⚠️ **Important Disclaimer:**\nThis integration is **not officially supported** by You Need A Budget (YNAB). Any issues or errors caused by this integration **are not covered by YNAB’s official support channels**.\n\nBy proceeding, you acknowledge and agree to the following:\n- YNAB does not provide support for this integration.\n- Any issues must be resolved through community support (**[GitHub Issues](https://github.com/DeLuca21/ynab-ha/issues)**, Home Assistant forums).\n- You take full responsibility for API usage and compliance with **YNAB’s API Terms of Use**.\n\n📜 **Review YNAB's API Terms of Use:**\n➡️ [YNAB API Terms](https://api.ynab.com/#terms)\n\n✅ **Check the box below to accept the terms and continue.**", + "data": { + "name": "Budget Name", + "access_token": "Access Token", + "accept_terms": "I acknowledge and accept the terms", + "include_closed_accounts": "Include closed accounts", + "include_hidden_categories": "Include hidden categories" + } + } + }, + "error": { + "cannot_connect": "⚠️ Failed to connect to YNAB. Please check your network and credentials.", + "invalid_auth": "❌ Invalid authentication. Please enter a valid access token.", + "rate_limited": "⏰ YNAB API rate limit exceeded. Please wait a few minutes and try again.", + "unknown": "⚠️ An unexpected error occurred. Please check your Home Assistant logs." + } + }, + "options": { + "step": { + "init": { + "title": "YNAB Integration Options", + "description": "Configure accounts, categories, and settings for your YNAB integration.", + "data": { + "selected_accounts": "Select Accounts to Sync", + "selected_categories": "Select Categories to Sync", + "update_interval": "Set Update Interval (in minutes)" + } + } + } + } + } +