# -*- coding: utf-8 -*-
"""Module for all the built-in container and other types that are supported by default"""
from abc import ABCMeta
import argparse
import collections
import collections.abc
import pathlib
import typing
import uuid
from . import base_savable
from .files import File, BaseFile
from . import helpers
from . import records
from . import refs
from .utils import sync
from . import types
from . import type_ids
__all__ = (
"List",
"LiveList",
"LiveRefList",
"RefList",
"Str",
"Dict",
"RefDict",
"LiveDict",
"LiveRefDict",
"BaseFile",
"File",
)
class _UserType(base_savable.SimpleSavable, metaclass=ABCMeta):
"""Mixin for helping user types to be compatible with the historian.
These typically have a .data member that stores the actual data (list, dict, str, etc)"""
ATTRS = ("data",)
data = None # placeholder
# This is an optional type for the data member. See save_instance_state type for information
# on when it might be useful
DATA_TYPE = None # type: type
def save_instance_state(self, saver):
# This is a convenient way of storing primitive data types directly as the state
# rather than having to be a 'data' member of a dictionary. This makes it much
# easier to search these types
if self.DATA_TYPE is not None and issubclass(
self.DATA_TYPE, saver.get_historian().primitives
):
self._on_save(
saver
) # Call this here are we aren't going to call up the hierarchy
return self.data
return super().save_instance_state(saver)
def load_instance_state(self, saved_state, loader):
# See save_instance_state
if self.DATA_TYPE is not None and issubclass(
self.DATA_TYPE, loader.get_historian().primitives
):
self._on_load(
loader
) # Call this here are we aren't going to call up the hierarchy
self.data = saved_state
else:
super().load_instance_state(saved_state, loader)
class ObjProxy(_UserType):
"""A simple proxy for any object/data type which can also be a primitive"""
TYPE_ID = uuid.UUID("d43c2db5-1e8c-428f-988f-8b198accde47")
def __init__(self, data=None):
super().__init__()
self.data = data
def __call__(self):
return self.data
def assign(self, value):
self.data = value
[docs]class Str(collections.UserString, _UserType):
TYPE_ID = uuid.UUID("350f3634-4a6f-4d35-b229-71238ce9727d")
DATA_TYPE = str
def __init__(self, seq):
collections.UserString.__init__(self, seq)
_UserType.__init__(self)
class Reffer:
"""Mixin for types that want to be able to create references non-primitive types"""
# pylint: disable=too-few-public-methods
def _ref(self, obj):
"""Create a reference for a non-primitive, otherwise use the value"""
if types.is_primitive(obj):
return obj
return self._create_ref(obj)
def _deref(self, obj):
"""Dereference a non-primitive, otherwise return the value"""
if types.is_primitive(obj):
return obj
return obj()
def _create_ref(self, obj):
"""Create a reference for a given object"""
return refs.ObjRef(obj)
# region lists
[docs]class List(collections.UserList, _UserType):
TYPE_ID = uuid.UUID("2b033f70-168f-4412-99ea-d1f131e3a25a")
DATA_TYPE = list
def __init__(self, initlist=None):
collections.UserList.__init__(self, initlist)
_UserType.__init__(self)
[docs]class RefList(collections.abc.MutableSequence, Reffer, _UserType):
"""A list that stores all entries as references in the database except primitives"""
TYPE_ID = uuid.UUID("091efff5-136d-4ac2-bd59-28f50f151263")
DATA_TYPE = list
def __init__(self, init_list=None):
super().__init__()
init_list = init_list or []
self.data = self.DATA_TYPE(self._ref(item) for item in init_list)
def __str__(self):
return str(self.data)
def __repr__(self):
return repr(self.data)
def __getitem__(self, item):
return self._deref(self.data[item])
def __setitem__(self, key, value):
self.data[key] = self._ref(value)
def __delitem__(self, key):
self.data.__delitem__(key)
def __len__(self):
return self.data.__len__()
[docs] def insert(self, index, value):
self.data.insert(index, self._ref(value))
[docs]class LiveList(collections.abc.MutableSequence, _UserType):
"""A list that is always in sync with the database"""
TYPE_ID = uuid.UUID("c83e6206-cd29-4fda-bf76-11fce1681cd9")
def __init__(self, init_list=None):
super().__init__()
init_list = init_list or []
self.data = RefList(self._create_proxy(item) for item in init_list)
@sync()
def __getitem__(self, item):
proxy = self.data[item] # type: ObjProxy
if self.is_saved():
proxy.sync()
return proxy()
@sync(save=True)
def __setitem__(self, key, value):
proxy = self.data[key] # type: ObjProxy
proxy.assign(value)
@sync(save=True)
def __delitem__(self, key):
proxy = self.data[key] # type: ObjProxy
del self.data[key]
if self._historian is not None:
self._historian.delete(proxy)
@sync()
def __len__(self):
return len(self.data)
[docs] @sync(save=True)
def insert(self, index, value):
proxy = self._create_proxy(value)
self.data.insert(index, proxy)
[docs] @sync(save=True)
def append(self, value):
proxy = self._create_proxy(value)
self.data.append(proxy)
def _create_proxy(self, obj):
return ObjProxy(obj)
[docs]class LiveRefList(Reffer, LiveList):
"""A live list that uses references to store objects"""
TYPE_ID = uuid.UUID("98454806-c587-4fcc-a514-65fdefb0180d")
@sync()
def __getitem__(self, item):
proxy: ObjProxy = self.data[item]
if self.is_saved():
proxy.sync()
ref = proxy()
return self._deref(ref)
@sync(save=True)
def __setitem__(self, key, value):
proxy: ObjProxy = self.data[key]
proxy.assign(self._ref(value))
def _create_proxy(self, obj):
return ObjProxy(self._ref(obj))
# endregion
# region Dicts
[docs]class Dict(collections.UserDict, _UserType):
TYPE_ID = uuid.UUID("a7584078-95b6-4e00-bb8a-b077852ca510")
DATA_TYPE = dict
def __init__(self, *args, **kwarg):
collections.UserDict.__init__(self, *args, **kwarg)
_UserType.__init__(self)
[docs]class RefDict(collections.abc.MutableMapping, Reffer, _UserType):
"""A dictionary that stores all values as references in the database."""
TYPE_ID = uuid.UUID("c95f4c4e-766b-4dda-a43c-5fca4fd7bdd0")
DATA_TYPE = dict
def __init__(self, *args, **kwargs):
super().__init__()
initial = dict(*args, **kwargs)
self.data = self.DATA_TYPE(
{key: self._ref(value) for key, value in initial.items()}
)
def __str__(self):
return str(self.data)
def __repr__(self):
return repr(self.data)
def __getitem__(self, item):
return self._deref(self.data[item])
def __setitem__(self, key, value):
self.data[key] = self._ref(value)
def __delitem__(self, key):
self.data.__delitem__(key)
def __iter__(self):
return self.data.__iter__()
def __len__(self):
return self.data.__len__()
[docs]class LiveDict(collections.abc.MutableMapping, _UserType):
TYPE_ID = uuid.UUID("740cc832-721c-4f85-9628-706257eb55b9")
DATA_TYPE = RefDict
def __init__(self, *args, **kwargs):
super().__init__()
initial = dict(*args, **kwargs)
self.data = RefDict(
{key: self._create_proxy(value) for key, value in initial.items()}
)
@sync()
def __getitem__(self, item):
proxy = self.data[item] # type: ObjProxy
proxy.sync()
return proxy()
@sync()
def __setitem__(self, key, value):
if key in self.data:
# Can simply update the proxy
proxy = self.data[key] # type: ObjProxy
proxy.assign(value)
if proxy.is_saved():
proxy.save()
else:
proxy = self._create_proxy(value)
self.data[key] = proxy
if self.is_saved():
self.save() # This will cause the proxy to be saved as well
@sync(save=True)
def __delitem__(self, key):
proxy = self.data[key]
del self.data[key]
self._historian.delete(proxy)
@sync()
def __iter__(self):
return self.data.__iter__()
@sync()
def __len__(self):
return len(self.data)
def _create_proxy(self, value):
return ObjProxy(value)
[docs]class LiveRefDict(Reffer, LiveDict):
"""A live dictionary that uses references to refer to contained objects"""
TYPE_ID = uuid.UUID("16e7e814-8268-46e0-8d8e-6f34132366b9")
def __init__(self, *args, **kwargs):
super().__init__()
initial = dict(*args, **kwargs)
self.data = RefDict(
{key: self._create_proxy(value) for key, value in initial.items()}
)
@sync()
def __getitem__(self, item):
proxy = self.data[item] # type: ObjProxy
proxy.sync()
ref = proxy()
return self._deref(ref)
@sync()
def __setitem__(self, key, value):
if key in self.data:
# Can simply update the proxy
proxy = self.data[key] # type: ObjProxy
proxy.assign(self._ref(value))
if proxy.is_saved():
proxy.save()
else:
proxy = self._create_proxy(value)
self.data[key] = proxy
if self.is_saved():
self.save() # This will cause the proxy to be saved as well
def _create_proxy(self, value):
return ObjProxy(self._ref(value))
class OrderedDictHelper(helpers.BaseHelper):
"""Enable saving of OrderedDicts. In the database, these will be stored as a list of (key, value) pairs and hence
preserve the order."""
TYPE = collections.OrderedDict
TYPE_ID = uuid.UUID("9e7714f8-8ecf-466f-a0e1-6c9fc1d92f51")
def yield_hashables(self, obj: collections.OrderedDict, hasher):
for entry in obj.items():
yield from hasher.yield_hashables(entry)
def save_instance_state(
self, obj: collections.OrderedDict, _saver
) -> typing.List[typing.Tuple]:
return list(obj.items())
def load_instance_state(self, obj, saved_state: typing.List[typing.Tuple], _loader):
obj.__init__(saved_state)
# endregion
class SetHelper(helpers.BaseHelper):
TYPE = set
TYPE_ID = uuid.UUID("3fb0db0e-e095-4829-928f-f72be46ff975")
def yield_hashables(self, obj: set, hasher):
# Yield hashes for all entries
for entry in obj:
yield from hasher.yield_hashables(entry)
def save_instance_state(self, obj: set, _saver) -> typing.List:
return list(obj)
def load_instance_state(self, obj: set, saved_state: List, _loader):
return obj.__init__(saved_state)
class SnapshotIdHelper(helpers.TypeHelper):
"""Add ability to store references"""
TYPE = records.SnapshotId
TYPE_ID = type_ids.SNAPSHOT_ID_TYPE_ID
def eq(self, one, other): # pylint: disable=invalid-name
if not (
isinstance(one, records.SnapshotId)
and isinstance(other, records.SnapshotId)
):
return False
return one.obj_id == other.obj_id and one.version == other.version
def yield_hashables(self, obj, hasher):
yield from hasher.yield_hashables(obj.obj_id)
yield from hasher.yield_hashables(obj.version)
def save_instance_state(self, obj, _saver):
return obj.to_dict()
def load_instance_state(self, obj, saved_state, _loader):
if isinstance(saved_state, list):
# Legacy version
obj.__init__(*saved_state) # pylint: disable=unnecessary-dunder-call
else:
# New version is a dictionary
obj.__init__(**saved_state) # pylint: disable=unnecessary-dunder-call
class PathHelper(helpers.BaseHelper):
TYPE = pathlib.Path
TYPE_ID = uuid.UUID("78e5c6b8-f194-41ae-aead-b231953318e1")
IMMUTABLE = True
def yield_hashables(self, obj: pathlib.Path, hasher):
yield from hasher.yield_hashables(str(obj))
def save_instance_state(self, obj: pathlib.Path, _saver):
return str(obj)
def new(self, encoded_saved_state):
return pathlib.Path(encoded_saved_state)
def load_instance_state(self, obj: pathlib.Path, saved_state, _loader):
pass # Done it all in new
class TupleHelper(helpers.BaseHelper):
TYPE = tuple
TYPE_ID = uuid.UUID("fd9d2f50-71d6-4e70-90b7-117f23d9cbaf")
IMMUTABLE = True
def save_instance_state(self, obj: tuple, _saver):
return list(obj)
def new(self, encoded_saved_state):
return self.TYPE(encoded_saved_state)
def load_instance_state(self, obj: pathlib.Path, saved_state, _loader):
pass # Done it all in new
def yield_hashables(self, obj, hasher):
for entry in obj:
yield from hasher.yield_hashables(entry)
class NamespaceHelper(helpers.BaseHelper):
TYPE = argparse.Namespace
TYPE_ID = uuid.UUID("c43f8329-0d68-4d12-9a35-af8f5ecc4f90")
def yield_hashables(self, obj, hasher):
yield from hasher.yield_hashables(vars(obj))
def save_instance_state(self, obj, _saver):
return vars(obj)
def load_instance_state(self, obj, saved_state, _loader):
obj.__init__(**saved_state) # pylint: disable=unnecessary-dunder-call
HISTORIAN_TYPES = (
Str,
List,
RefList,
LiveList,
LiveRefList,
Dict,
RefDict,
LiveDict,
LiveRefDict,
OrderedDictHelper,
SetHelper,
ObjProxy,
File,
PathHelper,
TupleHelper,
NamespaceHelper,
)