controller-wisetopdataserver/python/biopro/exp_pro/protocol.py

"""This module actually implement the content which declared at biopro.exp_pro.

"""

import inspect
from typing import Type, Callable

from biopro.util.json import JsonSerialize, JSON_OBJECT
from . import *

if True:
    # prevent import reorder
    pass

# noinspection PyUnresolvedReferences
from biopro.devlib.device import DeviceInfo


@overload
def exp_options(f) -> property:
    pass


@overload
def exp_options(name: str, desp: str = None) -> Callable[[Any], property]:
    pass


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
    """

    def _doc(doc: Optional[str]) -> str:
        if doc is None:
            return ''

        for line in doc.split('\n'):
            if len(line) > 0:
                break
        else:
            return ''

        if '.' in line:
            line = line[:line.index('.') + 1]

        return line

    def _inner(f):
        v = opt_type
        if v is None:
            s = inspect.signature(f)
            v = s.return_annotation

        _name = name if name is not None else f.__name__
        _desp = desp if desp is not None else _doc(f.__doc__)

        return _gen_exp_property(f, _name, _desp, v, readonly=readonly)

    if f is None:
        return _inner
    else:
        return _inner(f)


def _gen_exp_property(f, name: str, desp: str, opt_type=int, readonly=False):
    a = '__' + name

    # TODO do value type checking?

    def getter(self):
        if not hasattr(self, a):
            setattr(self, a, f(self))

        return getattr(self, a)

    def setter(self, value):
        setattr(self, a, value)

    def deleter(self):
        setattr(self, a, f(self))

    if readonly:
        ret = property(getter)
    else:
        ret = property(getter, setter, deleter)

    getter._exp_name = name
    getter._exp_desp = desp
    getter._exp_type = opt_type
    getter._exp_default = f

    return ret


def exp_protocol(name: str, template: str = None):
    """**decorator**, register ExpProtocol a export protocol"""

    def _inner(cls):
        cls._exp_name = name
        cls._exp_template = template
        return cls

    return _inner


class ExpOption(JsonSerialize):
    """
    **json format**

    ::

        exp_option = {
            'name': str
            'desp': str
        }
    """
    __slots__ = '__p', '__t'

    def __init__(self, p: property, t: 'ExpProtocol'):
        self.__p = p
        self.__t = t

    @property
    def name(self) -> str:
        """option name"""

        # noinspection PyProtectedMember
        return self.__p.fget._exp_name

    @property
    def desp(self) -> str:
        """option description"""

        # noinspection PyProtectedMember
        return self.__p.fget._exp_desp

    @property
    def value_type(self) -> Type[Any]:
        """option value type"""

        # noinspection PyProtectedMember
        return self.__p.fget._exp_type

    def get(self) -> Any:
        """get value from this option"""
        return getattr(self.__t, self.name)

    def set(self, value: Any):
        """set value to this option"""
        if self.__p.fset is not None:
            setattr(self.__t, self.name, value)

    def reset(self):
        """reset this option back to default value"""
        if self.__p.fdel is not None:
            delattr(self.__t, self.name)

    def default(self):
        """default value of this options"""

        # noinspection PyProtectedMember
        return self.__p.fget._exp_default(self)

    def __str__(self):
        return type(self.__t).__name__ + '.' + self.name

    def __repr__(self):
        return type(self.__t).__name__ + '.' + self.name

    def as_json(self) -> JSON_OBJECT:
        raise {
            'name': self.name,
            'desp': self.desp,
            'type': self.value_type.__name__
        }


class ReadonlyExpOption(ExpOption):
    __slots__ = ()

    def __init__(self, o: ExpOption):
        super().__init__(o.__p, o.__t)

    def set(self, value: Any):
        """set value to this option"""
        pass

    def reset(self):
        """reset this option back to default value"""
        pass


# noinspection PyMethodMayBeStatic, PyUnusedLocal
class ExpProtocol(metaclass=abc.ABCMeta):
    # protocol method (class method)

    @classmethod
    def exp_name(cls) -> str:
        return getattr(cls, '_exp_name')

    @classmethod
    def exp_desp(cls) -> Optional[str]:
        return cls.__doc__

    @classmethod
    def exp_template(cls) -> Optional[str]:
        return getattr(cls, '_exp_template', None)

    @classmethod
    def exp_filepath(cls) -> Optional[str]:
        return getattr(cls, '_exp_filepath', None)

    # protocol property, shortcut for protocol method

    @property
    def pro_name(self) -> str:
        return self.exp_name() + ':' + str(self.pro_id)

    @property
    def pro_class(self) -> str:
        return type(self).__name__

    @property
    def pro_template(self) -> Optional[str]:
        return self.exp_template()

    @property
    def pro_filepath(self) -> str:
        return self.exp_filepath()

    @property
    def pro_id(self) -> int:
        return getattr(self, '_exp_id', -1)

    @property
    def pro_error(self) -> Optional[BaseException]:
        return getattr(self, '_exp_error', None)

    # predefine protocol method

    def pro_ensure(self, api: ExpApi) -> bool:
        return True

    def pro_setup(self, api: ExpApi):
        self.pro_ensure(api)

    def pro_step(self, api: ExpApi) -> int:
        pass

    def pro_teardown(self, api: ExpApi):
        pass

    def pro_on_error(self, api: ExpApi, status: int) -> bool:
        return False

    # protocol utility methods

    def pro_reset(self, *options: str):
        for p in self.pro_options():
            if len(options) == 0 or p.name in options:
                p.reset()

    def pro_options(self) -> List[ExpOption]:
        ret = []

        cls = type(self)
        for attr_name in dir(cls):
            if attr_name.startswith('_'):
                continue

            attr_value = getattr(cls, attr_name)
            if isinstance(attr_value, property):
                if hasattr(attr_value.fget, '_exp_name'):
                    ret.append(ExpOption(attr_value, self))

        return ret

    def pro_option(self, option: str) -> ExpOption:
        cls = type(self)

        attr_value = getattr(cls, option, None)
        if isinstance(attr_value, property):
            if hasattr(attr_value.fget, '_exp_name'):
                return ExpOption(attr_value, self)

        raise RuntimeError('options not found : ' + option)

    def __str__(self):
        return self.pro_name

    def __repr__(self):
        return self.pro_name


class ExpProtocolInfo(JsonSerialize):
    """
    **json format**

    ::

        protocol_info = {
            'name': str
            'desp': str
            'template: str
        }
    """
    __slots__ = '__protocol',

    def __init__(self, protocol: Type[ExpProtocol]):
        self.__protocol = protocol

    @property
    def exp_name(self) -> str:
        return self.__protocol.exp_name()

    @property
    def exp_class(self) -> str:
        return self.__protocol.__name__

    @property
    def exp_desp(self) -> Optional[str]:
        return self.__protocol.exp_desp()

    @property
    def exp_template(self) -> Optional[str]:
        return self.__protocol.exp_template()

    @property
    def exp_filepath(self) -> Optional[str]:
        return self.__protocol.exp_filepath()

    def new_protocol(self) -> ExpProtocol:
        return self.__protocol()

    def as_json(self) -> JSON_OBJECT:
        return {
            'name':     self.__protocol.exp_name(),
            'desp':     self.__protocol.exp_desp(),
            'template': self.__protocol.exp_template()
        }


class CompletedExpApi(ExpApi, metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def shutdown(self):
        pass


class ExpApiBaseImpl(CompletedExpApi):
    def __init__(self, parent: ExpApi) -> None:
        self.__parent = parent

    def log_verbose(self, *message: Any):
        self.__parent.log_verbose(*message)

    def log_info(self, *message: Any):
        self.__parent.log_info(*message)

    def log_warn(self, *message: Any):
        self.__parent.log_warn(*message)

    @property
    def user(self) -> str:
        return self.__parent.user

    @property
    def root_directory(self) -> str:
        return self.__parent.root_directory

    @property
    def use_directory(self) -> str:
        return self.__parent.use_directory

    def change_directory(self, path: str) -> str:
        return self.__parent.change_directory(path)

    def shutdown(self):
        if isinstance(self.__parent, CompletedExpApi):
            self.__parent.shutdown()

    def current_timestamp(self) -> int:
        return self.__parent.current_timestamp()

    def current_datetime(self) -> datetime.datetime:
        return self.__parent.current_datetime()

    def schedule_time(self, timestamp: Union[float, datetime.datetime]):
        self.__parent.schedule_time(timestamp)

    def schedule_sleep(self, sleep: Union[float, datetime.timedelta]):
        self.__parent.schedule_sleep(sleep)

    def make_device_info(self, *, device_name: str = None, mac_address: Union[str, ADDRESS] = None) -> DeviceInfo:
        return self.__parent.make_device_info(device_name=device_name, mac_address=mac_address)

    def found_device(self) -> List[DeviceInfo]:
        return self.__parent.found_device()

    def connected_device(self) -> List[DeviceInfo]:
        return self.__parent.connected_device()

    def ensure_device(self, *device_list: DeviceInfo, status: int = STATUS_INIT) -> bool:
        # we ensure that all of the implement of ExpApi.ensure_device has parameter status.
        # noinspection PyArgumentList
        return self.__parent.ensure_device(*device_list, status=status)

    def all_device(self) -> List[DeviceInfo]:
        return self.__parent.all_device()

    def get_device(self, device: int) -> DeviceInfo:
        return self.__parent.get_device(device)

    def release_device(self, device_list: Optional[List[DeviceInfo]] = None, disconnect=True):
        self.__parent.release_device(device_list, disconnect=disconnect)

    def apply_setup(self, name: str):
        self.__parent.apply_setup(name)

    def device_get_parameter(self, device: Union[int, DeviceInfo], parameter: str) -> Any:
        return self.__parent.device_get_parameter(device, parameter)

    def device_set_parameter(self, device: Union[int, DeviceInfo], *parameter):
        self.__parent.device_set_parameter(device, *parameter)

    def device_to_value(self, device: Union[int, DeviceInfo], parameter: str, value: Any) -> Any:
        return self.__parent.device_to_value(device, parameter, value)

    def device_to_parameter(self, device: Union[int, DeviceInfo], parameter: str, value: Any) -> Any:
        return self.__parent.device_to_parameter(device, parameter, value)

    def device_call_instruction(self, device: Union[int, DeviceInfo], instruction: str):
        self.__parent.device_call_instruction(device, instruction)

    def device_use_filepath(self, device: Union[int, DeviceInfo]) -> Optional[str]:
        return self.__parent.device_use_filepath(device)

    def device_set_filepath(self, device: Union[int, DeviceInfo], filename: str) -> str:
        return self.__parent.device_set_filepath(device, filename)

    def device_unset_filepath(self, device: Union[int, DeviceInfo]) -> Optional[str]:
        return self.__parent.device_unset_filepath(device)

    def set_complete(self):
        self.__parent.set_complete()

    def set_error(self, message: str):
        self.__parent.set_error(message)

    def file_export(self,
                    ftype: str,
                    *filepath: Union[str, DeviceInfo],
                    **options) -> Promise[List[Union[str, DeviceInfo]]]:
        return self.__parent.file_export(ftype, *filepath, **options)

    def file_archive(self,
                     filename: str,
                     *filepath: Union[str, DeviceInfo],
                     ftype: Optional[str] = None) -> Promise[str]:
        return self.__parent.file_archive(filename, *filepath, ftype=ftype)