152 lines
4.1 KiB
Python
152 lines
4.1 KiB
Python
"""Support for N26 bank accounts."""
|
|
from datetime import datetime, timedelta, timezone
|
|
import logging
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.const import (
|
|
CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_USERNAME)
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.discovery import load_platform
|
|
from homeassistant.util import Throttle
|
|
|
|
from .const import DATA, DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEFAULT_SCAN_INTERVAL = timedelta(minutes=30)
|
|
|
|
# define configuration parameters
|
|
CONFIG_SCHEMA = vol.Schema({
|
|
DOMAIN: vol.Schema({
|
|
vol.Required(CONF_USERNAME): cv.string,
|
|
vol.Required(CONF_PASSWORD): cv.string,
|
|
vol.Optional(CONF_SCAN_INTERVAL,
|
|
default=DEFAULT_SCAN_INTERVAL): cv.time_period,
|
|
}),
|
|
}, extra=vol.ALLOW_EXTRA)
|
|
|
|
N26_COMPONENTS = [
|
|
'sensor',
|
|
'switch'
|
|
]
|
|
|
|
|
|
def setup(hass, config):
|
|
"""Set up N26 Component."""
|
|
user = config[DOMAIN][CONF_USERNAME]
|
|
password = config[DOMAIN][CONF_PASSWORD]
|
|
|
|
from n26 import api, config as api_config
|
|
api = api.Api(api_config.Config(user, password))
|
|
|
|
from requests import HTTPError
|
|
try:
|
|
api.get_token()
|
|
except HTTPError as err:
|
|
_LOGGER.error(str(err))
|
|
return False
|
|
|
|
api_data = N26Data(api)
|
|
api_data.update()
|
|
|
|
hass.data[DOMAIN] = {}
|
|
hass.data[DOMAIN][DATA] = api_data
|
|
|
|
# Load components for supported devices
|
|
for component in N26_COMPONENTS:
|
|
load_platform(hass, component, DOMAIN, {}, config)
|
|
|
|
return True
|
|
|
|
|
|
def timestamp_ms_to_date(epoch_ms) -> datetime or None:
|
|
"""Convert millisecond timestamp to datetime."""
|
|
if epoch_ms:
|
|
return datetime.fromtimestamp(epoch_ms / 1000, timezone.utc)
|
|
|
|
|
|
class N26Data:
|
|
"""Handle N26 API object and limit updates."""
|
|
|
|
def __init__(self, api):
|
|
"""Initialize the data object."""
|
|
self._api = api
|
|
|
|
self._account_info = {}
|
|
self._balance = {}
|
|
self._limits = {}
|
|
self._account_statuses = {}
|
|
|
|
self._cards = {}
|
|
self._spaces = {}
|
|
|
|
@property
|
|
def api(self):
|
|
"""Return N26 api client."""
|
|
return self._api
|
|
|
|
@property
|
|
def account_info(self):
|
|
"""Return N26 account info."""
|
|
return self._account_info
|
|
|
|
@property
|
|
def balance(self):
|
|
"""Return N26 account balance."""
|
|
return self._balance
|
|
|
|
@property
|
|
def limits(self):
|
|
"""Return N26 account limits."""
|
|
return self._limits
|
|
|
|
@property
|
|
def account_statuses(self):
|
|
"""Return N26 account statuses."""
|
|
return self._account_statuses
|
|
|
|
@property
|
|
def cards(self):
|
|
"""Return N26 cards."""
|
|
return self._cards
|
|
|
|
def card(self, card_id: str, default: dict = None):
|
|
"""Return a card by its id or the given default."""
|
|
return next((card for card in self.cards if card["id"] == card_id),
|
|
default)
|
|
|
|
@property
|
|
def spaces(self):
|
|
"""Return N26 spaces."""
|
|
return self._spaces
|
|
|
|
def space(self, space_id: str, default: dict = None):
|
|
"""Return a space by its id or the given default."""
|
|
return next((space for space in self.spaces["spaces"]
|
|
if space["id"] == space_id), default)
|
|
|
|
@Throttle(min_time=DEFAULT_SCAN_INTERVAL * 0.8)
|
|
def update_account(self):
|
|
"""Get the latest account data from N26."""
|
|
self._account_info = self._api.get_account_info()
|
|
self._balance = self._api.get_balance()
|
|
self._limits = self._api.get_account_limits()
|
|
self._account_statuses = self._api.get_account_statuses()
|
|
|
|
@Throttle(min_time=DEFAULT_SCAN_INTERVAL * 0.8)
|
|
def update_cards(self):
|
|
"""Get the latest cards data from N26."""
|
|
self._cards = self._api.get_cards()
|
|
|
|
@Throttle(min_time=DEFAULT_SCAN_INTERVAL * 0.8)
|
|
def update_spaces(self):
|
|
"""Get the latest spaces data from N26."""
|
|
self._spaces = self._api.get_spaces()
|
|
|
|
def update(self):
|
|
"""Get the latest data from N26."""
|
|
self.update_account()
|
|
self.update_cards()
|
|
self.update_spaces()
|