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.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 ...
總結
以上是生活随笔為你收集整理的python实现网关_用python实现与小米网关通讯的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： HDOJ树形DP专题之Centroid
- 下一篇： python自动交易源码_【硬核福利】量