a785a1ab5d
After six months the chevy website finally has been reimplemented to something that seems to work and is stable. The backend library has been updated thanks to upstream help, and now is working again.
147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
"""
|
|
MyChevy Component.
|
|
|
|
For more details about this platform, please refer to the documentation at
|
|
https://home-assistant.io/components/mychevy/
|
|
"""
|
|
from datetime import timedelta
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
|
from homeassistant.helpers import config_validation as cv
|
|
from homeassistant.helpers import discovery
|
|
from homeassistant.util import Throttle
|
|
|
|
REQUIREMENTS = ["mychevy==1.0.1"]
|
|
|
|
DOMAIN = 'mychevy'
|
|
UPDATE_TOPIC = DOMAIN
|
|
ERROR_TOPIC = DOMAIN + "_error"
|
|
|
|
MYCHEVY_SUCCESS = "success"
|
|
MYCHEVY_ERROR = "error"
|
|
|
|
NOTIFICATION_ID = 'mychevy_website_notification'
|
|
NOTIFICATION_TITLE = 'MyChevy website status'
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
|
|
ERROR_SLEEP_TIME = timedelta(minutes=30)
|
|
|
|
CONFIG_SCHEMA = vol.Schema({
|
|
DOMAIN: vol.Schema({
|
|
vol.Required(CONF_USERNAME): cv.string,
|
|
vol.Required(CONF_PASSWORD): cv.string,
|
|
}),
|
|
}, extra=vol.ALLOW_EXTRA)
|
|
|
|
|
|
class EVSensorConfig:
|
|
"""The EV sensor configuration."""
|
|
|
|
def __init__(self, name, attr, unit_of_measurement=None, icon=None):
|
|
"""Create new sensor configuration."""
|
|
self.name = name
|
|
self.attr = attr
|
|
self.unit_of_measurement = unit_of_measurement
|
|
self.icon = icon
|
|
|
|
|
|
class EVBinarySensorConfig:
|
|
"""The EV binary sensor configuration."""
|
|
|
|
def __init__(self, name, attr, device_class=None):
|
|
"""Create new binary sensor configuration."""
|
|
self.name = name
|
|
self.attr = attr
|
|
self.device_class = device_class
|
|
|
|
|
|
def setup(hass, base_config):
|
|
"""Set up the mychevy component."""
|
|
import mychevy.mychevy as mc
|
|
|
|
config = base_config.get(DOMAIN)
|
|
|
|
email = config.get(CONF_USERNAME)
|
|
password = config.get(CONF_PASSWORD)
|
|
hass.data[DOMAIN] = MyChevyHub(mc.MyChevy(email, password), hass,
|
|
base_config)
|
|
hass.data[DOMAIN].start()
|
|
|
|
return True
|
|
|
|
|
|
class MyChevyHub(threading.Thread):
|
|
"""MyChevy Hub.
|
|
|
|
Connecting to the mychevy website is done through a selenium
|
|
webscraping process. That can only run synchronously. In order to
|
|
prevent blocking of other parts of Home Assistant the architecture
|
|
launches a polling loop in a thread.
|
|
|
|
When new data is received, sensors are updated, and hass is
|
|
signaled that there are updates. Sensors are not created until the
|
|
first update, which will be 60 - 120 seconds after the platform
|
|
starts.
|
|
"""
|
|
|
|
def __init__(self, client, hass, hass_config):
|
|
"""Initialize MyChevy Hub."""
|
|
super().__init__()
|
|
self._client = client
|
|
self.hass = hass
|
|
self.hass_config = hass_config
|
|
self.cars = []
|
|
self.status = None
|
|
self.ready = False
|
|
|
|
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
|
def update(self):
|
|
"""Update sensors from mychevy website.
|
|
|
|
This is a synchronous polling call that takes a very long time
|
|
(like 2 to 3 minutes long time)
|
|
|
|
"""
|
|
self._client.login()
|
|
self._client.get_cars()
|
|
self.cars = self._client.cars
|
|
if self.ready is not True:
|
|
discovery.load_platform(self.hass, 'sensor', DOMAIN, {},
|
|
self.hass_config)
|
|
discovery.load_platform(self.hass, 'binary_sensor', DOMAIN, {},
|
|
self.hass_config)
|
|
self.ready = True
|
|
self.cars = self._client.update_cars()
|
|
|
|
def get_car(self, vid):
|
|
"""Compatibility to work with one car."""
|
|
if self.cars:
|
|
for car in self.cars:
|
|
if car.vid == vid:
|
|
return car
|
|
return None
|
|
|
|
def run(self):
|
|
"""Thread run loop."""
|
|
# We add the status device first outside of the loop
|
|
|
|
# And then busy wait on threads
|
|
while True:
|
|
try:
|
|
_LOGGER.info("Starting mychevy loop")
|
|
self.update()
|
|
self.hass.helpers.dispatcher.dispatcher_send(UPDATE_TOPIC)
|
|
time.sleep(MIN_TIME_BETWEEN_UPDATES.seconds)
|
|
except Exception: # pylint: disable=broad-except
|
|
_LOGGER.exception(
|
|
"Error updating mychevy data. "
|
|
"This probably means the OnStar link is down again")
|
|
self.hass.helpers.dispatcher.dispatcher_send(ERROR_TOPIC)
|
|
time.sleep(ERROR_SLEEP_TIME.seconds)
|