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.Queue进行通讯详解
python中利用队列asyncio.Queue进行通讯详解 本文主要给大家介绍了关于python用队列asyncio.Queue通讯的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细 …
【XMLRPC实现跨语言编程】Tcl <;—->; 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.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 …