Skip to content

Instantly share code, notes, and snippets.

@balloob balloob/pytradfri.py
Last active Aug 23, 2017

Embed
What would you like to do?
WIP library to control Ikea Tradfri
"""
This is the Ikea Tradfri code from @ggravlingen extracted into a lib.
https://github.com/ggravlingen/home-assistant/blob/master/custom_components/light/ikeatradfri.py
Depends on modified coap-client with dtls support. Build instructions here:
https://community.home-assistant.io/t/ikea-tradfri-gateway-zigbee-very-basic-working-implementation/14788/19?u=balloob
Run with python3 -i pytradfri.py IP KEY
Will give you an interactive Python shell:
- 'api' is method to call the hub
- 'hub' is hub object
- 'lights' is all devices that have lights
- 'light' is the first device that has a light
"""
from datetime import datetime
import json
import logging
import subprocess
import sys
PATH_ROOT = "15001"
ATTR_DEVICE_INFO = "3"
ATTR_MANUFACTURER = "0"
ATTR_MODEL_NAME = "1"
ATTR_NAME = "9001"
ATTR_CREATED_AT = "9002"
ATTR_ID = "9003"
ATTR_REACHABLE_STATE = "9019"
ATTR_LAST_SEEN = "9020"
ATTR_OTA_UPDATE_STATE = "9054"
ATTR_SWITCHES = "15009"
ATTR_LIGHTS = "3311" # array
ATTR_COLOR_X = "5709"
ATTR_COLOR_Y = "5710"
ATTR_BRIGHTNESS = "5851" # 0..254
ATTR_STATE = "5850" # 0 / 1
ATTR_DIMMER = "5851"
_LOGGER = logging.getLogger(__name__)
class PyTradFriError(Exception):
"""Base Error"""
pass
class CommandError(PyTradFriError):
pass
def api_factory(host, security_code):
"""Generate a request method."""
def request(method, path, data=None):
"""Make a request."""
path = '/'.join(str(v) for v in path)
command_string = 'coaps://{}:5684/{}'.format(host, path)
command = [
'/usr/local/bin/coap-client',
'-u',
'Client_identity',
'-k',
security_code,
'-v',
'0',
'-m',
method,
command_string
]
kwargs = {
'timeout': 10,
'stderr': subprocess.STDOUT,
}
if data is not None:
kwargs['input'] = json.dumps(data).encode('utf-8')
command.append('-f')
command.append('-')
_LOGGER.debug('Executing {} {} {}: {}'.format(
host, method, path, data))
else:
_LOGGER.debug('Executing {} {} {}'.format(host, method, path))
try:
return_value = subprocess.check_output(command, **kwargs)
out = return_value.strip().decode('utf-8')
except subprocess.CalledProcessError:
raise CommandError() from None
# Return only the last line, where there's JSON
lines = out.split('\n')
if len(lines) < 4:
return None
output = lines[3]
_LOGGER.debug('Received: %s', output)
return json.loads(output)
return request
class Hub(object):
"""This class connects to the IKEA Tradfri Gateway"""
def __init__(self, api):
self._api = api
self._devices = None
def get_devices(self):
"""Returns the devices linked to the gateway"""
devices = self._api('get', [PATH_ROOT])
return [Device(self._api, self._api('get', [PATH_ROOT, dev]))
for dev in devices]
def get_lights(self):
"""Return devices that contain lights connected to this hub."""
return [dev for dev in self.get_devices() if dev.has_lights]
class Device(object):
"""Base class for devices."""
def __init__(self, api, info):
self._api = api
self._info = info
@property
def id(self):
return self._info[ATTR_ID]
@property
def manufacturer(self):
return self._info[ATTR_DEVICE_INFO][ATTR_MANUFACTURER]
@property
def model_name(self):
return self._info[ATTR_DEVICE_INFO][ATTR_MODEL_NAME]
@property
def name(self):
return self._info[ATTR_NAME]
@property
def created_at(self):
return datetime.utcfromtimestamp(self._info[ATTR_CREATED_AT])
@property
def last_seen(self):
return datetime.utcfromtimestamp(self._info[ATTR_LAST_SEEN])
@property
def has_switches(self):
return ATTR_SWITCHES in self._info
@property
def has_lights(self):
return ATTR_LIGHTS in self._info
@property
def lights(self):
return [Light(self._api, self.id, index, light) for index, light
in enumerate(self._info.get(ATTR_LIGHTS, []))]
def set_light_brightness(self, brightness, *, index=0):
assert len(self._info.get(ATTR_LIGHTS, [])) == 1, \
'Only devices with 1 light supported'
self._api('put', [PATH_ROOT, self.id], {
ATTR_LIGHTS: [
{
ATTR_BRIGHTNESS: brightness
}
]
})
def set_light_xy_color(self, color_x, color_y, *, index=0):
assert len(self._info.get(ATTR_LIGHTS, [])) == 1, \
'Only devices with 1 light supported'
# TODO doesn't work yet
self._api('put', [PATH_ROOT, self.id], {
ATTR_LIGHTS: [
{
ATTR_COLOR_X: color_x,
ATTR_COLOR_Y: color_y,
}
]
})
def update(self):
self._info = self._api('get', [PATH_ROOT, self.id])
def __repr__(self):
return "<{} - {} ({})>".format(self.id, self.name, self.model_name)
class Light:
def __init__(self, api, parent_id, index, info):
self._api = api
self.parent_id = parent_id
self.index = index
self._info = info
@property
def is_on(self):
return self._info.get(ATTR_STATE) == 1
@property
def brightness(self):
return self._info.get(ATTR_BRIGHTNESS)
@property
def xy_color(self):
return self._info.get(ATTR_COLOR_X), self._info.get(ATTR_COLOR_Y)
def __repr__(self):
state = "on" if self.is_on else "off"
return "<Light #{} - {}>".format(self.index, state)
if __name__ == '__main__':
if len(sys.argv) != 3:
print('Call with {} <host> <key>'.format(sys.argv[0]))
sys.exit(1)
logging.basicConfig(level=logging.DEBUG)
api = api_factory(sys.argv[1], sys.argv[2])
hub = Hub(api)
lights = hub.get_lights()
light = lights[0]
print()
print("Example commands:")
print("> hub.get_devices()")
print("> light.lights")
print("> light.set_light_brightness(10)")
print("> light.set_light_brightness(254)")
print("> light.set_light_xy_color(254)")
{
"3": {
"0": "IKEA of Sweden",
"1": "TRADFRI bulb E26 WS opal 980lm",
"2": "",
"3": "1.1.1.1-5.7.2.0",
"6": 1
},
"3311": [
{
"5706": "f5faf6",
"5707": 0,
"5708": 0,
"5709": 24933,
"5710": 24691,
"5711": 0,
"5850": 1,
"5851": 167,
"9003": 0
}
],
"5750": 2,
"9001": "Living Room White",
"9002": 1491771330,
"9003": 65537,
"9019": 1,
"9020": 1491796224,
"9054": 0
}
{
"3": {
"0": "IKEA of Sweden",
"1": "TRADFRI remote control",
"2": "",
"3": "1.1.1.1-5.7.2.0",
"6": 3,
"9": 100
},
"5750": 0,
"9001": "Living Room Remote",
"9002": 1491771280,
"9003": 65536,
"9019": 1,
"9020": 1491783874,
"9054": 0,
"15009": [
{
"9003": 0
}
]
}
@balloob

This comment has been minimized.

Copy link
Owner Author

commented Apr 10, 2017

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.