import sys from collections import Counter from time import time, sleep from typing import Iterable from datetime import datetime import gc import base64 from biopro.data import * from biopro.devlib.data import * from biopro.generator import GeneratorOptions from biopro.impl.ext_mem import MultiExtMemSpiInterface, ExtMemManager from biopro.impl.selector import Selector from biopro.recording import RecordingMetaFile, RecordingFileWriter from biopro.util.address import EMPTY_ADDRESS, address_str from biopro.util.cli import * from biopro.util.console import WHITE, RED, YELLOW, MAGENTA, GREEN, pc from biopro.util.iter import counter_product_sum from biopro.util.leak import LeakSection from biopro.util.logger import logging_info, Logger from biopro.util.stack import print_exception from .cache import CacheClient from .socket import ServerThread, SocketServer from .mqtt import MqttDataMessageHandler, MqttThread from .database import DataBaseProcess from .recording import RecordingProcess from multiprocessing import Queue from json import loads as json_parse, dumps as _json_stringify import json def json_stringify(o) -> str: return _json_stringify(o, separators=(',', ':')) _RUNTIME_COMPILE = False _FLAG_MEMORY_LEAK_TEST_ = False _FLAG_DATA_LOST_STATS_ = True if _RUNTIME_COMPILE: from biopro.util.leak import NoLeakSection LEAK = NoLeakSection() elif _FLAG_MEMORY_LEAK_TEST_: from biopro.util.leak import ObjGraphLeakSection LEAK = ObjGraphLeakSection() else: from biopro.util.leak import NoLeakSection LEAK = NoLeakSection() class DataServer(SocketServer, DataAPI): def __init__(self, *options): """ **Server Options** 1. :class:`impl.data.DataServerOptions` :param options: """ global _FLAG_DATA_LOST_STATS_ super().__init__(DataAPI, 'DataServer', self.ADDR, socket_listen=20) data_options = DataServerOptions.get_options(*options) # device library self.repository = DeviceLibraryRepository(data_options.repository.library_path, no_system_default=True) self.repository.recursive = True # default enable # chip configuration self._configurations = {} # spi self._spi_thread = None self._spi_mode = self.MODE_DISABLE self._spi = None # rec file update self._rec_thread = None # available selector channel self._available_channel = [] # Cache Client self._cache_client = None # global flag _FLAG_DATA_LOST_STATS_ = not data_options.flag_disable_data_statistics # self.setup_spi(data_options.flag_spi_mode) # data server mqtt self.mqtt_thread = MqttThread(None, self.socket_mqtt_ip, self.socket_mqtt_port, 'data_server', self.log_verbose, 'MQTT-data-server') # api socket self._queue_ds_dict = {} self._queue_db = Queue() self._queue_msg = Queue() if len(self._queue_ds_dict) == 0: self._queue_ds_dict[4] = Queue() self._queue_ds_dict[5] = Queue() self._queue_ds_dict[6] = Queue() self._queue_ds_dict[7] = Queue() self.database_process = DataBaseProcess('data_server', self.log_verbose, self._queue_db, self._queue_ds_dict, self._queue_msg) self.database_process.start() # self._no_data_total_duration = 0 # self._recv_memory_cost_time = 0 @logging_info def setup(self): self._available_channel.clear() if self._spi is not None: self.reset_available_channel() LEAK.reset() return @logging_info def close(self, external_interrupt: bool): global _FLAG_DATA_LOST_STATS_ # force disable _FLAG_DATA_LOST_STATS_ = False try: self._stop_runtime() self.stop_sync() finally: self._configurations.clear() if self._spi is not None: self._spi.close() self._spi = None if _FLAG_MEMORY_LEAK_TEST_: LEAK.check_point() LEAK.dump((tuple, list, dict), 'dump_leak_object.txt') return @logging_info def reset(self, spi_mode: Optional[str] = None): # close all self.close(False) if self._spi_thread is not None: self._spi_thread.close() self._spi_thread = None if self._rec_thread is not None: self._rec_thread.close() self._rec_thread = None if self.mqtt_thread is not None: self.mqtt_thread.shutdown() if self.database_process is not None: self.database_process.shutdown() # close and clear data configuration for c in self._configurations: if self._configurations[c] is not None: self._configurations[c].shutdown() self._configurations.clear() # reopen spi, which has close at close() if spi_mode is None: spi_mode = self._spi_mode self._spi_mode = self.MODE_DISABLE self.setup_spi(spi_mode) # restart # self.setup() # data_server setup mqtt start if self.mqtt_thread is not None: self.mqtt_thread.start() # self.run_thread(self.mqtt_thread) # data_server setup api-socket start # if self.database_process is not None: # self.run_thread(self.database_process) # if self.database_process is not None: # self.database_process.start() if self._rec_thread is None or self._rec_thread.is_closed: self.log_verbose('start rec runtime thread') self._rec_thread = self.run_thread(RecRuntimeThread(self)) return @logging_info def setup_spi(self, spi_mode: str): if spi_mode == self._spi_mode: return elif spi_mode == self.MODE_DISABLE: spi = None elif spi_mode == self.MODE_SELECTOR: spi = MultiExtMemSpiInterface(Selector.get(Selector.MEM_SELECTOR)) else: raise ValueError('illegal spi mode : ' + spi_mode) if self._spi is not None: self._spi.close() self._spi_mode = spi_mode self._spi = spi if spi is not None: self.log_verbose('reset spi', spi.__class__.__name__) try: spi.reset() except IOError as e: raise RuntimeError('spi reset fail') from e return @logging_info def reset_available_channel(self): spi_mode = self._spi_mode if spi_mode == self.MODE_DISABLE: available_channel = [] elif spi_mode == self.MODE_SINGLE: available_channel = [0] elif spi_mode == self.MODE_SELECTOR: if isinstance(self._spi, MultiExtMemSpiInterface): m = ExtMemManager(self._spi) available_channel = m.get_available_channel() else: self.log_warn('unexpected spi interface in mode', spi_mode) available_channel = [] else: self.log_warn('unknown spi mode', spi_mode) available_channel = [] self._available_channel = available_channel self.log_verbose('available channel', self._available_channel) return @logging_info def shutdown(self): for r in self._configurations: if self._configurations[r] is not None: self._configurations[r].shutdown() if self.database_process is not None: self.database_process.shutdown() if self.database_process.is_alive: self.log_verbose('Database-Process shutdown') # stop a process gracefully self.database_process.terminate() self.database_process.join() self.database_process = None if self._spi_thread is not None: self._spi_thread.close() self._spi_thread = None if self._rec_thread is not None: self._rec_thread.close() self._rec_thread = None if self.mqtt_thread is not None: self.mqtt_thread.shutdown() self.mqtt_thread = None self.shutdown_server() return def spi_mode(self) -> str: return self._spi_mode def available_channel(self, reset=False) -> List[int]: if reset: self.reset_available_channel() return self._available_channel @logging_info def unset_device(self, device: int): if device < 0: raise ValueError('illegal device ID : ' + str(device)) self.log_verbose('unset device', device) while len(self._configurations) <= device: self._configurations.append(None) # check is using c = self._configurations[device] if c is not None and c.sync_started: raise RuntimeError('illegal state, unset device during sync') self._configurations[device] = None return @logging_info def update_device_configuration(self, device: str, meta_file: Union[str, Path, RecordingMetaFile], data_format: Union[AnyStr, DataDecodeFormat], project_info: str): # meta file device_id = -1 # print('&&&&', meta_file) if isinstance(meta_file, str): meta_file = Path(meta_file) # if isinstance(meta_file, Path): # meta_file = RecordingMetaFile.load(meta_file, database = self.database_process) # elif isinstance(meta_file, RecordingMetaFile): # meta_file.reload() # else: # raise TypeError('not a recording meta file : ' + meta_file.__class__.__name__) # if isinstance(device, Device): # meta_file.device = json_stringify(device.as_json()) # device_id = device.device_id # if isinstance(device, str): # print('****', device) _device = json.loads(device) _parameter = json_stringify(_device['configuration']) _parent = json_stringify(_device['parent']) _recording_file_name = _device['recording_file_name'] mac_address = json.loads(device)['device_address'] device_id = _device['device_id'] del _device['configuration'] del _device['parent'] del _device['recording_file_name'] del _device['battery'] del _device['device_id'] # meta_file.device = json_stringify(_device) # meta_file.parameter = _parameter # meta_file.parent = _parent # meta_file.recording_file_name = _recording_file_name if device_id < 0: raise ValueError('illegal device ID : ' + str(device_id)) self.log_verbose('device ID', device_id) # project binding meta file # project_id = None # _project = None # if project_info != None: # _project = json.loads(project_info) # self.database_process.put_queue(['project_insert', device_id, _project]) # result = self._queue_ds_dict[int(device_id)].get() # if result[0] == 'project_id': project_id = project_info # while len(self._configurations) <= device_id: # self._configurationsappend(None) # check runtime has existed and is working # c = self._configurations[device_id] # if c is not None and c.sync_started: # raise RuntimeError('illegal state, update configuration during sync') # data format if isinstance(data_format, (str, bytes)): # validate data_format = DataDecodeFormat.parse(data_format).name else: data_format = data_format.name self.log_verbose('device meta file', meta_file) # data runtime self.log_verbose('device mac address', address_str(mac_address)) if mac_address == EMPTY_ADDRESS: self.log_verbose('use GeneratorDataRuntime') # data_runtime = GeneratorDataRuntime(self, device, meta_file) else: self.log_verbose('use DeviceDataRuntime') # data_runtime = DeviceDataRuntime(self, device_id, meta_file, data_format) # cleaer ds queue try: while True: self._queue_ds_dict[int(device_id)].get_nowait() except: pass if device_id in self._configurations and self._configurations[device_id] is not None: c = self._configurations[device_id] c.shutdown() c.join() self._spi.set_wait_flag(c._device, True) data_runtime_process = RecordingProcess('data_server', self.log_verbose, device_id, Queue(), self._queue_ds_dict[int(device_id)], self._queue_msg, meta_file, json_stringify(_device), _parameter, _parent, _recording_file_name, data_format, self.socket_mqtt_ip, self.socket_mqtt_port, project_id = project_id, database = self.database_process) # check channel if isinstance(data_runtime_process, RecordingProcess) and device_id not in self._available_channel: raise RuntimeError('set device configuration on un-available channel', device_id) # done self._configurations[device_id] = data_runtime_process return @logging_info def command_test(self): print('********************************************************************') @logging_info def is_sync(self, device: Optional[int] = None) -> bool: if device is None: return self.sync_started else: try: if device in self._configurations: return self._configurations[device].sync_started except (IndexError, TypeError, AttributeError): return False @property def sync_started(self) -> bool: return any(map(lambda r: r is not None, self._configurations)) @logging_info def start_sync(self, *device: Union[int, Device]): if len(device) == 0: self.log_info('start_sync all') self.mqtt_thread.broadcast_command('start:all') r = filter(lambda it: it is not None, self._configurations) else: r = [] for d in device: if isinstance(d, Device): d = d.device_id # try: c = None if d in self._configurations: c = self._configurations[d] # except IndexError as e: # raise RuntimeError('device not found : ' + str(d)) from e # else: if c is None: raise RuntimeError('device no meta : ' + str(d)) else: r.append(c) if len(r) == 0: return # sync timer # XXX we comment sync timer, because it is impossible sync the time stamp between devices # in current hardware architecture. # timer: Optional[Timer] = None # # for runtime in r: # # for any not started data runtime # if not runtime.sync_started: # if timer is None: # timer = runtime.data_timer # else: # runtime.data_timer = timer client = None # self._ensure_cache_client() try: # try: # if client is not None: # client.open_socket() # except RuntimeError: # client = None # start sync for runtime in r: if runtime is not None and not runtime.sync_started: # if client is not None: # client.drop_data(runtime.device) try: runtime.start() self.log_info('start_sync', runtime.device, '->', runtime.filepath) self.mqtt_thread.broadcast_command('start:' + str(runtime.device)) except: print_exception(self) finally: self._start_runtime() # if client is not None: # client.close_socket() # start runtime # self._start_runtime() return @logging_info def stop_sync(self, *device: Union[int, Device]): if len(device) == 0 and self.mqtt_thread is not None: self.log_info('stop_sync all') self.mqtt_thread.broadcast_command('stop:all') r = filter(lambda it: it is not None, self._configurations) else: r = [] for d in device: if isinstance(d, Device): d = d.device_id # try: c = None if d in self._configurations: c = self._configurations[d] # except IndexError: # pass # else: if c is not None: c.shutdown() c.join() # self.spi_reset_ram_header(c._device) self._spi.set_wait_flag(c._device, True) del self._configurations[d] self.log_info('stop_sync', c._device) self.mqtt_thread.broadcast_command('stop:' + str(c._device)) # r.append(c) # stop runtime, none device working if not self.sync_started: self._stop_runtime() return @logging_info def sync_file(self): for c in self._configurations: if self._configurations[c] is not None: if self._configurations[c].sync_started: self._configurations[c].sync_file_request = True else: self._configurations[c].sync_file() return def hardware_test(self) -> JSON_OBJECT: runtime = {} for r in self._configurations: if self._configurations[r] is not None: runtime[self._configurations[r].device] = rc = { 'data_format': r.data_format.name, 'file_path': str(r.filepath) } if isinstance(r, GeneratorDataRuntime): rc.update({ 'library': r.library }) # noinspection PyTypeChecker return { 'path': list(map(lambda path: str(path), self.repository.library_path)), 'spi_mode': self._spi_mode, 'channel': self._available_channel, 'runtime': runtime } def _start_runtime(self): if self._spi_thread is None or self._spi_thread.is_closed: self._spi_thread = self.run_thread(SpiRuntimeThread(self)) # for c in self._configurations: # if c is not None and c.sync_started is False: # c.start() if self._rec_thread is None or self._rec_thread.is_closed: self.log_verbose('start rec runtime thread') self._rec_thread = self.run_thread(RecRuntimeThread(self)) return def _stop_runtime(self): if self._spi_thread is not None: self._spi_thread.close() self._spi_thread = None return def open_cache_client(self): self.log_verbose('open cache client') try: client = CacheClient() client.open_socket() except RuntimeError as e: self.log_warn(e) else: client.clear() self._cache_client = client def _ensure_cache_client(self) -> Optional[CacheClient]: if self._cache_client is None: self.open_cache_client() if not self._cache_client.socket_opened: self._cache_client.open_socket() return self._cache_client def close_cache_client(self): client = self._cache_client self._cache_client = None if client is not None: self.log_verbose('close cache client') client.clear() client.close_socket() def get_spi_obj(self): return self._spi def whether_to_record(self, device): # if user click "start", return True; if user click "stop", return False; if device in self._configurations.keys() and self._configurations[device] is not None: # print(self._configurations.keys(), ',', device, datetime.now()) return True else: return False def recv_data_form_spi_new(self, device) -> bool: ret = False sync = self.get_spi_obj() busy = sync.get_pin_busy() # print('pin_busy=', busy, device, datetime.now()) if sync.get_pin_mem_req() == sync.get_pin_ram_sel(): spi_data = sync.recv_memory(device) signal = sync.get_pin_mem_req() sync.set_pin_mem_req(not signal) data = spi_data else: data = None print('data=None, mem_req!=ram_sel, [', sync.get_pin_mem_req(), ', ', sync.get_pin_ram_sel(), ']', device, datetime.now()) if data is not None: if self._configurations.get(device, None) != None: if self._configurations[device].queue_flag: self._configurations[device].queue_flag = False print("q size= ", self._configurations[device]._queue_rec.qsize()) if self._configurations.get(device, None) != None: self._configurations[device].put_rec_queue(data) # self._no_data_total_duration = 0 ret = True # else: # time_diff = time() - self._recv_memory_cost_time # self._no_data_total_duration += time_diff # if self._no_data_total_duration > 1: # print('data is None', device, datetime.now(), self._no_data_total_duration) # print("sync.get_pin_mem_req() != sync.get_pin_ram_sel()") # print("no~", sync.select, sync.get_pin_mem_req(), sync.get_pin_ram_sel(), datetime.now()) # print() # self._recv_memory_cost_time = time() return ret def rec_update(self) -> bool: ds_db_msg = self._queue_msg.get() with CacheClient() as client: if isinstance(ds_db_msg, List): if ds_db_msg[0] == 'ds': client.send_command('device_instruction', ds_db_msg[1], ds_db_msg[2]) else: self.mqtt_thread.broadcast_command('runtime error: ' + str(ds_db_msg)) content = { 'header': 'device_instruction/0', 'device': ds_db_msg, 'instruction': 'interrupt' } self.mqtt_thread.publish('device_instruction', json_stringify(content), inter = True) return True def show_data(self, device): self._configurations[device].put_rec_queue('show_data') @logging_info def tag(self, device: int, data_points: int): self._configurations[device].put_rec_queue('tag_data') class DataRuntime(metaclass=abc.ABCMeta): __slots__ = ('_server', '_device', '_meta_file', '_data_format', '_sync_started', 'sync_file_request', '_writer') def __init__(self, server: DataServer, device: int, meta_file: RecordingMetaFile, data_format: AnyStr): self._server = server self._device = device self._meta_file = meta_file self._data_format = data_format if isinstance(data_format, RecDataDecoder): data_format.device = device self.sync_file_request = False self._sync_started = False self._writer = None @property def server(self) -> DataServer: return self._server @property def device(self) -> int: return self._device @property def sync_started(self) -> bool: return self._sync_started @sync_started.setter def sync_started(self, value: bool): self._sync_started = value if value: self._open() else: self._close() @property def data_format(self) -> DataDecodeFormat: if isinstance(self._data_format, (str, bytes)): self._data_format = DataDecodeFormat.parse(self._data_format) # add device information if isinstance(self._data_format, RecDataDecoder): self._data_format.device = self.device return self._data_format @property def data_timer(self) -> Optional[Timer]: data_format = self.data_format if isinstance(data_format, RecDataDecoder): return data_format.timer else: return None @data_timer.setter def data_timer(self, value: Optional[Timer]): data_format = self.data_format if isinstance(data_format, RecDataDecoder) and value is not None: data_format.timer = value def get_data_message(self) -> Optional[str]: return self.data_format.message() @abc.abstractmethod def sync_data(self, data = None) -> Optional[List[bytes]]: """fetch data and write to the file""" pass def sync_file(self): meta = self._meta_file if meta.is_dirty: meta.write(database = self._server.database_process) w = self._writer if w is not None: w.flush() self.sync_file_request = False return @property def filepath(self) -> Path: return self._meta_file.filepath @property def meta_file(self) -> RecordingMetaFile: return self._meta_file @property def file_writer(self) -> Optional[RecordingFileWriter]: return self._writer def _open(self): self._close() # reset data decoder if isinstance(self._data_format, DataDecodeFormat): self._data_format = self.data_format.name data_format = self.data_format # reopen meta file meta = self._meta_file if meta is not None: # clean old recording files meta.remove_file() # change data format self._meta_file = meta.reopen(data_format.format) self._meta_file.write() ### self._writer = RecordingFileWriter(self._meta_file, -1, self._server.database_process) return def _close(self): if self._writer is not None: self._writer.close() if self._meta_file is not None: if self._meta_file.is_dirty: self._meta_file.write(database = self._server.database_process) self._writer = None # close timer if isinstance(self._data_format, DataDecodeFormat): self._data_format = self.data_format.name return class GeneratorDataRuntime(DataRuntime): __slots__ = ('_generator', '_library', '_start_time') def __init__(self, server: DataServer, device: int, meta_file: RecordingMetaFile): super().__init__(server, device, meta_file, NulDataDecoder.NAME) configuration = meta_file.configuration self._library = library = configuration.library options = GeneratorOptions() options.repository = server.repository if library == 'GENERATOR': options.set_generator('sin') else: options.set_generator(library) options.device = self._device options.channel = configuration.channel options.sample_rate = configuration.sample_rate options.amplitude = configuration.amp_gain options.configuration = configuration self._generator = options.get_generator() server.log_verbose('use generator', self._generator.__class__.__name__) self._start_time = None return @property def library(self) -> str: return self._library @property def time_pass(self) -> float: t = time() if self._start_time is None: self._start_time = t return 0.0 else: return t - self._start_time def sync_data(self, data = None) -> Optional[List[bytes]]: r = self._generator.get_data(self.time_pass) del data if r is None or len(r) == 0: return None else: # write to the file w = self.file_writer if w is not None: w.write(r) return [d.encode() for d in r] def get_data_message(self) -> Optional[str]: return self._generator.message() class DeviceDataRuntime(DataRuntime): """ **SPI data format** :: | | 1 | 2 | 3 | 012345678901234567890123456789012 --------------------------------- | 0xFF |counter| length| - SPI header |device | length| - notify header | data .........................| - notify data """ __slots__ = ( '_start_time', '_data_counter', '_last_data_counter', '_data_ch', '_data_ch_time', '_mqtt_send_data_ch_level', '_timer', '_timer_of_not_send', '_get_rec_data', '_cycle_start_time', '_time_stamp_list', '_prev_data', '_prev_delta_time', '_prev_time_stamp', '_task_fininsh') def __init__(self, server: DataServer, device: int, meta_file: RecordingMetaFile, data_format: AnyStr): super().__init__(server, device, meta_file, data_format) # data statistics self._start_time = None self._data_counter = Counter() self._last_data_counter = None self._mqtt_send_data_ch_level = {} self._timer = None #int(time() * 10) self._timer_of_not_send = None self._get_rec_data = False # self._time_stamp_list = [] self._cycle_start_time = [] self._prev_data = [] self._prev_delta_time = [] self._prev_time_stamp = [] self._task_fininsh = None @property def data_format(self) -> DataDecodeFormat: decoder = super().data_format # transmit calibration gain level to decoder if isinstance(decoder, TDC4VAF2DataDecoder): decoder.amp_gain = self.meta_file.configuration.amp_gain elif isinstance(decoder, TDC4VCDataDecoder): decoder.amp_gain = self.meta_file.configuration.amp_gain decoder._channel = self.meta_file.configuration.channel decoder._adc_clock = self.meta_file.configuration.get_parameter('ADC_CLOCK') decoder._axis_ch = self.meta_file.configuration.get_parameter('AXIS_CH') decoder._prev_data = self._prev_data decoder._prev_delta_time = self._prev_delta_time decoder._prev_time_stamp = self._prev_time_stamp return decoder def sync_data(self, data = None) -> Optional[List[bytes]]: if self._timer_of_not_send is None: self._timer_of_not_send = int(time()) # if self._timer is None: # self._timer = int(time() * 10) current_time = time() if self._timer is not None: if current_time - self._timer > 1.5: print('time, sync_data routine_time', current_time, current_time - self._timer) self._timer = current_time decoder = self.data_format is_rec = decoder.format == RecordingFileDataFormat.REC_DATA # print('sync_data') # print('data',data) # print('server/data', self._prev_delta_time,self._prev_time_stamp, self._prev_data) if data is None or len(data) == 0: result = decoder.decode(b'') if result is not None: ret = result else: return None else: ret = [] # print('decoder data') for offset, section in self._foreach_data_section(data): # print(offset, '/', len(data), hex_line(section)) # print(offset, '/', len(data)) result = decoder.decode(section) if result is not None: try: if isinstance(decoder, I4V4Z4T4DataDecoder) or isinstance(decoder, EISZeroOneDataDecoder): if decoder.isFinishMode is not None and decoder.isFinishMode() == 1 and self._task_fininsh is None: content = {} content['header'] = 'device_instruction/0' content['device'] = result.device content['instruction'] = 'interrupt' self._task_fininsh = True self.server.mqtt_thread.publish_local('device_instruction',json_stringify(content), True) # self.server.stop_sync(self.device) except RuntimeError as e: print(e) # self._timer = self._timer + 1 ### if len(self._mqtt_send_data_ch_level) == 0: for ch in result.channels(): self._mqtt_send_data_ch_level[ch] = MqttDataMessageHandler(self.server.mqtt_thread, 'data_server/device_data_stream/' + str(result.device) + '/' + str(ch) ) # for t, c, v in result.entry_iter(): # print('t,c,v', t,c,v) self._timer_of_not_send = int(time()) ret.append(result) else: continue if ret is None: return None # write back w = self.file_writer if w is not None and len(ret) > 0: if len(w.channel_list) == 0: w.channels_update(ret[0].channels()) w.write(ret, self._mqtt_send_data_ch_level) # mqtt_sender_list = [] # if len(ret) > 0: # mqtt_sender = MqttDataMessageHandler(self.server.mqtt_thread, 'data_server/device_data_stream') # mqtt_sender.on_message(ret) del data return ret def _foreach_data_section(self, data: bytes) -> Iterable[Tuple[int, bytes, int]]: """ :param data: :return: """ offset = 0 while offset + 3 < len(data): data_header = data[offset] if data_header != 0xFF: offset += 1 continue head_counter = data[offset + 1] data_length = data[offset + 2] device_id = data[offset + 3] if device_id != self._device: offset += 1 continue if _FLAG_DATA_LOST_STATS_ and self._last_data_counter is not None: self._data_counter[(head_counter - self._last_data_counter)] += 1 self._last_data_counter = head_counter section = data[offset + 3:offset + 3 + data_length] section[0] = head_counter yield offset, section, head_counter offset += 3 + data_length return def log_data_receive_statistics(self, logger: Logger): c = self._data_counter # fix overflow c[1] += c[-255] del c[-255] total = counter_product_sum(c) + 1 missing = counter_product_sum(c, key_mapper=lambda k: k - 1) if total != 0: logger.log_info(pc('data', WHITE), pc('receive', GREEN), total, pc('missing', RED), missing, pc('rate', YELLOW), '%.1f%%' % (100 * missing / total)) logger.log_verbose(c) return class DataRuntimeThread(ServerThread): def __init__(self, server: DataServer, device_id: int): super().__init__('DataRuntime', MAGENTA) self._server = server self._device_id = device_id # routine period control self._start_time = 0.0 self._time_stamp = 0.0 self._interval = 0.3 self._counter = 0 def setup(self) -> None: super().setup() self._start_time = self._time_stamp = time() self._counter = 0 return def shutdown(self) -> None: # self._server.close_cache_client() return True def routine(self, event = None) -> None: del ret return class SpiRuntimeThread(ServerThread): def __init__(self, server: DataServer): super().__init__('SpiRunTime', MAGENTA) self._server = server # routine period control self._start_time = 0.0 self._time_stamp = 0.0 self._interval = 0.045 self._timer = time() self._read_memory_board_idx = 0 def setup(self) -> None: super().setup() self._start_time = self._time_stamp = time() # self._server.open_cache_client() def shutdown(self) -> None: # self._server.close_cache_client() return True def routine(self, event = None) -> None: # routine 50 ms server = self._server self._timer = time() sync = server.get_spi_obj() # select channel devicelist = [4, 5, 7, 6] c = devicelist[self._read_memory_board_idx] sync.select = c if server.whether_to_record(c): sync.set_pin_mem_sel(False) ret = server.recv_data_form_spi_new(c) # read ram & put data into queue, if has data, ret = true self._read_memory_board_idx = 0 if self._read_memory_board_idx == 3 else self._read_memory_board_idx + 1 run_time = time() - self._timer real_run_time = run_time if run_time > 0.045: # print('time, recv_data_form_spi_routine_time', time(), run_time) run_time = 0.045 if run_time >= 0.030: print('time, recv_data_form_spi_routine_time >= 0.030', 'device:', c, datetime.now(), time(), real_run_time) event.wait(self._interval - run_time) sync.set_pin_mem_sel(True) sleep(0.005) return class RecRuntimeThread(ServerThread): def __init__(self, server: DataServer): super().__init__('RecRunTime', MAGENTA) self._server = server def setup(self) -> None: super().setup() # self._server.open_cache_client() def shutdown(self) -> None: # self._server.close_cache_client() return True def routine(self, event = None) -> None: server = self._server ret = server.rec_update() return # noinspection PyUnusedLocal class Main(CliMain): """Data server/client command interface. """ def __init__(self): super().__init__() @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_command('help') def _help_command(self, cmd: str, argv: List[str]): """print action help""" if len(argv) == 0: self._help('') elif argv[0] in ('help', '-h', '--help'): print(sys.argv[0], 'help', 'COMMAND') else: self.parsing([argv[0], '-h']) @cli_command("runserver") def _run_server_command(self, cmd: str, argv: List[str]): """run data server""" return _ServerCommand @cli_command("reset") def _reset_command(self, cmd: str, argv: List[str]): """reset server""" return _ClientSimpleCommand @cli_command("file") def _file_command(self, cmd: str, argv: List[str]): """start sync""" return _ClientFileCommand @cli_command("start") def _start_sync_command(self, cmd: str, argv: List[str]): """start sync""" return _ClientSimpleCommand @cli_command("stop") def _stop_sync_command(self, cmd: str, argv: List[str]): """stop sync""" return _ClientSimpleCommand def run(self): self._help('') return # noinspection PyUnusedLocal class _ServerCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.server_options = self.extend_options(DataServerOptions()) @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() def run(self): server = DataServer(self.server_options) server.main() return # noinspection PyUnusedLocal class _ClientSimpleCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() return def run(self): if self.command == 'reset': with DataClient() as client: client.reset() elif self.command == 'start': with DataClient() as client: client.start_sync() elif self.command == 'stop': with DataClient() as client: client.stop_sync() else: raise RuntimeError('unknown command ; ' + self.command) return # noinspection PyUnusedLocal class _ClientFileCommand(CliSubCommandMain): def __init__(self, command: str): super().__init__(command) self.device = None self.current_file_path = None self.flag_rename = False @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() return @cli_flags('-r') def _rename(self, flag: str): """just rename file""" self.flag_rename = True return @cli_arguments(0, value='DEVICE') def _device(self, pos: int, value: str): """target device""" self.device = int(value) return @cli_arguments('?', value='FILE') def _current_file_path(self, pos: int, value: str): """rename file""" self.current_file_path = value return def run(self): if self.device is None: raise RuntimeError('lost target device') with DataClient() as client: path = client.get_current_filepath(self.device) if self.current_file_path is None: print(path) else: meta = RecordingMetaFile.load(path) src = meta.filepath if self.flag_rename: tar = src.with_name(self.current_file_path) else: tar = Path(self.current_file_path) meta.change_filepath(tar) return if __name__ == '__main__': Main().main()