fisrt commit
This commit is contained in:
peterlu14
@@ -0,0 +1,879 @@
|
||||
"""Device Library/Parameter
|
||||
|
||||
For the number or data structure used in this library, we can group them into there category:
|
||||
|
||||
1. raw instruction value (P value)
|
||||
|
||||
The value store in the device parameter and used in the instruction building.
|
||||
|
||||
#. raw parameter value (V value)
|
||||
|
||||
The value store in the device configuration which has a physical meaning.
|
||||
It will be saved into the recording meta file if has property *record_meta*.
|
||||
|
||||
#. human-readable (H value)
|
||||
|
||||
The value or text just to present to human and used in view module.
|
||||
|
||||
"""
|
||||
|
||||
from . import *
|
||||
from .domain import *
|
||||
|
||||
|
||||
class PropertySetter(JsonSerialize):
|
||||
"""
|
||||
|
||||
**json format**
|
||||
|
||||
::
|
||||
|
||||
PropertySetter
|
||||
= "exact"
|
||||
| "near"
|
||||
| "floor"
|
||||
| "ceil"
|
||||
| PropertySetterObject
|
||||
|
||||
PropertySetterObject = {
|
||||
"variable"? :variable_table
|
||||
"parameter" :parameter_table
|
||||
}
|
||||
|
||||
variable_table = {
|
||||
# with scope from caller
|
||||
variable_name = Expression[T]
|
||||
}
|
||||
|
||||
parameter_table = {
|
||||
# with scope from caller plus variable_table
|
||||
parameter_name = Expression[T]
|
||||
}
|
||||
|
||||
"""
|
||||
CAST_EXACT = "exact"
|
||||
CAST_NEAR = "near"
|
||||
CAST_FLOOR = "floor"
|
||||
CAST_CEIL = "ceil"
|
||||
CAST_NT = "NT" # XXX rename
|
||||
|
||||
__slots__ = ('_parameter', '_variable')
|
||||
|
||||
def __init__(self, parameter: Dict[str, Expression[Any]], variable: Dict[str, Expression[Any]] = None):
|
||||
self._parameter = parameter
|
||||
self._variable = variable
|
||||
|
||||
def cast_value(self, value: Any, context: Scope) -> Dict[str, Any]:
|
||||
"""
|
||||
|
||||
:param value: V value
|
||||
:param context: general case root scope
|
||||
:return: dict with parameter name map to P value
|
||||
"""
|
||||
scope = self._resolve_variable(value, context)
|
||||
|
||||
ret = {}
|
||||
|
||||
for par, exp in self._parameter.items():
|
||||
ret[par] = exp.value(scope)
|
||||
|
||||
return ret
|
||||
|
||||
def _resolve_variable(self, value: Any, context: Scope) -> Scope:
|
||||
scope = context.child(VALUE=value)
|
||||
|
||||
if self._variable is not None:
|
||||
var_set = set(self._variable.keys())
|
||||
|
||||
var_set_size = len(var_set)
|
||||
|
||||
while var_set_size != 0:
|
||||
for var, exp in self._variable.items():
|
||||
if var not in var_set:
|
||||
try:
|
||||
val = exp.value(scope)
|
||||
except NameError:
|
||||
pass
|
||||
else:
|
||||
scope.set_local(var, val)
|
||||
var_set.discard(var)
|
||||
|
||||
if len(var_set) == var_set_size:
|
||||
raise RuntimeError('recursive variable define : ' + ' '.join(var_set))
|
||||
|
||||
var_set_size = len(var_set)
|
||||
|
||||
return scope
|
||||
|
||||
def as_json(self):
|
||||
ret = {
|
||||
'parameter': {k: v.as_json() for k, v in self._parameter.items()}
|
||||
}
|
||||
|
||||
if self._variable is not None:
|
||||
ret['variable'] = {k: v.as_json() for k, v in self._variable.items()}
|
||||
|
||||
return ret
|
||||
|
||||
@classmethod
|
||||
def parse(cls, json: JSON_OBJECT) -> 'PropertySetter':
|
||||
par = _get(json, 'parameter', dict, 'value_set')
|
||||
var = _get(json, 'variable', dict, 'value_set', nullable=True)
|
||||
|
||||
par = _dict(par, lambda k, v: Expression.parse(v))
|
||||
|
||||
if var is not None:
|
||||
var = _dict(var, lambda k, v: Expression.parse(v))
|
||||
|
||||
return PropertySetter(par, var)
|
||||
|
||||
@classmethod
|
||||
def find_root(cls, strategy: str, f: Callable[[int], float], v: float, s: Tuple[int, int]) -> Optional[int]:
|
||||
"""
|
||||
|
||||
:param strategy: strategy method
|
||||
:param f: function (P) -> V
|
||||
:param v: V value
|
||||
:param s: seed range
|
||||
:return: P value, None if fail
|
||||
"""
|
||||
|
||||
if strategy == cls.CAST_EXACT:
|
||||
return cls.exact_root(f, v, s)
|
||||
elif strategy == cls.CAST_NEAR:
|
||||
return cls.near_root(f, v, s)
|
||||
elif strategy == cls.CAST_FLOOR:
|
||||
return cls.ceil_root(f, v, s)
|
||||
elif strategy == cls.CAST_CEIL:
|
||||
return cls.floor_root(f, v, s)
|
||||
elif strategy == cls.CAST_NT:
|
||||
return cls.nt_root(f, v, s)
|
||||
else:
|
||||
raise RuntimeError('unknown cast strategy : ' + strategy)
|
||||
|
||||
@staticmethod
|
||||
def find_identify(f: Callable[[int], str], value: str, seed_range: Tuple[int, int]) -> Optional[int]:
|
||||
for p in range(*seed_range):
|
||||
if f(p) == value:
|
||||
return p
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def exact_root(f: Callable[[int], float], value: float, seed_range: Tuple[int, int]) -> Optional[int]:
|
||||
# TODO implement binary search algo
|
||||
for p in range(*seed_range):
|
||||
if f(p) == value:
|
||||
return p
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def near_root(f: Callable[[int], float], value: float, seed_range: Tuple[int, int]) -> Optional[int]:
|
||||
ret = None
|
||||
min_d = None
|
||||
|
||||
for p in range(*seed_range):
|
||||
v = f(p)
|
||||
d = abs(value - v)
|
||||
|
||||
if min_d is None or d < min_d:
|
||||
ret = p
|
||||
min_d = d
|
||||
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def ceil_root(f: Callable[[int], float], value: float, seed_range: Tuple[int, int]) -> Optional[int]:
|
||||
ret = None
|
||||
|
||||
for p in range(seed_range[1], seed_range[0], -1):
|
||||
if value < f(p):
|
||||
ret = p
|
||||
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def floor_root(f: Callable[[int], float], value: float, seed_range: Tuple[int, int]) -> Optional[int]:
|
||||
ret = None
|
||||
|
||||
for p in range(*seed_range):
|
||||
if f(p) < value:
|
||||
ret = p
|
||||
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def nt_root(f: Callable[[int], float], value: float, seed: Tuple[int, int] = None) -> int:
|
||||
if seed is None:
|
||||
s = (0, 100)
|
||||
|
||||
else:
|
||||
s = (seed[0], seed[1])
|
||||
|
||||
v = [f(s[0]), f(s[1])]
|
||||
k = (v[1] - v[0]) / (s[1] - s[0])
|
||||
|
||||
a = (s[1] - s[0]) // 2
|
||||
|
||||
while True:
|
||||
b = f(a)
|
||||
c = value - b
|
||||
|
||||
if c == 0:
|
||||
return a
|
||||
|
||||
d = int(c / k)
|
||||
|
||||
if d == 0:
|
||||
return a - 1
|
||||
|
||||
a += d
|
||||
|
||||
|
||||
class ParameterInfo(JsonSerialize):
|
||||
"""
|
||||
**json format**
|
||||
|
||||
::
|
||||
|
||||
ParameterInfo = {
|
||||
"alias"? :str | list[str]
|
||||
"description"? :str
|
||||
"record_meta"? :bool
|
||||
"private"? :bool
|
||||
"domain"? :ParameterDomain
|
||||
"initial"? :int|Expression[T]
|
||||
"guard"? :GuardExpression|ListGuardExpression
|
||||
"value"? :Expression[T]
|
||||
"value_set"? :PropertySetter
|
||||
"on_change"? :str = instruction_name
|
||||
}
|
||||
|
||||
*record_meta* this parameter will be present in the device configuration by default and
|
||||
will be record in the recording meta file.
|
||||
Required parameters are always *record_meta*.
|
||||
|
||||
*private* this parameter will not present in the device configuration.
|
||||
*record_meta* parameter cannot be *private*.
|
||||
|
||||
*domain* parameter P value domain. Either *domain* or *value* should be existed.
|
||||
If only *value* present, *domain* would be considered :class:`ParameterConstantRangeDomain`
|
||||
with value ``[0, len(value)]``.
|
||||
If *domain* is "property", either *value* or *value_set* must present.
|
||||
If *domain* is "action", neither *value* nor *value_set* present.
|
||||
If parameter is ``CHANNEL``, the *domain* only allow "property" or set domain.
|
||||
|
||||
*guard* parameter guarder when parameter setting and modifying, the property *raise* can be ignored.
|
||||
|
||||
*value* V value getter,
|
||||
|
||||
*value_set* V value setter. If *domain* is not a property, it mean the strategy of casting V value to P value.
|
||||
omitted default to "exact". If *domain* is a property, it should be either ``null`` or a :class:`PropertySetter`.
|
||||
|
||||
*initial* initial P value for this parameter. It can be a expression which will be resolved at loading time with
|
||||
constant scope. If *domain* is :class:`ParameterCollectionDomain`, *initial* will be ignored.
|
||||
|
||||
**variable scope search chain**
|
||||
|
||||
* used when initialize
|
||||
|
||||
1. constant
|
||||
|
||||
* used when value evaluation
|
||||
|
||||
1. general case root (:class:`biopro.devlib.library.DeviceLibrary`)
|
||||
|
||||
"""
|
||||
|
||||
SAMPLE_RATE = 'SAMPLE_RATE'
|
||||
"""required parameter with type int"""
|
||||
|
||||
CHANNEL = 'CHANNEL'
|
||||
"""required parameter with type Set[int]"""
|
||||
|
||||
AMP_GAIN = 'AMP_GAIN'
|
||||
"""required parameter with type int"""
|
||||
|
||||
REQUIRED = (SAMPLE_RATE, CHANNEL, AMP_GAIN)
|
||||
"""required parameters"""
|
||||
|
||||
__slots__ = ('_name', '_domain', '_initial', '_value', '_value_set', '_guard', '_on_change',
|
||||
'_alias', '_description', '_record_meta', '_private')
|
||||
|
||||
def __init__(self,
|
||||
name: str,
|
||||
domain: ParameterDomain,
|
||||
initial: Optional[Any] = None,
|
||||
guard: Optional[ListGuardExpression] = None,
|
||||
value: Expression[Any] = None,
|
||||
value_set: Union[None, str, PropertySetter] = None,
|
||||
on_change: Optional[str] = None,
|
||||
alias: Union[str, List[str]] = None,
|
||||
description: Optional[str] = None,
|
||||
record_meta=False,
|
||||
private=False):
|
||||
"""
|
||||
|
||||
:param name: parameter name
|
||||
:param domain: parameter domain
|
||||
:param initial: initial P value
|
||||
:param guard: parameter guarded
|
||||
:param value: parameter present value
|
||||
:param value_set: parameter setter
|
||||
:param on_change: instruction to call when parameter changed
|
||||
:param alias: parameter name alias
|
||||
:param description: parameter description
|
||||
:param record_meta: record this parameter in recording meta file
|
||||
"""
|
||||
if domain is None:
|
||||
raise TypeError('NoneType domain for parameter : ' + name)
|
||||
|
||||
elif domain == ParameterConstantDomain:
|
||||
if initial is None:
|
||||
initial = 0
|
||||
|
||||
elif domain == ParameterPropertyDomain:
|
||||
if value is None and value_set is None:
|
||||
raise RuntimeError('parameter %s with property domain without value/value_set expression' % name)
|
||||
|
||||
if value_set is not None and not isinstance(value_set, PropertySetter):
|
||||
raise TypeError('value_set of the parameter %s not a PropertySetter' % name)
|
||||
|
||||
elif domain == ParameterActionDomain:
|
||||
if value is not None or value_set is not None:
|
||||
raise RuntimeError('parameter %s with action domain with value/value_set expression' % name)
|
||||
|
||||
record_meta = False
|
||||
private = True
|
||||
|
||||
if name == self.CHANNEL:
|
||||
if domain == ParameterPropertyDomain:
|
||||
pass
|
||||
elif isinstance(domain, ParameterSetDomain):
|
||||
if not isinstance(domain.element_domain, ParameterValueDomain):
|
||||
raise RuntimeError('parameter CHANNEL not a set of value domain')
|
||||
|
||||
else:
|
||||
raise RuntimeError('parameter CHANNEL not a set or property domain')
|
||||
|
||||
if isinstance(alias, str):
|
||||
alias = [alias]
|
||||
|
||||
self._name = name
|
||||
self._domain: ParameterDomain = domain
|
||||
self._initial: Optional[int] = initial
|
||||
self._value: Optional[Expression[Any]] = value
|
||||
self._value_set = value_set
|
||||
self._on_change = on_change
|
||||
self._guard = guard
|
||||
self._alias: List[str] = list(alias) if alias is not None else []
|
||||
self._description = description
|
||||
|
||||
# force record meta for required parameter
|
||||
self._record_meta = record_meta or name in self.REQUIRED
|
||||
|
||||
# record_meta parameter cannot be private
|
||||
self._private = private and not self._record_meta
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""parameter name """
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def domain(self) -> ParameterDomain:
|
||||
"""parameter domain."""
|
||||
return self._domain
|
||||
|
||||
@property
|
||||
def is_readable(self) -> bool:
|
||||
if self._domain == ParameterPropertyDomain and self._value is None:
|
||||
return False
|
||||
elif self._domain == ParameterActionDomain:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
@property
|
||||
def is_writable(self) -> bool:
|
||||
if self._domain == ParameterConstantDomain:
|
||||
return False
|
||||
elif self._domain == ParameterPropertyDomain and self._value_set is None:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
@property
|
||||
def element_domain(self) -> ParameterDomain:
|
||||
"""parameter domain."""
|
||||
if isinstance(self._domain, ParameterCollectionDomain):
|
||||
return self._domain.element_domain
|
||||
else:
|
||||
return self._domain
|
||||
|
||||
@property
|
||||
def description(self) -> Optional[str]:
|
||||
return self._description
|
||||
|
||||
@property
|
||||
def record_meta(self) -> bool:
|
||||
return self._record_meta
|
||||
|
||||
@property
|
||||
def private(self) -> bool:
|
||||
return self._private
|
||||
|
||||
@property
|
||||
def alias(self) -> List[str]:
|
||||
return list(self._alias)
|
||||
|
||||
@property
|
||||
def on_change(self) -> Optional[str]:
|
||||
return self._on_change
|
||||
|
||||
def init_para(self) -> Any:
|
||||
"""
|
||||
:return: initial P value
|
||||
:raises value: illegal initial value
|
||||
:raises PropertyDomainError: parameter is a property
|
||||
:raises ActionDomainError: parameter is a action
|
||||
"""
|
||||
if self._domain == ParameterConstantDomain:
|
||||
return int(self._initial)
|
||||
|
||||
elif self._domain == ParameterPropertyDomain:
|
||||
raise PropertyDomainError(self._name)
|
||||
|
||||
elif self._domain == ParameterActionDomain:
|
||||
raise ActionDomainError(self._name)
|
||||
|
||||
else:
|
||||
return self._domain.init_para(self._initial)
|
||||
|
||||
def ensure_guard(self, context: 'ParameterTableScope'):
|
||||
"""guarder ensure value in *context* is validated.
|
||||
|
||||
:param context: general case root scope
|
||||
:raises IllegalParameterValueError: guarder raise error
|
||||
"""
|
||||
if self._guard is None:
|
||||
return
|
||||
|
||||
guard = self._guard.test(context)
|
||||
|
||||
if guard is not None:
|
||||
raise IllegalParameterValueError(self._name, guard.error_message)
|
||||
|
||||
def get_value(self, value: Any, context: 'ParameterTableScope') -> Any:
|
||||
"""get the parameter present value.
|
||||
|
||||
:param value: P value
|
||||
:param context: general case root scope
|
||||
:return: V value
|
||||
"""
|
||||
if self._domain == ParameterActionDomain:
|
||||
raise ActionDomainError(self._name)
|
||||
|
||||
if self._domain == ParameterConstantDomain:
|
||||
# overwrite value if constant domain
|
||||
value = int(self._initial)
|
||||
|
||||
if self._value is None:
|
||||
# parameter no getter, which V value equals to P value. return directly
|
||||
return value
|
||||
|
||||
else:
|
||||
try:
|
||||
return self._get_value(value, context)
|
||||
except BaseException as e:
|
||||
raise DeviceParameterError(self._name, 'get_value error : ' + str(e)) from e
|
||||
|
||||
def _get_value(self, value: Any, context: 'ParameterTableScope') -> Any:
|
||||
if isinstance(self._domain, ParameterArrayDomain):
|
||||
if isinstance(value, list):
|
||||
size = self._domain.size
|
||||
index_para = self._domain.index_parameter
|
||||
|
||||
# keep old index_para value
|
||||
try:
|
||||
index_value_old = context.get_local(index_para)
|
||||
except KeyError:
|
||||
index_value_old = None
|
||||
|
||||
ret = []
|
||||
for i in range(size):
|
||||
# print(self._value.as_json(), 'VALUE=' + str(v))
|
||||
context.set_local(index_para, i)
|
||||
context.set_local('VALUE', value[i])
|
||||
ret.append(self._value.value(context))
|
||||
|
||||
# restore old index_para value
|
||||
if index_value_old is not None:
|
||||
context.set_local(index_para, index_value_old)
|
||||
else:
|
||||
context.del_local(index_para)
|
||||
|
||||
return ret
|
||||
|
||||
else:
|
||||
return self._value.value_local(context, VALUE=value)
|
||||
|
||||
elif isinstance(self._domain, ParameterCollectionDomain):
|
||||
if isinstance(value, (list, set)):
|
||||
ret = []
|
||||
|
||||
for v in value:
|
||||
# print(self._value.as_json(), 'VALUE=' + str(v))
|
||||
ret.append(self._value.value_local(context, VALUE=v))
|
||||
|
||||
if isinstance(value, set):
|
||||
return set(ret)
|
||||
else:
|
||||
return ret
|
||||
|
||||
else:
|
||||
return self._value.value_local(context, VALUE=value)
|
||||
|
||||
elif isinstance(self._value, ConstantExpression):
|
||||
return self._value.constant
|
||||
|
||||
elif isinstance(self._value, ArrayExpression):
|
||||
return self._value[int(value)]
|
||||
|
||||
else:
|
||||
return self._value.value_local(context, VALUE=value)
|
||||
|
||||
def cast_value(self, value: Any, context: 'ParameterTableScope') -> Any:
|
||||
"""get the parameter according to the present value. The parameter not always fit the present value.
|
||||
|
||||
:param value: V value
|
||||
:param context: general case root scope
|
||||
:return: P value
|
||||
"""
|
||||
|
||||
if self._domain == ParameterActionDomain:
|
||||
raise ActionDomainError(self._name)
|
||||
|
||||
setter: PropertySetter = self._value_set
|
||||
|
||||
if self._domain == ParameterConstantDomain:
|
||||
# constant parameter
|
||||
return int(self._initial)
|
||||
|
||||
elif isinstance(setter, PropertySetter):
|
||||
# raise an interrupt to indicate caller it is a special operation.
|
||||
raise PropertySetInterrupt(self._name, setter.cast_value(value, context))
|
||||
|
||||
elif self._domain == ParameterPropertyDomain:
|
||||
|
||||
if setter is None:
|
||||
raise PropertyDomainError(self._name, 'not a setter')
|
||||
|
||||
else:
|
||||
assert not isinstance(setter, PropertySetter)
|
||||
raise TypeError('parameter %s setter not a PropertySetter' % self._name)
|
||||
|
||||
elif self._value is None:
|
||||
# parameter no getter, which V value equals to P value. just return
|
||||
return value
|
||||
|
||||
elif isinstance(self._domain, ParameterArrayDomain):
|
||||
domain = self._domain.element_domain
|
||||
size = self._domain.size
|
||||
index_para = self._domain.index_parameter
|
||||
|
||||
assert index_para in context
|
||||
|
||||
ret = []
|
||||
for i in range(size):
|
||||
context.set_local(index_para, i)
|
||||
ret.append(self._cast_value(value[i], domain, context))
|
||||
|
||||
return ret
|
||||
|
||||
elif isinstance(self._domain, ParameterCollectionDomain):
|
||||
domain = self._domain.element_domain
|
||||
|
||||
ret = []
|
||||
|
||||
for v in value:
|
||||
ret.append(self._cast_value(v, domain, context))
|
||||
|
||||
if isinstance(value, list):
|
||||
return ret
|
||||
|
||||
elif isinstance(value, set):
|
||||
return set(ret)
|
||||
|
||||
else:
|
||||
raise RuntimeError('unknown type for parameter ' + self._name + ' : ' + str(type(value)))
|
||||
|
||||
else:
|
||||
return self._cast_value(value, self._domain, context)
|
||||
|
||||
def cast_value_dependency(self) -> List[str]:
|
||||
if self._value is None:
|
||||
return []
|
||||
|
||||
elif isinstance(self._value, ComplexExpression):
|
||||
return self._value.dependency
|
||||
|
||||
else:
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _cast_bool(value: Any) -> bool:
|
||||
if value is None:
|
||||
return False
|
||||
elif isinstance(value, bool):
|
||||
return value
|
||||
elif isinstance(value, int):
|
||||
return value > 0
|
||||
elif isinstance(value, float):
|
||||
return value > 0
|
||||
elif isinstance(value, str):
|
||||
value = value.lower()
|
||||
return value == '1' or value == 't' or value == 'true'
|
||||
|
||||
return False
|
||||
|
||||
def _cast_value(self, value: Any, domain: ParameterDomain, context: Scope) -> Any:
|
||||
if domain == ParameterBoolDomain:
|
||||
return self._cast_bool(value)
|
||||
|
||||
else:
|
||||
expr = self._value
|
||||
|
||||
assert expr is not None
|
||||
|
||||
def _f(_value):
|
||||
return expr.value_local(context, VALUE=_value)
|
||||
|
||||
strategy = self._value_set
|
||||
|
||||
if strategy is None:
|
||||
strategy = PropertySetter.CAST_EXACT
|
||||
|
||||
assert isinstance(strategy, str)
|
||||
|
||||
ret = None
|
||||
|
||||
if domain == ParameterIntDomain:
|
||||
seed_range = (int(value / 2), int(value * 3 / 2))
|
||||
ret = PropertySetter.find_root(PropertySetter.CAST_NT, _f, value, seed_range)
|
||||
|
||||
elif isinstance(domain, ParameterValueDomain):
|
||||
domain_range = domain.range
|
||||
test = _f(domain_range[0])
|
||||
|
||||
if isinstance(test, int):
|
||||
# XXX binary search with [floor, exact, ceil]
|
||||
ret = PropertySetter.find_root(strategy, _f, int(value), domain_range)
|
||||
|
||||
elif isinstance(test, float):
|
||||
# XXX nt root
|
||||
ret = PropertySetter.find_root(strategy, _f, float(value), domain_range)
|
||||
|
||||
elif isinstance(test, str):
|
||||
try:
|
||||
value = int(value)
|
||||
if domain_range[0] <= value < domain_range[1]:
|
||||
# return directory
|
||||
return value
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
ret = PropertySetter.find_identify(_f, str(value), domain_range)
|
||||
|
||||
else:
|
||||
raise DeviceParameterError(self._name,
|
||||
'unsupported cast value for domain : ' + domain.__class__.__name__)
|
||||
|
||||
if ret is None:
|
||||
raise DeviceParameterError(self._name, 'cannot cast value : ' + str(value))
|
||||
|
||||
# print(self.name, ret)
|
||||
|
||||
return ret
|
||||
|
||||
def as_json(self):
|
||||
ret = {
|
||||
'domain': self.domain.as_json(),
|
||||
'record_meta': self._record_meta,
|
||||
}
|
||||
|
||||
if len(self._alias) == 1:
|
||||
ret['alias'] = self._alias[0]
|
||||
|
||||
elif len(self._alias) > 1:
|
||||
ret['alias'] = self._alias
|
||||
|
||||
if self._description is not None:
|
||||
ret['description'] = self._description
|
||||
|
||||
if self._initial is not None:
|
||||
ret['initial'] = self._initial
|
||||
|
||||
if self._value is not None:
|
||||
ret['value'] = self._value.as_json()
|
||||
|
||||
if self._value_set is not None:
|
||||
ret['value'] = JsonSerialize.to_json(self._value_set)
|
||||
|
||||
if self._guard is not None:
|
||||
ret['guard'] = self._guard.as_json()
|
||||
|
||||
return ret
|
||||
|
||||
|
||||
class ParameterTable(DictNode[ParameterInfo]):
|
||||
"""
|
||||
**json format**
|
||||
|
||||
::
|
||||
|
||||
ParameterTable = {
|
||||
parameter_name :ParameterInfo
|
||||
}
|
||||
|
||||
**required parameter**
|
||||
|
||||
defined in :class:`biopro.devlib.device.DeviceParameter`
|
||||
|
||||
**variable scope search chain**
|
||||
|
||||
* used when parameter initializing.
|
||||
|
||||
1. "constant"
|
||||
|
||||
* used when parameter set/add guarding.
|
||||
|
||||
1. general case root (:class:`biopro.devlib.library.DeviceLibrary`)
|
||||
#. {VALUE=new value}
|
||||
|
||||
* used when parameter transform to present value
|
||||
|
||||
1. general case root (:class:`biopro.devlib.library.DeviceLibrary`)
|
||||
#. {VALUE=parameter value}
|
||||
|
||||
relate function :func:`ParameterInfo.get_value`, :func:`ParameterInfo.cast_value`
|
||||
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, parameters: Optional[Dict[str, ParameterInfo]] = None):
|
||||
super().__init__(parameters)
|
||||
|
||||
def keys_all(self, show_private=False) -> Iterable[str]:
|
||||
r = set()
|
||||
|
||||
for para in self.values():
|
||||
if show_private or not para.private:
|
||||
r.add(para.name)
|
||||
r.update(para.alias)
|
||||
|
||||
return r
|
||||
|
||||
def __contains__(self, key: Any) -> bool:
|
||||
if key in self._dict:
|
||||
return True
|
||||
|
||||
for para in self.values():
|
||||
if key in para.alias:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def __missing__(self, key: str) -> ParameterInfo:
|
||||
for para in self.values():
|
||||
if key in para.alias:
|
||||
return para
|
||||
|
||||
raise KeyError(key)
|
||||
|
||||
def get_value(self, name: Union[str, ParameterInfo], value: Any, context: Scope) -> Any:
|
||||
"""get parameter V value
|
||||
|
||||
:param name: parameter name or info
|
||||
:param value: P value
|
||||
:param context:
|
||||
:return: V value
|
||||
:raises KeyError: parameter name not found
|
||||
"""
|
||||
if isinstance(name, str):
|
||||
name = self[name]
|
||||
|
||||
return name.get_value(value, context.chain(ParameterTableScope(self)))
|
||||
|
||||
def cast_value(self, name: Union[str, ParameterInfo], value: Any, context: Scope) -> Optional[int]:
|
||||
if isinstance(name, str):
|
||||
name = self[name]
|
||||
|
||||
return name.cast_value(value, context.chain(ParameterTableScope(self)))
|
||||
|
||||
def ensure_guard(self, name: Union[str, ParameterInfo], context: Scope):
|
||||
"""invoke parameter *name*'s guarder ensure its value in *context* is validated.
|
||||
|
||||
:param name: parameter or its name
|
||||
:param context: general case root scope
|
||||
:raises IllegalParameterValueError: guarder raise error
|
||||
"""
|
||||
if isinstance(name, str):
|
||||
name = self[name]
|
||||
|
||||
return name.ensure_guard(context.chain(ParameterTableScope(self)))
|
||||
|
||||
|
||||
class ParameterTableScope(Scope):
|
||||
"""parameter table scope, make user access the parameter value through the context scope.
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = '_table', '_overwrite'
|
||||
|
||||
def __init__(self, table: ParameterTable):
|
||||
super().__init__()
|
||||
|
||||
self._table = table
|
||||
self._overwrite = {}
|
||||
|
||||
def locals(self) -> Iterable[str]:
|
||||
ret = list(self._overwrite.keys())
|
||||
ret.append('_value')
|
||||
|
||||
return ret
|
||||
|
||||
def contain_local(self, key: str) -> bool:
|
||||
return key == '_value' or key in self._overwrite
|
||||
|
||||
def get_local(self, key: str) -> Any:
|
||||
if key == '_value':
|
||||
return self._value
|
||||
|
||||
else:
|
||||
return self._overwrite[key]
|
||||
|
||||
def set_local(self, key: str, value: Any):
|
||||
self._overwrite[key] = value
|
||||
|
||||
def del_local(self, key: str):
|
||||
del self._overwrite[key]
|
||||
|
||||
def clear(self):
|
||||
self._overwrite.clear()
|
||||
|
||||
def _value(self, name: str):
|
||||
para = self._table[name]
|
||||
domain = para.domain
|
||||
|
||||
ret = para.get_value(self[name], self)
|
||||
|
||||
# print('_value(%s)' % name, '=', ret)
|
||||
|
||||
if isinstance(domain, ParameterArrayDomain):
|
||||
ret = domain.get_para(ret, self)
|
||||
|
||||
return ret
|
||||
Reference in New Issue
Block a user