peterlu14
1067 lines
28 KiB
Python
1067 lines
28 KiB
Python
from typing import Tuple, ItemsView
|
|
|
|
from biopro.util.text import part_suffix
|
|
from . import *
|
|
from .expression import *
|
|
|
|
import json
|
|
|
|
class DeviceParameterError(RuntimeError):
|
|
__slots__ = ()
|
|
|
|
def __init__(self, parameter_name: str, message: str):
|
|
super().__init__('parameter ' + parameter_name + ' ' + message)
|
|
|
|
|
|
class IllegalParameterValueError(DeviceParameterError):
|
|
__slots__ = ()
|
|
|
|
def __init__(self, parameter_name: str, message: Optional[str] = None):
|
|
super().__init__(parameter_name, message if message is not None else 'illegal value')
|
|
|
|
|
|
class ConstantDomainError(DeviceParameterError):
|
|
"""Any un-allowed operation on constant parameter will raise this error."""
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, parameter_name: str, message: Optional[str] = None):
|
|
m = message if message is not None else 'is a constant'
|
|
super().__init__(parameter_name, m)
|
|
|
|
|
|
class PropertyDomainError(DeviceParameterError):
|
|
"""Any un-allowed operation on property parameter will raise this error."""
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, parameter_name: str, message: Optional[str] = None):
|
|
m = message if message is not None else 'is a property'
|
|
super().__init__(parameter_name, m)
|
|
|
|
|
|
class ActionDomainError(DeviceParameterError):
|
|
"""Any operation on action parameter will raise this error"""
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, parameter_name: str, message: Optional[str] = None):
|
|
m = message if message is not None else 'is a action parameter'
|
|
super().__init__(parameter_name, m)
|
|
|
|
|
|
class PropertySetInterrupt(DeviceParameterError):
|
|
__slots__ = ('_parameter_update_table',)
|
|
|
|
def __init__(self, property_name: str, parameter_update_table: Dict[str, Any]):
|
|
super().__init__(property_name, 'property set')
|
|
self._parameter_update_table = parameter_update_table
|
|
|
|
def items(self) -> ItemsView[str, Any]:
|
|
return self._parameter_update_table.items()
|
|
|
|
|
|
class ParameterDomain(JsonSerialize, metaclass=abc.ABCMeta):
|
|
"""parameter value domain.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterDomain
|
|
= ParameterConstantDomain # type: "constant"
|
|
| ParameterTypeDomain
|
|
| ParameterBoolDomain # value: "bool"
|
|
| ParameterIntDomain # value: "int"
|
|
| ParameterValueDomain
|
|
| ParameterConstantRangeDomain # type: list[int]
|
|
| ParameterRangeDomain # type: list[Expression[int]]
|
|
| ParameterCollectionDomain
|
|
| ParameterListDomain # type: dict {"list": ParameterDomain}
|
|
| ParameterSetDomain # type: dict {"set": ParameterDomain}
|
|
| ParameterArrayDomain # type: dict {"array": ParameterDomain}
|
|
| PropertyDomain # value: "property"
|
|
| ActionDomain # value: "action"
|
|
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
@abc.abstractmethod
|
|
def init_para(self, initial: Optional[Any] = None) -> Any:
|
|
"""
|
|
|
|
:param initial: P value
|
|
:return: legal P value
|
|
:raise ValueError: illegal `initial` value
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def valid_para(self, value: int) -> bool:
|
|
"""exam **value** in this domain
|
|
|
|
:param value: P value
|
|
:return: valid
|
|
"""
|
|
pass
|
|
|
|
def __repr__(self):
|
|
return str(self)
|
|
|
|
@classmethod
|
|
def parse(cls, json: Optional[JSON], constant: ConstantScope) -> Optional['ParameterDomain']:
|
|
if json == 'constant':
|
|
return ParameterConstantDomain
|
|
|
|
elif json == 'bool':
|
|
return ParameterBoolDomain
|
|
|
|
elif json == 'int':
|
|
return ParameterIntDomain
|
|
|
|
elif json == 'property':
|
|
return ParameterPropertyDomain
|
|
|
|
elif json == 'action':
|
|
return ParameterActionDomain
|
|
|
|
elif isinstance(json, list):
|
|
return ParameterValueDomain.parse(json, constant)
|
|
|
|
elif isinstance(json, dict):
|
|
return ParameterCollectionDomain.parse(json, constant)
|
|
|
|
raise RuntimeError('illegal value domain : ' + _truncate(json))
|
|
|
|
|
|
class ParameterConstantDomainType(ParameterDomain):
|
|
"""constant P value domain."""
|
|
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> NoReturn:
|
|
"""
|
|
|
|
:param initial: P value
|
|
:raises: ConstantDomainError, always
|
|
"""
|
|
raise ConstantDomainError('<unknown>')
|
|
|
|
def valid_para(self, value: int) -> bool:
|
|
"""
|
|
|
|
:param value:
|
|
:return: always False
|
|
"""
|
|
return False
|
|
|
|
def as_json(self):
|
|
return 'constant'
|
|
|
|
def __str__(self):
|
|
return 'constant'
|
|
|
|
|
|
ParameterConstantDomain = ParameterConstantDomainType()
|
|
|
|
|
|
class ParameterPropertyDomainType(ParameterDomain):
|
|
"""Property parameter only has V value, but no P value, which means that the value only
|
|
present in the device configuration. The property parameter should has at least
|
|
*value* and *value_set* property which also represent as getter and setter respectively.
|
|
It give a way to get a V value (combine from one or more other parameter which has P value)
|
|
or set a V value (assign to one or more other parameter with calculated P value).
|
|
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> NoReturn:
|
|
"""
|
|
|
|
:param initial:
|
|
:raises: PropertyDomainError, always
|
|
"""
|
|
raise PropertyDomainError('<unknown>')
|
|
|
|
def valid_para(self, value: int) -> bool:
|
|
"""
|
|
|
|
:param value:
|
|
:return: always False
|
|
"""
|
|
return False
|
|
|
|
def as_json(self):
|
|
return 'property'
|
|
|
|
def __str__(self):
|
|
return 'property'
|
|
|
|
|
|
ParameterPropertyDomain = ParameterPropertyDomainType()
|
|
|
|
|
|
class ParameterActionDomainType(ParameterDomain):
|
|
"""Action parameter has neither P value nor V value. This parameter used to fire observer
|
|
event.
|
|
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> Any:
|
|
"""
|
|
|
|
:param initial:
|
|
:raises: ActionDomainError, always
|
|
"""
|
|
raise ActionDomainError('<unknown>')
|
|
|
|
def valid_para(self, value: int) -> bool:
|
|
"""
|
|
|
|
:param value:
|
|
:return: always False
|
|
"""
|
|
return False
|
|
|
|
def as_json(self):
|
|
return 'action'
|
|
|
|
def __str__(self):
|
|
return 'action'
|
|
|
|
|
|
ParameterActionDomain = ParameterActionDomainType()
|
|
|
|
|
|
class ParameterTypeDomain(ParameterDomain, metaclass=abc.ABCMeta):
|
|
"""parameter domain which value present a python primitive type."""
|
|
|
|
__slots__ = ()
|
|
|
|
@abc.abstractmethod
|
|
def init_para(self, initial: Optional[Any] = None):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def valid_para(self, value: int) -> bool:
|
|
pass
|
|
|
|
def as_json(self):
|
|
return str(self)
|
|
|
|
|
|
class ParameterBoolDomainType(ParameterTypeDomain):
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> bool:
|
|
if initial is None:
|
|
return False
|
|
else:
|
|
return int(initial) != 0
|
|
|
|
def valid_para(self, value: int) -> bool:
|
|
return True
|
|
|
|
def __str__(self):
|
|
return "bool"
|
|
|
|
|
|
ParameterBoolDomain = ParameterBoolDomainType()
|
|
|
|
|
|
class ParameterIntDomainType(ParameterTypeDomain):
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> int:
|
|
if initial is None:
|
|
return 0
|
|
else:
|
|
return int(initial)
|
|
|
|
def valid_para(self, value: int) -> bool:
|
|
return True
|
|
|
|
def __str__(self):
|
|
return "int"
|
|
|
|
|
|
ParameterIntDomain = ParameterIntDomainType()
|
|
|
|
|
|
class ParameterValueDomain(ParameterDomain, metaclass=abc.ABCMeta):
|
|
"""limited/ranged P value domain """
|
|
|
|
__slots__ = ()
|
|
|
|
def init_para(self, initial: Optional[Any] = None) -> int:
|
|
if initial is None:
|
|
return self.range[0]
|
|
else:
|
|
initial = int(initial)
|
|
f, t = self.range
|
|
|
|
if f <= initial < t:
|
|
return initial
|
|
else:
|
|
return f
|
|
|
|
def valid_para(self, value: Any) -> bool:
|
|
if isinstance(value, list):
|
|
return True
|
|
f, t = self.range
|
|
return f <= float(value) < t
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def range(self) -> Tuple[int, int]:
|
|
"""P values range
|
|
|
|
:return: P min value, P max value
|
|
"""
|
|
pass
|
|
|
|
def __str__(self):
|
|
return "[%d, %d)" % self.range
|
|
|
|
@classmethod
|
|
def parse(cls, json: JSON_ARRAY, constant: ConstantScope) -> 'ParameterValueDomain':
|
|
|
|
if len(json) == 0:
|
|
raise RuntimeError('empty range expression : ' + _truncate(json))
|
|
|
|
elif len(json) == 1:
|
|
_stop = json[0]
|
|
if isinstance(_stop, int):
|
|
return ParameterConstantRangeDomain(0, _stop)
|
|
|
|
_start = ConstantExpression(0)
|
|
_stop = Expression.parse(_stop)
|
|
|
|
elif len(json) == 2:
|
|
_start = json[0]
|
|
_stop = json[1]
|
|
|
|
if isinstance(_start, int) and isinstance(_stop, int):
|
|
return ParameterConstantRangeDomain(_start, _stop)
|
|
|
|
_start = Expression.parse(json[0])
|
|
_stop = Expression.parse(json[1])
|
|
|
|
else:
|
|
raise RuntimeError('not a range expression : ' + _truncate(json))
|
|
|
|
return ParameterRangeDomain(_start, _stop, constant)
|
|
|
|
|
|
class ParameterConstantRangeDomain(ParameterValueDomain):
|
|
"""constant limited/ranged P value domain
|
|
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterConstantValueDomain
|
|
= [int]
|
|
| [int, int]
|
|
|
|
The value would be resolve when this class be initialized.
|
|
The variable search scope is constant pool.
|
|
"""
|
|
|
|
__slots__ = ('_start', '_stop')
|
|
|
|
def __init__(self, start: int, stop: int):
|
|
"""
|
|
|
|
:param start: start P value
|
|
:param stop: stop P value (exclude)
|
|
"""
|
|
self._start: int = start
|
|
self._stop: int = stop
|
|
|
|
@property
|
|
def range(self) -> Tuple[int, int]:
|
|
"""P values range
|
|
|
|
:return: P min value, P max value
|
|
"""
|
|
return self._start, self._stop
|
|
|
|
def as_json(self):
|
|
return [self._start, self._stop]
|
|
|
|
|
|
class ParameterRangeDomain(ParameterValueDomain):
|
|
"""limited/ranged P value domain
|
|
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterRangeDomain
|
|
= [Expression[int]]
|
|
| [Expression[int], Expression[int]]
|
|
|
|
The value would be resolve when this class be initialized.
|
|
The variable search scope is constant pool.
|
|
"""
|
|
|
|
__slots__ = ('_constant',
|
|
'__start', '__stop',
|
|
'_start', '_stop')
|
|
|
|
def __init__(self, start: Expression, stop: Expression, constant: ConstantScope):
|
|
"""
|
|
|
|
:param start: value start expression
|
|
:param stop: value stop (exclude) expression
|
|
:param constant: scope
|
|
"""
|
|
|
|
self._constant = constant
|
|
self.__start: Expression = start
|
|
self.__stop: Expression = stop
|
|
|
|
# start
|
|
ret = self.__start.value(self._constant)
|
|
|
|
try:
|
|
self._start = int(ret)
|
|
except (ValueError, TypeError) as e:
|
|
raise RuntimeError('start not a number : ' + str(ret)) from e
|
|
|
|
# stop
|
|
ret = self.__stop.value(self._constant)
|
|
|
|
try:
|
|
self._stop = int(ret)
|
|
except (ValueError, TypeError) as e:
|
|
raise RuntimeError('stop not a number : ' + str(ret)) from e
|
|
|
|
@property
|
|
def range(self) -> Tuple[int, int]:
|
|
"""P values range
|
|
|
|
:return: P min value, P max value
|
|
"""
|
|
return self._start, self._stop
|
|
|
|
def as_json(self):
|
|
return [self.__start.as_json(), self.__stop.as_json()]
|
|
|
|
|
|
class ParameterCollectionValueOperator(metaclass=abc.ABCMeta):
|
|
"""
|
|
**format**
|
|
|
|
========================= ============= ========== ===========
|
|
expression set list array
|
|
========================= ============= ========== ===========
|
|
``VALUE`` add value add value add value
|
|
``*`` add all value
|
|
``:=`` clear all clear all
|
|
``INDEX=VALUE`` del value set value set value
|
|
``INDEX:TO=VALUE,...`` set value replace replace
|
|
========================= ============= ========== ===========
|
|
|
|
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
@abc.abstractmethod
|
|
def index(self, index_range: int) -> Union[None, int, slice]:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def value(self, domain: ParameterDomain) -> Union[int, List[int]]:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def __str__(self):
|
|
pass
|
|
|
|
def __repr__(self):
|
|
return str(self)
|
|
|
|
@classmethod
|
|
def parse(cls, expr: Union[int, str]) -> 'ParameterCollectionValueOperator':
|
|
# parse epression then create value_operator
|
|
if isinstance(expr, int):
|
|
return SingleValueOperator(None, expr)
|
|
|
|
elif isinstance(expr, list):
|
|
return CollectValueOperator((0, len(expr)), expr)
|
|
|
|
elif expr == '*':
|
|
return AddAllValueOperator
|
|
|
|
elif expr == ':=':
|
|
return CollectValueOperator((None, None), [])
|
|
|
|
elif '[' in expr and ']' in expr:
|
|
v = json.loads(expr)
|
|
return CollectValueOperator((0, len(v)), v)
|
|
|
|
elif '=' in expr:
|
|
a, v = part_suffix(expr, '=')
|
|
|
|
if ':' in a:
|
|
v = list(map(int, v.split(',')))
|
|
|
|
if a == ':':
|
|
return CollectValueOperator((None, None), v)
|
|
|
|
else:
|
|
# get list length (old way)
|
|
a, b = part_suffix(a, ':')
|
|
a = int(a) if len(a) else None
|
|
b = int(b) if len(b) else None
|
|
|
|
return CollectValueOperator((a, b), v)
|
|
|
|
else:
|
|
return SingleValueOperator(int(a), int(v))
|
|
else:
|
|
return SingleValueOperator(None, int(expr))
|
|
|
|
|
|
class SingleValueOperator(ParameterCollectionValueOperator):
|
|
__slots__ = ('_index', '_value')
|
|
|
|
def __init__(self, index: Optional[int], value: int):
|
|
self._index = index
|
|
self._value = value
|
|
|
|
def index(self, index_range: int) -> Optional[int]:
|
|
return self._index
|
|
|
|
def value(self, domain: ParameterDomain) -> int:
|
|
return self._value
|
|
|
|
def __str__(self):
|
|
if self._index is None:
|
|
return str(self._value)
|
|
else:
|
|
return '%d=%d' % (self._index, self._value)
|
|
|
|
|
|
class CollectValueOperator(ParameterCollectionValueOperator):
|
|
__slots__ = ('_index', '_value')
|
|
|
|
def __init__(self, index: Tuple[Optional[int], Optional[int]], value: List[int]):
|
|
self._index = index
|
|
self._value = value
|
|
|
|
def index(self, index_range: int) -> slice:
|
|
a, b = self._index
|
|
a = a if a is not None else 0
|
|
b = min(b, index_range) if b is not None else index_range
|
|
|
|
return slice(a, b)
|
|
|
|
def value(self, domain: ParameterDomain) -> List[int]:
|
|
return self._value
|
|
|
|
def __str__(self):
|
|
a, b = self._index
|
|
|
|
return '%s:%s=%s' % (str(a) if a is not None else '',
|
|
str(b) if b is not None else '',
|
|
','.join(map(str, self._value)))
|
|
|
|
|
|
class AllValueOperatorType(ParameterCollectionValueOperator):
|
|
__slots__ = ()
|
|
|
|
def index(self, index_range: int) -> slice:
|
|
return slice(index_range)
|
|
|
|
def value(self, domain: ParameterDomain) -> List[int]:
|
|
if isinstance(domain, ParameterValueDomain):
|
|
return list(range(*domain.range))
|
|
else:
|
|
raise RuntimeError()
|
|
|
|
def __str__(self):
|
|
return '*'
|
|
|
|
|
|
AddAllValueOperator = AllValueOperatorType()
|
|
|
|
|
|
class ParameterCollectionDomain(ParameterDomain, metaclass=abc.ABCMeta):
|
|
"""special collection with P values.
|
|
|
|
**special value**
|
|
|
|
===== ========
|
|
value describe
|
|
===== ========
|
|
'*' all
|
|
===== ========
|
|
|
|
"""
|
|
|
|
__slots__ = ('_domain',)
|
|
|
|
def __init__(self, domain: ParameterDomain):
|
|
super().__init__()
|
|
|
|
# if isinstance(domain, ParameterCollectionDomain):
|
|
# raise RuntimeError('cannot compose with parameter collection domain')
|
|
|
|
self._domain = domain
|
|
|
|
@property
|
|
@abc.abstractmethod
|
|
def collection_type(self):
|
|
pass
|
|
|
|
@property
|
|
def element_domain(self) -> ParameterDomain:
|
|
return self._domain
|
|
|
|
def init_para(self, initial: Optional[Any] = None):
|
|
"""
|
|
|
|
:param initial: ignore
|
|
:return: empty collection
|
|
"""
|
|
if initial is None:
|
|
return self.collection_type()
|
|
else:
|
|
ret = self.collection_type(initial)
|
|
|
|
for v in ret:
|
|
if not self.valid_para(v):
|
|
raise ValueError('illegal value domain : ' + str(v))
|
|
|
|
return ret
|
|
|
|
def valid_para(self, value: Any) -> bool:
|
|
"""
|
|
|
|
:param value: P value or special value
|
|
:return:
|
|
"""
|
|
if value == '*':
|
|
return True
|
|
|
|
return self._domain.valid_para(value)
|
|
|
|
@abc.abstractmethod
|
|
def oper_para(self, target: Any, oper: ParameterCollectionValueOperator, context: Scope):
|
|
"""operator parameter *value* for the collection *target*.
|
|
|
|
:param target: target collection
|
|
:param oper: value operator
|
|
:param context: general case root
|
|
"""
|
|
pass
|
|
|
|
@classmethod
|
|
def parse(cls, json: JSON_OBJECT, constant: ConstantScope) -> 'ParameterCollectionDomain':
|
|
if 'list' in json:
|
|
return ParameterListDomain.parse(json, constant)
|
|
elif 'set' in json:
|
|
return ParameterSetDomain.parse(json, constant)
|
|
elif 'array' in json:
|
|
return ParameterArrayDomain.parse(json, constant)
|
|
else:
|
|
raise RuntimeError('illegal parameter collection domain : ' + _truncate(json))
|
|
|
|
|
|
class ParameterListDomain(ParameterCollectionDomain):
|
|
"""P value list.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterListDomain = {
|
|
"list" :ParameterValueDomain
|
|
"limit"? :Expression[int] # list size
|
|
}
|
|
|
|
|
|
"""
|
|
|
|
__slots__ = ('__limit', '_limit')
|
|
|
|
def __init__(self,
|
|
domain: ParameterDomain,
|
|
constant: ConstantScope,
|
|
limit: Union[None, int, str, Expression[int]] = None):
|
|
"""
|
|
|
|
:param domain:
|
|
:param limit: list size limit
|
|
"""
|
|
super().__init__(domain)
|
|
|
|
self.__limit = limit
|
|
|
|
if limit is not None:
|
|
if isinstance(limit, Expression):
|
|
limit = limit.value(constant)
|
|
|
|
if isinstance(limit, str):
|
|
limit = constant[limit]
|
|
|
|
if not isinstance(limit, int):
|
|
raise TypeError('illegal limit type : ' + str(limit))
|
|
|
|
self._limit: Optional[int] = limit
|
|
|
|
@property
|
|
def collection_type(self):
|
|
return list
|
|
|
|
def oper_para(self, target: List[Any], oper: ParameterCollectionValueOperator, context: Scope):
|
|
"""operator *value* for the list *target*
|
|
|
|
:param target: target list
|
|
:param oper:
|
|
:param context: general case root
|
|
"""
|
|
if not isinstance(target, list):
|
|
raise TypeError('not a list : ' + str(target))
|
|
|
|
if oper == AddAllValueOperator:
|
|
raise ValueError("list parameter cannot accept '*' (all) value")
|
|
|
|
# d: domain, sz: lsit length, i: slice(a, b), v: list value
|
|
d = self.element_domain
|
|
sz = len(target)
|
|
i = oper.index(sz)
|
|
v = oper.value(d)
|
|
|
|
if i is None and v is None:
|
|
pass
|
|
|
|
elif i is None:
|
|
# add value
|
|
|
|
if isinstance(v, list):
|
|
if not self._valid_list_limit(target, len(v)):
|
|
raise RuntimeError('over size limit : ' + str(sz) + ' + ' + str(len(v)))
|
|
|
|
for _v in v:
|
|
if not d.valid_para(_v):
|
|
raise RuntimeError('illegal value : ' + str(_v))
|
|
|
|
target.extend(v)
|
|
|
|
else:
|
|
if not self._valid_list_limit(target, 1):
|
|
raise RuntimeError('over size limit : ' + str(sz) + ' + 1')
|
|
|
|
if not d.valid_para(v):
|
|
raise RuntimeError('illegal value : ' + str(v))
|
|
|
|
target.append(v)
|
|
|
|
elif len(v) == 0:
|
|
# delete value
|
|
del target[i]
|
|
|
|
else:
|
|
# replace value
|
|
|
|
if isinstance(i, int):
|
|
i = slice(i, i + 1)
|
|
|
|
if isinstance(v, int):
|
|
v = [v]
|
|
|
|
target[i] = v
|
|
|
|
def _valid_list_limit(self, target: List[Any], inc: int) -> bool:
|
|
if self._limit is not None:
|
|
if self._limit < len(target) + inc:
|
|
return False
|
|
|
|
return True
|
|
|
|
def as_json(self):
|
|
return {
|
|
'list': self._domain.as_json(),
|
|
'limit': JsonSerialize.to_json(self.__limit)
|
|
}
|
|
|
|
def __str__(self):
|
|
return "[" + str(self._domain) + "]"
|
|
|
|
@classmethod
|
|
def parse(cls, json: JSON_OBJECT, constant: ConstantScope) -> 'ParameterListDomain':
|
|
col = _get(json, 'list', None, 'parameter list domain')
|
|
|
|
limit = _get(json, 'limit', None, 'parameter list domain', nullable=True)
|
|
if limit is not None:
|
|
limit = Expression.parse(limit)
|
|
|
|
return ParameterListDomain(ParameterDomain.parse(col, constant), constant, limit)
|
|
|
|
|
|
class ParameterSetDomain(ParameterCollectionDomain):
|
|
"""P value set.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterSetDomain = {
|
|
"set" :ParameterValueDomain
|
|
}
|
|
|
|
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, domain: ParameterDomain):
|
|
super().__init__(domain)
|
|
|
|
@property
|
|
def collection_type(self):
|
|
return set
|
|
|
|
def oper_para(self, target: Set[Any], oper: ParameterCollectionValueOperator, context: Scope):
|
|
"""operator *value* for the set *target*
|
|
|
|
:param target: target set
|
|
:param oper: P value or special value
|
|
:param context: general case root
|
|
"""
|
|
|
|
if not isinstance(target, set):
|
|
raise TypeError('not a set : ' + str(target))
|
|
|
|
d = self.element_domain
|
|
|
|
if oper == AddAllValueOperator:
|
|
# add all
|
|
if not isinstance(d, ParameterValueDomain):
|
|
raise RuntimeError('domain not limited')
|
|
|
|
for p in range(*d.range):
|
|
target.add(p)
|
|
|
|
elif isinstance(oper, SingleValueOperator):
|
|
i = oper.index(0)
|
|
v = oper.value(d)
|
|
|
|
if i is None:
|
|
# add value
|
|
|
|
if not d.valid_para(v):
|
|
raise RuntimeError('illegal value : ' + str(v))
|
|
|
|
target.add(v)
|
|
else:
|
|
# remove value
|
|
|
|
try:
|
|
target.remove(v)
|
|
except KeyError:
|
|
pass
|
|
|
|
else:
|
|
assert isinstance(oper, CollectValueOperator)
|
|
# set value
|
|
v = oper.value(d)
|
|
|
|
for _v in v:
|
|
if not d.valid_para(_v):
|
|
raise RuntimeError('illegal value : ' + str(_v))
|
|
|
|
target.clear()
|
|
|
|
if len(v):
|
|
target.update(v)
|
|
|
|
def as_json(self):
|
|
return {'set': self._domain.as_json()}
|
|
|
|
def __str__(self):
|
|
return "{" + str(self._domain) + "}"
|
|
|
|
@classmethod
|
|
def parse(cls, json: JSON_OBJECT, constant: ConstantScope) -> 'ParameterSetDomain':
|
|
col = _get(json, 'set', None, 'parameter list domain')
|
|
return ParameterSetDomain(ParameterDomain.parse(col, constant))
|
|
|
|
|
|
class ParameterArrayDomain(ParameterCollectionDomain):
|
|
"""P value array. It work like as :class:`ParameterListDomain`, but different in view which only present
|
|
one V value instead a list of V values.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ParameterArrayDomain = {
|
|
"array" :ParameterValueDomain
|
|
"index" :parameter_name
|
|
"size" :int | str | Expression[int]
|
|
}
|
|
|
|
"""
|
|
|
|
__slots__ = ('_index', '__size', '_size')
|
|
|
|
def __init__(self,
|
|
domain: ParameterDomain,
|
|
constant: ConstantScope,
|
|
index: str,
|
|
size: Union[int, str, Expression[int]]):
|
|
"""
|
|
|
|
:param domain:
|
|
:param index:
|
|
:param size:
|
|
"""
|
|
super().__init__(domain)
|
|
|
|
self._index = index
|
|
self.__size = size
|
|
|
|
# evaluation size
|
|
if isinstance(size, Expression):
|
|
size = size.value(constant)
|
|
|
|
elif isinstance(size, str):
|
|
size = constant[size]
|
|
|
|
if not isinstance(size, int):
|
|
raise TypeError('illegal size type : ' + str(size))
|
|
|
|
self._size: int = size
|
|
|
|
@property
|
|
def collection_type(self):
|
|
return list
|
|
|
|
def init_para(self, initial: Optional[Any] = None):
|
|
"""
|
|
|
|
:param initial: ignore
|
|
:return: empty collection
|
|
"""
|
|
if initial is None:
|
|
z = self.element_domain.init_para()
|
|
ret = [z for _ in range(self._size)]
|
|
|
|
elif isinstance(initial, (list, tuple)):
|
|
if len(initial) != self._size:
|
|
raise ValueError('size not match : %d != %d' % (len(initial), self._size))
|
|
|
|
ret = list(initial)
|
|
|
|
else:
|
|
ret = [initial for _ in range(self._size)]
|
|
|
|
for v in ret:
|
|
if not self.valid_para(v):
|
|
raise ValueError('illegal value domain : ' + str(v))
|
|
|
|
return ret
|
|
|
|
@property
|
|
def index_parameter(self) -> str:
|
|
return self._index
|
|
|
|
def index(self, context: Scope) -> int:
|
|
return int(context[self._index])
|
|
|
|
@property
|
|
def size(self) -> int:
|
|
return self._size
|
|
|
|
def get_para(self, target: Any, context: Scope) -> Any:
|
|
"""get P value from array *target*
|
|
|
|
:param target: target array
|
|
:param context: general case root
|
|
:return: P' value
|
|
"""
|
|
if isinstance(target, list):
|
|
return target[self.index(context)]
|
|
else:
|
|
return target
|
|
|
|
def oper_para(self, target: List[Any], oper: Any, context: Scope):
|
|
"""set *value* into the set *target*
|
|
|
|
:param target: target array
|
|
:param oper: P' value or special value
|
|
:param context: general case root
|
|
"""
|
|
if not isinstance(target, list):
|
|
raise TypeError('not a array : ' + str(target))
|
|
|
|
if oper == AddAllValueOperator:
|
|
raise ValueError("array parameter cannot accept '*' (all) value")
|
|
|
|
sz = len(target)
|
|
d = self.element_domain
|
|
|
|
if isinstance(oper, SingleValueOperator):
|
|
# set value
|
|
|
|
i = oper.index(sz)
|
|
v = oper.value(d)
|
|
|
|
if i is None:
|
|
i = self.index(context)
|
|
|
|
if not d.valid_para(v):
|
|
raise RuntimeError('illegal value : ' + str(v))
|
|
|
|
target[i] = v
|
|
|
|
else:
|
|
# set array of value
|
|
|
|
assert isinstance(oper, CollectValueOperator)
|
|
i = oper.index(sz)
|
|
i = list(range(*i.indices(sz)))
|
|
|
|
v = oper.value(d)
|
|
|
|
if len(i) != len(v):
|
|
raise RuntimeError('array index range not match value size')
|
|
|
|
for _i, _j in enumerate(i):
|
|
target[_j] = v[_i]
|
|
|
|
def as_json(self):
|
|
return {
|
|
'list': self._domain.as_json(),
|
|
'index': self._index,
|
|
'size': JsonSerialize.to_json(self.__size)
|
|
}
|
|
|
|
def __str__(self):
|
|
return "[" + str(self._domain) + "][" + self._index + ']'
|
|
|
|
@classmethod
|
|
def parse(cls, json: JSON_OBJECT, constant: ConstantScope) -> 'ParameterArrayDomain':
|
|
col = _get(json, 'array', None, 'parameter array domain')
|
|
|
|
index = _get(json, 'index', str, 'parameter array domain')
|
|
|
|
size = _get(json, 'size', None, 'parameter array domain')
|
|
if isinstance(size, dict):
|
|
size = Expression.parse(size)
|
|
|
|
return ParameterArrayDomain(ParameterDomain.parse(col, constant),
|
|
constant, index, size)
|