forked from danielperna84/custom_homematic
-
Notifications
You must be signed in to change notification settings - Fork 0
/
update.py
173 lines (142 loc) · 6.13 KB
/
update.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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
"""switch for Homematic(IP) Local."""
from __future__ import annotations
import logging
from typing import Any
from hahomematic.const import DeviceFirmwareState, HmPlatform
from hahomematic.platforms.update import HmUpdate
from homeassistant.components.update import UpdateEntity, UpdateEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import CONTROL_UNITS, DOMAIN
from .control_unit import ControlUnit, signal_new_hm_entity
_LOGGER = logging.getLogger(__name__)
ATTR_FIRMWARE_UPDATE_STATE = "firmware_update_state"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Homematic(IP) Local update platform."""
control_unit: ControlUnit = hass.data[DOMAIN][CONTROL_UNITS][entry.entry_id]
@callback
def async_add_update(hm_entities: tuple[HmUpdate, ...]) -> None:
"""Add update from Homematic(IP) Local."""
_LOGGER.debug("ASYNC_ADD_UPDATE: Adding %i entities", len(hm_entities))
if entities := [
HaHomematicUpdate(
control_unit=control_unit,
hm_entity=hm_entity,
)
for hm_entity in hm_entities
]:
async_add_entities(entities)
entry.async_on_unload(
func=async_dispatcher_connect(
hass=hass,
signal=signal_new_hm_entity(entry_id=entry.entry_id, platform=HmPlatform.UPDATE),
target=async_add_update,
)
)
async_add_update(hm_entities=control_unit.get_new_entities(entity_type=HmUpdate))
class HaHomematicUpdate(UpdateEntity):
"""Representation of the HomematicIP update entity."""
_attr_supported_features = UpdateEntityFeature.PROGRESS | UpdateEntityFeature.INSTALL
_attr_has_entity_name = True
_attr_should_poll = False
_attr_entity_registry_enabled_default = True
_unrecorded_attributes = frozenset({ATTR_FIRMWARE_UPDATE_STATE})
def __init__(
self,
control_unit: ControlUnit,
hm_entity: HmUpdate,
) -> None:
"""Initialize the generic entity."""
self._cu: ControlUnit = control_unit
self._hm_entity: HmUpdate = hm_entity
self._attr_unique_id = f"{DOMAIN}_{hm_entity.unique_id}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, hm_entity.device.identifier)},
)
self._attr_extra_state_attributes = {
ATTR_FIRMWARE_UPDATE_STATE: hm_entity.firmware_update_state
}
_LOGGER.debug("init: Setting up %s", hm_entity.full_name)
@property
def available(self) -> bool:
"""Return if entity is available."""
return self._hm_entity.available
@property
def installed_version(self) -> str | None:
"""Version installed and in use."""
return self._hm_entity.firmware
@property
def in_progress(self) -> bool | int | None:
"""Update installation progress."""
return self._hm_entity.firmware_update_state in (
DeviceFirmwareState.DO_UPDATE_PENDING,
DeviceFirmwareState.PERFORMING_UPDATE,
)
@property
def latest_version(self) -> str | None:
"""Latest version available for install."""
if self._hm_entity.firmware_update_state in (
DeviceFirmwareState.READY_FOR_UPDATE,
DeviceFirmwareState.DO_UPDATE_PENDING,
DeviceFirmwareState.PERFORMING_UPDATE,
):
return self._hm_entity.available_firmware
return self._hm_entity.firmware
@property
def name(self) -> str | None:
"""Return the name of the entity."""
return self._hm_entity.name
async def async_install(self, version: str | None, backup: bool, **kwargs: Any) -> None:
"""Install an update."""
await self._hm_entity.update_firmware(refresh_after_update_intervals=(10, 60))
async def async_update(self) -> None:
"""Update entity."""
await self._hm_entity.refresh_firmware_data()
async def async_added_to_hass(self) -> None:
"""Register callbacks and load initial data."""
self._hm_entity.register_entity_updated_callback(
entity_updated_callback=self._entity_changed, custom_id=self.entity_id
)
self._hm_entity.register_entity_removed_callback(
entity_removed_callback=self._async_device_removed
)
def _entity_changed(self, *args: Any, **kwargs: Any) -> None:
"""Handle device state changes."""
# Don't update disabled entities
if self.enabled:
_LOGGER.debug("Update state changed event fired for %s", self.name)
self.schedule_update_ha_state()
else:
_LOGGER.debug(
"Update state changed event for %s not fired. Entity is disabled",
self.name,
)
async def async_will_remove_from_hass(self) -> None:
"""Run when hmip device will be removed from hass."""
# Remove callback from device.
self._hm_entity.unregister_entity_updated_callback(
entity_updated_callback=self._entity_changed, custom_id=self.entity_id
)
self._hm_entity.unregister_entity_removed_callback(
entity_removed_callback=self._async_device_removed
)
@callback
def _async_device_removed(self, *args: Any, **kwargs: Any) -> None:
"""Handle hm device removal."""
self.hass.async_create_task(self.async_remove(force_remove=True))
if not self.registry_entry:
return
if device_id := self.registry_entry.device_id:
# Remove from device registry.
device_registry = dr.async_get(self.hass)
if device_id in device_registry.devices:
# This will also remove associated entities from entity registry.
device_registry.async_remove_device(device_id)