peterlu14
462 lines
16 KiB
Python
462 lines
16 KiB
Python
import base64
|
|
import re
|
|
from json import loads as json_parse, dumps as _json_stringify
|
|
from typing import Dict, Optional, Any
|
|
|
|
from biopro.controller import ComplexResponse
|
|
from biopro.devlib.menu import Menu
|
|
from biopro.file.segment import FileSegment, FileSegmentContent
|
|
from biopro.file.util import RootInfo, DirectoryInfo, FileInfo
|
|
from biopro.recording.file import RecordingMetaFile
|
|
from biopro.message.request import DataMessageHandler
|
|
from biopro.util.console import MAGENTA
|
|
from biopro.util.json import JSON_OBJECT
|
|
from biopro.util.stack import print_exception
|
|
from biopro.util.text import part_prefix
|
|
from .cache import FileHandle
|
|
from .controller import ControlServerAPI, HardwareInfo
|
|
from .dispatcher import APIDispatcher
|
|
from .socket import ServerThread
|
|
import threading
|
|
|
|
# try:
|
|
import paho.mqtt.client as mqtt
|
|
# except ImportError:
|
|
# Client = None
|
|
# MQTTMessage = None
|
|
|
|
_RUNTIME_COMPILE = False
|
|
|
|
|
|
def json_stringify(o: Any) -> str:
|
|
return _json_stringify(o, separators=(',', ':'))
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class MqttThread(threading.Thread):
|
|
"""Mqtt server thread
|
|
|
|
**Subscribe Topic**
|
|
|
|
**Publish Topic**
|
|
|
|
**Publish Content Format**
|
|
|
|
::
|
|
|
|
response
|
|
= json_data
|
|
| error_message
|
|
| file_segment
|
|
|
|
json_data = {
|
|
'data'?: JSON,
|
|
'binary'?: str # base64 encode
|
|
'cookie'?: JSON_OBJECT
|
|
}
|
|
|
|
error_message = {
|
|
'error' = {
|
|
'type': str
|
|
'message': str
|
|
}
|
|
}
|
|
|
|
"""
|
|
|
|
def __init__(self, server: ControlServerAPI, url: str, port: int, client_id: str, log_verbose, name, alive=60, is_sub=True,):
|
|
# super().__init__('MQTT-' + client_id, MAGENTA)
|
|
super(MqttThread, self).__init__(name = name)
|
|
|
|
from ._identify import controller_device_id
|
|
self.__controller_ID = controller_device_id()
|
|
|
|
self._mqtt_url = url
|
|
self._mqtt_port = port
|
|
self._mqtt_alive = alive
|
|
self._mqtt_client = None
|
|
self._mqtt_client_local = None
|
|
self._client_id = client_id
|
|
self.log_verbose = log_verbose
|
|
|
|
if self._client_id is 'control_server':
|
|
self._handler = server
|
|
self._dispatcher = APIDispatcher(ControlServerAPI, server)
|
|
else:
|
|
self._handler = None
|
|
self._dispatcher = None
|
|
|
|
self.controller_available = False
|
|
self.is_sub = is_sub
|
|
|
|
|
|
@property
|
|
def mqtt_url(self) -> str:
|
|
return self._mqtt_url
|
|
|
|
@property
|
|
def mqtt_port(self) -> int:
|
|
return self._mqtt_port
|
|
|
|
def run(self) -> None:
|
|
if mqtt.Client is None:
|
|
if _RUNTIME_COMPILE:
|
|
raise RuntimeError('remote server is disabled')
|
|
else:
|
|
raise RuntimeError('python module paho.mqtt not found : '
|
|
'please type "sudo pip install paho-mqtt==1.4.0"')
|
|
|
|
self.log_verbose('use controller ID', self.__controller_ID)
|
|
|
|
# super().setup()
|
|
|
|
# self._mqtt_client = Client( self.__controller_ID + '_' + self._client_id )
|
|
# self._mqtt_client = mqtt.Client( self.__controller_ID + '_' + self._client_id )
|
|
|
|
# self.log_verbose('cloud-server: ' + '%s:%d' % (self._mqtt_url, self._mqtt_port))
|
|
|
|
# self._mqtt_client.connect_async(self._mqtt_url, self._mqtt_port, self._mqtt_alive)
|
|
# self._mqtt_client.on_connect = self.on_connect
|
|
# self._mqtt_client.on_disconnect = self.on_disconnect
|
|
# self._mqtt_client.on_message = self.on_message
|
|
|
|
# self._mqtt_client_local = Client( self.__controller_ID + '_' + self._client_id + '_local' )
|
|
self._mqtt_client_local = mqtt.Client( self.__controller_ID + '_' + self._client_id + '_local' )
|
|
self.log_verbose('localhost: 127.0.0.1:1883')
|
|
self._mqtt_client_local.connect_async('127.0.0.1', self._mqtt_port, self._mqtt_alive)
|
|
self._mqtt_client_local.on_connect = self.on_connect_local
|
|
self._mqtt_client_local.on_disconnect = self.on_disconnect_local
|
|
self._mqtt_client_local.on_message = self.on_message
|
|
|
|
self.routine()
|
|
|
|
def shutdown(self) -> None:
|
|
if self._mqtt_client is not None:
|
|
self._mqtt_client.loop_stop()
|
|
self._mqtt_client.disconnect()
|
|
if self._mqtt_client_local is not None:
|
|
self._mqtt_client_local.loop_stop()
|
|
self._mqtt_client_local.disconnect()
|
|
|
|
def routine(self, event = None) -> None:
|
|
if self._mqtt_client_local is not None:
|
|
self._mqtt_client_local.loop_start()
|
|
if self._mqtt_client is not None:
|
|
self._mqtt_client.loop_start()
|
|
# self._mqtt_client_local.loop_start()
|
|
# self._mqtt_client.loop_start()
|
|
|
|
def on_connect(self, client: mqtt.Client, user_data, flags: Dict[str, Any], rc: int):
|
|
self.log_verbose('cloud-server connected')
|
|
topic = self.__controller_ID + '_user'
|
|
if self.is_sub:
|
|
self.log_verbose('cloud-server subscribe', topic)
|
|
self._mqtt_client.subscribe(topic)
|
|
|
|
def on_connect_local(self, client: mqtt.Client, user_data, flags: Dict[str, Any], rc: int):
|
|
self.log_verbose('localhost connected')
|
|
topic = self.__controller_ID + '_user'
|
|
if self.is_sub:
|
|
self.log_verbose('localhost subscribe', topic)
|
|
self._mqtt_client_local.subscribe(topic)
|
|
|
|
def on_disconnect(self, client: mqtt.Client, user_data, rc):
|
|
self.log_verbose('cloud-server disconnected')
|
|
self.sleep(3)
|
|
self._mqtt_client.reconnect()
|
|
|
|
def on_disconnect_local(self, client: mqtt.Client, user_data, rc):
|
|
self.log_verbose('localhost disconnected')
|
|
self.sleep(3)
|
|
self._mqtt_client_local.reconnect()
|
|
|
|
def publish(self, topic: str, payload: str, inter = False, analysis=False, qos = 2):
|
|
if inter:
|
|
_topic = self.__controller_ID + '_user'
|
|
elif analysis:
|
|
_topic = self.__controller_ID + '_data_analysis/get_analysis_data'
|
|
else:
|
|
_topic = self.__controller_ID + '/' + topic
|
|
if self._mqtt_client_local is not None:
|
|
info = self._mqtt_client_local.publish(_topic, payload, qos = qos)
|
|
if self._mqtt_client is not None and inter == False:
|
|
info = self._mqtt_client.publish(_topic, payload, qos = qos)
|
|
|
|
def broadcast_command(self, command: str, device: Optional[int] = None):
|
|
if self._mqtt_client_local is not None:
|
|
if command == 'state:device_available':
|
|
self.controller_available = True
|
|
self._mqtt_client_local.publish(self.__controller_ID + '/broadcast', command)
|
|
if self._mqtt_client is not None:
|
|
if command == 'state:device_available':
|
|
self.controller_available = True
|
|
self._mqtt_client.publish(self.__controller_ID + '/broadcast', command)
|
|
|
|
def on_message(self, client: mqtt.Client, user_data, message):
|
|
json = json_parse(message.payload.decode())
|
|
|
|
method = json.get('header', None)
|
|
|
|
if self._client_id is 'data_server':
|
|
if 'hardware_scan' in str(method):
|
|
self.broadcast_command('scan:start')
|
|
return None
|
|
|
|
if self.controller_available is False:
|
|
return None
|
|
|
|
if 'file_info' in str(method):
|
|
if json.get('path', None) in '':
|
|
return None
|
|
|
|
device = json.get('device', None)
|
|
|
|
self.log_verbose('Request:', method, 'Data: ', str(json))
|
|
|
|
if method is None or len(method) == 0:
|
|
return
|
|
|
|
else:
|
|
argument = dict(json)
|
|
del argument['header']
|
|
|
|
response_topic = self.__controller_ID + '/' + method
|
|
|
|
try:
|
|
result = self._on_command(method, argument)
|
|
except BaseException as e:
|
|
print_exception()
|
|
|
|
response = {
|
|
'error': {
|
|
'type': e.__class__.__name__,
|
|
'message': str(e)
|
|
}
|
|
}
|
|
|
|
if self._mqtt_client_local is not None:
|
|
self._mqtt_client_local.publish(response_topic, json_stringify(response))
|
|
if self._mqtt_client is not None:
|
|
self._mqtt_client.publish(response_topic, json_stringify(response))
|
|
|
|
else:
|
|
cookie = None
|
|
|
|
if isinstance(result, ComplexResponse):
|
|
cookie = result.response_cookie
|
|
result = result.response
|
|
response = {
|
|
'data': result
|
|
}
|
|
|
|
elif isinstance(result, bytes):
|
|
response = {
|
|
'binary': base64.b64encode(result).decode()
|
|
}
|
|
elif isinstance(result, list):
|
|
result = [d.as_json() for d in result]
|
|
response = {
|
|
'data': result
|
|
}
|
|
elif isinstance(result, (HardwareInfo, Menu, RootInfo, DirectoryInfo, FileInfo, FileHandle, RecordingMetaFile)):
|
|
result = result.as_json()
|
|
response = {
|
|
'data': result
|
|
}
|
|
elif isinstance(result, (FileSegment, FileSegmentContent)):
|
|
result = result.as_json()
|
|
try:
|
|
_index = re.search(r'/export', str(result['path']) ).start() + 8
|
|
result['path'] = result['path'].as_posix()[_index:]
|
|
except BaseException as e:
|
|
result['path'] = result['path'].as_posix()[35:]
|
|
# print(e)
|
|
response = {
|
|
'data': result
|
|
}
|
|
else:
|
|
response = {
|
|
'data': result
|
|
}
|
|
|
|
if cookie is not None:
|
|
response['cookie'] = cookie
|
|
if device is not None:
|
|
response['device'] = device
|
|
|
|
if self._mqtt_client_local is not None:
|
|
self._mqtt_client_local.publish(response_topic, json_stringify(response))
|
|
if self._mqtt_client is not None:
|
|
self._mqtt_client.publish(response_topic, json_stringify(response))
|
|
|
|
# tmp = re.split('/', method)
|
|
# if tmp[0] == "device_parameter" and tmp[1] == "0":
|
|
# device_id = str(json.get('device', None))
|
|
# client.publish(self.__controller_ID + '/broadcast',
|
|
# "message:" + device_id + ":" + json_stringify(response))
|
|
self.log_verbose('*- mes: ' + str(json) + ', res: ' + json_stringify(response) + ', topic:' + response_topic)
|
|
|
|
BLACK_LIST = {
|
|
# 'hardware_datetime',
|
|
'broadcast_command',
|
|
'device_internal_command',
|
|
'websocket_data_request',
|
|
# 'file_handle_data',
|
|
# 'file_handle_data_start',
|
|
# 'file_handle_data_stop'
|
|
|
|
}
|
|
|
|
def _on_command(self, method: str, options: Dict[str, Any]) -> Any:
|
|
if self._client_id is not 'control_server':
|
|
return None
|
|
|
|
method, _ = part_prefix(method, '/', missing=method)
|
|
# print( method, options )
|
|
|
|
if method in self.BLACK_LIST:
|
|
return None
|
|
|
|
if method == 'file_segment_info':
|
|
return self._handler.file_segment('info', **options)
|
|
|
|
elif method == 'file_segment_new':
|
|
return self._handler.file_segment('new', **options)
|
|
|
|
elif method == 'file_segment_reload':
|
|
return self._handler.file_segment('reload', **options)
|
|
|
|
elif method == 'file_segment_invalid':
|
|
return self._handler.file_segment('invalid', **options)
|
|
|
|
elif method == 'file_handle_data':
|
|
return self._file_handle_data(**options)
|
|
|
|
elif method == 'file_handle_data_start':
|
|
return self._file_handle_stream(True, **options)
|
|
|
|
elif method == 'file_handle_data_stop':
|
|
return self._file_handle_stream(False, **options)
|
|
|
|
else:
|
|
return self._dispatcher.method_dispatch(method, (), options)
|
|
|
|
def _file_handle_stream(self, start: bool, user_session: Optional[str] = None):
|
|
if self._client_id is not 'control_server':
|
|
return None
|
|
|
|
if start:
|
|
session = self._handler.get_user_session(user_session, new_session=True, allow_empty=True)
|
|
else:
|
|
session = self._handler.get_user_session(user_session)
|
|
|
|
if session is None:
|
|
return None
|
|
|
|
data_handle = session.get_data('data_handle', type_check=MqttDataMessageHandler)
|
|
if data_handle is None or not isinstance(data_handle, MqttDataMessageHandler):
|
|
if not start:
|
|
return
|
|
|
|
data_handle = session.set_data('data_handle', MqttDataMessageHandler(self, 'file_handle_data'))
|
|
# self._handler.data_request_stream(start, data_handle)
|
|
|
|
# if start :
|
|
# data_handle = MqttDataMessageHandler(self, 'file_handle_data')
|
|
# self._handler.data_request_stream(start, data_handle)
|
|
# else:
|
|
# return
|
|
|
|
def _file_handle_data(self, expression: str, user_session: Optional[str] = None) -> ComplexResponse:
|
|
"""request data
|
|
|
|
**Request Url**
|
|
|
|
**Request Body**
|
|
|
|
data request expression
|
|
|
|
:class:`biopro.message.request.DataMessageRequest`
|
|
|
|
**response format**
|
|
|
|
websocket bytes data
|
|
|
|
**relative method**
|
|
|
|
:func:`biopro.server.controller.ControlServer.websocket_data_request`
|
|
|
|
:func:`biopro.server.cache.CacheManager.get_data_message`
|
|
|
|
:param expression:
|
|
:param user_session: user session ID
|
|
:return:
|
|
"""
|
|
|
|
if self._client_id is not 'control_server':
|
|
return None
|
|
|
|
session = self._handler.get_user_session(user_session, new_session=True, allow_empty=True)
|
|
|
|
if session is None:
|
|
return None
|
|
|
|
data_handle = session.get_data('data_handle',
|
|
type_check=MqttDataMessageHandler,
|
|
init=lambda: MqttDataMessageHandler(self, 'file_handle_data'))
|
|
|
|
ret = self._handler.websocket_data_request(data_handle, expression)
|
|
|
|
# ret = self._handler.websocket_data_request(MqttDataMessageHandler(self, 'file_handle_data'), expression)
|
|
|
|
resp = ComplexResponse(ret)
|
|
resp.set_cookie(user_session=session.user_session)
|
|
|
|
return resp
|
|
|
|
def hardware_test(self) -> JSON_OBJECT:
|
|
return {
|
|
'mqtt_url': self._mqtt_url,
|
|
'mqtt_port': self._mqtt_port,
|
|
'mqtt_alive': self._mqtt_alive,
|
|
# TODO move content for testing
|
|
'controller_id': self.__controller_ID
|
|
}
|
|
|
|
|
|
class MqttDataMessageHandler(DataMessageHandler):
|
|
__slots__ = '__client', '__topic'
|
|
|
|
def __init__(self, client: MqttThread, topic: str):
|
|
super().__init__()
|
|
self.__client = client
|
|
self.__topic = topic
|
|
|
|
def on_message(self, message, topic=None):
|
|
# response = {
|
|
# # 'binary': base64.b64encode(message).decode()
|
|
# 'binary': str(message)
|
|
# }
|
|
response = None
|
|
|
|
# if len(self.__topic) > 20:
|
|
# response = {
|
|
# 'binary': message
|
|
# }
|
|
# # print( 'mqtt_datastream: ' + str(json_stringify(response)) + ', topic:' + self.__topic)
|
|
# else:
|
|
# response = {
|
|
# 'binary': base64.b64encode(message).decode()
|
|
# }
|
|
if topic is None:
|
|
self.__client.publish(self.__topic, message, qos = 0)
|
|
else:
|
|
self.__client.publish(self.__topic + '/' + topic, message, qos = 0)
|
|
|
|
del response
|
|
del message
|
|
return
|
|
|
|
def __str__(self):
|
|
return 'DataMessageHandler[%s]' % ('mqtt:' + self.__topic)
|