python实现网关_用python实现与小米网关通讯

python 与小米网关通讯的三块内容：

以下内容的理解需要配合《绿米网关局域网通讯协议》使用

1、监听网关发出的组播信息：（有网关及连接设备的生命信号，事件信息）

2、读取需要获得的信息

3、控制连接设备（涉及了token加密部分）

1、upd广播监听小米网关的组播信息

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import socket

def get_gateway_heart():

SENDERIP = "0.0.0.0"

MYPORT = 9898

MYGROUP = '224.0.0.50'

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)

#allow multiple sockets to use the same PORT number

sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

#Bind to the port that we know will receive multicast data

sock.bind((SENDERIP,MYPORT))

#tell the kernel that we are a multicast socket

#sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 255)

#Tell the kernel that we want to add ourselves to a multicast group

#The address for the multicast group is the third param

status = sock.setsockopt(socket.IPPROTO_IP,

socket.IP_ADD_MEMBERSHIP,

socket.inet_aton(MYGROUP) + socket.inet_aton(SENDERIP));

#sock.setblocking(0)

#ts = time.time()

data, addr = sock.recvfrom(1024)

data_str=str(data,encoding='utf-8')

# sock.close()

return data_str

if __name__=='__main__':

while True:

tmp=get_gateway_heart()

print(tmp)

2、小米网关的初始密码向量转换为字符串的计算方法

from binascii import b2a_hex, a2b_hex

import sys

s='17996d093d28ddb3ba695a2e6f58562e' #初始向量

m=a2b_hex(s)

print(m)

#转换后的初始向量

#b'\x17\x99m\t=(\xdd\xb3\xbaiZ.oXV.'

3、加密token的方法

from Crypto.Cipher import AES

from binascii import b2a_hex, a2b_hex

class prpcrypt():

def __init__(self, key='cy5zmrpqws05vsqj'):

self.key = key #

self.mode = AES.MODE_CBC

# 加密函数，如果text不足16位就用空格补足为16位，

# 如果大于16当时不是16的倍数，那就补足为16的倍数。

def encrypt(self, text): #text是要加密的内容

cryptor = AES.new(self.key, self.mode,b'\x17\x99m\t=(\xdd\xb3\xbaiZ.oXV.')

# 这里密钥key 长度必须为16（AES-）,

# （AES-）,或者32 （AES-）Bytes 长度

# 目前AES- 足够目前使用

length =

count = len(text)

if count < length:

add = (length - count)

# \ backspace

text = text + ('\0' * add)

elif count > length:

add = (length - (count % length))

text = text + ('\0' * add)

self.ciphertext = cryptor.encrypt(text)

# 因为AES加密时候得到的字符串不一定是ascii字符集的，输出到终端或者保存时候可能存在问题

# 所以这里统一把加密后的字符串转化为16进制字符串

return str(b2a_hex(self.ciphertext),encoding='utf-8')

#return self.ciphertext

# 解密后，去掉补足的空格用strip() 去掉 b'

def decrypt(self, text):

cryptor = AES.new(self.key, self.mode, b'\x17\x99m\t=(\xdd\xb3\xbaiZ.oXV.')

plain_text = cryptor.decrypt(a2b_hex(text))

return plain_text.rstrip('\0')

if __name__ == '__main__':

pc = prpcrypt('0987654321qwerty') # 初始化密钥

e = pc.encrypt('1234567890abcdef') # 加密

# d = pc.decrypt(e) # 解密

print("加密:", e)

#print("解密:", d)

4、获取token，并对获取的token进行加密

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import socket

import json

from xm_gw.encrypty import prpcrypt

def get_token(): #通过get_id_list 获得token

ip_port_single = ("192.168.31.150", 9898)

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

comd = {'cmd': 'get_id_list'}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), ip_port_single)

data,addr=s.recvfrom(1024)

data_str=str(data,encoding='utf-8')

token=json.loads(data_str).get('token')

s.close()

return token

def get_token_encrypty():

tok = get_token() # 拿到当前token,要进行加密的内容

k = prpcrypt()

key_encrypt = k.encrypt(tok)

return key_encrypt

if __name__=='__main__':

tok=get_token()

tok_encry=get_token_encrypty()

print(tok)

print(tok_encry)

5、建立网关通讯并执行控制命令

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import socket

import json

from xm_gw import udp_token_key

class udp_gw():

def __init__(self, ip_gateway='192.168.8.100'):

self.ip_port_zu43 = ('224.0.0.50', 4321)

self.ip_port_single = (ip_gateway, 9898)

self.ip_port_zu9898=('224.0.0.50', 9898)

def whois(self):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

comd = {'cmd': 'whois'}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), self.ip_port_zu43)

