commit d19e19506c026b6af1a0896e2cbdd89bb8ce1955 Author: Your Name Date: Fri Jan 13 08:49:26 2017 +0300 Init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f70efb2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,93 @@ +#docker-compose file, because it contains passwords +config.yml +config.yaml + +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# IPython Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# dotenv +.env + +# virtualenv +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..0b066f5 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +#for x86 +FROM monster1025/alpine86-python + +#for x64 +#FROM jfloff/alpine-python:3.4 + +ENV LIBRARY_PATH=/lib:/usr/lib + +ADD src/requirements.txt / +RUN pip install --upgrade pip && pip install -r /requirements.txt + +WORKDIR /app +COPY src /app +ADD config.yaml /app/ + +CMD ["python", "-u", "/app/main.py"] \ No newline at end of file diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..0124494 --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,5 @@ +xiaomi: + build: . + container_name: xiaomi + net: host + restart: always diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..f26aba8 --- /dev/null +++ b/src/main.py @@ -0,0 +1,69 @@ +import logging +import time +import threading +import os +import json + +#mine +import mqtt +import yamlparser +from xiaomihub import XiaomiHub + +logging.basicConfig(level=logging.DEBUG) +_LOGGER = logging.getLogger(__name__) + +def process_gateway_messages(gateway, client): + while True: + try: + packet = gateway._queue.get() + _LOGGER.debug("data from queuee: " + format(packet)) + + sid = packet.get("sid", None) + model = packet.get("model", "") + data = packet.get("data", "") + + if (sid != None and data != ""): + data_decoded = json.loads(data) + client.publish(model, sid, data_decoded) + gateway._queue.task_done() + except Exception as e: + _LOGGER.error('Error while sending from gateway to mqtt: ', str(e)) + +def process_mqtt_messages(gateway, client): + while True: + try: + data = client._queue.get() + _LOGGER.debug("data from mqtt: " + format(data)) + + sid = data.get("sid", None) + param = data.get("param", None) + value = data.get("value", None) + + resp = gateway.write_to_hub(sid, param, value) + client._queue.task_done() + except Exception as e: + _LOGGER.error('Error while sending from mqtt to gateway: ', str(e)) + +if __name__ == "__main__": + _LOGGER.info("Loading config file...") + config=yamlparser.load_yaml('config.yaml') + gateway_pass = yamlparser.get_gateway_password(config) + + _LOGGER.info("Init mqtt client.") + client = mqtt.Mqtt(config) + client.connect() + #only this devices can be controlled from MQTT + client.subscribe("gateway", "+", "rgb", "set") + client.subscribe("plug", "+", "status", "set") + + gateway = XiaomiHub(gateway_pass) + t1 = threading.Thread(target=process_gateway_messages, args=[gateway, client]) + t1.daemon = True + t1.start() + + t2 = threading.Thread(target=process_mqtt_messages, args=[gateway, client]) + t2.daemon = True + t2.start() + + while True: + time.sleep(10) diff --git a/src/mqtt.py b/src/mqtt.py new file mode 100644 index 0000000..0ff6439 --- /dev/null +++ b/src/mqtt.py @@ -0,0 +1,140 @@ +import paho.mqtt.client as mqtt +import os +import logging +from queue import Queue +from threading import Thread + +_LOGGER = logging.getLogger(__name__) + +class Mqtt: + username = "" + password = "" + server = "localhost" + port = 1883 + prefix = "home" + + _client = None + _sids = None + _queue = None + _threads = None + + def __init__(self, config): + if (config == None): + raise "Config is null" + + #load sids dictionary + self._sids = config.get("sids", None) + if (self._sids == None): + raise "Config sids section is null" + + #load mqtt settings + mqttConfig = config.get("mqtt", None) + if (mqttConfig == None): + raise "Config mqtt section is null" + + self.username = mqttConfig.get("username", "") + self.password = mqttConfig.get("password", "") + self.server = mqttConfig.get("server", "localhost") + self.port = mqttConfig.get("port", 1883) + self.prefix = mqttConfig.get("prefix", "home") + self._queue = Queue() + self._threads = [] + + def connect(self): + _LOGGER.info("Connecting to MQTT server " + self.server + ":" + str(self.port) + " with username (" + self.username + ":" + self.password + ")") + self._client = mqtt.Client() + if (self.username != "" and self.password != ""): + self._client.username_pw_set(self.username, self.password) + self._client.on_message = self._mqtt_process_message + self._client.on_connect = self._mqtt_on_connect + self._client.connect(self.server, self.port, 60) + + #run message processing loop + t1 = Thread(target=self._mqtt_loop) + t1.start() + self._threads.append(t1) + + def subscribe(self, model="+", name="+", prop="+", command="set"): + topic = self.prefix + "/" + model + "/" + name + "/" + prop + "/" + command + _LOGGER.info("Subscibing to " + topic + ".") + self._client.subscribe(topic) + + def publish(self, model, sid, data, retain=True): + sidprops = self._sids.get(sid, None) + if (sidprops != None): + model = sidprops.get("model",model) + sid = sidprops.get("name",sid) + + # _LOGGER.info("data is " + format(data)) + PATH_FMT = self.prefix + "/{model}/{sid}/{prop}" + for key, value in data.items(): + # fix for latest motion value + if (model == "motion" and key == "no_motion"): + key="status" + value="no_motion" + + # fix for rgb format + # if (key == "rgb" and self._is_int(value)): + # intval = int(value) + # blue = (intval) & 255 + # green = (intval >> 8) & 255 + # red = (intval >> 16) & 255 + # value = str(red)+","+str(green)+","+str(blue) + + topic = PATH_FMT.format(model=model, sid=sid, prop=key) + _LOGGER.info("Publishing message to topic " + topic + ": " + str(value) + ".") + self._client.publish(topic, payload=value, qos=0, retain=retain) + + def _mqtt_on_connect(self, client, userdata, rc, unk): + _LOGGER.info("Connected to mqtt server.") + + def _mqtt_process_message(self, client, userdata, msg): + _LOGGER.info("Processing message in " + str(msg.topic) + ": " + str(msg.payload) + ".") + parts = msg.topic.split("/") + if (len(parts) != 5): + return + model = parts[1] + name = parts[2] #name part + found = False + for sid in self._sids: + if (sid == None): + continue + sidprops = self._sids.get(sid, None) + if sidprops == None: + continue + sidname = sidprops.get("name", sid) + sidmodel = sidprops.get("model", "") + if sidname != name or sidmodel != model: + _LOGGER.debug(sidmodel + "-" + sidname + " is not " + model + "-" + name + ".") + continue + param = parts[3] #param part + value = (msg.payload).decode('utf-8') + if self._is_int(value): + value = int(value) + + # fix for rgb format + if (param == "rgb" and "," in str(value)): + arr = value.split(",") + r = int(arr[0]) + g = int(arr[1]) + b = int(arr[2]) + value = int('%02x%02x%02x%02x' % (255, r, g, b), 16) + + data = {'sid': sid, 'model': model, 'name': name, 'param':param, 'value':value} + # put in process queuee + self._queue.put(data) + found = True + break + if (found == False): + _LOGGER.error("Device " + model + "/" + name + " not found =(.") + + def _mqtt_loop(self): + _LOGGER.info("Starting mqtt loop.") + self._client.loop_forever() + + def _is_int(self, x): + try: + tmp = int(x) + return True + except Exception as e: + return False diff --git a/src/requirements.txt b/src/requirements.txt new file mode 100644 index 0000000..ab7766c --- /dev/null +++ b/src/requirements.txt @@ -0,0 +1,3 @@ +paho-mqtt +pyyaml +pycrypto \ No newline at end of file diff --git a/src/xiaomihub.py b/src/xiaomihub.py new file mode 100644 index 0000000..352fac8 --- /dev/null +++ b/src/xiaomihub.py @@ -0,0 +1,251 @@ +import socket +import struct +import json +import logging +from collections import defaultdict +from queue import Queue +from threading import Thread + +_LOGGER = logging.getLogger(__name__) + +# MANDATORY!!!! NEED TO TURN OFF "_process_report" THREAD IF CODE IS UPDATED!!!! + +class XiaomiHub: + GATEWAY_KEY = None + GATEWAY_IP = None + GATEWAY_PORT = None + GATEWAY_SID = None + GATEWAY_TOKEN = None + + XIAOMI_DEVICES = defaultdict(list) + XIAOMI_HA_DEVICES = defaultdict(list) + + MULTICAST_ADDRESS = '224.0.0.50' + MULTICAST_PORT = 9898 + GATEWAY_DISCOVERY_ADDRESS = '224.0.0.50' + GATEWAY_DISCOVERY_PORT = 4321 + SOCKET_BUFSIZE = 1024 + + def __init__(self, key, gateway=None): + self.GATEWAY_KEY = key + self._listening = False + self._queue = None + self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self._mcastsocket = None + self._deviceCallbacks = defaultdict(list) + self._threads = [] + + if gateway is not None: + self.GATEWAY_DISCOVERY_ADDRESS = gateway + + try: + _LOGGER.info('Discovering Xiaomi Gateways using address {0}'.format(self.GATEWAY_DISCOVERY_ADDRESS)) + data = self._send_socket('{"cmd":"whois"}', "iam", self.GATEWAY_DISCOVERY_ADDRESS, self.GATEWAY_DISCOVERY_PORT) + if data["model"] == "gateway": + self.GATEWAY_IP = data["ip"] + self.GATEWAY_PORT = int(data["port"]) + self.GATEWAY_SID = data["sid"] + _LOGGER.info('Gateway found on IP {0}'.format(self.GATEWAY_IP)) + else: + _LOGGER.error('Error with gateway response : {0}'.format(data)) + except Exception as e: + raise + _LOGGER.error("Cannot discover hub using whois: {0}".format(e)) + + self._socket.close() + + if self.GATEWAY_IP is None: + _LOGGER.error('No Gateway found. Cannot continue') + return None + + self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + _LOGGER.info('Creating Multicast Socket') + self._mcastsocket = self._create_mcast_socket() + if self._listen() is True: + _LOGGER.info("Listening") + + _LOGGER.info('Discovering Xiaomi Devices') + self._discover_devices() + + def _discover_devices(self): + + cmd = '{"cmd" : "get_id_list"}' + resp = self._send_cmd(cmd, "get_id_list_ack") + self.GATEWAY_TOKEN = resp["token"] + sids = json.loads(resp["data"]) + + _LOGGER.info('Found {0} devices'.format(len(sids))) + + sensors = ['sensor_ht'] + binary_sensors = ['magnet', 'motion', 'switch', '86sw1', '86sw2', 'cube'] + switches = ['plug', 'ctrl_neutral1', 'ctrl_neutral2'] + + for sid in sids: + cmd = '{"cmd":"read","sid":"' + sid + '"}' + resp = self._send_cmd(cmd, "read_ack") + model = resp["model"] + + if model == '': + model = 'cube' + + xiaomi_device = { + "model":model, + "sid":resp["sid"], + "short_id":resp["short_id"], + "data":json.loads(resp["data"])} + + device_type = None + if model in sensors: + device_type = 'sensor' + elif model in binary_sensors: + device_type = 'binary_sensor' + elif model in switches: + device_type = 'switch' + + if device_type == None: + _LOGGER.error('Unsupported devices : {0}'.format(model)) + else: + self.XIAOMI_DEVICES[device_type].append(xiaomi_device) + + def _send_cmd(self, cmd, rtnCmd): + return self._send_socket(cmd, rtnCmd, self.GATEWAY_IP, self.GATEWAY_PORT) + + def _send_socket(self, cmd, rtnCmd, ip, port): + socket = self._socket + try: + socket.settimeout(30.0) + socket.sendto(cmd.encode(), (ip, port)) + socket.settimeout(30.0) + data, addr = socket.recvfrom(1024) + if len(data) is not None: + resp = json.loads(data.decode()) + if resp["cmd"] == rtnCmd: + return resp + else: + _LOGGER.error("Response from {0} does not match return cmd".format(ip)) + _LOGGER.error(data) + else: + _LOGGER.error("No response from Gateway") + except socket.timeout: + _LOGGER.error("Cannot connect to Gateway") + socket.close() + + def write_to_hub(self, sid, data_key, datavalue): + key = self._get_key() + if type(datavalue) == int: + datavalue_formatted = str(datavalue) + else: + datavalue_formatted = '"' + datavalue + '"' + cmd = '{ "cmd":"write","sid":"' + sid + '","data":"{"' + data_key + '":' + datavalue_formatted + ',"key":"' + key + '"}}' + return self._send_cmd(cmd, "write_ack") + + def get_from_hub(self, sid): + cmd = '{ "cmd":"read","sid":"' + sid + '"}' + return self._send_cmd(cmd, "read_ack") + + def _get_key(self): + from Crypto.Cipher import AES + IV = bytes(bytearray.fromhex('17996d093d28ddb3ba695a2e6f58562e')) + encryptor = AES.new(self.GATEWAY_KEY, AES.MODE_CBC, IV=IV) + ciphertext = encryptor.encrypt(self.GATEWAY_TOKEN) + return ''.join('{:02x}'.format(x) for x in ciphertext) + + def _create_mcast_socket(self): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((self.MULTICAST_ADDRESS, self.MULTICAST_PORT)) + mreq = struct.pack("4sl", socket.inet_aton(self.MULTICAST_ADDRESS), socket.INADDR_ANY) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + return sock + + def _listen(self): + """Start listening.""" + self._queue = Queue() + self._listening = True + + t1 = Thread(target=self._listen_to_msg, args=()) + self._threads.append(t1) + t1.daemon = True + t1.start() + + # t2 = Thread(target=self._process_report, args=()) + # self._threads.append(t2) + # t2.da = True + # t2.start() + + return True + + def stop(self): + """Stop listening.""" + self._listening = False + self._queue.put(None) + + for t in self._threads: + t.join() + + if self._mcastsocket is not None: + self._mcastsocket.close() + self._mcastsocket = None + + def _listen_to_msg(self): + while self._listening: + if self._mcastsocket is not None: + data, addr = self._mcastsocket.recvfrom(self.SOCKET_BUFSIZE) + try: + data = json.loads(data.decode("ascii")) + cmd = data['cmd'] + _LOGGER.debug(format(data)) + if cmd == 'heartbeat' and data['model'] == 'gateway': + self.GATEWAY_TOKEN = data['token'] + elif cmd == 'report' or cmd == 'heartbeat': + self._queue.put(data) + else: + _LOGGER.error('Unknown multicast data : {0}'.format(data)) + except Exception as e: + raise + _LOGGER.error('Cannot process multicast message : {0}'.format(data)) + + def _process_report(self): + while self._listening: + packet = self._queue.get(True) + if isinstance(packet, dict): + try: + sid = packet['sid'] + model = packet['model'] + data = json.loads(packet['data']) + + for device in self.XIAOMI_HA_DEVICES[sid]: + device.push_data(data) + + except Exception as e: + _LOGGER.error("Cannot process Report: {0}".format(e)) + + self._queue.task_done() + +class XiaomiDevice(): + """Representation a base Xiaomi device.""" + + def __init__(self, device, name, xiaomi_hub): + """Initialize the xiaomi device.""" + self._sid = device['sid'] + self._name = '{}_{}'.format(name, self._sid) + self.parse_data(device['data']) + + self.xiaomi_hub = xiaomi_hub + xiaomi_hub.XIAOMI_HA_DEVICES[self._sid].append(self) + + @property + def name(self): + """Return the name of the device.""" + return self._name + + @property + def should_poll(self): + return False + + def push_data(self, data): + return True + + def parse_data(self, data): + return True \ No newline at end of file diff --git a/src/yamlparser.py b/src/yamlparser.py new file mode 100644 index 0000000..a5cbfe2 --- /dev/null +++ b/src/yamlparser.py @@ -0,0 +1,24 @@ +import yaml +import logging + +_LOGGER = logging.getLogger(__name__) + +def load_yaml(file): + try: + stram = open(file, "r") + yaml_data = yaml.load(stram) + return yaml_data + except Exception as e: + raise + _LOGGER.error("Can't load yaml with sids %r (%r)" % (file, e)) + +def get_gateway_password(config, ip=""): + if (config == None): + raise "Config is null" + configGateway = config.get("gateway", None) + if (configGateway == None): + raise "Config gateway is null" + password = configGateway.get("password", None) + if (password == None): + raise "Config gateway passowrd is null" + return password