"""_____________________________________________________________________
:PROJECT: pythonLab
*resource base classes*
:details:
:authors: mark doerr (mark@uni-greifswald.de)
:date: (creation) 20210410
________________________________________________________________________
"""
__version__ = "0.0.1"
import importlib
import logging
import pkgutil
from enum import Enum
#from labdatareader.data_reader import DataReader
from typing import Optional
from abc import ABC, abstractmethod
[docs]
class DataDirection(Enum):
data_in = 1
data_out = 2
[docs]
class DataType(Enum):
single_value = 1 # scalar
structured_data = 2 # list, dict, or larger
data_stream = 3
[docs]
class Position:
def __init__(self, resource, position: int):
self._resource = resource
self._postion = position
self._postion_name = f"{str(resource.name)}_{str(position)}"
@property
def pos(self):
return self._postion
@property
def resource(self):
return self._resource
@property
def name(self):
return self._postion_name
[docs]
class Resource(ABC):
def __init__(self,
proc=None,
name: Optional[str] = None):
self.proc = proc
self._name = name
self.auto_register_resource()
[docs]
@abstractmethod
def init(self):
raise NotImplementedError
[docs]
def auto_register_resource(self):
"""auto register resource in corresponding process
"""
if isinstance(self, ServiceResource):
self.proc.register_service_resource(self)
elif isinstance(self, LabwareResource):
self.proc.register_labware_resource(self)
elif isinstance(self, DynamicLabwareResource):
self.proc.register_labware_resource(self)
elif isinstance(self, SubstanceResource):
self.proc.register_substance_resource(self)
elif isinstance(self, DataResource):
self.proc.register_data_resource(self)
[docs]
def import_resources(self, path):
"""Import resources from a given path
:param path: _description_
:type path: _type_
TODO: path to resources
"""
#TODO: Mark: What went wrong here?
#self._import_all_resources("TDOO: path to resource definitions")
[docs]
def _import_all_resources(self, namespace_pkg):
"""
recursively iterate through namespace
Specifying the second argument (prefix) to iter_modules makes the
returned name an absolute name instead of a relative one. This allows
import_module to work without having to do additional modification to
the name.
s. https://packaging.python.org/guides/creating-and-discovering-plugins/
TODO: recursive import !!
"""
for finder, name, ispkg in pkgutil.iter_modules(namespace_pkg.__path__, namespace_pkg.__name__ + "."):
submodule = importlib.import_module(name)
if ispkg:
self._import_all_resources(submodule)
# if a dictionary of discovered plugins is required, see LabDataReader
@property
def name(self):
return self._name
[docs]
class ServiceResource(Resource):
"""ServiceResource
is reflection SiLA's service oriented concept
(more general than a device, since a service can be much more than a device)
:param Resource: [description]
:type Resource: [type]
"""
def __init__(self,
proc,
name: Optional[str] = None):
super().__init__(proc=proc, name=name)
[docs]
class LabwareResource(Resource):
"""
multi-cavity / single cavity ?
associated labware, like labwares, lids, stacks
:param Resource: [description]
:type Resource: [type]
"""
def __init__(self,
proc,
name: str = None,
priority=None,
lidded: bool = False,
**kwargs):
super().__init__(proc=proc, name=name)
self._position = None
self.priority = priority
self.start_position = None
# this flag can be set in a process and will be 'consumed' when the next step with this labware is called
self._max_wait = None
self._min_wait = None
self._wait_cost = 0
self.lidded = lidded
self.kwargs = kwargs
[docs]
def init(self):
logging.debug(f"init {self.name}")
# returns the maximum waiting time until next step and resets it
[docs]
def consume_max_wait(self):
tmp = self._max_wait
self._max_wait = None
return tmp
[docs]
def max_wait(self, duration):
self._max_wait = duration
[docs]
def min_wait(self, duration):
self._min_wait = duration
[docs]
def consume_min_wait(self):
tmp = self._min_wait
self._min_wait = None
return tmp
# returns the cost for waiting until next step and resets it
[docs]
def consume_wait_cost(self):
tmp = self._wait_cost
self._wait_cost = 0
return tmp
[docs]
def wait_cost(self, cost_per_second):
self._wait_cost = cost_per_second
[docs]
def set_start_position(self, resource, position):
self._position = Position(resource=resource, position=position)
self.start_position = self._position
self.proc.set_starting_position(self, resource, position)
@property
def pos(self):
return self._position
[docs]
class DynamicLabwareResource(LabwareResource):
def __init__(self, proc, name: str, priority=None, lidded: bool = False, outside_cost=0, **kwargs):
"""Dynamic Labware is a special type of labware, where order of usage is (dynamically) determined during the process execution,
e.g. a reagent trough that is used when addition of a reagent to a certain labware is required, depending on an outcome of a descision,
like a pH measurement or induction after a certain Absorbtion is reached.
:param proc:
:param name:
:param priority:
:param lidded:
:param outside_cost: Some reagents need to be stored under special properties (e.g. cooled). These costs get
translated to waiting_costs between getting it out and putting it back
:param kwargs:
"""
super().__init__(proc, name=name, priority=priority, lidded=lidded, **kwargs)
self.outside_cost = outside_cost
[docs]
class SubstanceResource(Resource):
"""SubstanceResource
more general concept than sample
can be used to define substances / samples and their corresponding properties,
like liquid classes, or physical state (gas, liquid, solid/powder),
density, viscosity, vapour pressure
:param Resource: [description]
:type Resource: [type]
"""
def __init__(self,
proc,
name: str = None):
super().__init__(proc=proc, name=name)
[docs]
class DataResource(Resource):
def __init__(self,
proc,
name: str = None):
super().__init__(proc=proc, name=name)
self.direction = None # DataDirection.data_in
