142 lines
4.6 KiB
Python
142 lines
4.6 KiB
Python
"""Jackery Switch Platform."""
|
|
import logging
|
|
from typing import Any, TYPE_CHECKING
|
|
|
|
from homeassistant.components.switch import SwitchEntity
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from . import DOMAIN
|
|
|
|
if TYPE_CHECKING:
|
|
from .sensor import JackeryCoordinator
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
coordinator: "JackeryCoordinator" = hass.data[DOMAIN][config_entry.entry_id]
|
|
|
|
coordinator.add_switch_entities = async_add_entities
|
|
|
|
async_add_entities([
|
|
JackeryMainSwitch("isAutoStandby", "Auto Standby Allowed", coordinator),
|
|
JackeryMainSwitch("swEps", "EPS Switch", coordinator),
|
|
])
|
|
|
|
|
|
class JackeryMainSwitch(SwitchEntity):
|
|
"""On/off switch for a main-device boolean setting."""
|
|
|
|
_attr_should_poll = False
|
|
_attr_has_entity_name = True
|
|
|
|
def __init__(self, key: str, name: str, coordinator: "JackeryCoordinator") -> None:
|
|
self._key = key
|
|
self._coordinator = coordinator
|
|
self._attr_name = name
|
|
self._attr_unique_id = f"jackery_{coordinator.entry_id}_main_{key}"
|
|
self._attr_device_info = {
|
|
"identifiers": {(DOMAIN, coordinator.entry_id)},
|
|
"name": f"Jackery {coordinator.device_sn}",
|
|
"manufacturer": "Jackery",
|
|
"model": "Energy Monitor",
|
|
}
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
self._coordinator.register_entity(f"main_switch_{self._key}", self)
|
|
|
|
async def async_will_remove_from_hass(self) -> None:
|
|
self._coordinator.unregister_entity(f"main_switch_{self._key}")
|
|
await super().async_will_remove_from_hass()
|
|
|
|
def _update_from_coordinator(self, data: dict) -> None:
|
|
val = data.get(self._key)
|
|
if val is None:
|
|
return
|
|
self._attr_is_on = bool(int(val))
|
|
self._attr_available = True
|
|
self.async_write_ha_state()
|
|
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
await self._coordinator.control_main({self._key: 1})
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
await self._coordinator.control_main({self._key: 0})
|
|
|
|
|
|
class JackeryPlugSwitch(SwitchEntity):
|
|
"""On/off switch for a smart plug sub-device."""
|
|
|
|
_attr_should_poll = False
|
|
_attr_has_entity_name = True
|
|
|
|
def __init__(
|
|
self,
|
|
plug_sn: str,
|
|
dev_type: int,
|
|
coordinator: "JackeryCoordinator",
|
|
entry_id: str,
|
|
) -> None:
|
|
self._plug_sn = plug_sn
|
|
self._dev_type = dev_type
|
|
self._coordinator = coordinator
|
|
self._raw_data: dict = {}
|
|
|
|
self._attr_name = "Switch"
|
|
self._attr_unique_id = f"jackery_plug_{plug_sn}_switch"
|
|
self._attr_device_info = {
|
|
"identifiers": {(DOMAIN, f"sub_{plug_sn}")},
|
|
"via_device": (DOMAIN, entry_id),
|
|
"name": f"Jackery Plug {plug_sn}",
|
|
"manufacturer": "Jackery",
|
|
"model": f"Sub-device Type {dev_type}",
|
|
}
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
self._coordinator.register_entity(f"plug_switch_{self._plug_sn}", self)
|
|
|
|
async def async_will_remove_from_hass(self) -> None:
|
|
self._coordinator.unregister_entity(f"plug_switch_{self._plug_sn}")
|
|
await super().async_will_remove_from_hass()
|
|
|
|
def _update_from_coordinator(self, data: dict) -> None:
|
|
pool = data.get("plugs") or data.get("plug")
|
|
if not isinstance(pool, list):
|
|
return
|
|
my = next(
|
|
(p for p in pool
|
|
if p.get("sn") == self._plug_sn or p.get("deviceSn") == self._plug_sn),
|
|
None,
|
|
)
|
|
if not my:
|
|
return
|
|
self._raw_data = dict(my)
|
|
val = my.get("sysSwitch") if my.get("sysSwitch") is not None else my.get("switchSta")
|
|
if val is None:
|
|
return
|
|
self._attr_is_on = bool(int(val))
|
|
self._attr_available = True
|
|
self.async_write_ha_state()
|
|
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
await self._coordinator.control_subdevice(self._plug_sn, self._dev_type, True)
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
await self._coordinator.control_subdevice(self._plug_sn, self._dev_type, False)
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict:
|
|
return {
|
|
"plug_sn": self._plug_sn,
|
|
"dev_type": self._dev_type,
|
|
"raw_data": self._raw_data,
|
|
}
|