Roy
1400 lines
43 KiB
Python
1400 lines
43 KiB
Python
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=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]):
|
|
# 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)], self._queue_msg, 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 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')
|
|
|
|
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:
|
|
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()
|