-
-
Notifications
You must be signed in to change notification settings - Fork 30.5k
/
select.py
152 lines (127 loc) · 5 KB
/
select.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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
"""Platform for Kostal Plenticore select widgets."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import Plenticore, SelectDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True)
class PlenticoreSelectEntityDescription(SelectEntityDescription):
"""A class that describes plenticore select entities."""
module_id: str
SELECT_SETTINGS_DATA = [
PlenticoreSelectEntityDescription(
module_id="devices:local",
key="battery_charge",
name="Battery Charging / Usage mode",
options=[
"None",
"Battery:SmartBatteryControl:Enable",
"Battery:TimeControl:Enable",
],
)
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Add kostal plenticore Select widget."""
plenticore: Plenticore = hass.data[DOMAIN][entry.entry_id]
available_settings_data = await plenticore.client.get_settings()
select_data_update_coordinator = SelectDataUpdateCoordinator(
hass,
_LOGGER,
"Settings Data",
timedelta(seconds=30),
plenticore,
)
entities = []
for description in SELECT_SETTINGS_DATA:
assert description.options is not None
if description.module_id not in available_settings_data:
continue
needed_data_ids = {
data_id for data_id in description.options if data_id != "None"
}
available_data_ids = {
setting.id for setting in available_settings_data[description.module_id]
}
if not needed_data_ids <= available_data_ids:
continue
entities.append(
PlenticoreDataSelect(
select_data_update_coordinator,
description,
entry_id=entry.entry_id,
platform_name=entry.title,
device_info=plenticore.device_info,
)
)
async_add_entities(entities)
class PlenticoreDataSelect(
CoordinatorEntity[SelectDataUpdateCoordinator], SelectEntity
):
"""Representation of a Plenticore Select."""
_attr_entity_category = EntityCategory.CONFIG
entity_description: PlenticoreSelectEntityDescription
def __init__(
self,
coordinator: SelectDataUpdateCoordinator,
description: PlenticoreSelectEntityDescription,
entry_id: str,
platform_name: str,
device_info: DeviceInfo,
) -> None:
"""Create a new Select Entity for Plenticore process data."""
super().__init__(coordinator)
self.entity_description = description
self.entry_id = entry_id
self.platform_name = platform_name
self.module_id = description.module_id
self.data_id = description.key
self._attr_device_info = device_info
self._attr_unique_id = f"{entry_id}_{description.module_id}"
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self.coordinator.data is not None
and self.module_id in self.coordinator.data
and self.data_id in self.coordinator.data[self.module_id]
)
async def async_added_to_hass(self) -> None:
"""Register this entity on the Update Coordinator."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.start_fetch_data(
self.module_id, self.data_id, self.options
)
)
async def async_will_remove_from_hass(self) -> None:
"""Unregister this entity from the Update Coordinator."""
self.coordinator.stop_fetch_data(self.module_id, self.data_id, self.options)
await super().async_will_remove_from_hass()
async def async_select_option(self, option: str) -> None:
"""Update the current selected option."""
for all_option in self.options:
if all_option != "None":
await self.coordinator.async_write_data(
self.module_id, {all_option: "0"}
)
if option != "None":
await self.coordinator.async_write_data(self.module_id, {option: "1"})
self.async_write_ha_state()
@property
def current_option(self) -> str | None:
"""Return the selected entity option to represent the entity state."""
if self.available:
return self.coordinator.data[self.module_id][self.data_id]
return None