-
-
Notifications
You must be signed in to change notification settings - Fork 30.5k
/
switch.py
134 lines (110 loc) · 3.72 KB
/
switch.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
"""Support for customised Kankun SP3 Wifi switch."""
from __future__ import annotations
import logging
from typing import Any
import requests
import voluptuous as vol
from homeassistant.components.switch import (
PLATFORM_SCHEMA as SWITCH_PLATFORM_SCHEMA,
SwitchEntity,
)
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PATH,
CONF_PORT,
CONF_SWITCHES,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__)
DEFAULT_PORT = 80
DEFAULT_PATH = "/cgi-bin/json.cgi"
SWITCH_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_PATH, default=DEFAULT_PATH): cv.string,
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD): cv.string,
}
)
PLATFORM_SCHEMA = SWITCH_PLATFORM_SCHEMA.extend(
{vol.Required(CONF_SWITCHES): cv.schema_with_slug_keys(SWITCH_SCHEMA)}
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities_callback: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up Kankun Wifi switches."""
switches = config.get("switches", {})
devices = []
for dev_name, properties in switches.items():
devices.append(
KankunSwitch(
hass,
properties.get(CONF_NAME, dev_name),
properties.get(CONF_HOST),
properties.get(CONF_PORT, DEFAULT_PORT),
properties.get(CONF_PATH, DEFAULT_PATH),
properties.get(CONF_USERNAME),
properties.get(CONF_PASSWORD),
)
)
add_entities_callback(devices)
class KankunSwitch(SwitchEntity):
"""Representation of a Kankun Wifi switch."""
def __init__(self, hass, name, host, port, path, user, passwd):
"""Initialize the device."""
self._hass = hass
self._name = name
self._state = False
self._url = f"http://{host}:{port}{path}"
if user is not None:
self._auth = (user, passwd)
else:
self._auth = None
def _switch(self, newstate):
"""Switch on or off."""
_LOGGER.info("Switching to state: %s", newstate)
try:
req = requests.get(
f"{self._url}?set={newstate}", auth=self._auth, timeout=5
)
return req.json()["ok"]
except requests.RequestException:
_LOGGER.error("Switching failed")
def _query_state(self):
"""Query switch state."""
_LOGGER.info("Querying state from: %s", self._url)
try:
req = requests.get(f"{self._url}?get=state", auth=self._auth, timeout=5)
return req.json()["state"] == "on"
except requests.RequestException:
_LOGGER.error("State query failed")
@property
def name(self):
"""Return the name of the switch."""
return self._name
@property
def is_on(self):
"""Return true if device is on."""
return self._state
def update(self) -> None:
"""Update device state."""
self._state = self._query_state()
def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
if self._switch("on"):
self._state = True
def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self._switch("off"):
self._state = False