-
Notifications
You must be signed in to change notification settings - Fork 0
/
slxbmw.py
100 lines (82 loc) · 2.94 KB
/
slxbmw.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
""" slxmodule for bmw connect integratio"""
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers import entity_registry, device_registry
from homeassistant.helpers.entity_registry import EntityRegistry
from homeassistant.util import slugify
from homeassistant.loader import async_get_integrations, async_get_integration
import logging
import asyncio
from typing import Any
from enum import Enum
from homeassistant.core import (
HomeAssistant,
Event,
Callable,
State,
callback,
)
from .slxcar import SLXCar
_LOGGER = logging.getLogger(__name__)
class SLXBmw(SLXCar):
DOMAIN_NAME = "bmw_connected_drive"
# entities we subscribe to
WATCHED_ENTITIES = {
# type of value : entity_id
"soclevel": "sensor.{devicename}_remaining_battery_percent",
}
@staticmethod
def get_domain() -> str:
return SLXBmw.DOMAIN_NAME
@staticmethod
def get_required_entities() -> list[dict[str, str]]:
combined_list = [SLXBmw.WATCHED_ENTITIES]
return combined_list
def __init__(self, hass: HomeAssistant):
_LOGGER.info("Initialize SLXBMW")
super().__init__(hass)
self.dynamic_config[SLXCar.CONF_SOC_UPDATE_REQUIRED] = False
def connect(
self,
cb_soc: Callable[[Event], Any],
cb_soc_update: Callable[[Event], Any] | None = None,
device_id: str | None = None,
) -> bool:
found_devices = SLXCar.find_devices_check_entites(
self.hass, SLXBmw.get_domain(), SLXBmw.get_required_entities()
)
# self.device_id = device_id
if device_id in found_devices:
self.device_id = device_id
self.device_name = found_devices[self.device_id]
else:
_LOGGER.error("Device %s not found", device_id)
return False
self.slugified_name = SLXCar._slugify_device_name(self.device_name)
_LOGGER.info("Found BMW device : device_name %s", self.device_name)
self._subscribe_entity(
SLXCar._traslate_entity_name(
SLXBmw.WATCHED_ENTITIES["soclevel"], self.slugified_name
),
cb_soc,
)
super().connect()
async def disconnect(self) -> bool:
_LOGGER.info("Disconnect")
super().disconnect()
def request_soc_update(self) -> bool:
UPDATE_SERVICE_DOMAIN = "homeassistant"
UPDATE_SERVICE_REQUEST = "update_entity"
entity_name = SLXCar._traslate_entity_name(
SLXBmw.WATCHED_ENTITIES["soclevel"], self.slugified_name
)
if self.hass.services.has_service(
UPDATE_SERVICE_DOMAIN, UPDATE_SERVICE_REQUEST
):
self.hass.async_add_executor_job(
self.hass.services.call,
UPDATE_SERVICE_DOMAIN,
UPDATE_SERVICE_REQUEST,
{"entity_id": entity_name},
)
return True
return False