peterlu14
490 lines
12 KiB
Python
490 lines
12 KiB
Python
"""This module just define the biopro.exp_pro module interface.
|
|
The actual implement was put at another module
|
|
"""
|
|
|
|
import abc
|
|
import datetime
|
|
from typing import Optional, Any, overload, List, Dict, Union, Tuple, TypeVar
|
|
|
|
from biopro.util.promise import *
|
|
|
|
STATUS_INIT = 0
|
|
STATUS_SETUP = 1
|
|
STATUS_READY = 2
|
|
STATUS_START = 3
|
|
STATUS_SUSPEND = 4 # suspend
|
|
STATUS_COMPLETE = 10 # normal exit
|
|
STATUS_INTERRUPTED = 11 # interrupted
|
|
STATUS_ERROR = -1 # error exit
|
|
STATUS_UNKNOWN = -2 # scheduler not init
|
|
|
|
ADDRESS = Tuple[int, int, int, int, int, int]
|
|
|
|
T = TypeVar('T')
|
|
R = TypeVar('R')
|
|
|
|
|
|
class DeviceInfo(metaclass=abc.ABCMeta):
|
|
"""Device Information"""
|
|
|
|
__slots__ = ()
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def device_id(self) -> Optional[int]:
|
|
"""device number."""
|
|
pass
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def device_name(self) -> str:
|
|
"""device name."""
|
|
pass
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def mac_address(self) -> ADDRESS:
|
|
"""device max address"""
|
|
pass
|
|
|
|
|
|
class ExpApi(metaclass=abc.ABCMeta):
|
|
"""Experimental protocol API, make user can access/control the controller to do something.
|
|
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def log_verbose(self, *message: Any):
|
|
"""logging message at verbose level"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def log_info(self, *message: Any):
|
|
"""logging message at info level"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def log_warn(self, *message: Any):
|
|
"""logging message at warn level"""
|
|
pass
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def user(self) -> str:
|
|
"""get current user name"""
|
|
pass
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def root_directory(self) -> str:
|
|
pass
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def use_directory(self) -> str:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def change_directory(self, path: str) -> str:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def current_timestamp(self) -> int:
|
|
"""get current timestamp.
|
|
|
|
:return: timestamp [s]
|
|
"""
|
|
pass
|
|
|
|
def current_datetime(self) -> datetime.datetime:
|
|
"""get current datetime
|
|
|
|
:return:
|
|
"""
|
|
return datetime.datetime.fromtimestamp(self.current_timestamp())
|
|
|
|
@abc.abstractmethod
|
|
def schedule_time(self, timestamp: Union[float, datetime.datetime]):
|
|
"""set next wake up time at *timestamp*
|
|
|
|
:param timestamp: timestamp [s] or datetime or promise
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def schedule_sleep(self, sleep: Union[float, datetime.timedelta]):
|
|
"""set next wake up time after *sleep*
|
|
|
|
:param sleep:
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def make_device_info(self, *,
|
|
device_name: str = None,
|
|
mac_address: Union[str, ADDRESS] = None) -> DeviceInfo:
|
|
"""create a device info.
|
|
|
|
:param device_name: device name set in the user repository, alias allow
|
|
:param mac_address: device mac address
|
|
:return: device information
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def found_device(self) -> List[DeviceInfo]:
|
|
"""get scanned device.
|
|
|
|
:return: list of device response information
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def connected_device(self) -> List[DeviceInfo]:
|
|
"""get connected device.
|
|
|
|
:return: list of device information
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def ensure_device(self, *device_list: DeviceInfo) -> bool:
|
|
"""ensure device found and connected.
|
|
|
|
**In STATUS_INIT state**
|
|
|
|
It only check device connected and gain the lock.
|
|
|
|
**In STATUS_SETUP state**
|
|
|
|
It will set the device number according the *device_list* index, and make the parameter *device*
|
|
in some method available.
|
|
|
|
It will try to gain the experimental lock for user. (not implemented yet)
|
|
|
|
:param device_list: device list
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def all_device(self) -> List[DeviceInfo]:
|
|
"""all locked device, set by `ensure_device`_ .
|
|
|
|
:return: device used by this experimental
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def get_device(self, device: int) -> DeviceInfo:
|
|
"""
|
|
|
|
:param device: device number
|
|
:return: device used by this experimental
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def release_device(self, device_list: Optional[List[DeviceInfo]] = None, disconnect=True):
|
|
"""release locked device.
|
|
|
|
:param device_list: release device, None for all
|
|
:param disconnect: if True, also disconnect the device.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def apply_setup(self, name: str):
|
|
"""apply experimental setup from user setting.
|
|
|
|
It equals to do `ensure_device`_ and `device_set_parameter`_
|
|
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_get_parameter(self, device: Union[int, DeviceInfo], parameter: str) -> Any:
|
|
"""get the *parameter* from the *device*
|
|
|
|
:param device: device number
|
|
:param parameter: parameter name
|
|
:return: parameter P value
|
|
"""
|
|
pass
|
|
|
|
@overload
|
|
def device_set_parameter(self, device: Union[int, DeviceInfo], parameter: Dict[str, Any]):
|
|
pass
|
|
|
|
@overload
|
|
def device_set_parameter(self, device: Union[int, DeviceInfo], parameter: str):
|
|
pass
|
|
|
|
@overload
|
|
def device_set_parameter(self, device: Union[int, DeviceInfo], p: str, v: Any):
|
|
pass
|
|
|
|
@overload
|
|
def device_set_parameter(self,
|
|
device: Union[int, DeviceInfo],
|
|
p1: str, v1: Any,
|
|
p2: str, v2: Any):
|
|
pass
|
|
|
|
@overload
|
|
def device_set_parameter(self,
|
|
device: Union[int, DeviceInfo],
|
|
p1: str, v1: Any,
|
|
p2: str, v2: Any,
|
|
p3: str, v3: Any):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_set_parameter(self, device: Union[int, DeviceInfo], *parameter):
|
|
"""set device parameter.
|
|
|
|
the *parameter* could be:
|
|
|
|
1. **str** configuration name store in user repository.
|
|
2. **Dict[str, Any]** parameter table
|
|
3. **str, value, ...** parameter sequence.
|
|
|
|
**error raise**
|
|
|
|
1. general case for device
|
|
|
|
* `RuntimeError`_ device not found
|
|
* `RuntimeError`_ device un-locked
|
|
|
|
#. general case for parameter setting
|
|
|
|
see :func:`biopro.devlib.library.DeviceParameter.set_parameter` for more detail
|
|
|
|
#. for **str** configuration name case
|
|
|
|
* `RuntimeError`_ not a meta file
|
|
* `FileNotFoundError`_
|
|
|
|
#. for **str, value, ...** parameter sequence.
|
|
|
|
* `ValueError`_ illegal parameter count
|
|
* `ValueError`_ illegal parameter name type
|
|
|
|
:param device: device number
|
|
:param parameter:
|
|
:return:
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_to_value(self, device: Union[int, DeviceInfo], parameter: str, value: Any) -> Any:
|
|
"""cast P value to V value for device parameter.
|
|
|
|
:param device: device number
|
|
:param parameter: parameter name
|
|
:param value: P value
|
|
:return: V value
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_to_parameter(self, device: Union[int, DeviceInfo], parameter: str, value: Any) -> Any:
|
|
"""cast V value to P value for device parameter.
|
|
|
|
:param device: device number
|
|
:param parameter: parameter name
|
|
:param value: V value
|
|
:return: P value
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_call_instruction(self, device: Union[int, DeviceInfo], instruction: str):
|
|
"""device invoke instruction.
|
|
|
|
:param device: device number
|
|
:param instruction: instruction
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_use_filepath(self, device: Union[int, DeviceInfo]) -> Optional[str]:
|
|
"""get device saving filepath
|
|
|
|
:param device: device number
|
|
:return: filepath
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_set_filepath(self, device: Union[int, DeviceInfo], filename: str) -> str:
|
|
"""set device saving filepath
|
|
|
|
:param device: device number
|
|
:param filename: filename
|
|
:return: filepath
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def device_unset_filepath(self, device: Union[int, DeviceInfo]) -> Optional[str]:
|
|
"""unset/clean device saving filepath
|
|
|
|
:param device: device number
|
|
:return: filepath
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def set_complete(self):
|
|
"""mark protocol complete. """
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def set_error(self, message: str):
|
|
"""mark protocol terminated with error. """
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def file_export(self,
|
|
ftype: str,
|
|
*filepath: Union[str, DeviceInfo],
|
|
**options) -> Promise[List[Union[str, DeviceInfo]]]:
|
|
"""export file.
|
|
|
|
:param ftype: export file type
|
|
:param filepath: source filepath or device info
|
|
:param options: export options
|
|
:return: promise of output filepath
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def file_archive(self,
|
|
filename: str,
|
|
*filepath: Union[str, DeviceInfo],
|
|
ftype: Optional[str] = None) -> Promise[str]:
|
|
"""archive/compress file.
|
|
|
|
if *filepath* is empty, use the exported path for all device.
|
|
|
|
:param filename: output filename
|
|
:param filepath: source filepath or device info
|
|
:param ftype: archive file type
|
|
:return: promise of output filepath
|
|
"""
|
|
pass
|
|
|
|
|
|
DEVICE = str
|
|
'''available device name list'''
|
|
|
|
SETUP = str
|
|
'''available setup name list'''
|
|
|
|
CONFIG = str
|
|
'''available configuration name list'''
|
|
|
|
META = dict
|
|
'''meta file configuration'''
|
|
|
|
FILE_EXPORT = str
|
|
'''file export type'''
|
|
|
|
FILE_ARCHIVE = str
|
|
'''file archive type'''
|
|
|
|
FILE_NAME = str
|
|
'''file name'''
|
|
|
|
FILE_PATH = str
|
|
'''file path'''
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
def exp_options(f, *,
|
|
name: str = None,
|
|
desp: str = None,
|
|
opt_type=None,
|
|
readonly=False):
|
|
"""**decorator**, mark a function as a experimental protocol parameter.
|
|
This decorator will return a property. It will hide the original function *f*
|
|
into the property getter.
|
|
|
|
This property getter will get the hidden attribute from `ExpProtocol`_. If not found,
|
|
it will set this attribute a default value got from the result of *f*. getter will
|
|
set the value to the hidden attribute. deleter will reset the value.
|
|
|
|
:param f:
|
|
:param name: parameter name
|
|
:param desp: parameter description
|
|
:param opt_type:
|
|
:param readonly: readonly property, cannot not be set or reset
|
|
:return: property
|
|
"""
|
|
raise RuntimeError()
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
def exp_protocol(name: str, template: str = None):
|
|
"""**decorator**, register ExpProtocol a export protocol"""
|
|
raise NotImplementedError()
|
|
|
|
|
|
class ExpProtocol(metaclass=abc.ABCMeta):
|
|
# protocol property, shortcut for protocol method
|
|
|
|
@property
|
|
def pro_name(self) -> str:
|
|
"""protocol name"""
|
|
raise RuntimeError()
|
|
|
|
@property
|
|
def pro_class(self) -> str:
|
|
"""protocol class name"""
|
|
raise RuntimeError()
|
|
|
|
@property
|
|
def pro_template(self) -> Optional[str]:
|
|
"""protocol template html filepath"""
|
|
raise RuntimeError()
|
|
|
|
@property
|
|
def pro_filepath(self) -> str:
|
|
"""protocol source filepath"""
|
|
raise RuntimeError()
|
|
|
|
@property
|
|
def pro_id(self) -> int:
|
|
"""protocol ID"""
|
|
raise RuntimeError()
|
|
|
|
@property
|
|
def pro_error(self) -> Optional[BaseException]:
|
|
"""protocol current error information"""
|
|
raise RuntimeError()
|
|
|
|
# predefine protocol method
|
|
|
|
def pro_ensure(self, api: ExpApi) -> bool:
|
|
"""ensure experimental setup"""
|
|
pass
|
|
|
|
def pro_setup(self, api: ExpApi):
|
|
"""setup experimental environment, including required device, parameter setting,
|
|
protocol parameter setting, file saving path and etc.
|
|
|
|
"""
|
|
pass
|
|
|
|
def pro_teardown(self, api: ExpApi):
|
|
"""teardown experimental environment, including device disconnect, file post processing and etc."""
|
|
pass
|
|
|
|
def pro_step(self, api: ExpApi):
|
|
"""experimental step.
|
|
|
|
:param api:
|
|
:return: exit code
|
|
"""
|
|
pass
|