from time import sleep from typing import Iterable from datetime import datetime import biopro.impl.vcgencmd as vcg from biopro.data import DataServerOptions, DataAPI from biopro.device.manager import DeviceManagerOptions, DeviceHardwareInterface from biopro.devlib.data import DataDecodeFormat, RawDataDecoder from biopro.devlib.device import DeviceCommandAction, DeviceCommonInstruction from biopro.devlib.instruction import InternalInstruction, DeviceInstructionError # from biopro.exp_pro.manager import ExpManagerOption, ExpManager from biopro.file.manager import FileOptions from biopro.message import * from biopro.util.address import address_str from biopro.util.datetime import * from biopro.util.logger import logging_info, logging_verbose from biopro.util.sleep import long_sleep from biopro.util.sys import system_call, current_command_header from .cache import CacheAPI, CacheNotifyHandler, CacheManager from .controller import * from .data import DataServerProcess from .data_stream import DataStreamThread from .led import LEDControlThread from .mqtt import MqttThread from .connect import RoutineConnectDeviceThread from .socket import SocketServer, ServerThread, ServerShutdownException from .trash import RoutineTrashThread from .ups import UPSDetectThread from .api import ApiRequests from biopro.api.auth import AuthAPI from biopro.api.controller import ControllerAPI from biopro.api.device import DeviceAPI ### do not import Websocket mode ###from .websocket import WebsocketThread _RUNTIME_COMPILE = False T = TypeVar('T') def _fc(c): return c % len(FORE_COLOR) + FORE_COLOR[0] class ControlServer(SocketServer, ControlServerAPI, CacheAPI, CacheNotifyHandler): """Data Server""" def __init__(self, *options): """create data server. **Server Options** accept 1. :class:`biopro.server.controller.ControlServerOptions` #. :class:`biopro.data.DataServerOptions` #. :class:`biopro.file._api.FileOptions` #. :class:`biopro.device.manager.DeviceManagerOptions` :param options: server options """ super().__init__((ControlAPI, CacheAPI), 'Controller', self.ADDR, socket_listen=10, thread_pool_count=16) # controller ID from ._identify import controller_device_id self.__controller_ID = controller_device_id() # auth set jwt AuthAPI.set_key() # controller server self._controller_options = c = ControlServerOptions.get_options(*options) # file managers self._file_manager = self._init_file_manager(c) self._file_save_cache = {} # device manager device_option = DeviceManagerOptions.get_options(*options) self._device_manager = DeviceManager(device_option, self) # data server data_options = DataServerOptions.get_options(*options) 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: self.mqtt_thread = MqttThread(self, self.socket_mqtt_ip, self.socket_mqtt_port, 'control_server', 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, refresh_mode=not c.flag_disable_led_refresh) # device detect self._detect_device_thread = RoutineConnectDeviceThread(self._device_manager, self.mqtt_thread, self.led_thread, 3) if not c.flag_disable_led_control: self.led_thread.set_state(LED.COLOR_RED) # UPS detect self.ups_thread = None if not c.flag_disable_ups_detect: self.log_verbose('UPS_ENABLE', True) self.ups_thread = UPSDetectThread(self, c.flag_ups_wait_time) # user session 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 # scan max duration self._scan_max_timeout = 5 # _api request build self._api = ApiRequests() # init method def _init_file_manager(self, option: ControlServerOptions) -> FileManager: """initialize file manager :param option: control server cli options :return: file manager instance """ r = Path(option.storage_root).absolute() if not r.exists(): self.log_info('create storage root', r) r.mkdir(parents=True) file_manager_options = FileOptions.with_default_structure(r) e = option.export_root if e is not None: file_manager_options.export_root = str(Path(e).absolute()) # create manager return FileManager(file_manager_options) # abstract property override @property def options(self) -> ControlServerOptions: return self._controller_options @property def file_manager(self) -> FileManager: return self._file_manager @property def device_manager(self) -> DeviceManager: return self._device_manager # server setup def setup(self) -> None: """setup the control server """ # print allow host from biopro.util.inet import allow_host for host in allow_host(): self.log_info('allow_host', host) # UPS if self.ups_thread is not None: self.run_thread(self.ups_thread) # LED self.run_thread(self.led_thread) self.led_thread.set_state(LED.POWER_ON) # sha checking # if _RUNTIME_COMPILE or not self._controller_options.flag_disable_sha_check: # self.run_thread(_RandomCrashThread()) # data server interface = self.device_manager.interface options = self.data_server.options if interface == DeviceHardwareInterface.INTERFACE_DUMMY: options.flag_spi_mode = DataAPI.MODE_DISABLE elif interface == DeviceHardwareInterface.INTERFACE_CC2650_S: options.flag_spi_mode = DataAPI.MODE_SINGLE options.storage_root = str(self.file_manager.storage_root) self.data_server.start_process() # file manager self.log_verbose('setup file manager') self.file_manager.setup_storage() self.file_manager.update_storage() self.submit_thread(self._setup_file_manager) # device manager self.log_verbose('setup device manager') self.device_manager.setup() self.submit_thread(self._setup_library) # 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') # file manager clean external directory self.file_manager.umount_external() if not self._controller_options.flag_disable_external_storage: # file manager mount external storage self.file_manager.mount_external_auto() if len(self._controller_options.external_path_map): for f, t in self._controller_options.external_path_map: self.file_manager.mount_external(f, Path(t)) self.log_verbose('cache root info') ### init file info get # self.file_manager.info('').as_json() # self.file_manager.list('').as_json() return def _setup_library(self): self.log_verbose('reload library') self.device_manager.reload_library() def _setup_controller(self): hardware_info = self.hardware_info().as_json_1() from ._identify import controller_device_id mqtt_id = controller_device_id() hardware_info['mqtt_id'] = mqtt_id hardware_info['status'] = 'running' controller = ControllerAPI.getByMac(hardware_info['mac_address']) if len(controller) == 0: ControllerAPI.create(hardware_info) elif len(controller) > 0: if hardware_info['mac_address'] != controller[0]['mac_address'] or hardware_info['software_version'] != controller[0]['software_version'] or hardware_info['mqtt_id'] != controller[0]['mqtt_id'] or hardware_info['wifi_name'] != controller[0]['wifi_name']: ControllerAPI.updateByAttrs ( 'mac_address', hardware_info['mac_address'], { 'software_version': hardware_info['software_version'], 'mqtt_id': mqtt_id, 'name': hardware_info['name'], 'wifi_name': hardware_info['wifi_name'] } ) def _setup_device(self): # wait data server self.log_verbose('device hardware reset') self.device_manager.reset(software_reset=False) self.log_verbose('wait data server') if self.device_manager.interface == DeviceHardwareInterface.INTERFACE_DUMMY: sleep(1) else: sleep(3) try: available_channel = self._setup_get_available_channel() if available_channel is None: self.device_manager.setup(DeviceHardwareInterface.INTERFACE_DUMMY) else: self.broadcast_command('state:device_available') except ServerShutdownException as e: self.interrupt(e) return except RuntimeError as e: self.log_warn('reset fail', ':', str(e)) print_exception(self) if self._controller_options.shutdown_if_reset_fail: self.interrupt(ServerShutdownException()) return self._setup_restart_device_manager(DeviceHardwareInterface.INTERFACE_DUMMY) self._setup_reset_data_server() except: print_exception(self) finally: self.led_thread.set_state(LED.AVAILABLE) def _setup_get_available_channel(self) -> Optional[List[int]]: client = self.data_server.client() if client is None: raise RuntimeError('data server not available') retry = 0 while retry < 10: try: with client: client.reset() break except RuntimeError: self.log_verbose('wait data server', 'x%d' % retry) sleep(1) retry += 1 else: raise RuntimeError('data server not available') # reset data server with client: data_spi_mode = client.spi_mode() self.log_verbose('data server spi mode', data_spi_mode) available_channel = client.available_channel() # check channel if len(available_channel) == 0: if data_spi_mode == DataAPI.MODE_DISABLE: return None else: raise RuntimeError('empty channel') expect_channel = self._controller_options.expect_available_channel if expect_channel is not None: self.log_verbose('expect channel', expect_channel) retry = 0 while available_channel != expect_channel and retry < 3: sleep(1) with client: available_channel = client.available_channel(reset=True) retry += 1 self.log_verbose('available channel', available_channel) return available_channel def _setup_restart_device_manager(self, interface: str): self.log_info('shutdown device manager') self.device_manager.shutdown() # self.broadcast_command('shutdown') self.log_info('restart device manager with interface', interface) self.device_manager.setup(interface) def _setup_reset_data_server(self): """change data server spi mode according device manager hardware interface""" spi_mode = None interface = self.device_manager.interface if interface == DeviceHardwareInterface.INTERFACE_DUMMY: spi_mode = DataAPI.MODE_DISABLE elif interface == DeviceHardwareInterface.INTERFACE_CC2650_S: spi_mode = DataAPI.MODE_SINGLE if spi_mode is not None: self.log_verbose('reset data server to mode', spi_mode) client = self.data_server.client() if client is not None: with client: client.reset(spi_mode) # server close def close(self, external_interrupt: bool): 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. :return: connected device list """ ret = self.device_manager.list_device() if len(ret) == 0: # no connected device self.led_thread.set_state(LED.AVAILABLE) else: self.led_thread.set_state(LED.CONNECTED) return ret def _post_device_connected(self, device: Optional[int]): """general post action for any device has connected. :param device: device connected """ self.led_thread.set_state(LED.CONNECTED) self.broadcast_command(DataMessage.connected(device)) def _post_device_disconnected(self, device: Optional[int]): """general post action for any device has disconnected. :param device: device disconnected. """ client = self.data_server.client() if client is not None: with client: if client.is_sync(device): client.stop_sync(device) if len(self.device_manager.list_device()) == 0: self.led_thread.set_state(LED.AVAILABLE) data = { "connect_status": 'idle' } self._api.updateDeviceByAttr('connect_status', 'update', data) self.broadcast_command(DataMessage.disconnected(device)) # self.broadcast_command('device/disconnected/' + str(device)) # server provide method implement : hardware information @logging_info def hardware_info(self) -> HardwareInfo: return HardwareInfo( name=self.server_name, device_ID=self.__controller_ID, device_mac=self._controller_mac_address() ) def _controller_mac_address(self) -> Optional[str]: from biopro.util.inet import ip_addr if _RUNTIME_COMPILE: interface_list = ip_addr('eth0') else: interface_list = ip_addr() for interface in interface_list: if _RUNTIME_COMPILE: if interface.interface == 'eth0': break else: if interface.interface.startswith('e'): break else: return None return interface.link_addr def hardware_datetime(self, content: Optional[str]) -> str: if content is not None: frontend_time = datetime.fromtimestamp(int(content) / 1000) local_time = datetime.now() if frontend_time > local_time: self.log_verbose('timestamp update', local_time, frontend_time, frontend_time-local_time) set_datetime(frontend_time) # d = str_datetime(content) # print('d',d) # d = datetime # # set System datetime # self.file_manager.update_storage() return datetime_str() @logging_info def hardware_menu(self) -> Optional[Menu]: return super().hardware_menu() # server provide method implement : hardware functions @logging_info def hardware_scan(self) -> bool: self.hardware_scan_device() return True def hardware_scan_device(self): self.broadcast_command(DataMessage.scan('')) device_not_found = True # we want once find any device, update the information ASAP by broadcasting def callback(_: DeviceResponseInfo): nonlocal device_not_found device_not_found = False self.broadcast_command(DataMessage.scan('update')) if self.device_manager.scan_callback(callback, timeout=self._scan_max_timeout, all_device=True): self.broadcast_command(DataMessage.scan('done')) # if device_not_found: # # no device found, longer the max scan time # self._scan_max_timeout = min(10, self._scan_max_timeout + 2) @logging_info def hardware_device(self) -> List[DeviceResponseInfo]: return self.device_manager.found() @logging_info def hardware_device_found(self, content: Union[str, JSON_OBJECT, DeviceInfo], *, scan = False) -> Optional[DeviceResponseInfo]: return super().hardware_device_found(content, scan=scan) @logging_info def hardware_device_connect(self, content: Union[str, JSON_OBJECT, DeviceInfo]) -> CompletedDevice: if isinstance(content, DeviceResponseInfo): response = content else: response = to_device_info(content) try: ret = self.device_manager.connect(response) except DeviceInstructionError as e: '''when device reset fail error''' return None else: if ret is not None: # update device status by id DeviceAPI.updateByMac( address_str(ret.mac_address), { 'battery': ret.battery, 'connect_id': ret.device_id, 'connect_status': 'connect', 'device_version': ret.device_version, 'library': ret.library_name, 'library_version': ret.library_version, } ) self._post_device_connected(ret.device_id) return ret def hardware_device_direct_connect(self, content: Union[str, JSON_OBJECT, DeviceInfo]) -> CompletedDevice: if isinstance(content, DeviceResponseInfo): response = content else: response = to_device_info(content) try: ret = self.device_manager.connect(response) except DeviceInstructionError as e: '''when device reset fail error''' return None else: if ret is not None: self._post_device_connected(ret.device_id) return ret @logging_info def hardware_device_disconnect(self, device: Optional[int], force=False) -> bool: connect_device = self.device_manager.get_device(device) DeviceAPI.updateByMac( address_str(connect_device.mac_address), { 'connect_id': -1, 'connect_status': 'idle', 'auto_connect': False, } ) if device is None: self.device_manager.disconnect_all(force) ret = True else: ret = self.device_manager.disconnect(device, force) data = { "connect_status": 'idle', "connect_id": -1, "connect_time": 0, "auto_connect": False } self._api.updateDeviceByAttr('connect_id', str(device), data) self._post_device_disconnected(device) return ret @logging_info def hardware_send(self, command: str, **options: str): # hardware command for development mode if command == 'led': if not _RUNTIME_COMPILE: return self._hardware_send_led(options) elif command == 'power_off': return self._hardware_send_power_off(options) elif command == 'test': if not _RUNTIME_COMPILE: try: return self._hardware_send_test(options) except: print_exception(self) elif command == 'test_set': if not _RUNTIME_COMPILE: try: return self._hardware_send_test_set(options) except: print_exception(self) elif command == 'device': try: return self._hardware_send_device(options) except: print_exception(self) elif command == 'measure': try: return self._hardware_send_measure(options['item']) except: pass else: ret = self.device_manager.handle_command(None, command, options) self._post_hardware_send(command, None, ret) return None def _hardware_send_led(self, options: Dict[str, str]) -> bool: state = options.get('state', None) if state is not None: self.led_thread.set_state(state) return True def _hardware_send_power_off(self, options: Dict[str, str]) -> bool: s = options.get('s', 0) try: s = int(s) except ValueError: s = 0 def _power_off(_sec: int): if not long_sleep(_sec, lambda: self.is_closed): UPSDetectThread.power_off() self.submit_thread(_power_off, s) return True def _hardware_send_measure(self, item: str) -> Union[int, float]: if item == 'temp': return vcg.measure_temp() else: raise RuntimeError('unknown measure item') def _hardware_send_device(self, options: Dict[str, str]) -> JSON: command = options.get('manager', None) if command is not None: # manager command if command == 'all': # noinspection PyTypeChecker return ['all', 'reload'] elif command == 'reload': self.device_manager.reload_library() return True else: # TODO other manager command pass return False else: # TODO command for device pass def _post_hardware_send(self, command: str, device: Optional[int], action: DeviceCommandAction): """ :param command: send command :param device: target device :param action: what's happen """ self.log_verbose('hardware send', command, 'device', device, ':', action) if action == DeviceCommandAction.CONNECTED: self._post_device_connected(device) elif action == DeviceCommandAction.DISCONNECTED: self._post_device_disconnected(device) elif action == DeviceCommandAction.SCAN: self.broadcast_command(DataMessage.scan('done')) # server provide method implement : device functions @logging_info def get_device_info_all(self) -> List[CompletedDevice]: return self._post_device_list() @logging_info 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) @logging_info def device_all_instruction(self, instruction: str, content: str) -> bool: if len(content) == 0: # for all device device_id = list(map(lambda d: d.device_id, self.device_manager.list_device())) else: device_id = list(map(int, filter(len, content.split(' ')))) self.log_verbose('device', device_id, instruction) self._device_instruction_group = device_id try: return self._device_all_instruction(instruction, device_id) except: print_exception(self) finally: self._device_instruction_group = None def _device_all_instruction(self, instruction: str, device_id: List[int]) -> bool: if instruction == DeviceCommonInstruction.START: for device in device_id: self.device_manager.device_instruction(device, instruction) client = self.data_server.client() if client is not None: with client: client.start_sync(*device_id) for device in device_id: self.broadcast_command(DataMessage.start(device)) elif instruction == DeviceCommonInstruction.INTERRUPT or instruction == DeviceCommonInstruction.RESET: client = self.data_server.client() if client is not None: with client: client.stop_sync(*device_id) for device in device_id: self.device_manager.device_instruction(device, instruction) # for device in device_id: # self.broadcast_command(DataMessage.stop(device)) if instruction == DeviceCommonInstruction.RECORD_ALL: for device in device_id: self.device_manager.device_instruction(device, instruction) client = self.data_server.client() if client is not None: with client: client.start_sync(*device_id) # for device in device_id: # self.broadcast_command(DataMessage.start(device)) return True @logging_info def device_instruction(self, device: int, instruction: str) -> bool: self.device_manager.device_instruction(device, instruction) return True @logging_info def device_battery(self) -> List[int]: device_list = self.device_manager.list_device() if len(device_list) > 0: battery_list = list(map(lambda x: self.device_manager.get_device(x).battery, device_list)) return battery_list @logging_info def device_parent(self, device: int, content: Optional[str] = None) -> List[int]: if content is not None: self.device_manager.get_device(device).update_parent_info(JsonSerialize.to_json(content)) return self.device_manager.get_device(device).parent def device_recording_file_name(self, device: int, content: Optional[str] = None) -> List[int]: if content is not None: self.device_manager.get_device(device).update_recording_file_name_info(str(content)) return self.device_manager.get_device(device).recording_file_name @logging_info def device_sd_card_status(self, device:int) -> None: return None @logging_info def device_update_calibration(self, device: int) -> Union[bool, str]: connect_device = self.device_manager.get_device(device) mac_address = connect_device.mac_address try: if connect_device.library.name.startswith('Neulive3.'): connect_device.calibration_info('NeuliveThreeOne') elif connect_device.library.name.startswith('Neulive'): connect_device.calibration_info('TDC4VC') elif connect_device.library.name.startswith('EliteEIS'): connect_device.calibration_info('EISZeroOne') except BaseException as e: # reset device info in db DeviceAPI.updateByMac( address_str(mac_address), { 'connect_id': -1, 'connect_status': 'idle', 'auto_connect': False, } ) # disconnect device self.hardware_device_disconnect(device, True) print(e) return False else: # update_calibration_info DeviceAPI.updateByMac( address_str(mac_address), { 'calibration': str(connect_device.calibration) } ) # DeviceAPI.updateByAttrs('connect_status', 'update', {"connect_status": 'connect'}) return True @logging_info def device_update_radio_frequency(self, device: int, content:int) -> bool: d = self.device_manager.get_device(device) res = d.update_radio_frequency(content) return res @logging_info def device_update_led_brightness(self, device: int, content:int) -> bool: d = self.device_manager.get_device(device) res = d.update_led_brightness(content) return res @logging_info def device_update_accelerator_sensitivity(self, device: int, content:int) -> bool: d = self.device_manager.get_device(device) res = d.update_accelerator_sensitivity(content) return res @logging_info def device_set_central_rf(self, device: int, content: int) -> bool: d = self.device_manager.get_device(device) res = d.central_rf_set(content) @logging_info def device_get_central_version(self, device: int) -> Optional[list]: d = self.device_manager.get_device(device) res = d.central_version_get() return res @logging_info def device_parameter_last_update_timestamp(self, device: int) -> int: d = self.device_manager.get_device(device) if d is None: raise RuntimeError(DEVICE_NOT_FOUND, str(device)) return d.parameter.last_modified_timestamp @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 def device_parameter_value(self, device: int, parameter: str) -> Any: return self.device_manager.handle_device_parameter_value(device, parameter) @logging_info def device_save_path(self, device: int, unset: bool) -> Optional[str]: d = self.device_manager.get_device(device) if d is None: raise RuntimeError(DEVICE_NOT_FOUND, str(device)) if unset: try: del self._file_save_cache[device] except KeyError: pass return self.file_manager.unset(device) else: info = self.file_manager.use(device) if info.is_meta_file: # do not used file_path which contain file extension return info.parent_path + '/' + info.file_name directory = info.file_path content = self._file_save_cache.get(device, self.TEMP_FILENAME_EXPR) content = format_filename(content, d) if '/' not in content: content = directory + '/' + content self.log_verbose('device_save_path', device, content) return content @logging_info def device_save_path_set(self, device: int, content: str, dry_run: bool) -> str: d = self.device_manager.get_device(device) if d is None: raise RuntimeError(DEVICE_NOT_FOUND, str(device)) content = re.sub("^ +", "", content) content = re.sub(" +$", "", content) content = re.sub(" +", "_", content) ret = format_filename(content, d) if dry_run: return ret else: info = self.file_manager.use(device) if info.is_meta_file: directory = info.parent_path else: directory = info.file_path self._file_save_cache[d.device_id] = content if '/' not in ret: ret = directory + '/' + ret self.log_verbose('device_save_path_set', device, ret) return ret # server provide method implement : device history @logging_info def device_history_get(self, user: Optional[str]) -> List[DeviceResponseInfo]: raise NotImplementedError() @logging_info def device_history_found_all(self, user: str) -> List[DeviceResponseInfo]: raise NotImplementedError() @logging_info def device_history_found(self, user: str, device: int) -> Optional[DeviceResponseInfo]: raise NotImplementedError() @logging_info 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 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) if oper in (InternalInstruction.PREDEFINED_DATA_FORMAT, InternalInstruction.PREDEFINED_DATA_FORMAT_CALI): if device is not None: self._device_set_data_format(device, value) return True elif oper == InternalInstruction.PREDEFINED_SYNC: if device is not None: self._device_set_sync(device, bool(value)) return True elif oper == InternalInstruction.PREDEFINED_DISABLE_CACHE: if device is not None: self._device_set_disable_cache(device, value) return True else: return False def _device_set_data_format(self, device: CompletedDevice, value: Any): if isinstance(value, (str, bytes)): value = DataDecodeFormat.parse(value) if not isinstance(value, DataDecodeFormat): raise TypeError('not DataDecodeFormat : ' + value.__class__.__name__) self.log_verbose('update data format', value) client = self.data_server.client() if client is not None: info = self.file_manager.use(device) if info.is_meta_file: # overwrite info = self.file_manager.save(device, info, overwrite=True) else: filename = self._file_save_cache.get(device.device_id, self.TEMP_FILENAME_EXPR) filename = format_filename(filename, device) info = self.file_manager.save(device, filename) with client: client.update_device_configuration(device, info.meta_file, value) def _device_set_disable_cache(self, device: CompletedDevice, disable): if disable: self.log_verbose('disable cache', device.device_id) info = self.file_manager.use(device) if not info.is_meta_file: raise RuntimeError('%s should set after %s' % ( InternalInstruction.PREDEFINED_DISABLE_CACHE, InternalInstruction.PREDEFINED_DATA_FORMAT )) meta_file = info.meta_file assert meta_file is not None self.cache_manager.disable_cache(device, meta_file) else: self.cache_manager.enable_cache(device) def _device_set_sync(self, device: CompletedDevice, sync: bool): 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) with client: if not info.is_meta_file: # save data in raw format. filename = format_filename(self.TEMP_FILENAME_EXPR, device) info = self.file_manager.save(device, filename) client.update_device_configuration(device, info.meta_file, RawDataDecoder()) print('***456') if self._device_instruction_group is None: client.start_sync(device) if self._device_instruction_group is None: # broadcast self.broadcast_command(DataMessage.start(device.device_id)) else: self.log_verbose('stop sync', device.device_id) if self._device_instruction_group is None: start_sync = False client = self.data_server.client() if client is not None: with client: # start_sync = client.is_sync(device) start_sync = True if start_sync: client.stop_sync(device) if start_sync: # unset file info, but keep file path cache self.file_manager.unset(device.device_id) # broadcast self.broadcast_command(DataMessage.stop(device.device_id)) # server provide method implement : websocket broadcast functions @logging_info def broadcast_command(self, command: str, device: Optional[int] = None): self.log_verbose(pc('broadcast_command', GREEN), '(All)' if device is None else '(%d)' % device, command) ### self.websocket_thread.broadcast_command(command, device=device) 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]): """ :param command: websocket data request command :return: error message """ response = handler.update_command(command) if response is None: return if len(handler.listen_device) == 0: handler.on_error("empty channel") elif handler.sample_rate is None and handler.data_points is None: handler.on_error("sample rate not set") 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 def set_led_state(self, state: str): self.led_thread.set_state(state) # hardware send test if not _RUNTIME_COMPILE: def _hardware_send_test(self, options: Dict[str, str]): ret = { 'name': NAME, } if len(options) == 0 or 'software' in options: ret['software'] = self._hardware_send_test_software(options.get('software', None)) if len(options) == 0 or 'system' in options: ret['system'] = self._hardware_send_test_system(options.get('system', None)) if len(options) == 0 or 'device' in options: ret['device'] = self.device_manager.hardware_test() if len(options) == 0 or 'hardware' in options: ret['hardware'] = self._hardware_send_test_hardware(options.get('hardware', None)) return ret def _hardware_send_test_software(self, section: Optional[str] = None) -> Dict[str, Any]: if section is not None: if section == '': section = None else: section = section.split(',') ret = { 'version': VERSION, 'git-rev': REV_VERSION, 'vendor': VENDOR, 'copyright': COPYRIGHT, } if section is None or 'file' in section: ret['file'] = { 'storage_root': str(self.file_manager.storage_root), '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()), 'state': self.led_thread.state, 'disable': self.led_thread.is_disabled, 'count': self.led_thread.led_count, 'refresh': self.led_thread.refresh_mode } if section is None or 'ups' in section: ret['ups'] = self.ups_thread is not None if section is None or 'control' in section: ret['control'] = { 'socket_file': self.ADDR, 'cmds': current_command_header() } if section is None or 'data_process' in section: ret['data_process'] = { 'socket_file': str(self.data_server.socket_file), 'spi_mode': self.data_server.options.flag_spi_mode, 'disable_data_statistics': self.data_server.options.flag_disable_data_statistics, 'cmds': self.data_server.get_process_arguments() } if section is None or 'data_server' in section: client = self.data_server.client() if client is not None: with client: ret['data_server'] = client.hardware_test() if section is None or 'django' in section: from biopro.util.inet import allow_host ret['django'] = { 'allowed_hosts': allow_host() } if section is None or 'session' in section: ret['session'] = [{ 'user_session': s.user_session, 'time_stamp': s.time(update=False), 'property': list(filter(lambda attr: attr not in ('time', 'user_session'), filter(lambda attr: not attr.startswith('_'), dir(s)))) } for s in self._session.values()] return ret def _hardware_send_test_system(self, section: Optional[str] = None) -> Dict[str, Any]: if section is not None: if section == '': section = None else: section = section.split(',') ret = {} if section is None or 'uname' in section: ret['uname'] = system_call('uname', '-a') if section is None or 'volume' in section: root = self.file_manager.index() ret['volume'] = { 'total': root.volume_total, 'usage': root.volume_usage, 'free': root.volume_free, } if section is None or 'package_sys' in section: from biopro.util.package import Package ret['package_manager'] = Package.manager() ret['package_sys'] = [{ 'name': pkg.name, 'version': pkg.version, } for pkg in Package.list()] if section is None or 'package_pip' in section: from biopro.util.package import PipPackage ret['package_pip'] = [{ 'name': pkg.name, 'version': pkg.version, } for pkg in PipPackage.list()] return ret def _hardware_send_test_hardware(self, section: Optional[str] = None) -> Dict[str, Any]: if section is not None: if section == '': section = None else: section = section.split(',') ret = {} if section is None or 'measure' in section: ret['measure'] = { 'clock_arm': vcg.measure_clock('arm'), 'clock_core': vcg.measure_clock('core'), 'voltage': vcg.measure_volts('core'), 'temperature': vcg.measure_temp() } if section is None or 'ext_memory' in section: from biopro.impl.ext_mem import hardware_test ret['ext_memory'] = hardware_test() if section is None or 'gpio' in section: try: from biopro.impl.gpio import hardware_test except ImportError: result = None else: result = hardware_test() ret['gpio'] = result return ret def _hardware_send_test_set(self, options: Dict[str, str]) -> bool: # TODO write options files return False class _RandomCrashThread(ServerThread): def __init__(self): super().__init__('Crash') self.log_enable = False def setup(self) -> None: super().setup() from biopro.util.sha import sha256sum_validate try: from ._sha import SUM except ImportError: if _RUNTIME_COMPILE: return else: self.close() return for s, f in SUM: # type: str, str if not sha256sum_validate(f, s, ignore_missing=True): return self.close() def routine(self, event = None): from random import randint event.wait(randint(60, 600), disable_interrupt=True) UPSDetectThread.power_off()