data_bytes, addr = s.recvfrom(1024)

data_dic = json.loads(str(data_bytes, encoding='utf-8'))

s.close()

return data_dic

def get_id_list(self):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

comd = {'cmd': 'get_id_list'}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), self.ip_port_single)

data_bytes, addr = s.recvfrom(1024)

data_dic = json.loads(str(data_bytes, encoding='utf-8'))

s.close()

return data_dic

def read_sid(self, sid):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

comd = {'cmd': 'read', 'sid': sid}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), self.ip_port_single)

data_bytes, addr = s.recvfrom(1024)

data_dic = json.loads(str(data_bytes, encoding='utf-8'))

s.close()

return data_dic

def write_plug(self, status):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

key_encrypt = udp_token_key.get_token_encrypty()

comd = {"cmd": "write", "model": "plug", "sid": "158d0001b84d9a", "short_id": 46384,

"data": {"status": status, 'key': key_encrypt}}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), self.ip_port_single)

data_bytes, addr = s.recvfrom(1024)

data_dic = json.loads(str(data_bytes, encoding='utf-8'))

s.close()

return data_dic

def read_all_sid(self):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

ls = json.loads(self.get_id_list().get('data'))

ls_sensor_state = []

for sid in ls:

comd = {'cmd': 'read', 'sid': sid}

order = json.dumps(comd)

s.sendto(bytes(order, encoding="utf-8"), self.ip_port_single)

data_bytes, addr = s.recvfrom(1024)

data_dic = json.loads(str(data_bytes, encoding='utf-8'))

# print(data_dic)

ls_sensor_state.append(data_dic)

s.close()

return ls_sensor_state

def get_dict_model_sid(self):

dic_gw=self.whois()

ls=self.read_all_sid()

dic_model_sid = {}

for dic in ls:

model = dic.get('model')

sid = dic.get('sid')

dic_model_sid[model] = sid

dic_model_sid['gateway'] = dic_gw.get('sid')

return dic_model_sid

if __name__=='__main__':

import time

#{'plug': '158d0001b84d9a', 'switch': '158d0001c10bd7', 'sensor_ht': '158d0001e87bd9',

# 'magnet': '158d0001bb3daf', 'motion': '158d0001c2f110', 'gateway': '7811dcb38599'}

gw=udp_gw('192.168.31.150')

tmp = gw.read_sid('158d0001b84d9a')

# print(tmp1)

# time.sleep(5)

# gw.write_plug('off')

# time.sleep(5)

# tmp=gw.read_sid('158d0001b84d9a')

print(tmp)

python网络编程：TCP通讯模板、粘包及解决方案、自定义报头

一.TCP通讯模板二.远程CMD程序三.解决粘包问题四.解决粘包问题2 一.TCP通讯模板 TCP客户端 import socket c = socket.socket() # 连接服务器 c. ...

[Python]实现XMPP协议即时通讯发送消息功能

#-*- coding: utf-8 -*- __author__ = 'tsbc' import xmpp import time #注意帐号信息,必须加@域名格式 from_user = 'che ...

Python基础(四) socket简单通讯

socket:我们通常听过的套接字: 服务端: 1.创建socket对象 2.bing 绑定ip及端口 3.对该端口进行监听 4.消息阻塞(等待客户端消息) 客户端: 1.创建socket对象 2.连 ...

python处理多线程之间事件通讯方法

一.什么是事件每执行一个事情,肯定有该事情的执行后状态,那事件就是该事情发生的信号在程序中,多线程之间需要通讯,而事件就是方便线程之间的通讯案例: 1.服务器启动需要5秒 2.客服端启动后去链接 ...

用Robot Framework+python来测试基于socket通讯的C/S系统（网络游戏）

项目终于换了方案,改用socket来实现而不是之前的http了,所以测试工具就不能用以前的了,因为测试人手少,逼不得已的必须要挖掘更多的自动化方案来弥补.于是先研究了下python的socket解决方 ...

小米网关api

http://bbs.xiaomi.cn/t-13198850 https://github.com/snOOrz/homebridge-aqara/blob/master/README.md htt ...

TCP/IP协议，TCP与平台通信，通讯协议压力测试（python）

最近的项目来了一个需求,要求测试tcp网关通讯协议: 1.液压井盖通过TCP/IP TCP与平台通信: 2.硬件定期发送心跳包(10S)给平台,是平台与硬件保持长连接: 3.每台硬件有一个12字节的唯 ...

python中利用队列asyncio&period;Queue进行通讯详解

python中利用队列asyncio.Queue进行通讯详解本文主要给大家介绍了关于python用队列asyncio.Queue通讯的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细 ...

