controller-wisetopdataserver/python/biopro/devlib/lock.py

"""Device lock.

This module provide device lock let user can lock device(s). User can control the device if he/she own the lock.
Otherwise, others just only view/read the device setting and data.

Device Lock Level
=================

1. Experimental lock
2. General lock
3. Timeout lock
4. Public lock

"""
import abc
from time import time
from typing import Optional


class AbstractDeviceLock(metaclass=abc.ABCMeta):
    LOCK_EXPERIMENTAL = 1
    LOCK_GENERAL = 2
    LOCK_TIMEOUT = 3
    LOCK_PUBLIC = 4

    TIMEOUT_MIN = 10

    @property
    @abc.abstractmethod
    def experimental(self) -> Optional[str]:
        pass

    @property
    @abc.abstractmethod
    def user_name(self) -> Optional[str]:
        pass

    @property
    @abc.abstractmethod
    def priority(self) -> int:
        pass

    @abc.abstractmethod
    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        pass

    @abc.abstractmethod
    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        pass


class DeviceLock(AbstractDeviceLock):
    __slots__ = '__last_access_time', '__lock'

    def __init__(self):
        self.__last_access_time = time()
        self.__lock: AbstractDeviceLock = PublicLock()

    @property
    def experimental(self) -> Optional[str]:
        return self.__lock.experimental

    @property
    def user_name(self) -> Optional[str]:
        return self.__lock.user_name

    @property
    def priority(self) -> int:
        return self.__lock.priority

    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        return self.__lock.validate(lock, user_name=user_name, experimental=experimental)

    def gain_priority(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        if self.__lock.validate(lock, user_name=user_name, experimental=experimental):
            if self.__lock.priority == lock:
                return True
            elif lock == self.LOCK_EXPERIMENTAL:
                self.__lock = ExperimentalLock(experimental, user_name)
                return True
            elif lock == self.LOCK_GENERAL:
                self.__lock = GeneralLock(user_name)
                return True
        return False

    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        self.__lock: AbstractDeviceLock = self.__lock.flush(touch)
        return self

    def release(self, user_name: str, experimental: str) -> bool:
        if self.__lock.validate(self.LOCK_EXPERIMENTAL, user_name=user_name, experimental=experimental):
            self.__lock = GeneralLock(user_name)
            return True
        else:
            return False

    def public(self):
        if self.__lock == self.LOCK_EXPERIMENTAL:
            raise RuntimeError('experimental lock not release')
        else:
            self.__lock = PublicLock()


class ExperimentalLock(AbstractDeviceLock):
    PRIORITY = DeviceLock.LOCK_EXPERIMENTAL

    __slots__ = '__experimental', '__user_name'

    def __init__(self, experiment: str, user_name: str):
        """

        :param experiment: experimental protocol ID
        :param user_name: user name
        """
        self.__experimental = experiment
        self.__user_name = user_name

    @property
    def experimental(self) -> str:
        return self.__experimental

    @property
    def user_name(self) -> str:
        return self.__user_name

    @property
    def priority(self) -> int:
        return self.LOCK_EXPERIMENTAL

    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        if lock > self.LOCK_EXPERIMENTAL:
            return False

        return self.__experimental == experimental and self.__user_name == user_name

    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        return self


class GeneralLock(AbstractDeviceLock):
    __slots__ = '__user_name', '__last_access'

    def __init__(self, user_name: str):
        """

        :param user_name: user name
        """
        self.__user_name = user_name
        self.__last_access = time()

    @property
    def experimental(self) -> Optional[str]:
        return None

    @property
    def user_name(self) -> str:
        return self.__user_name

    @property
    def priority(self) -> int:
        return self.LOCK_GENERAL

    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        if lock == self.LOCK_EXPERIMENTAL:
            return self.__user_name == user_name
        elif lock == self.LOCK_GENERAL:
            return self.__user_name == user_name
        return False

    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        t = time()

        if touch:
            self.__last_access = t
            return self
        else:
            if t - self.__last_access > self.TIMEOUT_MIN * 60:
                return TimeoutLock(self.__user_name)
            else:
                return self


class TimeoutLock(AbstractDeviceLock):
    __slots__ = '__user_name',

    def __init__(self, user_name: str):
        """

        :param user_name: user name
        """
        self.__user_name = user_name

    @property
    def experimental(self) -> Optional[str]:
        return None

    @property
    def user_name(self) -> str:
        return self.__user_name

    @property
    def priority(self) -> int:
        return self.LOCK_TIMEOUT

    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        return lock <= self.LOCK_TIMEOUT

    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        return self


class PublicLock(AbstractDeviceLock):

    @property
    def experimental(self) -> Optional[str]:
        return None

    @property
    def user_name(self) -> Optional[str]:
        return None

    @property
    def priority(self) -> int:
        return self.LOCK_PUBLIC

    def validate(self, lock: int, *, user_name: str = None, experimental: str = None) -> bool:
        return True

    def flush(self, touch: bool) -> 'AbstractDeviceLock':
        return self