|
|
"""
|
|
|
A small Test application to show how to use Flask-MQTT.
|
|
|
|
|
|
"""
|
|
|
import os
|
|
|
import time
|
|
|
import logging
|
|
|
from datetime import datetime
|
|
|
import random
|
|
|
import eventlet
|
|
|
import json
|
|
|
from json import JSONEncoder
|
|
|
import numpy
|
|
|
import base64
|
|
|
import h5py
|
|
|
try:
|
|
|
import requests
|
|
|
except:
|
|
|
pass
|
|
|
from bs4 import BeautifulSoup
|
|
|
from flask import Flask, render_template, jsonify, request, redirect
|
|
|
from flask_mqtt import Mqtt
|
|
|
from flask_socketio import SocketIO
|
|
|
from flask_bootstrap import Bootstrap
|
|
|
|
|
|
eventlet.monkey_patch()
|
|
|
|
|
|
app = Flask(__name__)
|
|
|
app.config['SECRET'] = 'my secret key'
|
|
|
app.config['TEMPLATES_AUTO_RELOAD'] = True
|
|
|
app.config['MQTT_BROKER_URL'] = os.environ['BROKER_URL']
|
|
|
app.config['MQTT_BROKER_PORT'] = 1883
|
|
|
app.config['MQTT_CLIENT_ID'] = 'flask_mqtt_{}'.format(int(random.random()*100))
|
|
|
app.config['MQTT_CLEAN_SESSION'] = True
|
|
|
app.config['MQTT_USERNAME'] = ''
|
|
|
app.config['MQTT_PASSWORD'] = ''
|
|
|
app.config['MQTT_KEEPALIVE'] = 5
|
|
|
app.config['MQTT_TLS_ENABLED'] = False
|
|
|
app.config['MQTT_LAST_WILL_TOPIC'] = 'sophy/error'
|
|
|
app.config['MQTT_LAST_WILL_MESSAGE'] = 'Bye from flask'
|
|
|
app.config['MQTT_LAST_WILL_QOS'] = 1
|
|
|
|
|
|
mqtt = Mqtt(app)
|
|
|
|
|
|
socketio = SocketIO(app)
|
|
|
bootstrap = Bootstrap(app)
|
|
|
|
|
|
MODE_STOP = 0x00
|
|
|
MODE_SPEED = 0x01
|
|
|
MODE_POSITION = 0x02
|
|
|
RX_AZIMUTH = 0x27
|
|
|
RX_ELEVATION = 0x47
|
|
|
TX_AZIMUTH = 0x25
|
|
|
TX_ELEVATION = 0x45
|
|
|
NONE_AXIS = 0x65
|
|
|
RX_FUNCTION = 0x30
|
|
|
TX_FUNCTION = 0x40
|
|
|
HEADER = 0x7e
|
|
|
MIN_SPEED = -180.0
|
|
|
MAX_SPEED = 180.0
|
|
|
SHRT_MIN = -32768
|
|
|
SHRT_MAX = 32768
|
|
|
USHRT_MAX = 65535
|
|
|
|
|
|
DATA_PATH = '/data'
|
|
|
RUNNING = False
|
|
|
EXPERIMENT = {'name': 'test'}
|
|
|
DEBUG = False
|
|
|
|
|
|
class NumpyArrayEncoder(JSONEncoder):
|
|
|
def default(self, obj):
|
|
|
if isinstance(obj, numpy.ndarray):
|
|
|
return obj.tolist()
|
|
|
return JSONEncoder.default(self, obj)
|
|
|
|
|
|
class HDF5File():
|
|
|
|
|
|
def __init__(self):
|
|
|
self.first = True
|
|
|
self.data_pos = {'az':[], 'el':[], 'saz':[], 'sel':[], 'timestamp':[]}
|
|
|
self.pos_ready = True
|
|
|
self.pos_time = 0
|
|
|
self.raw = {}
|
|
|
self.timestamp = 0
|
|
|
self.metadata = None
|
|
|
|
|
|
def reset(self):
|
|
|
self.first = True
|
|
|
self.data_pos = {'az':[], 'el':[], 'saz':[], 'sel':[], 'timestamp':[]}
|
|
|
self.pos_time = 0
|
|
|
self.raw = {}
|
|
|
self.timestamp = 0
|
|
|
|
|
|
def update(self, data):
|
|
|
|
|
|
for var in ('az', 'el', 'saz', 'sel'):
|
|
|
tmp = numpy.array(data[var][:99])[::4].tolist()
|
|
|
while len(tmp)<25:
|
|
|
tmp.append(numpy.nan)
|
|
|
self.data_pos[var].extend(tmp)
|
|
|
|
|
|
self.data_pos['timestamp'].extend(numpy.linspace(int(data['timestamp'])-1, int(data['timestamp'])-0.04, 25).tolist())
|
|
|
self.timestamp = int(data['timestamp'])
|
|
|
|
|
|
def write(self, data):
|
|
|
|
|
|
if self.first:
|
|
|
self.pos_time = int(data['timestamp'])
|
|
|
self.first = False
|
|
|
|
|
|
self.raw = data
|
|
|
self.update(data)
|
|
|
|
|
|
if self.pos_ready:
|
|
|
filex = "pos@%10.3f.h5" % (self.pos_time)
|
|
|
date_folder = datetime.fromtimestamp(self.pos_time).strftime('%Y-%m-%dT%H-00-00')
|
|
|
filename = os.path.join(DATA_PATH, EXPERIMENT['name'], 'position', date_folder, filex)
|
|
|
if not os.path.exists(os.path.dirname(filename)):
|
|
|
path = os.path.dirname(filename)
|
|
|
os.makedirs(path)
|
|
|
self.fpos = h5py.File(filename, 'w')
|
|
|
self.pos_ready = False
|
|
|
|
|
|
if datetime.fromtimestamp(self.timestamp).second == 0:
|
|
|
self.pos_time = self.timestamp
|
|
|
grp = self.fpos.create_group("Data")
|
|
|
dset = grp.create_dataset("azi_pos", data=numpy.array(self.data_pos['az']))
|
|
|
dset = grp.create_dataset("ele_pos", data=numpy.array(self.data_pos['el']))
|
|
|
dset = grp.create_dataset("azi_speed", data=numpy.array(self.data_pos['saz']))
|
|
|
dset = grp.create_dataset("ele_speed", data=numpy.array(self.data_pos['sel']))
|
|
|
dset = grp.create_dataset("utc", data=numpy.array(self.data_pos['timestamp']))
|
|
|
self.fpos.close()
|
|
|
self.data_pos['az'] = []
|
|
|
self.data_pos['el'] = []
|
|
|
self.data_pos['saz'] = []
|
|
|
self.data_pos['sel'] = []
|
|
|
self.data_pos['timestamp'] = []
|
|
|
self.pos_ready = True
|
|
|
|
|
|
if DEBUG:
|
|
|
date_folder = datetime.fromtimestamp(self.timestamp).strftime('%Y-%m-%dT%H-00-00')
|
|
|
filelog = os.path.join(DATA_PATH, EXPERIMENT['name'], 'position', date_folder, "pos@%10.3f.txt" % (self.timestamp))
|
|
|
f = open(filelog, 'w')
|
|
|
f.write(json.dumps(self.raw, cls=NumpyArrayEncoder))
|
|
|
f.close()
|
|
|
|
|
|
|
|
|
HDF = HDF5File()
|
|
|
|
|
|
def getSpeedPosition(msg_b64):
|
|
|
AzPosition=[]
|
|
|
AzSpeed=[]
|
|
|
ElPosition=[]
|
|
|
ElSpeed=[]
|
|
|
raw = numpy.frombuffer(base64.decodebytes(msg_b64.encode()),numpy.dtype('B'))
|
|
|
raw_size = len(raw)
|
|
|
Timestamp = (raw[raw_size-4])|(raw[raw_size-3]<<8)|(raw[raw_size-2]<<16)|(raw[raw_size-1]<<24)
|
|
|
# Timestamp = time.time()
|
|
|
counter = 0
|
|
|
while counter < raw_size-4:
|
|
|
if raw[counter]==HEADER:
|
|
|
if raw[counter+1]==RX_FUNCTION:
|
|
|
if raw[counter+2]==RX_AZIMUTH or raw[counter+2]==RX_ELEVATION:
|
|
|
iangle = 0.0
|
|
|
ispeed = 0.0
|
|
|
hadstuffing = 0
|
|
|
|
|
|
if (counter+hadstuffing+4<raw_size-4):
|
|
|
if raw[counter+hadstuffing+4]==HEADER-1:
|
|
|
hadstuffing+=1
|
|
|
iangle = int(0x70|raw[counter+hadstuffing+4]&0x0F)
|
|
|
else:
|
|
|
iangle = int(raw[counter+hadstuffing+4])
|
|
|
else:
|
|
|
logging.debug("Warning: Index out of bounds. The packet is incomplete or corrupted.")
|
|
|
break
|
|
|
|
|
|
if (counter+hadstuffing+5<raw_size-4):
|
|
|
if raw[counter+hadstuffing+5]==HEADER-1:
|
|
|
hadstuffing+=1
|
|
|
iangle = iangle + int((0x70|raw[counter+hadstuffing+5]&0x0F)<<8)
|
|
|
else:
|
|
|
iangle = iangle + int(raw[counter+hadstuffing+5]<<8)
|
|
|
else:
|
|
|
logging.debug("Warning: Index out of bounds. The packet is incomplete or corrupted.")
|
|
|
break
|
|
|
|
|
|
if (counter+hadstuffing+6<raw_size-4):
|
|
|
if raw[counter+hadstuffing+6]==HEADER-1:
|
|
|
hadstuffing+=1
|
|
|
ispeed = int(0x70|raw[counter+hadstuffing+6]&0x0F)
|
|
|
else:
|
|
|
ispeed = int(raw[counter+hadstuffing+6])
|
|
|
else:
|
|
|
logging.debug("Warning: Index out of bounds. The packet is incomplete or corrupted.")
|
|
|
break
|
|
|
|
|
|
if (counter+hadstuffing+7<raw_size-4):
|
|
|
if raw[counter+hadstuffing+7]==HEADER-1:
|
|
|
hadstuffing+=1
|
|
|
ispeed = ispeed + int((0x70|raw[counter+hadstuffing+7]&0x0F)<<8)
|
|
|
else:
|
|
|
ispeed = ispeed + int(raw[counter+hadstuffing+7]<<8)
|
|
|
else:
|
|
|
logging.debug("Warning: Index out of bounds. The packet is incomplete or corrupted.")
|
|
|
break
|
|
|
|
|
|
if (counter+2<raw_size-4):
|
|
|
if raw[counter+2]==RX_AZIMUTH:
|
|
|
AzPosition.append(iangle*360.0/USHRT_MAX)
|
|
|
AzSpeed.append(((ispeed-SHRT_MIN)*(MAX_SPEED-MIN_SPEED)/(SHRT_MAX-SHRT_MIN))+MIN_SPEED)
|
|
|
elif raw[counter+2]==RX_ELEVATION:
|
|
|
ElPosition.append(iangle*360.0/USHRT_MAX)
|
|
|
ElSpeed.append(((ispeed-SHRT_MIN)*(MAX_SPEED-MIN_SPEED)/(SHRT_MAX-SHRT_MIN))+MIN_SPEED)
|
|
|
else:
|
|
|
counter+=1
|
|
|
continue
|
|
|
else:
|
|
|
logging.debug("Warning: Index out of bounds. The packet is incomplete or corrupted.")
|
|
|
break
|
|
|
|
|
|
counter = counter+hadstuffing+13
|
|
|
else:
|
|
|
counter+=1
|
|
|
continue
|
|
|
else:
|
|
|
counter+=1
|
|
|
continue
|
|
|
else:
|
|
|
counter+=1
|
|
|
continue
|
|
|
|
|
|
az = numpy.array(AzPosition)
|
|
|
az[numpy.where(az>=359.8)] = 0
|
|
|
az = numpy.round(az, 2)
|
|
|
|
|
|
saz = numpy.array(AzSpeed)
|
|
|
saz[numpy.where(saz>=28)] = saz[(saz>=28)]-360
|
|
|
saz = numpy.round(saz, 2)
|
|
|
|
|
|
el = numpy.array(ElPosition)
|
|
|
el[numpy.where((190 <= el) & (el <= 360))] = el[(190 <= el) & (el<= 360)] - 360
|
|
|
el = numpy.round(el, 2)
|
|
|
|
|
|
sel = numpy.array(ElSpeed)
|
|
|
sel[numpy.where(sel>=28)] = sel[(sel>=28)]-360
|
|
|
sel = numpy.round(sel, 2)
|
|
|
|
|
|
return az, saz, el, sel, int(Timestamp)
|
|
|
|
|
|
|
|
|
@app.route('/')
|
|
|
def index():
|
|
|
global RUNNING, EXPERIMENT
|
|
|
running = 'Running' if RUNNING else 'Not Running'
|
|
|
name = '-' if EXPERIMENT['name']=='test' else EXPERIMENT['name']
|
|
|
return render_template('index.html', status=running, experiment_name=name)
|
|
|
|
|
|
@app.route('/start', methods=['POST'])
|
|
|
def start_proc():
|
|
|
|
|
|
global EXPERIMENT, RUNNING
|
|
|
|
|
|
EXPERIMENT.update(request.get_json())
|
|
|
RUNNING = True
|
|
|
path = os.path.join(DATA_PATH, EXPERIMENT['name'])
|
|
|
if not os.path.exists(path):
|
|
|
os.makedirs(path)
|
|
|
fo = open(os.path.join(path, 'experiment.conf'), 'w')
|
|
|
fo.write(json.dumps(EXPERIMENT, indent=2))
|
|
|
fo.close()
|
|
|
|
|
|
mqtt.publish('sophy/control', json.dumps(EXPERIMENT))
|
|
|
socketio.emit('mqtt_message', data={'topic':'monitor', 'status': 'Running', 'name': EXPERIMENT['name']})
|
|
|
|
|
|
return jsonify({'start': 'ok'})
|
|
|
|
|
|
@app.route('/stop')
|
|
|
def stop_proc():
|
|
|
|
|
|
global RUNNING, DEBUG, HDF, EXPERIMENT
|
|
|
|
|
|
RUNNING = False
|
|
|
DEBUG = False
|
|
|
HDF.reset()
|
|
|
socketio.emit('mqtt_message', data={'topic':'monitor', 'status': 'Not Running'})
|
|
|
mqtt.publish('sophy/control', json.dumps({'stop': 'ok'}))
|
|
|
|
|
|
return jsonify({'stop': 'ok'})
|
|
|
|
|
|
@app.route('/status')
|
|
|
def status_proc():
|
|
|
|
|
|
global RUNNING
|
|
|
|
|
|
return jsonify({'status': RUNNING})
|
|
|
|
|
|
@app.route('/run')
|
|
|
def run_proc():
|
|
|
|
|
|
global RUNNING, EXPERIMENT, DEBUG
|
|
|
if request.args.get('debug', False):
|
|
|
DEBUG = True
|
|
|
name = request.args.get('name', 'TEST')
|
|
|
path = os.path.join(DATA_PATH, name)
|
|
|
if not os.path.exists(path):
|
|
|
os.makedirs(path)
|
|
|
EXPERIMENT['name'] = name
|
|
|
RUNNING = True
|
|
|
|
|
|
mqtt.publish('sophy/control', json.dumps(EXPERIMENT))
|
|
|
socketio.emit('mqtt_message', data={'topic':'monitor', 'status': 'Running'})
|
|
|
|
|
|
return redirect('/')
|
|
|
|
|
|
@app.route('/mqtt')
|
|
|
def mqtt_log():
|
|
|
|
|
|
return render_template('mqtt.html')
|
|
|
|
|
|
@app.route('/acq')
|
|
|
def acq_log():
|
|
|
|
|
|
return render_template('acq.html')
|
|
|
|
|
|
@socketio.on('publish')
|
|
|
def handle_publish(json_str):
|
|
|
data = json.loads(json_str)
|
|
|
mqtt.publish(data['topic'], data['message'], data['qos'])
|
|
|
|
|
|
@socketio.on('subscribe')
|
|
|
def handle_subscribe(json_str):
|
|
|
data = json.loads(json_str)
|
|
|
mqtt.subscribe(data['topic'], data['qos'])
|
|
|
|
|
|
@socketio.on('unsubscribe_all')
|
|
|
def handle_unsubscribe_all():
|
|
|
mqtt.unsubscribe_all()
|
|
|
|
|
|
@mqtt.on_connect()
|
|
|
def handle_connect(client, userdata, flags, rc):
|
|
|
mqtt.subscribe(os.environ.get('SOPHY_TOPIC', 'sophy/#'))
|
|
|
|
|
|
@mqtt.on_message()
|
|
|
def handle_mqtt_message(client, userdata, message):
|
|
|
|
|
|
global RUNNING, HDF
|
|
|
|
|
|
payload = message.payload.decode()
|
|
|
data = {}
|
|
|
|
|
|
if 'pedestal' in message.topic:
|
|
|
|
|
|
az, saz, el, sel, tm = getSpeedPosition(payload)
|
|
|
times = numpy.linspace(tm, tm+0.9, 10)
|
|
|
data['az'] = az[:100][::10].tolist()
|
|
|
data['el'] = el[:100][::10].tolist()
|
|
|
data['saz'] = saz[:100][::10].tolist()
|
|
|
data['sel'] = sel[:100][::10].tolist()
|
|
|
data['timestamp'] = times.tolist()
|
|
|
|
|
|
if RUNNING:
|
|
|
HDF.write({'az':az.tolist(), 'el':el.tolist(), 'saz':saz.tolist(), 'sel':sel.tolist(), 'timestamp':tm})
|
|
|
|
|
|
data['payload'] = payload
|
|
|
data['topic'] = message.topic
|
|
|
data['qos'] = message.qos
|
|
|
|
|
|
socketio.emit('mqtt_message', data=data)
|
|
|
|
|
|
@mqtt.on_log()
|
|
|
def handle_logging(client, userdata, level, buf):
|
|
|
# print(level, buf)
|
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
socketio.run(app, host='0.0.0.0', port=5000, use_reloader=False, debug=True)
|
|
|
|
|
|
|
