Skip to content

Instantly share code, notes, and snippets.

@Nikolay-Kha
Last active January 31, 2018 12:32
Show Gist options
  • Save Nikolay-Kha/c07243c838a74ed0c9a2305d29560f97 to your computer and use it in GitHub Desktop.
Save Nikolay-Kha/c07243c838a74ed0c9a2305d29560f97 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python
import uuid
import json
import time
import random
import paho.mqtt.client as mqtt
SERVER_HOST = 'example.com'
ACCESS_TOKEN = 'AccessToken='
DEVICE_ID = 'mqtt-demo-device-' + ACCESS_TOKEN[0:4]
MPU6050_ADDRESS = 0x68 # default MPU6050 address
try:
import mpu6050
sensor = mpu6050.mpu6050(MPU6050_ADDRESS)
print("Using real MPU6050 sensor.")
except ImportError:
class FakeMPU6050(object):
def rand(self):
return round(-0.25 + 0.5 * random.random(), 2)
def get_temp(self):
return 25 + self.rand()
def get_gyro_data(self):
return {'x': self.rand(),
'y': self.rand(),
'z': self.rand()}
def get_accel_data(self):
return {'x': 9.8 + self.rand(),
'y': self.rand(),
'z': self.rand()}
sensor = FakeMPU6050()
print("Using MPU6050 emulator.")
class MQTTDemo(object):
def __init__(self, url, access_token, device_id):
self._client_id = str(uuid.uuid4())
self._connected = False
self._device_id = device_id
self._accessToken = access_token
self._client = mqtt.Client(self._client_id)
self._client.connect(url)
self._client.on_connect = self._on_connect
self._client.on_message = self._on_message
self._client.on_disconnect = self._on_disconnect
self._client.loop_start()
def _on_connect(self, client, userdata, flags, rc):
print('Connected with rc=%s' % rc)
client.subscribe('dh/response/authenticate@%s' % self._client_id)
self._publish('dh/request', {
'action': 'authenticate',
'token': self._accessToken,
'requestId': str(uuid.uuid4())
})
def _on_message(self, client, userdata, message):
print('New message: %s' % message.payload)
js = json.loads(message.payload)
if js['action'] == 'authenticate':
if js['status'] == 'success':
client.subscribe('dh/response/device/save@%s' % self._client_id)
client.subscribe('dh/response/notification/insert@%s' % self._client_id)
self._connected = True
else:
print("Failed to authenticate.")
def _on_disconnect(self, client, userdata, rc):
print('Disconnected with rc=%s' % rc)
self._connected = False
def _publish(self, topic, payload):
payload['requestId'] = str(uuid.uuid4())
self._client.publish(topic, json.dumps(payload))
def run(self):
while not self._connected:
time.sleep(0.01)
time.sleep(1.0)
self._publish('dh/request', {
'action': 'device/save',
'deviceId': self._device_id,
'device': {
'name': self._device_id
}
})
while self._connected:
self._publish('dh/request', {
'action': 'notification/insert',
'deviceId': self._device_id,
'notification': {
'notification': 'mpu6050',
'parameters':
{
'temperature': sensor.get_temp(),
'acceleration': sensor.get_accel_data(),
'gyro': sensor.get_gyro_data()
}
}
})
time.sleep(5)
if __name__ == '__main__':
d = MQTTDemo(SERVER_HOST, ACCESS_TOKEN, DEVICE_ID)
d.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment