# Copyright (c) Fraunhofer MEVIS, Germany. All rights reserved.
# **InsertLicense** code author="Jan-Martin Kuhnigk"
from parameter_info import utils
from parameter_info.parameter_info import ParameterInfo
def raise_InformationSourceKeyError(msg):
raise InformationSource.InformationSourceKeyError( msg )
[docs]class IndirectParameterInfo( ParameterInfo ):
"""
Extension of ParameterInfo which is basically an adapter for data retrieved from a user-defined information source.
As writing to a remote source usually makes no sense, any attempts to write items will result
in an AssignmentOnReadOnlyKeyError.
NOTE: Mixed dictionaries containing both queries and elemental values that may also be written could be supported,\
but that would make this class more complex, so it will not be done without a concrete use case.
"""
[docs] class AssignmentOnReadOnlyKeyError( RuntimeError ):
""" Raised if value assignment is tried on a read-only key."""
pass
def __init__( self, information_source ):
"""
:param information_source: Object to ask for information, see InformationSource
"""
super(IndirectParameterInfo, self).__init__()
self._information_source = information_source
[docs] def to_dict( self ):
"""
DEPRECATED: *Deprecated, use parameter_info.utils.to_dict() instead.*
:return: pure dict
"""
import warnings
warnings.warn( "d.to_dict() is deprecated, use parameter_info.utils.to_dict( d ) instead!" )
# Uncomment the following two lines to find out where to_dict is used:
# import traceback
# traceback.print_stack()
return utils.to_dict( self )
[docs] def register_key(self, key, source_key=None, default=None):
"""
Adds a source query for the given source_key into the dict under the given key (overwriting any pre-existing
entry for that key).
:param key: External key, use to access data in self
:param source_key: Key relayed to self._information_source. If None, the external key is also used\
internally.
:param default: Default value to assign if the key is not supported (and no exception is raised by\
self._information_source)
:return: None
"""
if source_key is None:
source_key = key
self._set_internal_value( key, self._information_source.create_query( source_key, default=default ) )
[docs] def register_keys(self, keys=None):
"""
Adds given keys without values to the dictionary, overwriting any pre-existing keys of the same name.
If keys is a dictionary, it is interpreted as a mapping of external keys (those to be used with self) and
internal keys (those to be interpreted by self._information_source).
See register_key() for further info.
:param keys: Keys to be supported (must be supported by information source at time of read access). If None\
is given, all keys declared as supported by self._information_source are registered.
:return: None
"""
register_all_supported_keys = keys is None
if register_all_supported_keys:
keys = self._information_source.get_supported_keys()
has_source_key_mapping = False
else:
has_source_key_mapping = isinstance(keys, dict)
for key in keys:
source_key = keys[key] if has_source_key_mapping else key
self.register_key(key, source_key=source_key)
# Now search for sub-info groups and initiate recursive registering
if register_all_supported_keys:
for key in self:
if isinstance( self._get_internal_value( key ), IndirectParameterInfo ):
self._get_internal_value( key ).register_keys()
def __setitem__(self, key, value):
if key in self and self._is_indirect_key( key ):
raise self.AssignmentOnReadOnlyKeyError("Direct assignment not allowed on registered key {}!".format( key ) )
self._set_internal_value( key, value )
[docs] def setdefault(self, key, default ):
if key in self and self._is_indirect_key( key ):
raise self.AssignmentOnReadOnlyKeyError("Direct assignment not allowed on registered key {}!".format( key ) )
super().setdefault( key, default )
def _set_internal_value( self, key, value ):
"""
This method must be used to set values, as the public []=... (__set_item__) is not allowed.
"""
super().__setitem__( key, value )
def __getitem__(self, key):
"""
Returns value for the given key.
:raises: KeyError if the key cannot be retrieved from the information source query.
"""
query_or_value = self._get_internal_value( key )
if self._is_indirect_value( query_or_value ):
return query_or_value()
else:
return query_or_value
def _is_indirect_key( self, key ):
"""
Checks if the key refers to a value that is accessed indirectly (i.e. through a query).
:param key: Key to check
:return: True or False
"""
return self._is_indirect_value( self._get_internal_value( key ) )
def _is_indirect_value( self, value ):
return isinstance( value, Query )
def _get_internal_value(self, key):
"""
Convenience access to the query object for the provided key
"""
return super().__getitem__( key )
"""
Methods to override for better dict emulation (the outside world should not have to know about Queries).
"""
[docs] def copy(self): # don't delegate w/ super - dict.copy() -> dict :(
"""
Unlike the other methods below, this one will actually copy references to the queries
"""
result = type(self)( self._information_source )
result.clear()
for key in self:
result._set_internal_value( key, self._get_internal_value(key) )
return result
[docs] def get(self, key, default=None):
value = default
if key in self:
value = self[key]
return value
[docs] def items(self):
for key in self:
yield key, self[key]
[docs] def values(self):
for _, value in self.items():
yield value
def iteritems(self):
for key, value in self.items():
yield key, value
def __repr__(self):
return utils.to_dict( self ).__repr__()
[docs]class Query(object):
"""
Class encapsulating a call to a source getter function that can be deferred. No caching is performed here.
"""
def __init__(self, getter_cb, *getter_args, **getter_kwargs):
"""
:param getter_cb: "Source getter" callback to execute for retrieving the value.
:param getter_args: Arguments required by the source getter callback
:param getter_kwargs: Keyword arguments required by the source getter callback
"""
self.__source_getter = getter_cb
self.__source_getter_args = getter_args
self.__source_getter_kwargs = getter_kwargs
def __call__(self):
"""
Executes the query
:return: Result value
"""
return self.__source_getter(*self.__source_getter_args, **self.__source_getter_kwargs)