Source code for parameter_info.indirect_info
# 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 )
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
def get(self, key, default=None):
value = default
if key in self:
value = self[key]
return value
def items(self):
for key in self:
yield key, self[key]
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 InformationSource( object ):
"""
Wrapper for some information source for indirect/lazy data retrieval. Once set up, new query objects can be created
with create_query( source_key[, default] ). The query returned can be executed via (), resulting in actual
value computation via _get_from_source.
"""
def __init__(self, source_key_failure_handler=raise_InformationSourceKeyError, **source_query_data):
"""
:param source_key_failure_handler: Executed on query execution if the key is not supported.
:param source_query_data: Any additional, non key-specific information required to retrieve the information.
"""
super( InformationSource, self ).__init__()
self.query_class = Query
self._source_query_data = source_query_data
self._source_key_failure_handler = source_key_failure_handler
[docs] def create_query(self, source_key=None, default=None):
"""
Returns a query object for the given source_key using the currently set query_class and source data.
:param source_key: Key to query the source for (once executed).
:param default: If key_failure_handler does not raise an exception, this is the value returned when the query\
is executed, but the key does not exist at that time.
:return: Object of type query_class encapsulating all information to retrieve the data when executed.
"""
return self.query_class(
self.get_from_source,
source_key,
default=default,
key_failure_handler=self._source_key_failure_handler,
**self._source_query_data
)
[docs] def get_supported_keys(self):
"""
Should return a comprehensive list of all supported keys. As it may not be possible to know all supported
keys in advance (i.e. before requesting them from the actual source), there is no guarantee to the caller
to receive ALL supported keys. However, all returned keys MUST be supported.
:return: Tuple with all (known) supported keys.
"""
return ()
@classmethod
[docs] def get_from_source(cls, source_key, default=None,
key_failure_handler=raise_InformationSourceKeyError, **source_query_data):
"""
Queries a value from the information source. Is also used by the query objects.
If self._get_from_source() is not successful, the key_failure_handler will be
called with the received InformationSourceKeyError. If this does not raise an exception, default is returned.
:param source_key: Key to query from source
:param default: Value to return if source_key is not found (and no exception is raised by key_failure_handler)
:param key_failure_handler: What to do with the InformationSourceKeyError thrown by _get_from_source on\
key not found.
:param source_query_data: Additional arguments the source may need.
:return: Queried value for source_key
:raises: cls.InformationSourceValueRetrievalError if key was valid, but value could not be computed, or whatever\
exception key_failure_handler raises on key error.
"""
info = default
try:
info = cls._get_from_source(source_key, **source_query_data)
except cls.InformationSourceKeyError as e:
key_failure_handler(e)
return info
@classmethod
def _get_from_source(cls, source_key, **source_query_data):
"""
To be implemented by derived class. Must raise cls.InformationSourceKeyError(source_key) if source_key cannot
be retrieved. Should raise cls.InformationSourceValueRetrievalError() if the key was fine, but value could not
be retrieved.
:param source_key: Key to query from source
:param source_query_data: Additional arguments the source may need.
:return: Queried value for source_key
:raises: cls.InformationSourceKeyError(source_key) if source_key not found,\
cls.InformationSourceValueRetrievalError(source_key, **source_query_data), if the value could not be\
retrieved.
"""
raise NotImplementedError()
[docs] class InformationSourceKeyError(RuntimeError):
"""
Raised if a key could not be retrieved by the information source.
"""
pass
[docs] class InformationSourceValueRetrievalError(RuntimeError):
"""
Raised if a value could not be retrieved from the information source, although the key was valid
"""
def __init__(self, source_key, **source_query_data):
message = "Could not retrieve data for key '{}' with source_query_data={}".format(source_key,
source_query_data)
super( InformationSource.InformationSourceValueRetrievalError, self ).__init__( message )
[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)