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
|
# pyHomee
> pyHomee is the backbone of the [Home Assistant homee integration](https://github.com/Taraman17/hass-homee).
pyHomee is an unofficial python library for interacting with the [homee](https://hom.ee) smart home/home automation platform. It uses the [websockets](https://github.com/aaugustin/websockets) library to connect to a local homee cube and maintains a list of nodes (devices), attributes, groups, users and more that are updated whenever new data arrives from homee.
Large parts of this library are directly ported from the awesome [homeeApi](https://github.com/stfnhmplr/homee-api) javascript library.
This library was initially developed as "pymee" by [FreshlyBrewedCode](https://github.com/FreshlyBrewedCode/pymee).
## Installation
Install from [PyPI](https://pypi.org/project/pyHomee/):
```
pip install pyHomee
```
## Usage
### Getting started
pyHomee should be used with `asyncio`:
```python
from pyHomee import Homee
import asyncio
import logging
# Set debug level so we get verbose logs
logging.getLogger().setLevel(logging.DEBUG)
# Define a simple async entry point function
async def main():
# Create an instance of Homee
homee = Homee("<HOMEE IP>", "<USERNAME>", "<PASSWORD>")
# Connect and start listening on a new task
homeeTask = asyncio.create_task(homee.run())
# Wait until the connection is live and all data has been received
await homee.wait_until_connected()
# Do something here...
# Close the connection and wait until we are disconnected
homee.disconnect()
await homee.wait_until_disconnected()
# Start our entry point function
asyncio.run(main())
```
### Access devices and attributes
Devices are represented as "nodes" in the api. All nodes are available in the list `Homee.nodes` and are represented by the `HomeeNode` class.
Each node has a list of attributes accessible from `HomeeNode.attributes`. The attributes on a node represent the different attributes on a device, i.e. if a light is turned on or the target temperature of a thermostat. Attributes can be identified by the `HomeeAttribute.type` property. You can compare the type with the values from `pyHomee.const.AttributeType` to figure out what each attribute represents. The value can be accessed with the `HomeeAttribute.current_value` property.
If you need to change the value of an attribute you can use `Homee.set_value()`:
```python
# Get some node, for example using get_node_by_id
node = homee.get_node_by_id(5)
# Turn on the device. You need to pass the id of the node and the attribute as well as the value.
# Using get_attribute_by_type you can quickly find the desired attribute.
await homee.set_value(node.id, node.get_attribute_by_type(AttributeType.ON_OFF).id, 1)
```
### Receiving updates
The `Homee` class can be inherited to receive events:
```python
class MyHomee(Homee):
# Called once the websocket connection has been established.
async def on_connected(self):
pass
# Called after the websocket connection has been closed.
async def on_disconnected(self):
pass
# Called after an error has occurred.
async def on_error(self, error: str):
pass
# Called when the websocket receives a message.
# The message is automatically parsed from json into a dictionary.
async def on_message(self, msg: dict):
pass
# Called when an 'attribute' message was received and an attribute was updated.
# Contains the parsed json attribute data and the corresponding node instance.
async def on_attribute_updated(self, attribute_data: dict, node: HomeeNode):
pass
```
You can also add a listener to specific nodes to receive attribute updates:
```python
# A listener is just a function that takes a node and an attribute
def my_node_handler(node: HomeeNode, attribute: HomeeAttribute):
logging.info(f"Attribute {attribute.id} of node {node.name} was updated!")
node = homee.get_node_by_id(5)
# Adding the listener will return a function that can be called to remove the listener again
remove_listener = node.add_on_changed_listener(my_node_handler)
# If you don't need the listener anymore...
remove_listener()
```
To manually request updates from Homee, you can use the following functions:
```python
homee.update_node(self, nodeId: int)
"""Request current data for a node."""
homee.update_attribute(self, nodeId: int, attributeId: int)
"""Request current data for an attribute"""
```
### More examples
Example implementation that dumps all info into a json file and logs whenever a light is turned on or off:
```python
from pyHomee.const import NodeProfile, AttributeType
from pyHomee.model import HomeeAttribute
class JsonHomee(Homee):
async def on_message(self, msg: dict):
# Homee sends an "all" message at the beginning of each connection
# or after 'GET:all' was send.
if list(msg)[0] == "all":
f = open("homee.json", "w")
f.write(json.dumps(msg))
f.close()
async def on_attribute_updated(self, attribute_data, node):
# Wrap the attribute data with the HomeeAttribute class for easier access
attribute = HomeeAttribute(attribute_data)
# We only care for changes
if attribute.current_value == attribute.target_value:
return
# Check node profile (the type of device) and attribute type
if (
node.profile == NodeProfile.DIMMABLE_EXTENDED_COLOR_LIGHT
and attribute.type == AttributeType.ON_OFF
):
self._log(
f"[Light] {node.name} turned {'on' if attribute.target_value == 1 else 'off'}"
)
```
## License
MIT
|
