Compare commits

..

4 Commits

15 changed files with 234 additions and 65 deletions
+5 -3
View File
@@ -87,7 +87,8 @@ class DataAPI(metaclass=abc.ABCMeta):
def update_device_configuration(self,
device: Union[int, Device],
meta_file: Union[str, Path, RecordingMetaFile],
data_format: Union[AnyStr, DataDecodeFormat]):
data_format: Union[AnyStr, DataDecodeFormat],
project_info: str):
"""set device configuration.
:param device: device ID
@@ -154,7 +155,8 @@ class DataClient(SocketClient, DataAPI, metaclass=SocketClientMacro(DataAPI)):
def update_device_configuration(self,
device: Union[int, Device],
meta_file: Union[str, Path, RecordingMetaFile],
data_format: Union[AnyStr, DataDecodeFormat]):
data_format: Union[AnyStr, DataDecodeFormat],
project_info: str):
# meta file
if isinstance(meta_file, str):
@@ -186,7 +188,7 @@ class DataClient(SocketClient, DataAPI, metaclass=SocketClientMacro(DataAPI)):
raise TypeError('not a data decode format : ' + data_format.__class__.__name__)
# send
self.send_command('update_device_configuration', device, meta_file, data_format)
self.send_command('update_device_configuration', device, meta_file, data_format, project_info)
def is_sync(self, device: Union[None, int, Device] = None) -> bool:
if isinstance(device, Device):
+10
View File
@@ -0,0 +1,10 @@
from sqlalchemy import create_engine
# from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, declarative_base
SQLALCHEMY_DATABASE_URL = "postgresql://biopro:BioProControlBox@127.0.0.1:5432/postgres"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
Session = sessionmaker(bind=engine)
Base = declarative_base()
+21
View File
@@ -0,0 +1,21 @@
from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
from sqlalchemy.sql import select, func
from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
from sqlalchemy.dialects.postgresql import JSONB
from .base import Base
class MetaProjectInfo(Base):
__tablename__ = "meta_project_info"
id = Column(Integer, primary_key=True)
project = Column(String(36))
cycle = Column(String)
task = Column(JSONB)
serial_number = Column(Integer)
deleted = Column(Boolean, default = False)
created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
# def __repr__(self):
# return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"
+24
View File
@@ -0,0 +1,24 @@
from sqlalchemy import Table, Column, String, MetaData, ForeignKey, JSON
from sqlalchemy.sql import select, func
from sqlalchemy.types import Integer, BigInteger, String, Boolean, TIMESTAMP, Numeric
from sqlalchemy.dialects.postgresql import JSONB
from .base import Base
class ProjectReport(Base):
__tablename__ = "project_report"
id = Column(Integer, primary_key=True)
name = Column(String)
desc = Column(String)
task = Column(JSONB)
cycle = Column(JSONB)
device = Column(JSONB)
uuid = Column(String(36))
user_auth = Column(JSONB)
deleted = Column(Boolean, default = False)
created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
updated_at = Column(TIMESTAMP(timezone=True), onupdate=func.now())
# def __repr__(self):
# return f"User(id={self.id!r}, name={self.name!r}, fullname={self.task!r})"
+15 -12
View File
@@ -574,20 +574,23 @@ class DeviceManager(MasterDevice, Synchronized):
:param device: device ID or response info
:return: CompletedDevice, None if not found.
"""
if isinstance(device, int):
for slave in self._device:
if slave.device_id == device:
return slave
try:
if isinstance(device, int):
for slave in self._device:
if slave.device_id == device:
return slave
elif isinstance(device, str):
for slave in self._device:
if ':'.join('{:02x}'.format(b) for b in slave.mac_address) == device.lower():
return slave
elif isinstance(device, str):
for slave in self._device:
if ':'.join('{:02x}'.format(b) for b in slave.mac_address) == device.lower():
return slave
elif isinstance(device, DeviceInfo):
for slave in self._device:
if device.match(slave):
return slave
elif isinstance(device, DeviceInfo):
for slave in self._device:
if device.match(slave):
return slave
except BaseException as e:
self.log_error('device not found', str(device))
return None
+1 -1
View File
@@ -738,7 +738,7 @@ class DeviceCommonInstruction:
elif instruction == cls.START:
return '_data_format("TDC4VAF2")', '_notify(True)', 'VIS_STI', '_sync(True)'
elif instruction == cls.INTERRUPT:
return '_sync(False)', '_notify(False)', 'VIS_INT'
return '_notify(False)', 'VIS_INT', '_sync(False)'
elif instruction == cls.CLOSE:
return '_sync(False)', '_notify(False)', 'VIS_INT'
elif instruction == cls.CALL:
+12 -1
View File
@@ -968,7 +968,7 @@ class CompletedDevice(Device):
"""
__slots__ = ('_master', '_device_id', '_device', '_library', '_context',
'_parameter', '_configuration', '_lock', '_feature_mask',
'_cache_battery', '_cache_battery_timestamp', '_coeff', '_status')
'_cache_battery', '_cache_battery_timestamp', '_coeff', '_status', '_occupied_by_project')
def __init__(self, master: MasterDevice, library: DeviceLibrary, device_id: int, device: Device):
"""
@@ -986,6 +986,8 @@ class CompletedDevice(Device):
# device status ( 0: idle, 1: working, 2: error, 3: power off)
self._status = 0
# device be occupied by project
self._occupied_by_project = None
# context
s1 = library.constant
@@ -1082,6 +1084,14 @@ class CompletedDevice(Device):
def status(self, new_status):
self._status = new_status
@property
def occupied_by_project(self) -> str:
return self._occupied_by_project
@occupied_by_project.setter
def occupied_by_project(self, new_occupied_by_project):
self._occupied_by_project = new_occupied_by_project
# device parameter getter/setter
def parameters(self) -> List[str]:
@@ -1303,6 +1313,7 @@ class CompletedDevice(Device):
'recording_file_name': self.recording_file_name,
'library_name': self._library.name,
'library_version': str(self._library.version),
'occupied_by_project': self._occupied_by_project,
'configuration': self.configuration.as_json(list_hide=True),
}
+26 -2
View File
@@ -66,6 +66,7 @@ class Project(threading.Thread):
for device in device_list:
mac_address = device_list[device]['pair']
complete_device = self._device_manager.get_device(mac_address)
complete_device.occupied_by_project = self._uuid
self._complete_device[device] = complete_device
@property
@@ -136,6 +137,10 @@ class Project(threading.Thread):
def running_task(self):
return self._task_manager.running_task
def get_device_parameter_set(self, device_address):
parameter_set = self._task_manager.running_task.parameter_set
return next((x for x in parameter_set if parameter_set[x]['target'] == device_address), None)
def run(self):
self._status = 1
self._start_time = time()
@@ -161,6 +166,7 @@ class Project(threading.Thread):
for condition in match_condition_list:
match_action_list = task.get_match_action(condition.id)
print('match_action_list', match_action_list, match_action_list)
for action in match_action_list:
if action.type == 'cycle' and condition.type == 'previous_task_done':
self.mqtt_thread.broadcast_command('project:task ' + task.name + ' start at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
@@ -184,6 +190,8 @@ class Project(threading.Thread):
elif action.type == 'idle':
self._task_manager.running_task.stop()
# elif action.type == 'stop':
# self._task_manager.running_task.stop()
instruction_set = None
@@ -223,7 +231,7 @@ class Project(threading.Thread):
task_info['file_name'] += '#' + str(cycle_info['name']) + '-' + str(cycle_info['count'])
self.log_verbose('file name ' + task_info['file_name'])
if instruction_set != None and len(task.parameter_set) > 0:
if instruction_set != None and self.get_device_parameter_set(action.target) != None:
if action.type == 'start':
self._count += 1
for instruction in instruction_set:
@@ -253,6 +261,8 @@ class Project(threading.Thread):
pass
def stop(self):
for device in self._complete_device:
self._complete_device[device].occupied_by_project = None
self._task_manager.running_task.stop()
self.mqtt_thread.broadcast_command('project:task ' + str(self._task_manager.running_task.name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
self._status = 2
@@ -261,6 +271,8 @@ class Project(threading.Thread):
self.mqtt_thread.broadcast_command('project:project ' + str(self._name) + ' stop at ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
def close(self):
for device in self._complete_device:
self._complete_device[device].occupied_by_project = None
self._status = 2
self._end_time = time()
self._stop_flag = True
@@ -319,6 +331,18 @@ class Project(threading.Thread):
'device': self._device,
'task': self.task_list,
'running_task': running_task,
'cycle': self._cycle
'cycle': self._cycle,
'count': self._count,
}
return data
def info_pass_data_server(self):
if self._task_manager.running_task is not None:
running_task = self._task_manager.running_task.as_json()
data = {
'project': self._uuid,
'task': running_task,
'cycle': self._cycle,
'serial_number': self._count,
}
return data
+15 -4
View File
@@ -118,7 +118,7 @@ class RecordingMetaFile(JsonSerialize):
'_dirty', '_last_modify_time', '_file_size_cache',
'_recording_file_ch', '_recording_sub_file', '_size',
'_parameter', '_parent', '_recording_file_name', '_recording_sub_mini', '_recording_mini_ch',
'_raw_serial_number', '_mini_serial_number', '_device_id')
'_raw_serial_number', '_mini_serial_number', '_device_id', '_project_id')
def __init__(self, filepath: Union[str, Path], data_format: bytes = None, id_db = 0, database = None):
if isinstance(filepath, str):
@@ -153,6 +153,7 @@ class RecordingMetaFile(JsonSerialize):
self._parent = ''
self._recording_file_name = ''
self._device_id = -1
self._project_id = None
# recording parameter
self._configuration = None
@@ -297,6 +298,7 @@ class RecordingMetaFile(JsonSerialize):
'uuid': str(self._file_uuid),
'data_format': 'rec',
'device': self._device,
'project': self._project_id,
}
if len(self._parent) > 0:
@@ -462,7 +464,7 @@ class RecordingMetaFile(JsonSerialize):
def parameter(self) -> str:
return self._parameter
@device.setter
@parameter.setter
def parameter(self, parameter):
self._parameter = parameter
return None
@@ -471,7 +473,7 @@ class RecordingMetaFile(JsonSerialize):
def parent(self) -> str:
return self._parent
@device.setter
@parent.setter
def parent(self, parent):
self._parent = parent
@@ -479,7 +481,7 @@ class RecordingMetaFile(JsonSerialize):
def recording_file_name(self) -> str:
return self._recording_file_name
@device.setter
@recording_file_name.setter
def recording_file_name(self, recording_file_name):
self._recording_file_name = recording_file_name
@@ -492,6 +494,15 @@ class RecordingMetaFile(JsonSerialize):
self._configuration = conf
self._dirty = True
return None
@property
def project_id(self) -> int:
return self._project_id
@project_id.setter
def project_id(self, project: int):
self._project_id = project
return None
@property
def channels(self) -> List[int]:
+18 -10
View File
@@ -4,7 +4,6 @@ from time import time, sleep
from typing import Iterable
from datetime import datetime
import gc
import base64
from biopro.data import *
@@ -27,7 +26,6 @@ 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
@@ -317,7 +315,8 @@ class DataServer(SocketServer, DataAPI):
def update_device_configuration(self,
device: str,
meta_file: Union[str, Path, RecordingMetaFile],
data_format: Union[AnyStr, DataDecodeFormat]):
data_format: Union[AnyStr, DataDecodeFormat],
project_info: str):
# meta file
device_id = -1
@@ -345,6 +344,8 @@ class DataServer(SocketServer, DataAPI):
_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']
@@ -354,13 +355,22 @@ class DataServer(SocketServer, DataAPI):
# 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)
# 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 = result[1]
# while len(self._configurations) <= device_id:
# self._configurationsappend(None)
@@ -380,8 +390,6 @@ class DataServer(SocketServer, DataAPI):
else:
data_format = data_format.name
# self.log_verbose('device data format', data_format)
self.log_verbose('device meta file', meta_file)
# data runtime
@@ -406,15 +414,15 @@ class DataServer(SocketServer, DataAPI):
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()
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)
+30 -1
View File
@@ -550,4 +550,33 @@ class DataBaseProcess(Process):
except psycopg2.Error as e:
print('e', e)
return True
return True
def project_insert(self, device_id, _data):
sql_str = 'INSERT INTO "public"."meta_project_info" ('
sql_set = []
for item in _data.keys():
sql_str = sql_str + str(item) + ', '
if isinstance(_data[item], dict) or isinstance(_data[item], list):
sql_set.append(json_stringify(_data[item]))
else:
sql_set.append(_data[item])
sql_str = sql_str[0:-2] + ') VALUES ('
for item in _data.keys():
sql_str = sql_str + '%s, '
sql_str = sql_str[0:-2] + ') RETURNING id'
sql_cursor = self._psql_conn.cursor()
try:
sql_cursor.execute(sql_str, sql_set)
except BaseException as e:
print('insert error', e)
self._psql_conn.commit()
sql_cursor.close()
self._queue_error.put()
finally:
self._queue_ds[int(device_id)].put(['project_id', int(sql_cursor.fetchone()[0])])
self._psql_conn.commit()
sql_cursor.close()
+39 -17
View File
@@ -1,6 +1,7 @@
from time import sleep
from typing import Iterable
from datetime import datetime
import json
import biopro.impl.vcgencmd as vcg
from biopro.data import DataServerOptions, DataAPI
@@ -31,8 +32,10 @@ from biopro.api.auth import AuthAPI
from biopro.api.controller import ControllerAPI
from biopro.api.device import DeviceAPI
from biopro.project.project_manager import ProjectManager
### do not import Websocket mode
###from .websocket import WebsocketThread
from biopro.db.base import Base, Session, engine
from biopro.db.project_report import ProjectReport
from biopro.db.meta_project_info import MetaProjectInfo
_RUNTIME_COMPILE = False
@@ -117,8 +120,8 @@ class ControlServer(SocketServer, ControlServerAPI):
self.ups_thread = UPSDetectThread(self, c.flag_ups_wait_time)
# user session
self._session = {}
self._user_session_expire = c.user_session_expire_days * 24 * 60 * 60 # sec
# self._session = {}
# self._user_session_expire = c.user_session_expire_days * 24 * 60 * 60 # sec
# indicate those device together
self._device_instruction_group = None
@@ -129,6 +132,10 @@ class ControlServer(SocketServer, ControlServerAPI):
# _api request build
self._api = ApiRequests()
# sqlalchemy
Base.metadata.create_all(engine)
self._session = Session()
# init method
def _init_file_manager(self, option: ControlServerOptions) -> FileManager:
@@ -820,16 +827,19 @@ class ControlServer(SocketServer, ControlServerAPI):
@logging_info
def device_battery(self, device:int) -> List[int]:
battery = {}
if device == 'all':
device_list = self.device_manager.list_device()
if len(device_list) > 0:
battery = list(map(lambda x: {"device": x.memory_board, "battery":self.device_manager.get_device(x).battery}, device_list))
else:
battery = {
"device": device,
"battery": self.device_manager.get_device(device).battery
}
try:
battery = {}
if device == 'all':
device_list = self.device_manager.list_device()
if len(device_list) > 0:
battery = list(map(lambda x: {"device": x.memory_board, "battery":self.device_manager.get_device(x).battery}, device_list))
else:
battery = {
"device": device,
"battery": self.device_manager.get_device(device).battery
}
except AttributeError as e:
print('device_battery', device, 'not found', e)
return battery
@logging_info
@@ -1021,8 +1031,17 @@ class ControlServer(SocketServer, ControlServerAPI):
@logging_info
def run_project(self, project) -> bool:
if project is not None:
# print(project)
project = self._project_manager.create(project)
with self._session as session:
project_report = ProjectReport(
name = project.name,
uuid = project.uuid,
cycle = project.cycle,
task = project.task_list,
device = project.device
)
session.add(project_report)
session.commit()
self._project_manager.run_project(project)
return project.as_json()
@@ -1078,6 +1097,7 @@ class ControlServer(SocketServer, ControlServerAPI):
self.log_verbose('update data format', value)
client = self.data_server.client()
project = None
if client is not None:
info = self.file_manager.use(device)
@@ -1088,9 +1108,12 @@ class ControlServer(SocketServer, ControlServerAPI):
filename = self._file_save_cache.get(device.device_id, self.TEMP_FILENAME_EXPR)
filename = format_filename(filename, device)
info = self.file_manager.save(device, filename)
if device.occupied_by_project != None:
project = json.dumps(self.project_manager.get(device.occupied_by_project).info_pass_data_server())
with client:
client.update_device_configuration(device, info.meta_file, value)
client.update_device_configuration(device, info.meta_file, value, project)
def _device_set_disable_cache(self, device: CompletedDevice, disable):
if disable:
@@ -1124,7 +1147,6 @@ class ControlServer(SocketServer, ControlServerAPI):
filename = format_filename(self.TEMP_FILENAME_EXPR, device)
info = self.file_manager.save(device, filename)
client.update_device_configuration(device, info.meta_file, RawDataDecoder())
if self._device_instruction_group is None:
client.start_sync(device)
+4 -1
View File
@@ -44,7 +44,9 @@ class RecordingProcess(Process):
def __init__(self, client_id: str, log_verbose, device_id, queue_rec, queue_ds, queue_msg, meta_path,
device_info, parameter, parent, recording_file_name, data_format,
mqtt_ip, mqtt_port,
name = 'Recording-Process', database = None):
name = 'Recording-Process',
project_id = None,
database = None):
super(RecordingProcess, self).__init__(name = name)
self._database = database
@@ -57,6 +59,7 @@ class RecordingProcess(Process):
self._meta_file._parent = parent
self._meta_file._recording_file_name = recording_file_name
self._meta_file._device_id = device_id
self._meta_file._project_id = project_id
print(self._meta_file._device, self._meta_file._parameter, self._meta_file._parent, self._meta_file._filepath)
self._meta_file.write(self._database)
@@ -33,6 +33,7 @@
0,
1,
1,
1,
1
],
"domain": {
@@ -617,7 +618,8 @@
"Open Circuit Potential",
"Pulse Sensing",
"Differential Pulse Voltammetry (DPV)",
"Chronopotentiometry"
"Chronopotentiometry",
"Idle"
]
},
"VOLT_ORIGIN": {
@@ -949,6 +951,7 @@
"10": "cali_dac_test",
"11": "cali_adc_test",
"12": "",
"17": "start_data_with_empty_ins",
"*": "start_data"
}
}
@@ -976,6 +979,11 @@
"VIS_STI",
"_cdr('1X;4X>ADC_VALUE_I')"
],
"start_data_with_empty_ins": [
"data_format",
"_sync(True)",
"VIS_STI"
],
"start_data": [
"data_format",
"_notify(True)",
@@ -290,7 +290,7 @@
"auto"
]
},
"RT_VOLT_SET": {
"DAC_VOLT": {
"description": "DAC output Voltage",
"record_meta": true,
"initial": 37500,
@@ -299,8 +299,7 @@
],
"value": {
"expression": "VALUE"
},
"on_change": "set_para_RT_VOLT_SET"
}
},
"BLE_WRITE": {
"description": "send msg to elite",
@@ -354,7 +353,7 @@
"1": "curve_cv3",
"2": "curve_const_vscan",
"3": "curve_vt",
"4": "curve_rt"
"4": "curve_zt"
}
},
{
@@ -487,10 +486,10 @@
"2B>pe"
]
},
"curve_rt": {
"curve_zt": {
"type": "RIS",
"parameter": {
"va": "RT_VOLT_SET",
"va": "DAC_VOLT",
"pa": "ADC_LEVEL_I_15",
"pb": "ADC_LEVEL_V_IN_15",
"pd": "CTRL_HIGH_Z_15",
@@ -521,12 +520,6 @@
"1B>va"
]
},
"set_para_RT_VOLT_SET": {
"type": "RIS",
"data": [
"XE2;X01;2B>RT_VOLT_SET"
]
},
"ble_instru_send": [
"ble_write",
"_cdr('20X>ADC_VALUE_I')"