【XMLRPC实现跨语言编程】Tcl &lt&semi;----&gt&semi; python

转载请声明出处,谢谢合作. # 期望一种能实现tcl.python两者解释器能双向通讯的结合体# py->tcl: from Tkinter import Tcl; tcl = Tcl(); t ...

随机推荐

MyISAM 和InnoDB区别

MyISAM 和InnoDB 讲解 InnoDB和MyISAM是许多人在使用MySQL时最常用的两个表类型,这两个表类型各有优劣,视具体应用而定.基本的差别为:MyISAM类型不支持事务处理等高级处理 ...

9款经典华丽的CSS3分享按钮

如果你经常活跃在一些社交网站上,那么你肯定会看到过很多形式各异的分享按钮,目前由于HTML5和CSS3的普及,很多分享按钮也都应用了CSS3样式,甚至会有很多带有动画的CSS3分享按钮.本文就向大家介 ...

mysqldump的常用语句及各参数详解

mysqldump的常用语句及各参数详解分类: MySQL 2011-01-11 17:55 1368人阅读评论(0) 收藏举报数据库mysql服务器tableinsertdatabase m ...

Chapter 2 Open Book——33

My chin raised a fraction. 我的下巴抬起来了一点. 我略微抬起下颚. "No, she did not send me here. I sent myself.&q ...

【ASP&period;NET MVC 学习笔记】- 11 Controller和Action（2）

本文参考:http://www.cnblogs.com/willick/p/3331513.html 1.MVC一个请求的发出至action返回结果的流程图如下: 重点是Controller Fact ...

XCopy 小技巧

使用XCOPY Copy 一个文件时,如果目标地址没有对应的文件, 系统会提示选择是文件,还是目录,如下图所示. 有时我们不想出现这个提示,这是只需要修改目标文件的写法.如下将 "D:\t ...

xpath详细讲解

什么是XML XML 指可扩展标记语言(EXtensible Markup Language) XML 是一种标记语言,很类似 HTML XML 的设计宗旨是传输数据,而非显示数据 XML 的标签需要 ...

JDK 在 Linux 上安装的详细过程

1.下载JDK Linux上一般会安装Open JDK,如果有系统预装了OpenJDK的话需要先卸载掉OpenJDK, 卸载过程可以参考这里:https://www.cnblogs.com/sxdcg ...

MYSQL 查看最大连接数和修改最大连接数

MySQL查看最大连接数和修改最大连接数 1.查看最大连接数show variables like '%max_connections%';2.修改最大连接数set GLOBAL max_connec ...

生产者与消费者代码实现 java

首先,我利用忙测试写出了第一次版本的代码 package How; //自写代码缺陷无法完全实现pv操作线程处于忙测试状态 public class bin_1_1 { public static ...

python实现网关_用python实现与小米网关通讯相关推荐

python 时间序列预测_使用Python进行动手时间序列预测
python 时间序列预测 Time series analysis is the endeavor of extracting meaningful summary and statistical ...
python 概率分布模型_使用python的概率模型进行公司估值
python 概率分布模型 Note from Towards Data Science's editors: While we allow independent authors to publis ...
python集群_使用Python集群文档
python集群 Natural Language Processing has made huge advancements in the last years. Currently, variou ...
python 网页编程_通过Python编程检索网页
python 网页编程 The internet and the World Wide Web (WWW), is probably the most prominent source of info ...
python机器学习预测_使用Python和机器学习预测未来的股市趋势
python机器学习预测 Note from Towards Data Science's editors: While we allow independent authors to publish ...
python高斯求和_利用Python进行数据分析(3)- 列表、元组、字典、集合
本文主要是对Python的数据结构进行了一个总结,常见的数据结构包含:列表list.元组tuple.字典dict和集合set. image 索引左边0开始,右边-1开始通过index()函数查看索 ...
python 免费空间_用python做大数据
不学Python迟早会被淘汰?Python真有这么好的前景? 最近几年Python编程语言在国内引起不小的轰动,有超越Java之势,本来在美国这个编程语言就是最火的,应用的非常非常的广泛,而Pytho ...
python希腊字母字符串_#10 Python字符串
前言通过上一节可知,Python6个序列的内置类型中,最常见的是列表和元组,但在Python中,最常用的数据类型却不是列表和元组,而是字符串.要想深入了解字符串,必须先掌握字符编码问题.因此本篇博文 ...
使用python预测基金_使用python先知3 1创建预测
使用python预测基金 This tutorial was created to democratize data science for business users (i.e., minimiz ...

python实现网关_用python实现与小米网关通讯

python实现网关_用python实现与小米网关通讯相关推荐

最新文章

热门文章