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
|
import logging
from collections import defaultdict
from typing import Any, Dict, Optional
import click
from .click_common import command, format_output
from .device import Device, DeviceStatus
_LOGGER = logging.getLogger(__name__)
MODEL_PWZN_RELAY_APPLE = "pwzn.relay.apple"
MODEL_PWZN_RELAY_BANANA = "pwzn.relay.banana"
AVAILABLE_PROPERTIES = {
MODEL_PWZN_RELAY_APPLE: [
"relay_status",
"on_count",
"name0",
"name1",
"name2",
"name3",
"name4",
"name5",
"name6",
"name7",
"name8",
"name9",
"name10",
"name11",
"name12",
"name13",
"name14",
"name15",
],
MODEL_PWZN_RELAY_BANANA: [
"relay_status",
"on_count",
"name0",
"name1",
"name2",
"name3",
"name4",
"name5",
"name6",
"name7",
"name8",
"name9",
"name10",
"name11",
"name12",
"name13",
"name14",
"name15",
],
}
class PwznRelayStatus(DeviceStatus):
"""Container for status reports from the plug."""
def __init__(self, data: Dict[str, Any]) -> None:
"""Response of a PWZN Relay Apple (pwzn.relay.apple)
{ 'relay_status': 9, 'on_count': 2, 'name0': 'channel1', 'name1': '',
'name2': '', 'name3': '', 'name4': '', 'name5': '', 'name6': '',
'name7': '', 'name8': '', 'name9': '', 'name10': '', 'name11': '',
'name12': '', 'name13': '', 'name14': '', 'name15': '' }
"""
self.data = data
@property
def relay_state(self) -> Optional[int]:
"""Current relay state."""
if "relay_status" in self.data:
return self.data["relay_status"]
return None
@property
def relay_names(self) -> Dict[int, str]:
def _extract_index_from_key(name) -> int:
"""extract the index from the variable."""
return int(name[4:])
return {
_extract_index_from_key(name): value
for name, value in self.data.items()
if name.startswith("name")
}
@property
def on_count(self) -> Optional[int]:
"""Number of on relay."""
if "on_count" in self.data:
return self.data["on_count"]
return None
class PwznRelay(Device):
"""Main class representing the PWZN Relay."""
_supported_models = list(AVAILABLE_PROPERTIES.keys())
@command(default_output=format_output("", "on_count: {result.on_count}\n"))
def status(self) -> PwznRelayStatus:
"""Retrieve properties."""
properties = AVAILABLE_PROPERTIES.get(
self.model, AVAILABLE_PROPERTIES[MODEL_PWZN_RELAY_APPLE]
).copy()
values = self.get_properties(properties)
return PwznRelayStatus(defaultdict(lambda: None, zip(properties, values)))
@command(
click.argument("number", type=int),
default_output=format_output("Turn on relay {number}"),
)
def relay_on(self, number: int = 0):
"""Relay X on."""
if self.send("power_on", [number]) == [0]:
return ["ok"]
@command(
click.argument("number", type=int),
default_output=format_output("Turn off relay {number}"),
)
def relay_off(self, number: int = 0):
"""Relay X off."""
if self.send("power_off", [number]) == [0]:
return ["ok"]
@command(default_output=format_output("Turn on all relay"))
def all_relay_on(self):
"""Relay all on."""
return self.send("power_all", [1])
@command(default_output=format_output("Turn off all relay"))
def all_relay_off(self):
"""Relay all off."""
return self.send("power_all", [0])
@command(
click.argument("number", type=int),
click.argument("name", type=str),
default_output=format_output("Set relay {number} name to {name}"),
)
def set_name(self, number: int = 0, name: str = ""):
"""Set relay X name."""
return self.send("set_name", [number, name])
|