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 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=3) 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_spi = queue.Queue() # self._queue_spi_dict = {} self._queue_ds_dict = {} self._queue_db = Queue() self._queue_db_error = Queue() # self._queue_ds = Queue() # if len(self._queue_spi_dict) == 0: # self._queue_spi_dict[4] = Queue() # self._queue_spi_dict[5] = Queue() # self._queue_spi_dict[6] = Queue() # self._queue_spi_dict[7] = 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_db_error) self.database_process.start() @logging_info def setup(self): self._available_channel.clear() if self._spi is not None: self.log_verbose('spi flush') self._spi.flush_all() 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() self._queue_spi.queue.clear() 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() with CacheClient() as client: client.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]): # 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'] 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 device_id = int(json.loads(device)['device_id']) if device_id < 0: raise ValueError('illegal device ID : ' + str(device_id)) self.log_verbose('device ID', device_id) # 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 data format', data_format) 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)], meta_file, json_stringify(_device), _parameter, _parent, _recording_file_name, data_format, self.socket_mqtt_ip, self.socket_mqtt_port, database = self.database_process ) # data_runtime_process.start() # 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 isinstance(runtime, DeviceDataRuntime): # self.spi_flush(runtime.device) # 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) self._configurations[d] = None self.log_info('stop_sync', c._device) self.mqtt_thread.broadcast_command('stop:' + str(c._device)) # r.append(c) # stop sync # for runtime in r: # if runtime is not None and runtime.sync_started: # self.log_info('stop_sync', runtime.device) # self.mqtt_thread.broadcast_command('stop:' + str(runtime.device)) # try: # runtime.shutdown() # except: # print_exception(self) # # finally: # # # self.spi_flush(runtime.device) # # if _FLAG_DATA_LOST_STATS_: # # if isinstance(runtime, DeviceDataRuntime): # # runtime.log_data_receive_statistics(self) # # runtime.shutdown() # 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._cache_client = None return 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 recv_data_form_spi(self) -> bool: """ """ current_time = time() ret = False devicelist = self._configurations.keys() devicelist = [4, 5, 7, 6] for c in devicelist: sync = self._spi sync.select = c # delay 120us ~ 270us count = 0 for i in range(300): count = count + 1 if c in self._configurations.keys() and self._configurations[c] is not None: data = self.spi_recv(c) if time() - current_time > 0.1: print('time, spi_recv', current_time, time() - current_time) current_time = time() if data is not None: try: if self._configurations[c].queue_flag: self._configurations[c].queue_flag = False print("q size= ", self._configurations[c]._queue_rec.qsize()) self._configurations[c].put_rec_queue(data) if time() - current_time > 0.05: print('time, queue', current_time, time() - current_time) current_time = time() # print('done put queue') except: return else: ret = True # else: # print('****1') return ret def rec_update(self) -> bool: device_id = self._queue_db_error.get() self.mqtt_thread.broadcast_command('runtime error: ' + str(device_id)) content = { 'header': 'device_instruction/0', 'device': device_id, 'instruction': 'interrupt' } self.mqtt_thread.publish('device_instruction', json_stringify(content), inter = True) return True def recv_data_foreach_runtime(self, device_id: int) -> bool: # """foreach data runtime to receive data. called by :class:`DataRuntimeThread`. # :return: has runtime received data # """ # q = None # data = None # try: # q = self._queue_spi_dict[device_id].get() # except: # return False # ret = False # if self._queue_spi_dict[device_id].qsize() > 10: # print('qsize: ', device_id, self._queue_spi_dict[device_id].qsize()) # for c in self._configurations: # if self._configurations[c] is not None and self._configurations[c].sync_started: # if self._configurations[c]._device == device_id: # try: # data = self._configurations[c].sync_data(q) # except EODInterrupt: # self._configurations[c].sync_started = False # except: # # catch any error and ignore it # print_exception(self) # self._configurations[c].sync_started = False # else: # if data is not None and len(data) > 0: # ret = True # # if self._configurations[c].sync_file_request: # # self._configurations[c].sync_file() # # for c in self._configurations: # # # print('_configurations', c) # # if c is not None and c.sync_started: # # if q[0] == c._device: # # try: # # # print('c.time', c._prev_data, c._prev_time_stamp, c._prev_delta_time) # # # c._timer_of_not_send = int(time()) # # data = c.sync_data(q[1]) # # except EODInterrupt: # # c.sync_started = False # # except: # # # catch any error and ignore it # # print_exception(self) # # c.sync_started = False # # else: # # if data is not None and len(data) > 0: # # ret = True # # if c.sync_file_request: # # c.sync_file() # del q # del data return def spi_flush(self, device: int): """clean spi buffer""" if self._spi is not None: self._spi.select = device self._spi.flush() return def spi_reset_ram_header(self, device: int): if self._spi is not None: self._spi.select = device self._spi.request_data() sleep(0.01) self._spi.reset() self._spi.flush() return def spi_send(self, device: int, address: int, data: bytes): sync = self._spi if sync is None: return sync.select = device sync.write_memory(address, data) return def spi_recv(self, device: int) -> Optional[bytes]: sync = self._spi if sync is None: return None if self._is_usb_port(device): # if self._is_demo_usb_port(device): # self._usb will be a list when suporting multi usb device sync = self._usb if sync is None: return None data = sync.recv_memory() if data is None: return None data_length = len(data) + 1 header = 0xFF data_counter = 0x00 data_header = [header, data_counter, data_length, device] data_header.extend(data) return bytes(data_header) else: sync = self._spi if sync is None: return None # sync.select = device # sleep(0.003) if sync.changed(flip=True): spi_data = sync.recv_memory(device) return spi_data # if sync.get_wait_flag(device) is False: # return spi_data # else: # sync.set_wait_flag(device, False) # print("discard first data") # return None # return sync.recv_memory() else: sync.request_data() return None def _is_usb_port(self, device: int) -> bool: max_spi_port_number = 8 if device >= max_spi_port_number: return True else: return False sync.select = device if sync.changed(flip=True): # print('recv_memory~~~') return sync.recv_memory() else: # print('request_data___') return sync.request_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): # get amp_gain from meta file decoder.amp_gain = self.meta_file.configuration.amp_gain elif isinstance(decoder, TDC4VCDataDecoder): # get amp_gain from meta file 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._time_stamp_list = self._time_stamp_list 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 # elif isinstance(decoder, I4V4Z4T4DataDecoder): # get cycle_time from meta file # decoder._mode = self.meta_file.configuration.MODE # decoder._cycle_start_time = self._cycle_start_time elif isinstance(decoder, NeuliveThreeOneDataDecoder): # get amp_gain from meta file 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._time_stamp_list = self._time_stamp_list 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: # start = time() # ret = self._server.recv_data_foreach_runtime(self._device_id) # self._counter += 1 # if self._counter > 10000: # gc.collect() # self._counter = 0 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.15 self._timer = time() 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: server = self._server self._timer = time() ret = server.recv_data_form_spi() run_time = time() - self._timer if run_time > self._interval: print('time, recv_data_form_spi_routine_time', time(), run_time) if run_time > 0.15: run_time = 0.15 if server.sync_started: event.wait((self._interval - run_time)) else: self.close() return self._start_time = time() 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()