Skip to content
Snippets Groups Projects
Commit 1ee9a3bd authored by KERDREUX Jerome's avatar KERDREUX Jerome
Browse files

PowerRelay work in progress

I get powerrelay to work partially (able to get the right state).. but
this *** ZwaveJS API use 2 values (one for get/one for set..). So still
unable to switch the relays ;(
parent 90dda015
No related branches found
No related tags found
No related merge requests found
from xaal.schemas import devices
import pdb
from logging import getLogger
from zwave_js_server.event import Event
from zwave_js_server.model.node import Node
from zwave_js_server.model.value import Value
from zwave_js_server.event import Event
from logging import getLogger
from .cmdclass import COMMAND_CLASS
from xaal.schemas import devices
from xaal.lib import tools, bindings, AsyncEngine, Device
import pdb
from .cmdclass import COMMAND_CLASS
from . import const
logger = getLogger(__name__)
BASE_ADDR = tools.get_uuid('74fd7cf2-a349-46af-bbaf-41d98fab0000')
def get_property_id(value: Value):
if value.property_ in ["value", "currentValue", "targetValue"]:
......@@ -20,22 +23,44 @@ def get_property_id(value: Value):
return "--" + str(value.property_name)
def dump_value(value: Value):
def value_dump(value: Value):
print(f"[{value}] {'0x%x'%value.command_class} {value.endpoint} {get_property_id(value)} => {value.value}")
def build_devices(node: Node):
def value_is_command_class(value: Value, command_class: COMMAND_CLASS):
return value.command_class == command_class.value
def value_is_value(value: Value):
return value.property_ in ["value", "currentValue"]
def build_devices(node: Node, eng: AsyncEngine):
config = node.device_config
logger.info(f"{node.node_id} {config.manufacturer}/{config.label} ")
# node.on("value updated", self.value_updated)
# pdb.set_trace()
#
base_addr = BASE_ADDR + node.node_id * 128
for k in node.values:
value = node.values.get(k)
assert value
if value.command_class == COMMAND_CLASS.SWITCH_BINARY.value and value.property_ == 'currentValue':
dump_value(value)
if value.command_class == COMMAND_CLASS.METER.value and value.property_ == 'value':
dump_value(value)
if value is None or not value_is_value(value):
continue
base_addr = base_addr + 1
value_dump(value)
obj = None
if value_is_command_class(value, COMMAND_CLASS.SWITCH_BINARY):
obj = PowerRelay(node, value, base_addr)
if value_is_command_class(value, COMMAND_CLASS.METER):
# obj = PowerMeter(node, value, base_addr)
pass
if obj is not None and obj.dev:
obj.setup()
eng.add_device(obj.dev)
# if node.node_id == 5:
# pdb.set_trace()
......@@ -44,25 +69,47 @@ def build_devices(node: Node):
class ZwaveDevice(object):
def __init__(self, node: Node):
self.node = node
self.dev: Device | None = None
def setup(self):
pass
assert self.dev
dev = self.dev
dev.vendor_id = f"ZwaveJS/{self.node.device_config.manufacturer}"
dev.product_id = f"{self.node.device_config.label}"
dev.info = f"{self.node.node_id}"
self.node.on(const.EVT_VALUE_UPDATED, self.update)
def update(self, event: Event):
pass
print(event)
class PowerRelay(ZwaveDevice):
def __init__(self, node):
def __init__(self, node: Node, value: Value, base_addr: bindings.UUID):
super().__init__(node)
self.dev = devices.powerrelay()
dev = devices.powerrelay(base_addr)
dev.methods["turn_on"] = self.turn_on
dev.methods["turn_off"] = self.turn_off
dev.attributes["power"] = value.value
self.dev = dev
self.state = value
def update(self, event: Event):
if self.dev:
value = event['value']
if self.state == value:
self.dev.attributes["power"] = value.value
async def turn_on(self):
pdb.set_trace()
await self.node.async_set_value(self.state, True)
async def turn_off(self):
await self.node.async_set_value(self.state, False)
class PowerMeter(ZwaveDevice):
def __init__(self, node):
def __init__(self, node: Node, value: Value, base_addr: bindings.UUID):
super().__init__(node)
self.dev = devices.powermeter()
self.state = value
self.dev = devices.powermeter(base_addr)
self.dev.unsupported_attributes = ['devices']
def update(self, event: Event):
print(event)
......@@ -58,9 +58,9 @@ class GW:
nodes = self.client.driver.controller.nodes
for node in nodes.values():
if node.ready:
build_devices(node)
if DEBUG:
node.on(EVT_VALUE_UPDATED, self.dump_event)
build_devices(node, self.engine)
# if DEBUG:
# node.on(EVT_VALUE_UPDATED, self.dump_event)
def dump_event(self, event):
cmd_class = int(event["args"]["commandClass"])
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment