"""
CcpNmr version of the Trailets; all subclassed for added functionalities:
- _traitOrder
- fixing of default_value issues (see also https://github.com/ipython/traitlets/issues/165)
- json handlers
"""
#=========================================================================================
# Licence, Reference and Credits
#=========================================================================================
__copyright__ = "Copyright (C) CCPN project (https://www.ccpn.ac.uk) 2014 - 2022"
__credits__ = ("Ed Brooksbank, Joanna Fox, Victoria A Higman, Luca Mureddu, Eliza Płoskoń",
"Timothy J Ragan, Brian O Smith, Gary S Thompson & Geerten W Vuister")
__licence__ = ("CCPN licence. See http://www.ccpn.ac.uk/v3-software/downloads/license",
)
__reference__ = ("Skinner, S.P., Fogh, R.H., Boucher, W., Ragan, T.J., Mureddu, L.G., & Vuister, G.W.",
"CcpNmr AnalysisAssign: a flexible platform for integrated NMR analysis",
"J.Biomol.Nmr (2016), 66, 111-124, http://doi.org/10.1007/s10858-016-0060-y"
)
#=========================================================================================
# Last code modification
#=========================================================================================
__modifiedBy__ = "$modifiedBy: Geerten Vuister $"
__dateModified__ = "$dateModified: 2022-03-16 16:19:26 +0000 (Wed, March 16, 2022) $"
__version__ = "$Revision: 3.1.0 $"
#=========================================================================================
# Created
#=========================================================================================
__author__ = "$Author: geertenv $"
__date__ = "$Date: 2018-05-14 10:28:41 +0000 (Fri, April 07, 2017) $"
#=========================================================================================
# Start of code
#=========================================================================================
import sys
import pathlib
from collections import OrderedDict
from traitlets import \
Long, Complex, CComplex, Bytes, CBytes, \
ObjectName, DottedObjectName, \
Type, This, ForwardDeclaredInstance, ForwardDeclaredType, \
Enum, CaselessStrEnum, TCPAddress, CRegExp, \
TraitType, default, validate, observe, Undefined, HasTraits, TraitError
from traitlets import Any as _Any
from traitlets import Instance as _Instance
from traitlets import Int as _Int
from traitlets import CInt as _CInt
from traitlets import Float as _Float
from traitlets import CFloat as _CFloat
from traitlets import Unicode as _Unicode
from traitlets import CUnicode as _CUnicode
from traitlets import Bool as _Bool
from traitlets import CBool as _CBool
from traitlets import List as _List
from traitlets import Set as _Set
from traitlets import Dict as _Dict
from traitlets import Tuple as _Tuple
from ccpn.util.traits.TraitJsonHandlerBase import TraitJsonHandlerBase, RecursiveDictHandlerABC, \
RecursiveListHandlerABC
from ccpn.util.AttributeDict import AttributeDict
from ccpn.util.Path import aPath, Path
from ccpn.util.Logging import getLogger
from ccpn.framework.Application import getApplication
class _Ordered(object):
"""A class that maintains and sets trait-order
"""
_globalTraitOrder = 0
def __init__(self):
self._traitOrder = _Ordered._globalTraitOrder
_Ordered._globalTraitOrder += 1
[docs]class Any(_Any, _Ordered):
def __init__(self, *args, **kwargs):
if not 'default_value' in kwargs:
raise ValueError('%s Traitlet without explicit default_value' % self.__class__.__name__)
_Any.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class Instance(_Instance, _Ordered):
def __init__(self, *args, **kwargs):
if not 'default_value' in kwargs:
raise ValueError('%s Traitlet without explicit default_value' % self.__class__.__name__)
_Instance.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class Int(_Int, _Ordered):
def __init__(self, *args, **kwargs):
_Int.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class CInt(_CInt, _Ordered):
def __init__(self, *args, **kwargs):
_CInt.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class Float(_Float, _Ordered):
def __init__(self, *args, **kwargs):
_Float.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class CFloat(_CFloat, _Ordered):
def __init__(self, *args, **kwargs):
_CFloat.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class Unicode(_Unicode, _Ordered):
def __init__(self, *args, **kwargs):
_Unicode.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class CUnicode(_CUnicode, _Ordered):
def __init__(self, *args, **kwargs):
_CUnicode.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class Bool(_Bool, _Ordered):
def __init__(self, *args, **kwargs):
_Bool.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class CBool(_CBool, _Ordered):
def __init__(self, *args, **kwargs):
_CBool.__init__(self, *args, **kwargs)
_Ordered.__init__(self)
[docs]class List(_List, _Ordered):
"""Fixing default_value problem"""
def __init__(self, trait=None, default_value=[], minlen=0, maxlen=sys.maxsize, **kwargs):
_List.__init__(self, trait=trait, default_value=default_value, minlen=minlen, maxlen=maxlen, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs]class CList(List, _Ordered):
"""Casting list, any iterable"""
[docs] def validate(self, obj, values):
# local import, because isotopeRecords in Common cause circular imports £%%$$GRr
from ccpn.util.Common import isIterable
if isinstance(values, list):
pass
elif isIterable(values):
values = [val for val in values]
values = self.validate_elements(obj, values)
return values
[docs]class RecursiveList(List):
"""A list trait that implements recursion of any of the values that are a CcpNmrJson (sub)type
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveListHandlerABC):
klass = list
recursion = True
[docs]class Set(_Set, _Ordered):
"""Fixing default_value problem"""
def __init__(self, trait=None, default_value=None, minlen=0, maxlen=sys.maxsize, **kwargs):
_Set.__init__(self, trait=trait, default_value=default_value, minlen=minlen, maxlen=maxlen, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs]class RecursiveSet(Set):
"""A Set trait that implements recursion of any of the values that are a CcpNmrJson (sub)type
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveListHandlerABC):
klass = set
recursion = True
[docs]class Tuple(_Tuple, _Ordered):
"""Fixing default_value problem
"""
def __init__(self, *traits, **kwargs):
default_value = kwargs.setdefault('default_value', None)
_Tuple.__init__(self, *traits, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs]class CTuple(Tuple):
"""Casting tuple, any iterable
"""
[docs] def validate(self, obj, values):
# local import, because isotopeRecords in Common cause circular imports £%%$$GRr
from ccpn.util.Common import isIterable
if isinstance(values, (tuple, list)):
pass
elif isIterable(values):
values = [val for val in values]
values = self.validate_elements(obj, values)
return tuple(values)
[docs]class RecursiveTuple(Tuple):
"""A tuple trait that implements recursion of any of the values that are a CcpNmrJson (sub)type
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveListHandlerABC):
klass = tuple
recursion = True
[docs]class Dict(_Dict, _Ordered):
"""Fixing default_value problem"""
def __init__(self, trait=None, traits=None, default_value={}, **kwargs):
_Dict.__init__(self, trait=trait, traits=traits, default_value=default_value, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs]class RecursiveDict(Dict):
"""A dict trait that implements recursion of any of the values that are a CcpNmrJson (sub)type
Recursion is active by default, unless tagged with .tag(recursion=False)
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveDictHandlerABC):
klass = dict
[docs]class Adict(TraitType, _Ordered):
"""A trait that defines a json serialisable AttributeDict;
dicts or (key,value) iterables are automatically cast into AttributeDict
Recursion is not active
"""
default_value = AttributeDict()
info_text = "'an AttributeDict'"
def __init__(self, default_value={}, allow_none=False, read_only=None, **kwargs):
TraitType.__init__(self, default_value=default_value, allow_none=allow_none, read_only=read_only, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs] def validate(self, obj, value):
"""Assure a AttributeDict instance
"""
if isinstance(value, AttributeDict):
return value
elif isinstance(value, dict):
return AttributeDict(**value)
elif isinstance(value, list) or isinstance(value, tuple):
return AttributeDict(value)
else:
self.error(obj, value)
# trait-specific json handler
[docs] class jsonHandler(RecursiveDictHandlerABC):
klass = AttributeDict
recursion = False
# end class
[docs]class RecursiveAdict(Adict):
"""A trait that defines a json serialisable AttributeDict;
dicts or (key,value) iterables are automatically cast into AttributeDict
Recursion is active
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveDictHandlerABC):
klass = AttributeDict
recursion = True
# end class
[docs]class Odict(TraitType, _Ordered):
"""A trait that defines a json serialisable OrderedDict;
dicts are automatically cast into OrderedDict
Recursion is not active
"""
default_value = OrderedDict()
info_text = "'an OrderedDict'"
def __init__(self, default_value={}, allow_none=False, read_only=False, **kwargs):
TraitType.__init__(self, default_value=default_value, allow_none=allow_none, read_only=read_only, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs] def validate(self, obj, value):
"""Assure a OrderedDict instance
"""
if isinstance(value, OrderedDict):
return value
elif isinstance(value, dict):
return OrderedDict(list(value.items()))
else:
self.error(obj, value)
# trait-specific json handler
[docs] class jsonHandler(RecursiveDictHandlerABC):
klass = OrderedDict
recursion = False
# end class
[docs]class RecursiveOdict(Odict):
"""A trait that defines a json serialisable OrderedDict;
dicts are automatically cast into OrderedDict
Recursion is active
"""
# trait-specific json handler
[docs] class jsonHandler(RecursiveDictHandlerABC):
klass = OrderedDict
recursion = True
# end class
[docs]class Immutable(Any, _Ordered):
info_text = 'an immutable object, intended to be used as constant'
def __init__(self, value):
TraitType.__init__(self, default_value=value, read_only=True)
_Ordered.__init__(self)
# trait-specific json handler
[docs] class jsonHandler(TraitJsonHandlerBase):
"""Serialise Immutable to be json compatible.
"""
# def encode(self, obj, trait): # inherits from base class
# return getattr(obj, trait)
[docs] def decode(self, obj, trait, value):
# force set value
obj.setTraitValue(trait, value, force=True)
# end class
#end class
[docs]class CPath(TraitType, _Ordered):
"""A trait that defines a casting Path object and is json serialisable
"""
default_value = aPath('.')
info_text = "'an Path object'"
def __init__(self, default_value='', allow_none=False, read_only=False, **kwargs):
TraitType.__init__(self, default_value=default_value, allow_none=allow_none, read_only=read_only, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs] def validate(self, obj, value):
"""Assure a Path instance
"""
if isinstance(value, Path):
pass
elif isinstance(value, pathlib.Path) or isinstance(value, str):
value = Path(value)
else:
self.error(obj, value)
return value
# trait-specific json handler
[docs] class jsonHandler(TraitJsonHandlerBase):
"""Serialise Path to be json compatible.
"""
[docs] def encode(self, obj, trait):
# stores as a str for json if not None
value = getattr(obj, trait)
if value is not None:
value = str(value)
return value
[docs] def decode(self, obj, trait, value):
# needs conversion from str into Path if not None
if value is not None:
value = Path(value)
setattr(obj, trait, value)
# end class
# end class
[docs]class CString(TraitType, _Ordered):
"""A trait that defines a string object, casts from bytes object and is json serialisable
"""
default_value = ''
info_text = "'an string'"
NONE_VALUE = '__CSTRING_NONE_VALUE__'
def __init__(self, default_value='', encoding='utf8', allow_none=False, read_only=None, **kwargs):
TraitType.__init__(self, default_value=default_value, allow_none=allow_none, read_only=read_only, **kwargs)
_Ordered.__init__(self)
self.encoding = encoding
if default_value is not None:
self.default_value = default_value
[docs] def asBytes(self, value):
"""Return value encoded as a bytes object; encode None"""
if value is None:
value = self.NONE_VALUE
return bytes(value, self.encoding)
[docs] def fromBytes(self, value):
"""Return value decoded from bytes object; decode NONE_VALUE to None"""
# 3.1.0.alpha2: encountered error that value was of type str
if isinstance(value, bytes):
value = value.decode(self.encoding)
if value == self.NONE_VALUE:
value = None
return value
[docs] def validate(self, obj, value):
"""Assure a str instance
"""
if isinstance(value, str):
pass
elif isinstance(value, bytes):
value = self.fromBytes(value)
# Test again if None is allowed, as this was missed if it was encoded as NONE_VALUE
if value is None and not self.allow_none:
self.error(obj, value)
else:
self.error(obj, value)
return value
# trait-specific json handler
[docs] class jsonHandler(TraitJsonHandlerBase):
"""json compatible;
"""
pass
# def encode(self, obj, trait):
# "returns a json serialisable object"
# value = getattr(obj, trait)
# return CString.asBytes(value)
#
# def decode(self, obj, trait, value):
# "uses value to generate and set the new (or modified) obj"
# value = CString.fromBytes
# setattr(obj, trait, value)
# end class
[docs]class V3Object(TraitType, _Ordered):
"""A trait that defines a V3-object, json serialisable through its Pid
"""
default_value = None
info_text = "A V3-Object"
def __init__(self, default_value = None, allow_none=True, **kwargs):
TraitType.__init__(self, default_value=default_value, allow_none=allow_none, **kwargs)
_Ordered.__init__(self)
if default_value is not None:
self.default_value = default_value
[docs] def validate(self, obj, value):
"""Assure a str instance
"""
from ccpn.core._implementation.AbstractWrapperObject import AbstractWrapperObject
from ccpn.core._implementation.V3CoreObjectABC import V3CoreObjectABC
if isinstance(value, (AbstractWrapperObject, V3CoreObjectABC)):
pass
else:
self.error(obj, value)
return value
# trait-specific json handler
[docs] class jsonHandler(TraitJsonHandlerBase):
"""json compatible;
"""
[docs] def encode(self, obj, trait):
"returns a json serialisable object"
value = getattr(obj, trait)
if value is None:
return None
else:
return value.pid
[docs] def decode(self, obj, trait, value):
"uses value to generate and set the new (or modified) obj"
if value is None:
result = None
else:
_app = getApplication()
if (result := _app.get(value)) is None:
getLogger().debug('Error decoding %r; set to None' % value)
setattr(obj, trait, result)
# end class
[docs]class V3ObjectList(List):
"""A trait that defines a list of V3-objects, json serialisable through their Pid's
"""
default_value = []
info_text = "A V3-ObjectList"
def __init__(self, default_value = [], **kwargs):
List.__init__(self, default_value=default_value, allow_none=False, **kwargs)
if default_value is not None:
self.default_value = default_value
def validate_elements(self, obj, value):
"""Assure a str instance
"""
from ccpn.core._implementation.AbstractWrapperObject import AbstractWrapperObject
from ccpn.core._implementation.V3CoreObjectABC import V3CoreObjectABC
for val in value:
if isinstance(val, (AbstractWrapperObject, V3CoreObjectABC)):
pass
else:
self.error(obj, value)
return value
# trait-specific json handler
[docs] class jsonHandler(TraitJsonHandlerBase):
"""json compatible;
"""
[docs] def encode(self, obj, trait):
"returns a json serialisable object"
value = getattr(obj, trait)
# make a list of pids
pids = [val.pid for val in value]
return pids
[docs] def decode(self, obj, trait, value):
"uses value to generate and set the new (or modified) obj"
_app = getApplication()
# get obj's for the list of pids
result = [_app.get(val) for val in value]
if None in result:
getLogger().warning('Unable to decode some pid\'s to objects; %r' % value)
setattr(obj, trait, result)
# end class