wisetop/controller-wisetopdataserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[update] remove duplicate code

Browse Source
This commit is contained in:
peterlu14
2022-04-13 15:49:04 +08:00
parent 5dbe231c73
commit f9ab0bc59b
1 changed files with 4 additions and 516 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/biopro/server/main.py
+4 -516
View File
@@ -43,7 +43,7 @@ def _fc(c):
return c % len(FORE_COLOR) + FORE_COLOR[0]
class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler):
class ControlServer(SocketServer, ControlServerAPI):
"""Data Server"""
def __init__(self, *options):
@@ -60,7 +60,7 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
:param options: server options
"""
super().__init__((ControlAPI, CacheAPI), 'Controller', self.ADDR,
super().__init__((ControlAPI), 'Controller', self.ADDR,
socket_listen=10,
thread_pool_count=16)
@@ -87,21 +87,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
data_options.repository = device_option.repository
self.data_server = DataServerProcess(data_options)
# experimental protocol
# exp_options = ExpManagerOption.get_options(*options)
# self.exp_manager = ExpManager(self, exp_options)
# cache
self.cache_manager = CacheManager(self,
c.cache_server_cache_size,
# time unit from sec to ms
c.cache_server_cache_time * 1000)
# websocket server ###
### self.websocket_thread = WebsocketThread(self,
### c.websocket_address,
### c.websocket_port)
# remote server / MQTT
self.mqtt_thread = None
if not c.flag_disable_remote_server:
@@ -112,17 +97,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
self.log_verbose,
'MQTT-control-server')
# routine trash thread
self._routine_trash_thread = None
# if not c.flag_disable_routine_trash:
# self._routine_trash_thread = RoutineTrashThread(self._file_manager,
# c.trash_routine_time,
# c.trash_expire_time,
# c.trash_remove_expire_time)
# routine data routine
self._data_stream_thread = None
# LED control
self.led_thread = LEDControlThread(disable=c.flag_disable_led_control,
led_count=c.led_count,
@@ -131,6 +105,8 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
# device detect
self._detect_device_thread = RoutineConnectDeviceThread(self._device_manager, self.mqtt_thread, self.led_thread, 3)
self._project_manager = ProjectManager(self.device_manager, self.mqtt_thread)
if not c.flag_disable_led_control:
self.led_thread.set_state(LED.COLOR_RED)
@@ -144,8 +120,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
self._session = {}
self._user_session_expire = c.user_session_expire_days * 24 * 60 * 60 # sec
# temp use variable
# indicate those device together
self._device_instruction_group = None
@@ -155,9 +129,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
# _api request build
self._api = ApiRequests()
self._project_manager = ProjectManager(self.device_manager, self.mqtt_thread)
# init method
def _init_file_manager(self, option: ControlServerOptions) -> FileManager:
@@ -249,32 +220,16 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
# device manager reset
self.submit_thread(self._setup_device)
# self.submit_thread(self._setup_controller)
self._setup_controller()
# self.run_thread(self._detect_device_thread)
# experimental protocol manager
# if not self._controller_options.flag_disable_exp_manager:
# self.exp_manager.setup()
# self.exp_manager.reload_protocol()
# websocket server
### self.log_verbose('setup websocket server', pc(self.websocket_thread.websocket_address, YELLOW))
# websocket server thread run ###
### self.run_thread(self.websocket_thread)
# broadcast greeting
self.broadcast_command(DataMessage.greeting())
# routine trashing
if self._routine_trash_thread is not None:
self.run_thread(self._routine_trash_thread)
# remote server
if self.mqtt_thread is not None:
self.mqtt_thread.start()
# self.run_thread(self.mqtt_thread)
def _setup_file_manager(self):
self.log_verbose('setup file manager')
@@ -449,90 +404,14 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
self.broadcast_command('shutdown')
self.data_server.stop_process(external_interrupt)
self._stop_data_stream_thread()
self.device_manager.shutdown()
self.file_manager.shutdown(umount_external=True)
self.cache_manager.shutdown()
### self.websocket_thread.close()
self.led_thread.close()
if self.mqtt_thread is not None:
self.mqtt_thread.shutdown()
# self.mqtt_thread.close()
# user session
def new_user_session(self) -> str:
session = UserSession(None)
self._session[session.user_session] = session
self.log_info('new user session', session.user_session)
return session.user_session
@logging_info
def del_user_session(self, user_session: Optional[str] = None):
if user_session is None:
self._session.clear()
else:
try:
del self._session[user_session]
except KeyError:
pass
def get_user_session(self,
user_session: Optional[str],
new_session = False,
allow_empty = False) -> Optional[UserSession]:
"""
:param user_session: user session ID
:param new_session: new a user session if *user_session* not found.
:param allow_empty: consider empty *user_session* None
:return: user session
"""
if user_session is None:
session = None
elif len(user_session) == 0:
if allow_empty:
session = None
else:
raise None
else:
session = self._session.get(user_session, None)
if session is not None:
session.time()
elif new_session:
session = UserSession(user_session)
self._session[session.user_session] = session
self.log_info('new user session', session.user_session)
return session
def expire_user_session(self):
current_time = time()
for user_session in list(self._session.keys()):
session = self._session[user_session]
if session.time(update=False) - current_time > self._user_session_expire:
del self._session[user_session]
# general post action
def _post_device_list(self) -> List[CompletedDevice]:
"""general post action for getting list of connected devices.
@@ -869,12 +748,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
def get_device_info(self, device: int) -> Optional[CompletedDevice]:
return super().get_device_info(device)
# @logging_info
# def device_call(self, device: int) -> bool:
# self.device_manager.call_device(device)
#
# return True
@logging_info
def device_instruction_all(self, device: int) -> List[str]:
return self.device_manager.list_device_instruction(device)
@@ -1048,8 +921,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
@logging_info
def device_parameter(self, device: int, parameter: str, content: Optional[str] = None) -> Any:
# if content is not None:
# self.broadcast_command('parameter:' + str(device) + '/' + parameter + '/' + content)
return self.device_manager.handle_device_parameter_command(device, parameter, content)
@logging_info
@@ -1141,256 +1012,12 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
def device_history_delete(self, user: str, device: Optional[int]) -> bool:
raise NotImplementedError()
# server provide method implement : file manager functions
@logging_info
def file_clean_storage(self) -> bool:
return super().file_clean_storage()
@logging_info
def file_index(self) -> RootInfo:
return super().file_index()
@logging_info
def file_list(self, path: str) -> DirectoryInfo:
return super().file_list(path)
@logging_info
def file_list_range(self, path: str, start: int, end: int) -> DirectoryInfo:
return super().file_list_range(path, start, end)
@logging_info
def file_rename(self, path: str, content: str) -> FileInfo:
return super().file_rename(path, content)
@logging_info
def file_mkdir(self, path: str) -> DirectoryInfo:
return super().file_mkdir(path)
@logging_info
def file_delete(self, path: str) -> bool:
return super().file_delete(path)
@logging_info
def file_info(self, path: str) -> FileInfo:
return super().file_info(path)
@logging_info
def file_meta_info(self, path: str) -> RecordingMetaFile:
return super().file_meta_info(path)
@logging_info
def file_export_info(self, path: str) -> Dict[str, Optional[Path]]:
return super().file_export_info(path)
@logging_info
def file_export_functions(self) -> Tuple[ExternalExportFunction, ...]:
return super().file_export_functions()
@logging_info
def file_export_parameters(self, ftype: str) -> Tuple[ExternalParameter, ...]:
return super().file_export_parameters(ftype)
@logging_info
def file_export(self, ftype: str, path: str, **options: str) -> str:
return super().file_export(ftype, path, **options)
@logging_info
def file_segment(self, method: str, path: str) -> Union[None, bool, FileSegment]:
return super().file_segment(method, path)
@logging_info
def file_segment_get(self, segment: int, path: str):
return super().file_segment_get(segment, path)
@logging_info
def file_handle_open(self, content: str) -> Optional[FileHandle]:
info = self.file_manager.info(content)
meta = info.meta_file
if meta is None:
raise RuntimeError('not a meta file')
handle = self.cache_manager.file_open(meta)
return self.file_handle_info(handle)
@logging_info
def file_handle_info(self, handle: int) -> Optional[FileHandle]:
info = self.cache_manager.file_info(handle)
if info is None:
return None
# modify file path
info.shadow_path(self.file_manager.shadow_path(Path(info.filepath)))
return info
@logging_info
def file_handle_close(self, handle: int):
self.cache_manager.file_close(handle)
@logging_info
def user_all(self) -> List[str]:
raise super().user_all()
@logging_info
def user_info(self, user: str, *, user_session: Optional[str] = None) -> Optional[AbstractUserSetting]:
return super().user_info(user, user_session=user_session)
@logging_info
def user_login(self, user: str, content: str, *,
user_session = None) -> ComplexResponse[AbstractUserSetting]:
raise super().user_login(user, content, user_session=user_session)
@logging_info
def user_logout(self, user_session: Optional[str] = None) -> None:
super().user_logout(user_session)
@logging_info
def user_add(self, user: str, content: str, *,
user_session = None) -> ComplexResponse[AbstractUserSetting]:
return super().user_add(user, content, user_session=user_session)
@logging_info
def user_del(self, user: str, content: str) -> bool:
return super().user_del(user, content)
@logging_info
def user_sync(self, user_session: Optional[str]) -> bool:
return super().user_sync(user_session)
@logging_info
def user_configuration_list(self, user_session: Optional[str], **options: str) -> List[str]:
return super().user_configuration_list(user_session, **options)
@logging_info
def user_configuration_get(self, name: str, user_session: Optional[str] = None) -> Optional[RecordingMetaFile]:
return super().user_configuration_get(name, user_session)
@logging_info
def user_configuration_add(self, name: str, content: str, user_session: Optional[str] = None) -> bool:
return super().user_configuration_add(name, content, user_session)
@logging_info
def user_configuration_del(self, name: str, user_session: Optional[str] = None) -> bool:
return super().user_configuration_del(name, user_session)
@logging_info
def user_configuration_clone(self, name: str, content: str, user_session: Optional[str] = None) -> bool:
return super().user_configuration_clone(name, content, user_session)
@logging_info
def user_setup_list(self, user_session: Optional[str] = None) -> List[str]:
return super().user_setup_list(user_session)
@logging_info
def user_setup_info(self, name: str, user_session: Optional[str] = None) -> Optional[UserSetup]:
return super().user_setup_info(name, user_session)
@logging_info
def user_setup_new(self, name: str, user_session: Optional[str] = None) -> Optional[UserSetup]:
return super().user_setup_new(name, user_session)
@logging_info
def user_setup_del(self, name: str, user_session: Optional[str] = None) -> bool:
return super().user_setup_del(name, user_session)
@logging_info
def user_alias_list(self, user_session: Optional[str] = None) -> List[UserDeviceAlias]:
return super().user_alias_list(user_session)
@logging_info
def user_alias_info(self, name: str, user_session: Optional[str] = None) -> Optional[UserDeviceAlias]:
return super().user_alias_info(name, user_session)
@logging_info
def user_alias_edit(self, name: str, content: str, user_session: Optional[str] = None) -> Optional[UserDeviceAlias]:
return super().user_alias_edit(name, content, user_session)
# @logging_info
# def exp_pro_list(self) -> List[ExpProtocolInfo]:
# return self.exp_manager.list_available_protocol()
@logging_info
def exp_pro_create(self, content: str, *,
user_session = None) -> int:
user_name = self._session_user_name(user_session)
if user_name is None:
raise RuntimeError('require user login')
self.file_manager.user_setting.user_info(user_name)
# p = self.exp_manager.create_protocol(content, user_name)
return
@logging_info
def run_project(self, project) -> bool:
project = self._project_manager.create(project)
self._project_manager.run(project)
return False
# @logging_info
# def exp_pro_options(self, protocol: int) -> List[ExpOption]:
# return self.exp_manager.protocol_options(protocol)
# @logging_info
# def exp_pro_option(self, protocol: int, option: str, content: Optional[str] = None) -> bool:
# if content == '':
# # option reset
# if option == '':
# # reset all options
# self.exp_manager.protocol_reset(protocol)
# else:
# self.exp_manager.protocol_reset(protocol, *option.split(','))
# else:
# # option get/set
# self.exp_manager.protocol_option(protocol, option, content)
# return True
# @logging_info
# def exp_pro_status_all(self) -> List[AbstractExpProtocolState]:
# ret = []
# for p in self.exp_manager.list_protocol():
# s = self.exp_manager.protocol_status(p)
# if s is not None:
# ret.append(s)
# else:
# ret.append(AbstractExpProtocolState(p))
# return ret
# @logging_info
# def exp_pro_status(self, protocol: int) -> Optional[ExpProtocolState]:
# return self.exp_manager.protocol_status(protocol)
# @logging_info
# def exp_pro_command(self, protocol: int, content: str) -> bool:
# if content == 'ensure':
# return self.exp_manager.ensure_protocol(protocol)
# elif content == 'start':
# self.exp_manager.submit_protocol(protocol)
# elif content == 'interrupt':
# self.exp_manager.interrupt_protocol(protocol)
# elif content == 'suspend':
# self.exp_manager.suspend_protocol(protocol)
# elif content == 'resume':
# self.exp_manager.resume_protocol(protocol)
# else:
# raise RuntimeError('unknown protocol command : ' + content)
# return True
@logging_info
def device_internal_command(self, device: int, oper: str, value: Any) -> bool:
device = self.device_manager.get_device(device)
@@ -1406,8 +1033,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
return True
elif oper == InternalInstruction.PREDEFINED_DISABLE_CACHE:
if device is not None:
self._device_set_disable_cache(device, value)
return True
else:
@@ -1461,8 +1086,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
if sync:
# self.log_verbose('start sync', device.device_id)
self.cache_manager.drop_data(device)
client = self.data_server.client()
if client is not None:
info = self.file_manager.use(device)
@@ -1515,96 +1138,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
if self.mqtt_thread is not None:
self.mqtt_thread.broadcast_command(command, device=device)
# data stream thread
def data_request_stream(self, start: bool, handler: DataMessageHandler):
if start:
t = self._start_data_stream_thread()
t.add_handle(handler)
else:
t = self._data_stream_thread
if t is not None:
t.del_handle(handler)
if not t.has_handle():
self._stop_data_stream_thread()
def _start_data_stream_thread(self) -> DataStreamThread:
t = self._data_stream_thread
if self._data_stream_thread is None:
self._data_stream_thread = t = self.run_thread(DataStreamThread(self))
return t
def _stop_data_stream_thread(self):
t = self._data_stream_thread
self._data_stream_thread = None
if t is not None:
t.close()
# noinspection PyShadowingNames
@logging_info
def websocket_command(self, command: str) -> Union[None, str, bytes, List[str]]:
if command == DataMessage.GREETING:
return DataMessage.greeting()
elif command == DataMessage.SYNC:
return self._websocket_command_sync(map(lambda d: d.device_id, self._post_device_list()))
elif command.startswith(DataMessage.SYNC):
_, device_list = part_suffix(command, ':')
if ',' in device_list:
return self._websocket_command_sync(map(int, device_list.split(',')))
else:
return self._websocket_command_sync([int(device_list)])
elif command.startswith(DataMessage.BROADCAST):
_, message = part_suffix(command, ':')
if message is not None:
self.broadcast_command(DataMessage.broadcast(message))
return None
# noinspection PyShadowingNames
def _websocket_command_sync(self, device_list: Iterable[int]) -> Union[str, List[str]]:
client = self.data_server.client()
if client is None:
return DataMessage.sync(False)
else:
result_start = []
result_stop = []
with client:
for device in device_list:
if client.is_sync(device):
result_start.append(device)
else:
result_stop.append(device)
return [DataMessage.start(result_start), DataMessage.stop(result_stop)]
# noinspection PyShadowingNames
def websocket_data_request(self, handler: DataMessageHandler, command: Optional[str]):
try:
self._handle_command_response(handler, command)
except (ValueError, IndexError) as e:
self.log_warn(e)
handler.on_error(DataMessage.warn(str(e)))
except BaseException as e:
self.log_warn(e)
handler.on_error(DataMessage.error(str(e)))
# noinspection PyShadowingNames
def _handle_command_response(self, handler: DataMessageHandler, command: Optional[str]):
"""
@@ -1626,29 +1159,7 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
else:
self.cache_manager.get_data_message(handler, response)
# server provide method implement : cache API
@logging_verbose
def clear(self):
self.cache_manager.clear()
def push_data(self, data: Union[bytes, List[bytes]]):
self.cache_manager.push_data(data)
@logging_verbose
def drop_data(self, *device: int):
self.cache_manager.drop_data(*device)
def stop_data(self, device: int):
self.cache_manager.stop_data(device)
# noinspection PyShadowingNames
@logging_verbose
def notify_data(self, device: int, message: str):
self.broadcast_command(message, device=device)
# led function
def get_led_state(self) -> str:
return self.led_thread.state
@@ -1656,7 +1167,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
self.led_thread.set_state(state)
# hardware send test
if not _RUNTIME_COMPILE:
def _hardware_send_test(self, options: Dict[str, str]):
@@ -1699,30 +1209,9 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
'export_root': str(self.file_manager.export_root)
}
if section is None or 'cache' in section:
ret['cache'] = {
'cache_size': self.cache_manager.cache_size,
'cache_time': self.cache_manager.cache_time,
}
# if section is None or 'websocket' in section:
# ret['websocket'] = {
# 'address': self.websocket_thread.websocket_address,
# 'port': self.websocket_thread.websocket_port,
# }
if section is None or 'mqtt' in section:
ret['mqtt'] = None if self.mqtt_thread is None else self.mqtt_thread.hardware_test()
if section is None or 'trash' in section:
trash_manager = self._routine_trash_thread
ret['trash'] = None if trash_manager is None else {
'routine_interval': trash_manager.routine_interval,
'trash_expire': trash_manager.trash_expire,
'remove_expire': trash_manager.remove_expire,
}
if section is None or 'led' in section:
ret['led'] = {
'state_all': list(LED.COLOR.keys()),
@@ -1851,7 +1340,6 @@ class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler
# TODO write options files
return False
class _RandomCrashThread(ServerThread):
def __init__(self):
super().__init__('Crash')