signature:
def contextmanager ( func ) :
    """@ contextmanager decorator .
 Typical usage :
 @ contextmanager
 def some _ generator ( < arguments > ) :
 < setup >
 try :
 yield < value >
 finally :
 < cleanup >
 This makes this :
 with some _ generator ( < arguments > ) as < variable > :
 < body >
 equivalent to this :
 < setup >
 try :
 < variable > = < value >
 < body >
 finally :
 < cleanup >"""

implementation:
@ wraps ( func )
def helper ( * args , ** kwds ) :
    return GeneratorContextManager ( func ( * args , ** kwds ) )
return helper


signature:
def nested ( * managers ) :
    """Combine multiple context managers into a single nested context manager .
 This function has been deprecated in favour of the multiple manager form
 of the with statement .
 The one advantage of this function over the multiple manager form of the
 with statement is that argument unpacking allows it to be
 used with a variable number of context managers as follows :
 with nested ( * managers ) :
 do _ something ( )"""

implementation:
warn ( "With-statements now directly support multiple context managers" , DeprecationWarning , 3 )
exits = [ ]
vars = [ ]
exc = ( None , None , None )
try :
    for mgr in managers :
        exit = mgr . __exit__
        enter = mgr . __enter__
        vars . append ( enter ( ) )
        exits . append ( exit )
    yield vars
except :
    exc = sys . exc_info ( )
finally :
    while exits :
        exit = exits . pop ( )
        try :
            if exit ( * exc ) :
                exc = ( None , None , None )
        except :
            exc = sys . exc_info ( )
    if exc != ( None , None , None ) : # Don ' t rely on sys . exc _ info ( ) still containing
    # the right information . Another exception may
    # have been raised and caught by an exit method
        raise exc [ 0 ] , exc [ 1 ] , exc [ 2 ]


signature:
def decode ( self , s , _w = WHITESPACE . match ) :
    """Return the Python representation of ` ` s ` ` ( a ` ` str ` ` or ` ` unicode ` `
 instance containing a JSON document )"""

implementation:
obj , end = self . raw_decode ( s , idx = _w ( s , 0 ) . end ( ) )
end = _w ( s , end ) . end ( )
if end != len ( s ) :
    raise ValueError ( errmsg ( "Extra data" , s , end , len ( s ) ) )
return obj


signature:
def tf_loss ( self , states , internals , reward , update , reference = None ) :
    """Creates the TensorFlow operations for calculating the L2 loss between predicted
 state values and actual rewards .
 Args :
 states : Dict of state tensors .
 internals : List of prior internal state tensors .
 reward : Reward tensor .
 update : Boolean tensor indicating whether this call happens during an update .
 reference : Optional reference tensor ( s ) , in case of a comparative loss .
 Returns :
 Loss tensor"""

implementation:
prediction = self . predict ( states = states , internals = internals , update = update )
return tf . nn . l2_loss ( t = ( prediction - reward ) )


signature:
def get_variables ( self , include_nontrainable = False ) :
    """Returns the TensorFlow variables used by the baseline .
 Returns :
 List of variables"""

implementation:
if include_nontrainable :
    return [ self . all_variables [ key ] for key in sorted ( self . all_variables ) ]
else :
    return [ self . variables [ key ] for key in sorted ( self . variables ) ]


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates a baseline from a specification dict ."""

implementation:
baseline = util . get_object ( obj = spec , predefined_objects = tensorforce . core . baselines . baselines , kwargs = kwargs )
assert isinstance ( baseline , Baseline )
return baseline


signature:
def reset ( self ) :
    """same as step ( no kwargs to pass ) , but needs to block and return observation _ dict
 - stores the received observation in self . last _ observation"""

implementation:
# Send command .
self . protocol . send ( { "cmd" : "reset" } , self . socket )
# Wait for response .
response = self . protocol . recv ( self . socket )
# Extract observations .
return self . extract_observation ( response )


signature:
def execute ( self , action ) :
    """Executes a single step in the UE4 game . This step may be comprised of one or more actual game ticks for all of
 which the same given
 action - and axis - inputs ( or action number in case of discretized actions ) are repeated .
 UE4 distinguishes between action - mappings , which are boolean actions ( e . g . jump or dont - jump ) and axis - mappings ,
 which are continuous actions
 like MoveForward with values between - 1.0 ( run backwards ) and 1.0 ( run forwards ) , 0.0 would mean : stop ."""

implementation:
action_mappings , axis_mappings = [ ] , [ ]
# TODO : what if more than one actions are passed ?
# Discretized - > each action is an int
if self . discretize_actions : # Pull record from discretized _ actions , which will look like : [ A , Right , SpaceBar ] .
    combination = self . discretized_actions [ action ]
    # Translate to { " axis _ mappings " : [ ( ' A ' , 1.0 ) , ( Right , 1.0 ) ] , " action _ mappings " : [ ( SpaceBar , True ) ] }
    for key , value in combination : # Action mapping ( True or False ) .
        if isinstance ( value , bool ) :
            action_mappings . append ( ( key , value ) )
        # Axis mapping : always use 1.0 as value as UE4 already multiplies with the correct scaling factor .
        else :
            axis_mappings . append ( ( key , value ) )
# Non - discretized : Each action is a dict of action - and axis - mappings defined in UE4 game ' s input settings .
# Re - translate Incoming action names into keyboard keys for the server .
elif action :
    try :
        action_mappings , axis_mappings = self . translate_abstract_actions_to_keys ( action )
    except KeyError as e :
        raise TensorForceError ( "Action- or axis-mapping with name '{}' not defined in connected UE4 game!" . format ( e ) )
# message = { " cmd " : " step " , ' delta _ time ' : 0.33,
# ' actions ' : [ ( ' X ' , True ) , ( ' Y ' , False ) ] ,
# ' axes ' : [ ( ' Left ' : 1.0 ) , ( ' Up ' : - 1.0 ) ]
message = dict ( cmd = "step" , delta_time = self . delta_time , num_ticks = self . num_ticks , actions = action_mappings , axes = axis_mappings )
self . protocol . send ( message , self . socket )
# Wait for response ( blocks ) .
response = self . protocol . recv ( self . socket )
r = response . pop ( b"_reward" , 0.0 )
is_terminal = response . pop ( b"_is_terminal" , False )
obs = self . extract_observation ( response )
# Cache last observation
self . last_observation = obs
return obs , is_terminal , r


signature:
def translate_abstract_actions_to_keys ( self , abstract ) :
    """Translates a list of tuples ( [ pretty mapping ] , [ value ] ) to a list of tuples ( [ some key ] , [ translated value ] )
 each single item in abstract will undergo the following translation :
 Example1:
 we want : " MoveRight " : 5.0
 possible keys for the action are : ( " Right " , 1.0 ) , ( " Left " , - 1.0)
 result : " Right " : 5.0 * 1.0 = 5.0
 Example2:
 we want : " MoveRight " : - 0.5
 possible keys for the action are : ( " Left " , - 1.0 ) , ( " Right " , 1.0)
 result : " Left " : - 0.5 * - 1.0 = 0.5 ( same as " Right " : - 0.5)"""

implementation:
# Solve single tuple with name and value - > should become a list ( len = 1 ) of this tuple .
if len ( abstract ) >= 2 and not isinstance ( abstract [ 1 ] , ( list , tuple ) ) :
    abstract = list ( ( abstract , ) )
# Now go through the list and translate each axis into an actual keyboard key ( or mouse event / etc . . ) .
actions , axes = [ ] , [ ]
for a in abstract : # first _ key = key - name ( action mapping or discretized axis mapping ) OR tuple ( key - name , scale ) ( continuous
# axis mapping )
    first_key = self . action_space_desc [ a [ 0 ] ] [ "keys" ] [ 0 ]
    # action mapping
    if isinstance ( first_key , ( bytes , str ) ) :
        actions . append ( ( first_key , a [ 1 ] ) )
    # axis mapping
    elif isinstance ( first_key , tuple ) :
        axes . append ( ( first_key [ 0 ] , a [ 1 ] * first_key [ 1 ] ) )
    else :
        raise TensorForceError ( "action_space_desc contains unsupported type for key {}!" . format ( a [ 0 ] ) )
return actions , axes


signature:
def discretize_action_space_desc ( self ) :
    """Creates a list of discrete action ( - combinations ) in case we want to learn with a discrete set of actions ,
 but only have action - combinations ( maybe even continuous ) available from the env .
 E . g . the UE4 game has the following action / axis - mappings :
 ` ` ` javascript
 ' Fire ' :
 { ' type ' : ' action ' , ' keys ' : ( ' SpaceBar ' , ) } ,
 ' MoveRight ' :
 { ' type ' : ' axis ' , ' keys ' : ( ( ' Right ' , 1.0 ) , ( ' Left ' , - 1.0 ) , ( ' A ' , - 1.0 ) , ( ' D ' , 1.0 ) ) } ,
 - > this method will discretize them into the following 6 discrete actions :
 ` ` ` javascript
 [ ( Right , 0.0 ) , ( SpaceBar , False ) ] ,
 [ ( Right , 0.0 ) , ( SpaceBar , True ) ]
 [ ( Right , - 1.0 ) , ( SpaceBar , False ) ] ,
 [ ( Right , - 1.0 ) , ( SpaceBar , True ) ] ,
 [ ( Right , 1.0 ) , ( SpaceBar , False ) ] ,
 [ ( Right , 1.0 ) , ( SpaceBar , True ) ] ,"""

implementation:
# Put all unique _ keys lists in one list and itertools . product that list .
unique_list = [ ]
for nice , record in self . action_space_desc . items ( ) :
    list_for_record = [ ]
    if record [ "type" ] == "axis" : # The main key for this record ( always the first one )
        head_key = record [ "keys" ] [ 0 ] [ 0 ]
        # The reference value ( divide by this one to get the others )
        head_value = record [ "keys" ] [ 0 ] [ 1 ]
        # The zero key ( idle action ; axis scale = 0.0)
        list_for_record . append ( ( head_key , 0.0 ) )
        set_ = set ( )
        for key_and_scale in self . action_space_desc [ nice ] [ "keys" ] : # Build unique lists of mappings ( each axis value should only be represented once ) .
            if key_and_scale [ 1 ] not in set_ :
                list_for_record . append ( ( head_key , key_and_scale [ 1 ] / head_value ) )
                set_ . add ( key_and_scale [ 1 ] )
    else : # Action - mapping
        list_for_record = [ ( record [ "keys" ] [ 0 ] , False ) , ( record [ "keys" ] [ 0 ] , True ) ]
    unique_list . append ( list_for_record )
def so ( in_ ) : # in _ is List [ Tuple [ str , any ] ] - > sort by concat ' d sequence of str ( any ' s )
    st = ""
    for i in in_ :
        st += str ( i [ 1 ] )
    return st
# Then sort and get the entire list of all possible sorted meaningful key - combinations .
combinations = list ( itertools . product ( * unique_list ) )
combinations = list ( map ( lambda x : sorted ( list ( x ) , key = lambda y : y [ 0 ] ) , combinations ) )
combinations = sorted ( combinations , key = so )
# Store that list as discretized _ actions .
self . discretized_actions = combinations


signature:
def reset ( self ) :
    """Resets the environment to its initialization state . This method needs to be called to start a
 new episode after the last episode ended .
 : return : initial state"""

implementation:
self . level . reset ( )
# optional : episode = - 1 , seed = None
return self . level . observations ( ) [ self . state_attribute ]


signature:
def execute ( self , action ) :
    """Pass action to universe environment , return reward , next step , terminal state and
 additional info .
 : param action : action to execute as numpy array , should have dtype np . intc and should adhere to
 the specification given in DeepMindLabEnvironment . action _ spec ( level _ id )
 : return : dict containing the next state , the reward , and a boolean indicating if the
 next state is a terminal state"""

implementation:
adjusted_action = list ( )
for action_spec in self . level . action_spec ( ) :
    if action_spec [ 'min' ] == - 1 and action_spec [ 'max' ] == 1 :
        adjusted_action . append ( action [ action_spec [ 'name' ] ] - 1 )
    else :
        adjusted_action . append ( action [ action_spec [ 'name' ] ] )
        # clip ?
action = np . array ( adjusted_action , dtype = np . intc )
reward = self . level . step ( action = action , num_steps = self . repeat_action )
state = self . level . observations ( ) [ 'RGB_INTERLACED' ]
terminal = not self . level . is_running ( )
return state , terminal , reward


signature:
def tf_solve ( self , fn_x , x_init , b ) :
    """Iteratively solves the system of linear equations $ A x = b $ .
 Args :
 fn _ x : A callable returning the left - hand side $ A x $ of the system of linear equations .
 x _ init : Initial solution guess $ x _ 0 $ , zero vector if None .
 b : The right - hand side $ b $ of the system of linear equations .
 Returns :
 A solution $ x $ to the problem as given by the solver ."""

implementation: return super ( ConjugateGradient , self ) . tf_solve ( fn_x , x_init , b )


signature:
def tf_initialize ( self , x_init , b ) :
    """Initialization step preparing the arguments for the first iteration of the loop body :
 $ x _ 0 , 0 , p _ 0 , r _ 0 , r _ 0 ^ 2 $ .
 Args :
 x _ init : Initial solution guess $ x _ 0 $ , zero vector if None .
 b : The right - hand side $ b $ of the system of linear equations .
 Returns :
 Initial arguments for tf _ step ."""

implementation:
if x_init is None : # Initial guess is zero vector if not given .
    x_init = [ tf . zeros ( shape = util . shape ( t ) ) for t in b ]
initial_args = super ( ConjugateGradient , self ) . tf_initialize ( x_init )
# r _ 0 : = b - A * x _ 0
# c _ 0 : = r _ 0
conjugate = residual = [ t - fx for t , fx in zip ( b , self . fn_x ( x_init ) ) ]
# r _ 0 ^ 2 : = r ^ T * r
squared_residual = tf . add_n ( inputs = [ tf . reduce_sum ( input_tensor = ( res * res ) ) for res in residual ] )
return initial_args + ( conjugate , residual , squared_residual )


signature:
def tf_step ( self , x , iteration , conjugate , residual , squared_residual ) :
    """Iteration loop body of the conjugate gradient algorithm .
 Args :
 x : Current solution estimate $ x _ t $ .
 iteration : Current iteration counter $ t $ .
 conjugate : Current conjugate $ c _ t $ .
 residual : Current residual $ r _ t $ .
 squared _ residual : Current squared residual $ r _ t ^ 2 $ .
 Returns :
 Updated arguments for next iteration ."""

implementation:
x , next_iteration , conjugate , residual , squared_residual = super ( ConjugateGradient , self ) . tf_step ( x , iteration , conjugate , residual , squared_residual )
# Ac : = A * c _ t
A_conjugate = self . fn_x ( conjugate )
# TODO : reference ?
if self . damping > 0.0 :
    A_conjugate = [ A_conj + self . damping * conj for A_conj , conj in zip ( A_conjugate , conjugate ) ]
# cAc : = c _ t ^ T * Ac
conjugate_A_conjugate = tf . add_n ( inputs = [ tf . reduce_sum ( input_tensor = ( conj * A_conj ) ) for conj , A_conj in zip ( conjugate , A_conjugate ) ] )
# \ alpha : = r _ t ^ 2 / cAc
alpha = squared_residual / tf . maximum ( x = conjugate_A_conjugate , y = util . epsilon )
# x _ { t + 1 } : = x _ t + \ alpha * c _ t
next_x = [ t + alpha * conj for t , conj in zip ( x , conjugate ) ]
# r _ { t + 1 } : = r _ t - \ alpha * Ac
next_residual = [ res - alpha * A_conj for res , A_conj in zip ( residual , A_conjugate ) ]
# r _ { t + 1 } ^ 2 : = r _ { t + 1 } ^ T * r _ { t + 1}
next_squared_residual = tf . add_n ( inputs = [ tf . reduce_sum ( input_tensor = ( res * res ) ) for res in next_residual ] )
# \ beta = r _ { t + 1 } ^ 2 / r _ t ^ 2
beta = next_squared_residual / tf . maximum ( x = squared_residual , y = util . epsilon )
# c _ { t + 1 } : = r _ { t + 1 } + \ beta * c _ t
next_conjugate = [ res + beta * conj for res , conj in zip ( next_residual , conjugate ) ]
return next_x , next_iteration , next_conjugate , next_residual , next_squared_residual


signature:
def tf_next_step ( self , x , iteration , conjugate , residual , squared_residual ) :
    """Termination condition : max number of iterations , or residual sufficiently small .
 Args :
 x : Current solution estimate $ x _ t $ .
 iteration : Current iteration counter $ t $ .
 conjugate : Current conjugate $ c _ t $ .
 residual : Current residual $ r _ t $ .
 squared _ residual : Current squared residual $ r _ t ^ 2 $ .
 Returns :
 True if another iteration should be performed ."""

implementation:
next_step = super ( ConjugateGradient , self ) . tf_next_step ( x , iteration , conjugate , residual , squared_residual )
return tf . logical_and ( x = next_step , y = ( squared_residual >= util . epsilon ) )


signature:
def tf_step ( self , time , variables , ** kwargs ) :
    """Creates the TensorFlow operations for performing an optimization step .
 Args :
 time : Time tensor .
 variables : List of variables to optimize .
 * * kwargs : Additional arguments passed on to the internal optimizer .
 Returns :
 List of delta tensors corresponding to the updates for each optimized variable ."""

implementation:
deltas = self . optimizer . step ( time = time , variables = variables , ** kwargs )
with tf . control_dependencies ( control_inputs = deltas ) :
    clipped_deltas = list ( )
    exceeding_deltas = list ( )
    for delta in deltas :
        clipped_delta = tf . clip_by_value ( t = delta , clip_value_min = - self . clipping_value , clip_value_max = self . clipping_value )
        clipped_deltas . append ( clipped_delta )
        exceeding_deltas . append ( clipped_delta - delta )
applied = self . apply_step ( variables = variables , deltas = exceeding_deltas )
with tf . control_dependencies ( control_inputs = ( applied , ) ) :
    return [ delta + 0.0 for delta in clipped_deltas ]


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates a layer from a specification dict ."""

implementation:
layer = util . get_object ( obj = spec , predefined_objects = tensorforce . core . networks . layers , kwargs = kwargs )
assert isinstance ( layer , Layer )
return layer


signature:
def tf_q_delta ( self , q_value , next_q_value , terminal , reward ) :
    """Creates the deltas ( or advantage ) of the Q values .
 : return : A list of deltas per action"""

implementation:
for _ in range ( util . rank ( q_value ) - 1 ) :
    terminal = tf . expand_dims ( input = terminal , axis = 1 )
    reward = tf . expand_dims ( input = reward , axis = 1 )
multiples = ( 1 , ) + util . shape ( q_value ) [ 1 : ]
terminal = tf . tile ( input = terminal , multiples = multiples )
reward = tf . tile ( input = reward , multiples = multiples )
zeros = tf . zeros_like ( tensor = next_q_value )
next_q_value = tf . where ( condition = terminal , x = zeros , y = ( self . discount * next_q_value ) )
return reward + next_q_value - q_value


signature:
def target_optimizer_arguments ( self ) :
    """Returns the target optimizer arguments including the time , the list of variables to
 optimize , and various functions which the optimizer might require to perform an update
 step .
 Returns :
 Target optimizer arguments as dict ."""

implementation:
variables = self . target_network . get_variables ( ) + [ variable for name in sorted ( self . target_distributions ) for variable in self . target_distributions [ name ] . get_variables ( ) ]
source_variables = self . network . get_variables ( ) + [ variable for name in sorted ( self . distributions ) for variable in self . distributions [ name ] . get_variables ( ) ]
arguments = dict ( time = self . global_timestep , variables = variables , source_variables = source_variables )
if self . global_model is not None :
    arguments [ 'global_variables' ] = self . global_model . target_network . get_variables ( ) + [ variable for name in sorted ( self . global_model . target_distributions ) for variable in self . global_model . target_distributions [ name ] . get_variables ( ) ]
return arguments


signature:
def from_spec ( spec , kwargs ) :
    """Creates an environment from a specification dict ."""

implementation:
env = tensorforce . util . get_object ( obj = spec , predefined_objects = tensorforce . environments . environments , kwargs = kwargs )
assert isinstance ( env , Environment )
return env


signature:
def setup ( app ) :
    """When used for spinx extension ."""

implementation:
global _is_sphinx
_is_sphinx = True
app . add_config_value ( 'no_underscore_emphasis' , False , 'env' )
app . add_source_parser ( '.md' , M2RParser )
app . add_directive ( 'mdinclude' , MdInclude )


signature:
def output_image_link ( self , m ) :
    """Pass through rest role ."""

implementation: return self . renderer . image_link ( m . group ( 'url' ) , m . group ( 'target' ) , m . group ( 'alt' ) )


signature:
def output_eol_literal_marker ( self , m ) :
    """Pass through rest link ."""

implementation:
marker = ':' if m . group ( 1 ) is None else ''
return self . renderer . eol_literal_marker ( marker )


signature:
def header ( self , text , level , raw = None ) :
    """Rendering header / heading tags like ` ` < h1 > ` ` ` ` < h2 > ` ` .
 : param text : rendered text content for the header .
 : param level : a number for the header level , for example : 1.
 : param raw : raw text content of the header ."""

implementation: return '\n{0}\n{1}\n' . format ( text , self . hmarks [ level ] * len ( text ) )


signature:
def list ( self , body , ordered = True ) :
    """Rendering list tags like ` ` < ul > ` ` and ` ` < ol > ` ` .
 : param body : body contents of the list .
 : param ordered : whether this list is ordered or not ."""

implementation:
mark = '#. ' if ordered else '* '
lines = body . splitlines ( )
for i , line in enumerate ( lines ) :
    if line and not line . startswith ( self . list_marker ) :
        lines [ i ] = ' ' * len ( mark ) + line
return '\n{}\n' . format ( '\n' . join ( lines ) ) . replace ( self . list_marker , mark )


signature:
def table ( self , header , body ) :
    """Rendering table element . Wrap header and body in it .
 : param header : header part of the table .
 : param body : body part of the table ."""

implementation:
table = '\n.. list-table::\n'
if header and not header . isspace ( ) :
    table = ( table + self . indent + ':header-rows: 1\n\n' + self . _indent_block ( header ) + '\n' )
else :
    table = table + '\n'
table = table + self . _indent_block ( body ) + '\n\n'
return table


signature:
def table_row ( self , content ) :
    """Rendering a table row . Like ` ` < tr > ` ` .
 : param content : content of current table row ."""

implementation:
contents = content . splitlines ( )
if not contents :
    return ''
clist = [ '* ' + contents [ 0 ] ]
if len ( contents ) > 1 :
    for c in contents [ 1 : ] :
        clist . append ( '  ' + c )
return '\n' . join ( clist ) + '\n'


signature:
def codespan ( self , text ) :
    """Rendering inline ` code ` text .
 : param text : text content for inline code ."""

implementation:
if '``' not in text :
    return '\ ``{}``\ ' . format ( text )
else : # actually , docutils split spaces in literal
    return self . _raw_html ( '' '{}' '' . format ( text . replace ( '`' , '`' ) ) )


signature:
def link ( self , link , title , text ) :
    """Rendering a given link with content and title .
 : param link : href link for ` ` < a > ` ` tag .
 : param title : title content for ` title ` attribute .
 : param text : text content for description ."""

implementation:
if title :
    raise NotImplementedError ( 'sorry' )
return '\ `{text} <{target}>`_\ ' . format ( target = link , text = text )


signature:
def image ( self , src , title , text ) :
    """Rendering a image with title and text .
 : param src : source link of the image .
 : param title : title text of the image .
 : param text : alt text of the image ."""

implementation:
# rst does not support title option
# and I couldn ' t find title attribute in HTML standard
return '\n' . join ( [ '' , '.. image:: {}' . format ( src ) , '   :target: {}' . format ( src ) , '   :alt: {}' . format ( text ) , '' , ] )


signature:
def run ( self ) :
    """Most of this method is from ` ` docutils . parser . rst . Directive ` ` .
 docutils version : 0.12"""

implementation:
if not self . state . document . settings . file_insertion_enabled :
    raise self . warning ( '"%s" directive disabled.' % self . name )
source = self . state_machine . input_lines . source ( self . lineno - self . state_machine . input_offset - 1 )
source_dir = os . path . dirname ( os . path . abspath ( source ) )
path = rst . directives . path ( self . arguments [ 0 ] )
path = os . path . normpath ( os . path . join ( source_dir , path ) )
path = utils . relative_path ( None , path )
path = nodes . reprunicode ( path )
# get options ( currently not use directive - specific options )
encoding = self . options . get ( 'encoding' , self . state . document . settings . input_encoding )
e_handler = self . state . document . settings . input_encoding_error_handler
tab_width = self . options . get ( 'tab-width' , self . state . document . settings . tab_width )
# open the inclding file
try :
    self . state . document . settings . record_dependencies . add ( path )
    include_file = io . FileInput ( source_path = path , encoding = encoding , error_handler = e_handler )
except UnicodeEncodeError as error :
    raise self . severe ( 'Problems with "%s" directive path:\n' 'Cannot encode input file path "%s" ' '(wrong locale?).' % ( self . name , SafeString ( path ) ) )
except IOError as error :
    raise self . severe ( 'Problems with "%s" directive path:\n%s.' % ( self . name , ErrorString ( error ) ) )
# read from the file
try :
    rawtext = include_file . read ( )
except UnicodeError as error :
    raise self . severe ( 'Problem with "%s" directive:\n%s' % ( self . name , ErrorString ( error ) ) )
config = self . state . document . settings . env . config
converter = M2R ( no_underscore_emphasis = config . no_underscore_emphasis )
include_lines = statemachine . string2lines ( converter ( rawtext ) , tab_width , convert_whitespace = True )
self . state_machine . insert_input ( include_lines , path )
return [ ]


signature:
def clone_worker_agent ( agent , factor , environment , network , agent_config ) :
    """Clones a given Agent ( ` factor ` times ) and returns a list of the cloned Agents with the original Agent
 in the first slot .
 Args :
 agent ( Agent ) : The Agent object to clone .
 factor ( int ) : The length of the final list .
 environment ( Environment ) : The Environment to use for all cloned agents .
 network ( LayeredNetwork ) : The Network to use ( or None ) for an Agent ' s Model .
 agent _ config ( dict ) : A dict of Agent specifications passed into the Agent ' s c ' tor as kwargs .
 Returns :
 The list with ` factor ` cloned agents ( including the original one ) ."""

implementation:
ret = [ agent ]
for i in xrange ( factor - 1 ) :
    worker = WorkerAgentGenerator ( type ( agent ) ) ( states = environment . states , actions = environment . actions , network = network , model = agent . model , ** agent_config )
    ret . append ( worker )
return ret


signature:
def run ( self , num_episodes = - 1 , max_episode_timesteps = - 1 , episode_finished = None , summary_report = None , summary_interval = 0 , num_timesteps = None , deterministic = False , episodes = None , max_timesteps = None , testing = False , sleep = None ) :
    """Executes this runner by starting all Agents in parallel ( each one in one thread ) .
 Args :
 episodes ( int ) : Deprecated ; see num _ episodes .
 max _ timesteps ( int ) : Deprecated ; see max _ episode _ timesteps ."""

implementation:
# Renamed episodes into num _ episodes to match BaseRunner ' s signature ( fully backw . compatible ) .
if episodes is not None :
    num_episodes = episodes
    warnings . warn ( "WARNING: `episodes` parameter is deprecated, use `num_episodes` instead." , category = DeprecationWarning )
assert isinstance ( num_episodes , int )
# Renamed max _ timesteps into max _ episode _ timesteps to match single Runner ' s signature ( fully backw . compatible ) .
if max_timesteps is not None :
    max_episode_timesteps = max_timesteps
    warnings . warn ( "WARNING: `max_timesteps` parameter is deprecated, use `max_episode_timesteps` instead." , category = DeprecationWarning )
assert isinstance ( max_episode_timesteps , int )
if summary_report is not None :
    warnings . warn ( "WARNING: `summary_report` parameter is deprecated, use `episode_finished` callback " "instead to generate summaries every n episodes." , category = DeprecationWarning )
self . reset ( )
# Reset counts / stop - condition for this run .
self . global_episode = 0
self . global_timestep = 0
self . should_stop = False
# Create threads .
threads = [ threading . Thread ( target = self . _run_single , args = ( t , self . agent [ t ] , self . environment [ t ] , ) , kwargs = { "deterministic" : deterministic , "max_episode_timesteps" : max_episode_timesteps , "episode_finished" : episode_finished , "testing" : testing , "sleep" : sleep } ) for t in range ( len ( self . agent ) ) ]
# Start threads .
self . start_time = time . time ( )
[ t . start ( ) for t in threads ]
# Stay idle until killed by SIGINT or a global stop condition is met .
try :
    next_summary = 0
    next_save = 0 if self . save_frequency_unit != "s" else time . time ( )
    while any ( [ t . is_alive ( ) for t in threads ] ) and self . global_episode < num_episodes or num_episodes == - 1 :
        self . time = time . time ( )
        # This is deprecated ( but still supported ) and should be covered by the ` episode _ finished ` callable .
        if summary_report is not None and self . global_episode > next_summary :
            summary_report ( self )
            next_summary += summary_interval
        if self . save_path and self . save_frequency is not None :
            do_save = True
            current = None
            if self . save_frequency_unit == "e" and self . global_episode > next_save :
                current = self . global_episode
            elif self . save_frequency_unit == "s" and self . time > next_save :
                current = self . time
            elif self . save_frequency_unit == "t" and self . global_timestep > next_save :
                current = self . global_timestep
            else :
                do_save = False
            if do_save :
                self . agent [ 0 ] . save_model ( self . save_path )
                # Make sure next save is later than right now .
                while next_save < current :
                    next_save += self . save_frequency
        time . sleep ( 1 )
except KeyboardInterrupt :
    print ( 'Keyboard interrupt, sending stop command to threads' )
self . should_stop = True
# Join threads .
[ t . join ( ) for t in threads ]
print ( 'All threads stopped' )


signature:
def _run_single ( self , thread_id , agent , environment , deterministic = False , max_episode_timesteps = - 1 , episode_finished = None , testing = False , sleep = None ) :
    """The target function for a thread , runs an agent and environment until signaled to stop .
 Adds rewards to shared episode rewards list .
 Args :
 thread _ id ( int ) : The ID of the thread that ' s running this target function .
 agent ( Agent ) : The Agent object that this particular thread uses .
 environment ( Environment ) : The Environment object that this particular thread uses .
 max _ episode _ timesteps ( int ) : Max . number of timesteps per episode . Use - 1 or 0 for non - limited episodes .
 episode _ finished ( callable ) : Function called after each episode that takes an episode summary spec and
 returns False , if this single run should terminate after this episode .
 Can be used e . g . to set a particular mean reward threshold ."""

implementation:
# figure out whether we are using the deprecated way of " episode _ finished " reporting
old_episode_finished = False
if episode_finished is not None and len ( getargspec ( episode_finished ) . args ) == 1 :
    old_episode_finished = True
episode = 0
# Run this single worker ( episode loop ) as long as global count thresholds have not been reached .
while not self . should_stop :
    state = environment . reset ( )
    agent . reset ( )
    self . global_timestep , self . global_episode = agent . timestep , agent . episode
    episode_reward = 0
    # Time step ( within episode ) loop
    time_step = 0
    time_start = time . time ( )
    while True :
        action , internals , states = agent . act ( states = state , deterministic = deterministic , buffered = False )
        reward = 0
        for repeat in xrange ( self . repeat_actions ) :
            state , terminal , step_reward = environment . execute ( action = action )
            reward += step_reward
            if terminal :
                break
        if not testing : # agent . observe ( reward = reward , terminal = terminal )
        # Insert everything at once .
            agent . atomic_observe ( states = state , actions = action , internals = internals , reward = reward , terminal = terminal )
        if sleep is not None :
            time . sleep ( sleep )
        time_step += 1
        episode_reward += reward
        if terminal or time_step == max_episode_timesteps :
            break
        # Abort the episode ( discard its results ) when global says so .
        if self . should_stop :
            return
    self . global_timestep += time_step
    # Avoid race condition where order in episode _ rewards won ' t match order in episode _ timesteps .
    self . episode_list_lock . acquire ( )
    self . episode_rewards . append ( episode_reward )
    self . episode_timesteps . append ( time_step )
    self . episode_times . append ( time . time ( ) - time_start )
    self . episode_list_lock . release ( )
    if episode_finished is not None : # old way of calling episode _ finished
        if old_episode_finished :
            summary_data = { "thread_id" : thread_id , "episode" : episode , "timestep" : time_step , "episode_reward" : episode_reward }
            if not episode_finished ( summary_data ) :
                return
        # New way with BasicRunner ( self ) and thread - id .
        elif not episode_finished ( self , thread_id ) :
            return
    episode += 1


signature:
def _int_to_pos ( self , flat_position ) :
    """Returns x , y from flat _ position integer .
 Args :
 flat _ position : flattened position integer
 Returns : x , y"""

implementation: return flat_position % self . env . action_space . screen_shape [ 0 ] , flat_position % self . env . action_space . screen_shape [ 1 ]


signature:
def _wait_state ( self , state , reward , terminal ) :
    """Wait until there is a state ."""

implementation:
while state == [ None ] or not state :
    state , terminal , reward = self . _execute ( dict ( key = 0 ) )
return state , terminal , reward


signature:
def apply_step ( self , variables , deltas ) :
    """Applies the given ( and already calculated ) step deltas to the variable values .
 Args :
 variables : List of variables .
 deltas : List of deltas of same length .
 Returns :
 The step - applied operation . A tf . group of tf . assign _ add ops ."""

implementation:
if len ( variables ) != len ( deltas ) :
    raise TensorForceError ( "Invalid variables and deltas lists." )
return tf . group ( * ( tf . assign_add ( ref = variable , value = delta ) for variable , delta in zip ( variables , deltas ) ) )


signature:
def minimize ( self , time , variables , ** kwargs ) :
    """Performs an optimization step .
 Args :
 time : Time tensor .
 variables : List of variables to optimize .
 * * kwargs : Additional optimizer - specific arguments . The following arguments are used
 by some optimizers :
 - arguments : Dict of arguments for callables , like fn _ loss .
 - fn _ loss : A callable returning the loss of the current model .
 - fn _ reference : A callable returning the reference values , in case of a comparative
 loss .
 - fn _ kl _ divergence : A callable returning the KL - divergence relative to the
 current model .
 - sampled _ loss : A sampled loss ( integer ) .
 - return _ estimated _ improvement : Returns the estimated improvement resulting from
 the natural gradient calculation if true .
 - source _ variables : List of source variables to synchronize with .
 - global _ variables : List of global variables to apply the proposed optimization
 step to .
 Returns :
 The optimization operation ."""

implementation:
# # Add training variable gradient histograms / scalars to summary output
# # if ' gradients ' in self . summary _ labels :
# if any ( k in self . summary _ labels for k in [ ' gradients ' , ' gradients _ histogram ' , ' gradients _ scalar ' ] ) :
# valid = True
# if isinstance ( self , tensorforce . core . optimizers . TFOptimizer ) :
# gradients = self . optimizer . compute _ gradients ( kwargs [ ' fn _ loss ' ] ( ) )
# elif isinstance ( self . optimizer , tensorforce . core . optimizers . TFOptimizer ) :
# # This section handles " Multi _ step " and may handle others
# # if failure is found , add another elif to handle that case
# gradients = self . optimizer . optimizer . compute _ gradients ( kwargs [ ' fn _ loss ' ] ( ) )
# else :
# # Didn ' t find proper gradient information
# valid = False
# # Valid gradient data found , create summary data items
# if valid :
# for grad , var in gradients :
# if grad is not None :
# if any ( k in self . summary _ labels for k in ( ' gradients ' , ' gradients _ scalar ' ) ) :
# axes = list ( range ( len ( grad . shape ) ) )
# mean , var = tf . nn . moments ( grad , axes )
# tf . contrib . summary . scalar ( name = ' gradients / ' + var . name + " / mean " , tensor = mean )
# tf . contrib . summary . scalar ( name = ' gradients / ' + var . name + " / variance " , tensor = var )
# if any ( k in self . summary _ labels for k in ( ' gradients ' , ' gradients _ histogram ' ) ) :
# tf . contrib . summary . histogram ( name = ' gradients / ' + var . name , tensor = grad )
deltas = self . step ( time = time , variables = variables , ** kwargs )
with tf . control_dependencies ( control_inputs = deltas ) :
    return tf . no_op ( )


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates an optimizer from a specification dict ."""

implementation:
optimizer = util . get_object ( obj = spec , predefined_objects = tensorforce . core . optimizers . optimizers , kwargs = kwargs )
assert isinstance ( optimizer , Optimizer )
return optimizer


signature:
def np_dtype ( dtype ) :
    """Translates dtype specifications in configurations to numpy data types .
 Args :
 dtype : String describing a numerical type ( e . g . ' float ' ) or numerical type primitive .
 Returns : Numpy data type"""

implementation:
if dtype == 'float' or dtype == float or dtype == np . float32 or dtype == tf . float32 :
    return np . float32
elif dtype == np . float64 or dtype == tf . float64 :
    return np . float64
elif dtype == np . float16 or dtype == tf . float16 :
    return np . float16
elif dtype == 'int' or dtype == int or dtype == np . int32 or dtype == tf . int32 :
    return np . int32
elif dtype == np . int64 or dtype == tf . int64 :
    return np . int64
elif dtype == np . int16 or dtype == tf . int16 :
    return np . int16
elif dtype == 'bool' or dtype == bool or dtype == np . bool_ or dtype == tf . bool :
    return np . bool_
else :
    raise TensorForceError ( "Error: Type conversion from type {} not supported." . format ( str ( dtype ) ) )


signature:
def get_tensor_dependencies ( tensor ) :
    """Utility method to get all dependencies ( including placeholders ) of a tensor ( backwards through the graph ) .
 Args :
 tensor ( tf . Tensor ) : The input tensor .
 Returns : Set of all dependencies ( including needed placeholders ) for the input tensor ."""

implementation:
dependencies = set ( )
dependencies . update ( tensor . op . inputs )
for sub_op in tensor . op . inputs :
    dependencies . update ( get_tensor_dependencies ( sub_op ) )
return dependencies


signature:
def get_object ( obj , predefined_objects = None , default_object = None , kwargs = None ) :
    """Utility method to map some kind of object specification to its content ,
 e . g . optimizer or baseline specifications to the respective classes .
 Args :
 obj : A specification dict ( value for key ' type ' optionally specifies
 the object , options as follows ) , a module path ( e . g . ,
 my _ module . MyClass ) , a key in predefined _ objects , or a callable
 ( e . g . , the class type object ) .
 predefined _ objects : Dict containing predefined set of objects ,
 accessible via their key
 default _ object : Default object is no other is specified
 kwargs : Arguments for object creation
 Returns : The retrieved object"""

implementation:
args = ( )
kwargs = dict ( ) if kwargs is None else kwargs
if isinstance ( obj , str ) and os . path . isfile ( obj ) :
    with open ( obj , 'r' ) as fp :
        obj = json . load ( fp = fp )
if isinstance ( obj , dict ) :
    kwargs . update ( obj )
    obj = kwargs . pop ( 'type' , None )
if predefined_objects is not None and obj in predefined_objects :
    obj = predefined_objects [ obj ]
elif isinstance ( obj , str ) :
    if obj . find ( '.' ) != - 1 :
        module_name , function_name = obj . rsplit ( '.' , 1 )
        module = importlib . import_module ( module_name )
        obj = getattr ( module , function_name )
    else :
        raise TensorForceError ( "Error: object {} not found in predefined objects: {}" . format ( obj , list ( predefined_objects or ( ) ) ) )
elif callable ( obj ) :
    pass
elif default_object is not None :
    args = ( obj , )
    obj = default_object
else : # assumes the object is already instantiated
    return obj
return obj ( * args , ** kwargs )


signature:
def prepare_kwargs ( raw , string_parameter = 'name' ) :
    """Utility method to convert raw string / diction input into a dictionary to pass
 into a function . Always returns a dictionary .
 Args :
 raw : string or dictionary , string is assumed to be the name of the activation
 activation function . Dictionary will be passed through unchanged .
 Returns : kwargs dictionary for * * kwargs"""

implementation:
kwargs = dict ( )
if isinstance ( raw , dict ) :
    kwargs . update ( raw )
elif isinstance ( raw , str ) :
    kwargs [ string_parameter ] = raw
return kwargs


signature:
def register_saver_ops ( self ) :
    """Registers the saver operations to the graph in context ."""

implementation:
variables = self . get_savable_variables ( )
if variables is None or len ( variables ) == 0 :
    self . _saver = None
    return
base_scope = self . _get_base_variable_scope ( )
variables_map = { strip_name_scope ( v . name , base_scope ) : v for v in variables }
self . _saver = tf . train . Saver ( var_list = variables_map , reshape = False , sharded = False , max_to_keep = 5 , keep_checkpoint_every_n_hours = 10000.0 , name = None , restore_sequentially = False , saver_def = None , builder = None , defer_build = False , allow_empty = True , write_version = tf . train . SaverDef . V2 , pad_step_number = False , save_relative_paths = True )


signature:
def save ( self , sess , save_path , timestep = None ) :
    """Saves this component ' s managed variables .
 Args :
 sess : The session for which to save the managed variables .
 save _ path : The path to save data to .
 timestep : Optional , the timestep to append to the file name .
 Returns :
 Checkpoint path where the model was saved ."""

implementation:
if self . _saver is None :
    raise TensorForceError ( "register_saver_ops should be called before save" )
return self . _saver . save ( sess = sess , save_path = save_path , global_step = timestep , write_meta_graph = False , write_state = True , # Do we need this ?
)


signature:
def restore ( self , sess , save_path ) :
    """Restores the values of the managed variables from disk location .
 Args :
 sess : The session for which to save the managed variables .
 save _ path : The path used to save the data to ."""

implementation:
if self . _saver is None :
    raise TensorForceError ( "register_saver_ops should be called before restore" )
self . _saver . restore ( sess = sess , save_path = save_path )


signature:
def reset ( self ) :
    """Calls ` reset ` on all our Preprocessor objects .
 Returns :
 A list of tensors to be fetched ."""

implementation:
fetches = [ ]
for processor in self . preprocessors :
    fetches . extend ( processor . reset ( ) or [ ] )
return fetches


signature:
def process ( self , tensor ) :
    """Process state .
 Args :
 tensor : tensor to process
 Returns : processed state"""

implementation:
for processor in self . preprocessors :
    tensor = processor . process ( tensor = tensor )
return tensor


signature:
def processed_shape ( self , shape ) :
    """Shape of preprocessed state given original shape .
 Args :
 shape : original state shape
 Returns : processed state shape"""

implementation:
for processor in self . preprocessors :
    shape = processor . processed_shape ( shape = shape )
return shape


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates a preprocessing stack from a specification dict ."""

implementation:
if isinstance ( spec , dict ) :
    spec = [ spec ]
stack = PreprocessorStack ( )
for preprocessor_spec in spec : # need to deep copy , otherwise will add first processors spec _ to kwargs to second processor
    preprocessor_kwargs = copy . deepcopy ( kwargs )
    preprocessor = util . get_object ( obj = preprocessor_spec , predefined_objects = tensorforce . core . preprocessors . preprocessors , kwargs = preprocessor_kwargs )
    assert isinstance ( preprocessor , Preprocessor )
    stack . preprocessors . append ( preprocessor )
return stack


signature:
def tf_solve ( self , fn_x , x_init , * args ) :
    """Iteratively solves an equation / optimization for $ x $ involving an expression $ f ( x ) $ .
 Args :
 fn _ x : A callable returning an expression $ f ( x ) $ given $ x $ .
 x _ init : Initial solution guess $ x _ 0 $ .
 * args : Additional solver - specific arguments .
 Returns :
 A solution $ x $ to the problem as given by the solver ."""

implementation:
self . fn_x = fn_x
# Initialization step
args = self . initialize ( x_init , * args )
# args = util . map _ tensors ( fn = tf . stop _ gradient , tensors = args )
# Iteration loop with termination condition
if self . unroll_loop : # Unrolled for loop
    for _ in range ( self . max_iterations ) :
        next_step = self . next_step ( * args )
        step = ( lambda : self . step ( * args ) )
        do_nothing = ( lambda : args )
        args = tf . cond ( pred = next_step , true_fn = step , false_fn = do_nothing )
else : # TensorFlow while loop
    args = tf . while_loop ( cond = self . next_step , body = self . step , loop_vars = args )
# First argument contains solution
return args [ 0 ]


signature:
def execute ( self , action ) :
    """Executes action , observes next state and reward .
 Args :
 actions : Actions to execute .
 Returns :
 Tuple of ( next state , bool indicating terminal , reward )"""

implementation:
next_state , rew , done , _ = self . env . step ( action )
return next_state , rew , done


signature:
def as_local_model ( self ) :
    """Makes sure our optimizer is wrapped into the global _ optimizer meta . This is only relevant for distributed RL ."""

implementation:
super ( MemoryModel , self ) . as_local_model ( )
self . optimizer_spec = dict ( type = 'global_optimizer' , optimizer = self . optimizer_spec )


signature:
def setup_components_and_tf_funcs ( self , custom_getter = None ) :
    """Constructs the memory and the optimizer objects .
 Generates and stores all template functions ."""

implementation:
custom_getter = super ( MemoryModel , self ) . setup_components_and_tf_funcs ( custom_getter )
# Memory
self . memory = Memory . from_spec ( spec = self . memory_spec , kwargs = dict ( states = self . states_spec , internals = self . internals_spec , actions = self . actions_spec , summary_labels = self . summary_labels ) )
# Optimizer
self . optimizer = Optimizer . from_spec ( spec = self . optimizer_spec , kwargs = dict ( summary_labels = self . summary_labels ) )
# TensorFlow functions
self . fn_discounted_cumulative_reward = tf . make_template ( name_ = 'discounted-cumulative-reward' , func_ = self . tf_discounted_cumulative_reward , custom_getter_ = custom_getter )
self . fn_reference = tf . make_template ( name_ = 'reference' , func_ = self . tf_reference , custom_getter_ = custom_getter )
self . fn_loss_per_instance = tf . make_template ( name_ = 'loss-per-instance' , func_ = self . tf_loss_per_instance , custom_getter_ = custom_getter )
self . fn_regularization_losses = tf . make_template ( name_ = 'regularization-losses' , func_ = self . tf_regularization_losses , custom_getter_ = custom_getter )
self . fn_loss = tf . make_template ( name_ = 'loss' , func_ = self . tf_loss , custom_getter_ = custom_getter )
self . fn_optimization = tf . make_template ( name_ = 'optimization' , func_ = self . tf_optimization , custom_getter_ = custom_getter )
self . fn_import_experience = tf . make_template ( name_ = 'import-experience' , func_ = self . tf_import_experience , custom_getter_ = custom_getter )
return custom_getter


signature:
def tf_discounted_cumulative_reward ( self , terminal , reward , discount = None , final_reward = 0.0 , horizon = 0 ) :
    """Creates and returns the TensorFlow operations for calculating the sequence of discounted cumulative rewards
 for a given sequence of single rewards .
 Example :
 single rewards = 2.0 1.0 0.0 0.5 1.0 - 1.0
 terminal = False , False , False , False True False
 gamma = 0.95
 final _ reward = 100.0 ( only matters for last episode ( r = - 1.0 ) as this episode has no terminal signal )
 horizon = 3
 output = 2.95 1.45 1.38 1.45 1.0 94.0
 Args :
 terminal : Tensor ( bool ) holding the is - terminal sequence . This sequence may contain more than one
 True value . If its very last element is False ( not terminating ) , the given ` final _ reward ` value
 is assumed to follow the last value in the single rewards sequence ( see below ) .
 reward : Tensor ( float ) holding the sequence of single rewards . If the last element of ` terminal ` is False ,
 an assumed last reward of the value of ` final _ reward ` will be used .
 discount ( float ) : The discount factor ( gamma ) . By default , take the Model ' s discount factor .
 final _ reward ( float ) : Reward value to use if last episode in sequence does not terminate ( terminal sequence
 ends with False ) . This value will be ignored if horizon = = 1 or discount = = 0.0.
 horizon ( int ) : The length of the horizon ( e . g . for n - step cumulative rewards in continuous tasks
 without terminal signals ) . Use 0 ( default ) for an infinite horizon . Note that horizon = 1 leads to the
 exact same results as a discount factor of 0.0.
 Returns :
 Discounted cumulative reward tensor with the same shape as ` reward ` ."""

implementation:
# By default - > take Model ' s gamma value
if discount is None :
    discount = self . discount
# Accumulates discounted ( n - step ) reward ( start new if terminal )
def cumulate ( cumulative , reward_terminal_horizon_subtract ) :
    rew , is_terminal , is_over_horizon , sub = reward_terminal_horizon_subtract
    return tf . where ( # If terminal , start new cumulation .
    condition = is_terminal , x = rew , y = tf . where ( # If we are above the horizon length ( H ) - > subtract discounted value from H steps back .
    condition = is_over_horizon , x = ( rew + cumulative * discount - sub ) , y = ( rew + cumulative * discount ) ) )
# Accumulates length of episodes ( starts new if terminal )
def len_ ( cumulative , term ) :
    return tf . where ( condition = term , # Start counting from 1 after is - terminal signal
    x = tf . ones ( shape = ( ) , dtype = tf . int32 ) , # Otherwise , increase length by 1
    y = cumulative + 1 )
# Reverse , since reward cumulation is calculated right - to - left , but tf . scan only works left - to - right .
reward = tf . reverse ( tensor = reward , axis = ( 0 , ) )
# e . g . - 1.0 1.0 0.5 0.0 1.0 2.0
terminal = tf . reverse ( tensor = terminal , axis = ( 0 , ) )
# e . g . F T F F F F
# Store the steps until end of the episode ( s ) determined by the input terminal signals ( True starts new count ) .
lengths = tf . scan ( fn = len_ , elems = terminal , initializer = 0 )
# e . g . 1 1 2 3 4 5
off_horizon = tf . greater ( lengths , tf . fill ( dims = tf . shape ( lengths ) , value = horizon ) )
# e . g . F F F F T T
# Calculate the horizon - subtraction value for each step .
if horizon > 0 :
    horizon_subtractions = tf . map_fn ( lambda x : ( discount ** horizon ) * x , reward , dtype = tf . float32 )
    # Shift right by size of horizon ( fill rest with 0.0 ) .
    horizon_subtractions = tf . concat ( [ np . zeros ( shape = ( horizon , ) ) , horizon_subtractions ] , axis = 0 )
    horizon_subtractions = tf . slice ( horizon_subtractions , begin = ( 0 , ) , size = tf . shape ( reward ) )
    # e . g . 0.0 , 0.0 , 0.0 , - 1.0 * g ^ 3 , 1.0 * g ^ 3 , 0.5 * g ^ 3
# all 0.0 if infinite horizon ( special case : horizon = 0)
else :
    horizon_subtractions = tf . zeros ( shape = tf . shape ( reward ) )
# Now do the scan , each time summing up the previous step ( discounted by gamma ) and
# subtracting the respective ` horizon _ subtraction ` .
reward = tf . scan ( fn = cumulate , elems = ( reward , terminal , off_horizon , horizon_subtractions ) , initializer = final_reward if horizon != 1 else 0.0 )
# Re - reverse again to match input sequences .
return tf . reverse ( tensor = reward , axis = ( 0 , ) )


signature:
def tf_reference ( self , states , internals , actions , terminal , reward , next_states , next_internals , update ) :
    """Creates the TensorFlow operations for obtaining the reference tensor ( s ) , in case of a
 comparative loss .
 Args :
 states : Dict of state tensors .
 internals : List of prior internal state tensors .
 actions : Dict of action tensors .
 terminal : Terminal boolean tensor .
 reward : Reward tensor .
 next _ states : Dict of successor state tensors .
 next _ internals : List of posterior internal state tensors .
 update : Boolean tensor indicating whether this call happens during an update .
 Returns :
 Reference tensor ( s ) ."""

implementation: return None


signature:
def tf_loss_per_instance ( self , states , internals , actions , terminal , reward , next_states , next_internals , update , reference = None ) :
    """Creates the TensorFlow operations for calculating the loss per batch instance .
 Args :
 states : Dict of state tensors .
 internals : Dict of prior internal state tensors .
 actions : Dict of action tensors .
 terminal : Terminal boolean tensor .
 reward : Reward tensor .
 next _ states : Dict of successor state tensors .
 next _ internals : List of posterior internal state tensors .
 update : Boolean tensor indicating whether this call happens during an update .
 reference : Optional reference tensor ( s ) , in case of a comparative loss .
 Returns :
 Loss per instance tensor ."""

implementation: raise NotImplementedError


signature:
def tf_loss ( self , states , internals , actions , terminal , reward , next_states , next_internals , update , reference = None ) :
    """Creates the TensorFlow operations for calculating the full loss of a batch .
 Args :
 states : Dict of state tensors .
 internals : List of prior internal state tensors .
 actions : Dict of action tensors .
 terminal : Terminal boolean tensor .
 reward : Reward tensor .
 next _ states : Dict of successor state tensors .
 next _ internals : List of posterior internal state tensors .
 update : Boolean tensor indicating whether this call happens during an update .
 reference : Optional reference tensor ( s ) , in case of a comparative loss .
 Returns :
 Loss tensor ."""

implementation:
# Mean loss per instance
loss_per_instance = self . fn_loss_per_instance ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward , next_states = next_states , next_internals = next_internals , update = update , reference = reference )
# Returns no - op .
updated = self . memory . update_batch ( loss_per_instance = loss_per_instance )
with tf . control_dependencies ( control_inputs = ( updated , ) ) :
    loss = tf . reduce_mean ( input_tensor = loss_per_instance , axis = 0 )
    # Loss without regularization summary .
    if 'losses' in self . summary_labels :
        tf . contrib . summary . scalar ( name = 'loss-without-regularization' , tensor = loss )
    # Regularization losses .
    losses = self . fn_regularization_losses ( states = states , internals = internals , update = update )
    if len ( losses ) > 0 :
        loss += tf . add_n ( inputs = [ losses [ name ] for name in sorted ( losses ) ] )
        if 'regularization' in self . summary_labels :
            for name in sorted ( losses ) :
                tf . contrib . summary . scalar ( name = ( 'regularization/' + name ) , tensor = losses [ name ] )
    # Total loss summary .
    if 'losses' in self . summary_labels or 'total-loss' in self . summary_labels :
        tf . contrib . summary . scalar ( name = 'total-loss' , tensor = loss )
    return loss


signature:
def optimizer_arguments ( self , states , internals , actions , terminal , reward , next_states , next_internals ) :
    """Returns the optimizer arguments including the time , the list of variables to optimize ,
 and various functions which the optimizer might require to perform an update step .
 Args :
 states ( dict ) : Dict of state tensors .
 internals ( dict ) : Dict of prior internal state tensors .
 actions ( dict ) : Dict of action tensors .
 terminal : 1D boolean is - terminal tensor .
 reward : 1D ( float ) rewards tensor .
 next _ states ( dict ) : Dict of successor state tensors .
 next _ internals ( dict ) : Dict of posterior internal state tensors .
 Returns :
 Optimizer arguments as dict to be used as * * kwargs to the optimizer ."""

implementation:
arguments = dict ( time = self . global_timestep , variables = self . get_variables ( ) , arguments = dict ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward , next_states = next_states , next_internals = next_internals , update = tf . constant ( value = True ) ) , fn_reference = self . fn_reference , fn_loss = self . fn_loss )
if self . global_model is not None :
    arguments [ 'global_variables' ] = self . global_model . get_variables ( )
return arguments


signature:
def tf_optimization ( self , states , internals , actions , terminal , reward , next_states = None , next_internals = None ) :
    """Creates the TensorFlow operations for performing an optimization update step based
 on the given input states and actions batch .
 Args :
 states : Dict of state tensors .
 internals : List of prior internal state tensors .
 actions : Dict of action tensors .
 terminal : Terminal boolean tensor .
 reward : Reward tensor .
 next _ states : Dict of successor state tensors .
 next _ internals : List of posterior internal state tensors .
 Returns :
 The optimization operation ."""

implementation:
arguments = self . optimizer_arguments ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward , next_states = next_states , next_internals = next_internals )
return self . optimizer . minimize ( ** arguments )


signature:
def tf_observe_timestep ( self , states , internals , actions , terminal , reward ) :
    """Creates and returns the op that - if frequency condition is hit - pulls a batch from the memory
 and does one optimization step ."""

implementation:
# Store timestep in memory
stored = self . memory . store ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward )
# Periodic optimization
with tf . control_dependencies ( control_inputs = ( stored , ) ) :
    unit = self . update_mode [ 'unit' ]
    batch_size = self . update_mode [ 'batch_size' ]
    frequency = self . update_mode . get ( 'frequency' , batch_size )
    first_update = self . update_mode . get ( 'first_update' , 0 )
    if unit == 'timesteps' : # Timestep - based batch
        optimize = tf . logical_and ( x = tf . equal ( x = ( self . timestep % frequency ) , y = 0 ) , y = tf . logical_and ( x = tf . greater_equal ( x = self . timestep , y = batch_size ) , y = tf . greater_equal ( x = self . timestep , y = first_update ) ) )
    elif unit == 'episodes' : # Episode - based batch
        optimize = tf . logical_and ( x = tf . equal ( x = ( self . episode % frequency ) , y = 0 ) , y = tf . logical_and ( # Only update once per episode increment .
        x = tf . greater ( x = tf . count_nonzero ( input_tensor = terminal ) , y = 0 ) , y = tf . logical_and ( x = tf . greater_equal ( x = self . episode , y = batch_size ) , y = tf . greater_equal ( x = self . episode , y = first_update ) ) ) )
    elif unit == 'sequences' : # Timestep - sequence - based batch
        sequence_length = self . update_mode . get ( 'length' , 8 )
        optimize = tf . logical_and ( x = tf . equal ( x = ( self . timestep % frequency ) , y = 0 ) , y = tf . logical_and ( x = tf . greater_equal ( x = self . timestep , y = ( batch_size + sequence_length - 1 ) ) , y = tf . greater_equal ( x = self . timestep , y = first_update ) ) )
    else :
        raise TensorForceError ( "Invalid update unit: {}." . format ( unit ) )
    def true_fn ( ) :
        if unit == 'timesteps' : # Timestep - based batch
            batch = self . memory . retrieve_timesteps ( n = batch_size )
        elif unit == 'episodes' : # Episode - based batch
            batch = self . memory . retrieve_episodes ( n = batch_size )
        elif unit == 'sequences' : # Timestep - sequence - based batch
            batch = self . memory . retrieve_sequences ( n = batch_size , sequence_length = sequence_length )
        # Do not calculate gradients for memory - internal operations .
        batch = util . map_tensors ( fn = ( lambda tensor : tf . stop_gradient ( input = tensor ) ) , tensors = batch )
        optimize = self . fn_optimization ( ** batch )
        with tf . control_dependencies ( control_inputs = ( optimize , ) ) :
            return tf . logical_and ( x = True , y = True )
    return tf . cond ( pred = optimize , true_fn = true_fn , false_fn = tf . no_op )


signature:
def tf_import_experience ( self , states , internals , actions , terminal , reward ) :
    """Imports experiences into the TensorFlow memory structure . Can be used to import
 off - policy data .
 : param states : Dict of state values to import with keys as state names and values as values to set .
 : param internals : Internal values to set , can be fetched from agent via agent . current _ internals
 if no values available .
 : param actions : Dict of action values to import with keys as action names and values as values to set .
 : param terminal : Terminal value ( s )
 : param reward : Reward value ( s )"""

implementation: return self . memory . store ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward )


signature:
def import_experience ( self , states , internals , actions , terminal , reward ) :
    """Stores experiences ."""

implementation:
fetches = self . import_experience_output
feed_dict = self . get_feed_dict ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward )
self . monitored_session . run ( fetches = fetches , feed_dict = feed_dict )


signature:
def tf_step ( self , time , variables , arguments , fn_loss , fn_reference , ** kwargs ) :
    """Creates the TensorFlow operations for performing an optimization step .
 Args :
 time : Time tensor .
 variables : List of variables to optimize .
 arguments : Dict of arguments for callables , like fn _ loss .
 fn _ loss : A callable returning the loss of the current model .
 fn _ reference : A callable returning the reference values , in case of a comparative loss .
 * * kwargs : Additional arguments passed on to the internal optimizer .
 Returns :
 List of delta tensors corresponding to the updates for each optimized variable ."""

implementation:
# Set reference to compare with at each optimization step , in case of a comparative loss .
arguments [ 'reference' ] = fn_reference ( ** arguments )
# Negative value since line search maximizes .
loss_before = - fn_loss ( ** arguments )
with tf . control_dependencies ( control_inputs = ( loss_before , ) ) :
    deltas = self . optimizer . step ( time = time , variables = variables , arguments = arguments , fn_loss = fn_loss , return_estimated_improvement = True , ** kwargs )
    if isinstance ( deltas , tuple ) : # If ' return _ estimated _ improvement ' argument exists .
        if len ( deltas ) != 2 :
            raise TensorForceError ( "Unexpected output of internal optimizer." )
        deltas , estimated_improvement = deltas
        # Negative value since line search maximizes .
        estimated_improvement = - estimated_improvement
    else :
        estimated_improvement = None
with tf . control_dependencies ( control_inputs = deltas ) : # Negative value since line search maximizes .
    loss_step = - fn_loss ( ** arguments )
with tf . control_dependencies ( control_inputs = ( loss_step , ) ) :
    def evaluate_step ( deltas ) :
        with tf . control_dependencies ( control_inputs = deltas ) :
            applied = self . apply_step ( variables = variables , deltas = deltas )
        with tf . control_dependencies ( control_inputs = ( applied , ) ) : # Negative value since line search maximizes .
            return - fn_loss ( ** arguments )
    return self . solver . solve ( fn_x = evaluate_step , x_init = deltas , base_value = loss_before , target_value = loss_step , estimated_improvement = estimated_improvement )


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates a distribution from a specification dict ."""

implementation:
distribution = util . get_object ( obj = spec , predefined_objects = tensorforce . core . distributions . distributions , kwargs = kwargs )
assert isinstance ( distribution , Distribution )
return distribution


signature:
def reset ( self ) :
    """Resets the agent to its initial state ( e . g . on experiment start ) . Updates the Model ' s internal episode and
 time step counter , internal states , and resets preprocessors ."""

implementation:
self . episode , self . timestep , self . next_internals = self . model . reset ( )
self . current_internals = self . next_internals


signature:
def act ( self , states , deterministic = False , independent = False , fetch_tensors = None , buffered = True , index = 0 ) :
    """Return action ( s ) for given state ( s ) . States preprocessing and exploration are applied if
 configured accordingly .
 Args :
 states ( any ) : One state ( usually a value tuple ) or dict of states if multiple states are expected .
 deterministic ( bool ) : If true , no exploration and sampling is applied .
 independent ( bool ) : If true , action is not followed by observe ( and hence not included
 in updates ) .
 fetch _ tensors ( list ) : Optional String of named tensors to fetch
 buffered ( bool ) : If true ( default ) , states and internals are not returned but buffered
 with observes . Must be false for multi - threaded mode as we need atomic inserts .
 Returns :
 Scalar value of the action or dict of multiple actions the agent wants to execute .
 ( fetched _ tensors ) Optional dict ( ) with named tensors fetched"""

implementation:
self . current_internals = self . next_internals
if self . unique_state :
    self . current_states = dict ( state = np . asarray ( states ) )
else :
    self . current_states = { name : np . asarray ( states [ name ] ) for name in sorted ( states ) }
if fetch_tensors is not None : # Retrieve action
    self . current_actions , self . next_internals , self . timestep , self . fetched_tensors = self . model . act ( states = self . current_states , internals = self . current_internals , deterministic = deterministic , independent = independent , fetch_tensors = fetch_tensors , index = index )
    if self . unique_action :
        return self . current_actions [ 'action' ] , self . fetched_tensors
    else :
        return self . current_actions , self . fetched_tensors
# Retrieve action .
self . current_actions , self . next_internals , self . timestep = self . model . act ( states = self . current_states , internals = self . current_internals , deterministic = deterministic , independent = independent , index = index )
# Buffered mode only works single - threaded because buffer inserts
# by multiple threads are non - atomic and can cause race conditions .
if buffered :
    if self . unique_action :
        return self . current_actions [ 'action' ]
    else :
        return self . current_actions
else :
    if self . unique_action :
        return self . current_actions [ 'action' ] , self . current_states , self . current_internals
    else :
        return self . current_actions , self . current_states , self . current_internals


signature:
def observe ( self , terminal , reward , index = 0 ) :
    """Observe experience from the environment to learn from . Optionally pre - processes rewards
 Child classes should call super to get the processed reward
 EX : terminal , reward = super ( ) . . .
 Args :
 terminal ( bool ) : boolean indicating if the episode terminated after the observation .
 reward ( float ) : scalar reward that resulted from executing the action ."""

implementation:
self . current_terminal = terminal
self . current_reward = reward
if self . batched_observe : # Batched observe for better performance with Python .
    self . observe_terminal [ index ] . append ( self . current_terminal )
    self . observe_reward [ index ] . append ( self . current_reward )
    if self . current_terminal or len ( self . observe_terminal [ index ] ) >= self . batching_capacity :
        self . episode = self . model . observe ( terminal = self . observe_terminal [ index ] , reward = self . observe_reward [ index ] , index = index )
        self . observe_terminal [ index ] = list ( )
        self . observe_reward [ index ] = list ( )
else :
    self . episode = self . model . observe ( terminal = self . current_terminal , reward = self . current_reward )


signature:
def atomic_observe ( self , states , actions , internals , reward , terminal ) :
    """Utility method for unbuffered observing where each tuple is inserted into TensorFlow via
 a single session call , thus avoiding race conditions in multi - threaded mode .
 Observe full experience tuplefrom the environment to learn from . Optionally pre - processes rewards
 Child classes should call super to get the processed reward
 EX : terminal , reward = super ( ) . . .
 Args :
 states ( any ) : One state ( usually a value tuple ) or dict of states if multiple states are expected .
 actions ( any ) : One action ( usually a value tuple ) or dict of states if multiple actions are expected .
 internals ( any ) : Internal list .
 terminal ( bool ) : boolean indicating if the episode terminated after the observation .
 reward ( float ) : scalar reward that resulted from executing the action ."""

implementation:
# TODO probably unnecessary here .
self . current_terminal = terminal
self . current_reward = reward
# print ( ' action = { } ' . format ( actions ) )
if self . unique_state :
    states = dict ( state = states )
if self . unique_action :
    actions = dict ( action = actions )
self . episode = self . model . atomic_observe ( states = states , actions = actions , internals = internals , terminal = self . current_terminal , reward = self . current_reward )


signature:
def save_model ( self , directory = None , append_timestep = True ) :
    """Save TensorFlow model . If no checkpoint directory is given , the model ' s default saver
 directory is used . Optionally appends current timestep to prevent overwriting previous
 checkpoint files . Turn off to be able to load model from the same given path argument as
 given here .
 Args :
 directory ( str ) : Optional checkpoint directory .
 append _ timestep ( bool ) : Appends the current timestep to the checkpoint file if true .
 If this is set to True , the load path must include the checkpoint timestep suffix .
 For example , if stored to models / and set to true , the exported file will be of the
 form models / model . ckpt - X where X is the last timestep saved . The load path must
 precisely match this file name . If this option is turned off , the checkpoint will
 always overwrite the file specified in path and the model can always be loaded under
 this path .
 Returns :
 Checkpoint path were the model was saved ."""

implementation: return self . model . save ( directory = directory , append_timestep = append_timestep )


signature:
def restore_model ( self , directory = None , file = None ) :
    """Restore TensorFlow model . If no checkpoint file is given , the latest checkpoint is
 restored . If no checkpoint directory is given , the model ' s default saver directory is
 used ( unless file specifies the entire path ) .
 Args :
 directory : Optional checkpoint directory .
 file : Optional checkpoint file , or path if directory not given ."""

implementation: self . model . restore ( directory = directory , file = file )


signature:
def from_spec ( spec , kwargs ) :
    """Creates an agent from a specification dict ."""

implementation:
agent = util . get_object ( obj = spec , predefined_objects = tensorforce . agents . agents , kwargs = kwargs )
assert isinstance ( agent , Agent )
return agent


signature:
def get_named_tensor ( self , name ) :
    """Returns a named tensor if available .
 Returns :
 valid : True if named tensor found , False otherwise
 tensor : If valid , will be a tensor , otherwise None"""

implementation:
if name in self . named_tensors :
    return True , self . named_tensors [ name ]
else :
    return False , None


signature:
def from_spec ( spec , kwargs = None ) :
    """Creates a network from a specification dict ."""

implementation:
network = util . get_object ( obj = spec , default_object = LayeredNetwork , kwargs = kwargs )
assert isinstance ( network , Network )
return network


signature:
def put ( self , item , priority = None ) :
    """Stores a transition in replay memory .
 If the memory is full , the oldest entry is replaced ."""

implementation:
if not self . _isfull ( ) :
    self . _memory . append ( None )
position = self . _next_position_then_increment ( )
old_priority = 0 if self . _memory [ position ] is None else ( self . _memory [ position ] . priority or 0 )
row = _SumRow ( item , priority )
self . _memory [ position ] = row
self . _update_internal_nodes ( position , ( row . priority or 0 ) - old_priority )


signature:
def move ( self , external_index , new_priority ) :
    """Change the priority of a leaf node"""

implementation:
index = external_index + ( self . _capacity - 1 )
return self . _move ( index , new_priority )


signature:
def _move ( self , index , new_priority ) :
    """Change the priority of a leaf node ."""

implementation:
item , old_priority = self . _memory [ index ]
old_priority = old_priority or 0
self . _memory [ index ] = _SumRow ( item , new_priority )
self . _update_internal_nodes ( index , new_priority - old_priority )


signature:
def _update_internal_nodes ( self , index , delta ) :
    """Update internal priority sums when leaf priority has been changed .
 Args :
 index : leaf node index
 delta : change in priority"""

implementation:
# Move up tree , increasing position , updating sum
while index > 0 :
    index = ( index - 1 ) // 2
    self . _memory [ index ] += delta


signature:
def _next_position_then_increment ( self ) :
    """Similar to position + + ."""

implementation:
start = self . _capacity - 1
position = start + self . _position
self . _position = ( self . _position + 1 ) % self . _capacity
return position


signature:
def _sample_with_priority ( self , p ) :
    """Sample random element with priority greater than p ."""

implementation:
parent = 0
while True :
    left = 2 * parent + 1
    if left >= len ( self . _memory ) : # parent points to a leaf node already .
        return parent
    left_p = self . _memory [ left ] if left < self . _capacity - 1 else ( self . _memory [ left ] . priority or 0 )
    if p <= left_p :
        parent = left
    else :
        if left + 1 >= len ( self . _memory ) :
            raise RuntimeError ( 'Right child is expected to exist.' )
        p -= left_p
        parent = left + 1


signature:
def sample_minibatch ( self , batch_size ) :
    """Sample minibatch of size batch _ size ."""

implementation:
pool_size = len ( self )
if pool_size == 0 :
    return [ ]
delta_p = self . _memory [ 0 ] / batch_size
chosen_idx = [ ]
# if all priorities sum to ~ 0 choose randomly otherwise random sample
if abs ( self . _memory [ 0 ] ) < util . epsilon :
    chosen_idx = np . random . randint ( self . _capacity - 1 , self . _capacity - 1 + len ( self ) , size = batch_size ) . tolist ( )
else :
    for i in xrange ( batch_size ) :
        lower = max ( i * delta_p , 0 )
        upper = min ( ( i + 1 ) * delta_p , self . _memory [ 0 ] )
        p = random . uniform ( lower , upper )
        chosen_idx . append ( self . _sample_with_priority ( p ) )
return [ ( i , self . _memory [ i ] ) for i in chosen_idx ]


signature:
def get_batch ( self , batch_size , next_states = False ) :
    """Samples a batch of the specified size according to priority .
 Args :
 batch _ size : The batch size
 next _ states : A boolean flag indicating whether ' next _ states ' values should be included
 Returns : A dict containing states , actions , rewards , terminals , internal states ( and next states )"""

implementation:
if batch_size > len ( self . observations ) :
    raise TensorForceError ( "Requested batch size is larger than observations in memory: increase config.first_update." )
# Init empty states
states = { name : np . zeros ( ( batch_size , ) + tuple ( state [ 'shape' ] ) , dtype = util . np_dtype ( state [ 'type' ] ) ) for name , state in self . states_spec . items ( ) }
internals = [ np . zeros ( ( batch_size , ) + shape , dtype ) for shape , dtype in self . internals_spec ]
actions = { name : np . zeros ( ( batch_size , ) + tuple ( action [ 'shape' ] ) , dtype = util . np_dtype ( action [ 'type' ] ) ) for name , action in self . actions_spec . items ( ) }
terminal = np . zeros ( ( batch_size , ) , dtype = util . np_dtype ( 'bool' ) )
reward = np . zeros ( ( batch_size , ) , dtype = util . np_dtype ( 'float' ) )
if next_states :
    next_states = { name : np . zeros ( ( batch_size , ) + tuple ( state [ 'shape' ] ) , dtype = util . np_dtype ( state [ 'type' ] ) ) for name , state in self . states_spec . items ( ) }
    next_internals = [ np . zeros ( ( batch_size , ) + shape , dtype ) for shape , dtype in self . internals_spec ]
# Start with unseen observations
unseen_indices = list ( xrange ( self . none_priority_index + self . observations . _capacity - 1 , len ( self . observations ) + self . observations . _capacity - 1 ) )
self . batch_indices = unseen_indices [ : batch_size ]
# Get remaining observations using weighted sampling
remaining = batch_size - len ( self . batch_indices )
if remaining :
    samples = self . observations . sample_minibatch ( remaining )
    sample_indices = [ i for i , o in samples ]
    self . batch_indices += sample_indices
# Shuffle
np . random . shuffle ( self . batch_indices )
# Collect observations
for n , index in enumerate ( self . batch_indices ) :
    observation , _ = self . observations . _memory [ index ]
    for name , state in states . items ( ) :
        state [ n ] = observation [ 0 ] [ name ]
    for k , internal in enumerate ( internals ) :
        internal [ n ] = observation [ 1 ] [ k ]
    for name , action in actions . items ( ) :
        action [ n ] = observation [ 2 ] [ name ]
    terminal [ n ] = observation [ 3 ]
    reward [ n ] = observation [ 4 ]
    if next_states :
        for name , next_state in next_states . items ( ) :
            next_state [ n ] = observation [ 5 ] [ name ]
        for k , next_internal in enumerate ( next_internals ) :
            next_internal [ n ] = observation [ 6 ] [ k ]
if next_states :
    return dict ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward , next_states = next_states , next_internals = next_internals )
else :
    return dict ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward )


signature:
def update_batch ( self , loss_per_instance ) :
    """Computes priorities according to loss .
 Args :
 loss _ per _ instance :"""

implementation:
if self . batch_indices is None :
    raise TensorForceError ( "Need to call get_batch before each update_batch call." )
# if len ( loss _ per _ instance ) ! = len ( self . batch _ indices ) :
# raise TensorForceError ( " For all instances a loss value has to be provided . " )
for index , loss in zip ( self . batch_indices , loss_per_instance ) : # Sampling priority is proportional to the largest absolute temporal difference error .
    new_priority = ( np . abs ( loss ) + self . prioritization_constant ) ** self . prioritization_weight
    self . observations . _move ( index , new_priority )
    self . none_priority_index += 1


signature:
def import_experience ( self , experiences ) :
    """Imports experiences .
 Args :
 experiences :"""

implementation:
if isinstance ( experiences , dict ) :
    if self . unique_state :
        experiences [ 'states' ] = dict ( state = experiences [ 'states' ] )
    if self . unique_action :
        experiences [ 'actions' ] = dict ( action = experiences [ 'actions' ] )
    self . model . import_experience ( ** experiences )
else :
    if self . unique_state :
        states = dict ( state = list ( ) )
    else :
        states = { name : list ( ) for name in experiences [ 0 ] [ 'states' ] }
    internals = [ list ( ) for _ in experiences [ 0 ] [ 'internals' ] ]
    if self . unique_action :
        actions = dict ( action = list ( ) )
    else :
        actions = { name : list ( ) for name in experiences [ 0 ] [ 'actions' ] }
    terminal = list ( )
    reward = list ( )
    for experience in experiences :
        if self . unique_state :
            states [ 'state' ] . append ( experience [ 'states' ] )
        else :
            for name in sorted ( states ) :
                states [ name ] . append ( experience [ 'states' ] [ name ] )
        for n , internal in enumerate ( internals ) :
            internal . append ( experience [ 'internals' ] [ n ] )
        if self . unique_action :
            actions [ 'action' ] . append ( experience [ 'actions' ] )
        else :
            for name in sorted ( actions ) :
                actions [ name ] . append ( experience [ 'actions' ] [ name ] )
        terminal . append ( experience [ 'terminal' ] )
        reward . append ( experience [ 'reward' ] )
    self . model . import_experience ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward )


signature:
def connect ( self , timeout = 600 ) :
    """Starts the server tcp connection on the given host : port .
 Args :
 timeout ( int ) : The time ( in seconds ) for which we will attempt a connection to the remote
 ( every 5sec ) . After that ( or if timeout is None or 0 ) , an error is raised ."""

implementation:
# If we are already connected , return error .
if self . socket :
    raise TensorForceError ( "Already connected to {}:{}. Only one connection allowed at a time. " + "Close first by calling `close`!" . format ( self . host , self . port ) )
self . socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM )
if timeout < 5 or timeout is None :
    timeout = 5
err = 0
start_time = time . time ( )
while time . time ( ) - start_time < timeout :
    self . socket . settimeout ( 5 )
    err = self . socket . connect_ex ( ( self . host , self . port ) )
    if err == 0 :
        break
    time . sleep ( 1 )
if err != 0 :
    raise TensorForceError ( "Error when trying to connect to {}:{}: errno={} errcode='{}' '{}'" . format ( self . host , self . port , err , errno . errorcode [ err ] , os . strerror ( err ) ) )


signature:
def disconnect ( self ) :
    """Ends our server tcp connection ."""

implementation:
# If we are not connected , return error .
if not self . socket :
    logging . warning ( "No active socket to close!" )
    return
# Close our socket .
self . socket . close ( )
self . socket = None


signature:
def send ( self , message , socket_ ) :
    """Sends a message ( dict ) to the socket . Message consists of a 8 - byte len header followed by a msgpack - numpy
 encoded dict .
 Args :
 message : The message dict ( e . g . { " cmd " : " reset " } )
 socket _ : The python socket object to use ."""

implementation:
if not socket_ :
    raise TensorForceError ( "No socket given in call to `send`!" )
elif not isinstance ( message , dict ) :
    raise TensorForceError ( "Message to be sent must be a dict!" )
message = msgpack . packb ( message )
len_ = len ( message )
# prepend 8 - byte len field to all our messages
socket_ . send ( bytes ( "{:08d}" . format ( len_ ) , encoding = "ascii" ) + message )


signature:
def recv ( self , socket_ , encoding = None ) :
    """Receives a message as msgpack - numpy encoded byte - string from the given socket object .
 Blocks until something was received .
 Args :
 socket _ : The python socket object to use .
 encoding ( str ) : The encoding to use for unpacking messages from the socket .
 Returns : The decoded ( as dict ) message received ."""

implementation:
unpacker = msgpack . Unpacker ( encoding = encoding )
# Wait for an immediate response .
response = socket_ . recv ( 8 )
# get the length of the message
if response == b"" :
    raise TensorForceError ( "No data received by socket.recv in call to method `recv` " + "(listener possibly closed)!" )
orig_len = int ( response )
received_len = 0
while True :
    data = socket_ . recv ( min ( orig_len - received_len , self . max_msg_len ) )
    # There must be a response .
    if not data :
        raise TensorForceError ( "No data of len {} received by socket.recv in call to method `recv`!" . format ( orig_len - received_len ) )
    data_len = len ( data )
    received_len += data_len
    unpacker . feed ( data )
    if received_len == orig_len :
        break
# Get the data .
for message in unpacker :
    sts = message . get ( "status" , message . get ( b"status" ) )
    if sts :
        if sts == "ok" or sts == b"ok" :
            return message
        else :
            raise TensorForceError ( "RemoteEnvironment server error: {}" . format ( message . get ( "message" , "not specified" ) ) )
    else :
        raise TensorForceError ( "Message without field 'status' received!" )
raise TensorForceError ( "No message encoded in data stream (data stream had len={})" . format ( orig_len ) )


signature:
def is_action_available ( self , action ) :
    """Determines whether action is available .
 That is , executing it would change the state ."""

implementation:
temp_state = np . rot90 ( self . _state , action )
return self . _is_action_available_left ( temp_state )


signature:
def _is_action_available_left ( self , state ) :
    """Determines whether action ' Left ' is available ."""

implementation:
# True if any field is 0 ( empty ) on the left of a tile or two tiles can
# be merged .
for row in range ( 4 ) :
    has_empty = False
    for col in range ( 4 ) :
        has_empty |= state [ row , col ] == 0
        if state [ row , col ] != 0 and has_empty :
            return True
        if ( state [ row , col ] != 0 and col > 0 and state [ row , col ] == state [ row , col - 1 ] ) :
            return True
return False


signature:
def do_action ( self , action ) :
    """Execute action , add a new tile , update the score & return the reward ."""

implementation:
temp_state = np . rot90 ( self . _state , action )
reward = self . _do_action_left ( temp_state )
self . _state = np . rot90 ( temp_state , - action )
self . _score += reward
self . add_random_tile ( )
return reward


signature:
def _do_action_left ( self , state ) :
    """Executes action ' Left ' ."""

implementation:
reward = 0
for row in range ( 4 ) : # Always the rightmost tile in the current row that was already moved
    merge_candidate = - 1
    merged = np . zeros ( ( 4 , ) , dtype = np . bool )
    for col in range ( 4 ) :
        if state [ row , col ] == 0 :
            continue
        if ( merge_candidate != - 1 and not merged [ merge_candidate ] and state [ row , merge_candidate ] == state [ row , col ] ) : # Merge tile with merge _ candidate
            state [ row , col ] = 0
            merged [ merge_candidate ] = True
            state [ row , merge_candidate ] += 1
            reward += 2 ** state [ row , merge_candidate ]
        else : # Move tile to the left
            merge_candidate += 1
            if col != merge_candidate :
                state [ row , merge_candidate ] = state [ row , col ]
                state [ row , col ] = 0
return reward


signature:
def add_random_tile ( self ) :
    """Adds a random tile to the grid . Assumes that it has empty fields ."""

implementation:
x_pos , y_pos = np . where ( self . _state == 0 )
assert len ( x_pos ) != 0
empty_index = np . random . choice ( len ( x_pos ) )
value = np . random . choice ( [ 1 , 2 ] , p = [ 0.9 , 0.1 ] )
self . _state [ x_pos [ empty_index ] , y_pos [ empty_index ] ] = value


signature:
def setup ( self ) :
    """Sets up the TensorFlow model graph , starts the servers ( distributed mode ) , creates summarizers
 and savers , initializes ( and enters ) the TensorFlow session ."""

implementation:
# Create / get our graph , setup local model / global model links , set scope and device .
graph_default_context = self . setup_graph ( )
# Start a tf Server ( in case of distributed setup ) . Only start once .
if self . execution_type == "distributed" and self . server is None and self . is_local_model :
    self . start_server ( )
# build the graph
with tf . device ( device_name_or_function = self . device ) :
    with tf . variable_scope ( name_or_scope = self . scope , reuse = False ) : # Variables and summaries
        self . variables = dict ( )
        self . all_variables = dict ( )
        self . registered_variables = set ( )
        # Build the graph ' s placeholders , tf _ functions , etc
        self . setup_placeholders ( )
        # Create model ' s " external " components .
        # Create tensorflow functions from " tf _ " - methods .
        self . setup_components_and_tf_funcs ( )
        # Create core variables ( timestep , episode counters , buffers for states / actions / internals ) .
        self . fn_initialize ( )
        if self . summarizer_spec is not None :
            with tf . name_scope ( name = 'summarizer' ) :
                self . summarizer = tf . contrib . summary . create_file_writer ( logdir = self . summarizer_spec [ 'directory' ] , max_queue = None , flush_millis = ( self . summarizer_spec . get ( 'flush' , 10 ) * 1000 ) , filename_suffix = None , name = None )
                default_summarizer = self . summarizer . as_default ( )
                # Problem : not all parts of the graph are called on every step
                assert 'steps' not in self . summarizer_spec
                # if ' steps ' in self . summarizer _ spec :
                # record _ summaries = tf . contrib . summary . record _ summaries _ every _ n _ global _ steps (
                # n = self . summarizer _ spec [ ' steps ' ] ,
                # global _ step = self . global _ timestep
                # else :
                record_summaries = tf . contrib . summary . always_record_summaries ( )
            default_summarizer . __enter__ ( )
            record_summaries . __enter__ ( )
        # Input tensors
        states = util . map_tensors ( fn = tf . identity , tensors = self . states_input )
        internals = util . map_tensors ( fn = tf . identity , tensors = self . internals_input )
        actions = util . map_tensors ( fn = tf . identity , tensors = self . actions_input )
        terminal = tf . identity ( input = self . terminal_input )
        reward = tf . identity ( input = self . reward_input )
        # Probably both deterministic and independent should be the same at some point .
        deterministic = tf . identity ( input = self . deterministic_input )
        independent = tf . identity ( input = self . independent_input )
        episode_index = tf . identity ( input = self . episode_index_input )
        states , actions , reward = self . fn_preprocess ( states = states , actions = actions , reward = reward )
        self . create_operations ( states = states , internals = internals , actions = actions , terminal = terminal , reward = reward , deterministic = deterministic , independent = independent , index = episode_index )
        # Add all summaries specified in summary _ labels
        if 'inputs' in self . summary_labels or 'states' in self . summary_labels :
            for name in sorted ( states ) :
                tf . contrib . summary . histogram ( name = ( 'states-' + name ) , tensor = states [ name ] )
        if 'inputs' in self . summary_labels or 'actions' in self . summary_labels :
            for name in sorted ( actions ) :
                tf . contrib . summary . histogram ( name = ( 'actions-' + name ) , tensor = actions [ name ] )
        if 'inputs' in self . summary_labels or 'reward' in self . summary_labels :
            tf . contrib . summary . histogram ( name = 'reward' , tensor = reward )
        if 'graph' in self . summary_labels :
            with tf . name_scope ( name = 'summarizer' ) :
                graph_def = self . graph . as_graph_def ( )
                graph_str = tf . constant ( value = graph_def . SerializeToString ( ) , dtype = tf . string , shape = ( ) )
                self . graph_summary = tf . contrib . summary . graph ( param = graph_str , step = self . global_timestep )
                if 'meta_param_recorder_class' in self . summarizer_spec :
                    self . graph_summary = tf . group ( self . graph_summary , * self . summarizer_spec [ 'meta_param_recorder_class' ] . build_metagraph_list ( ) )
        if self . summarizer_spec is not None :
            record_summaries . __exit__ ( None , None , None )
            default_summarizer . __exit__ ( None , None , None )
            with tf . name_scope ( name = 'summarizer' ) :
                self . flush_summarizer = tf . contrib . summary . flush ( )
                self . summarizer_init_op = tf . contrib . summary . summary_writer_initializer_op ( )
                assert len ( self . summarizer_init_op ) == 1
                self . summarizer_init_op = self . summarizer_init_op [ 0 ]
# If we are a global model - > return here .
# Saving , syncing , finalizing graph , session is done by local replica model .
if self . execution_type == "distributed" and not self . is_local_model :
    return
# Saver / Summary - > Scaffold .
self . setup_saver ( )
self . setup_scaffold ( )
# Create necessary hooks for the upcoming session .
hooks = self . setup_hooks ( )
# We are done constructing : Finalize our graph , create and enter the session .
self . setup_session ( self . server , hooks , graph_default_context )


signature:
def setup_graph ( self ) :
    """Creates our Graph and figures out , which shared / global model to hook up to .
 If we are in a global - model ' s setup procedure , we do not create
 a new graph ( return None as the context ) . We will instead use the already existing local replica graph
 of the model .
 Returns : None or the graph ' s as _ default ( ) - context ."""

implementation:
graph_default_context = None
# Single ( non - distributed ) mode .
if self . execution_type == "single" :
    self . graph = tf . Graph ( )
    graph_default_context = self . graph . as_default ( )
    graph_default_context . __enter__ ( )
    self . global_model = None
# Distributed tf
elif self . execution_type == "distributed" : # Parameter - server - > Do not build any graph .
    if self . distributed_spec [ "job" ] == "ps" :
        return None
    # worker - > construct the global ( main ) model ; the one hosted on the ps ,
    elif self . distributed_spec [ "job" ] == "worker" : # The local replica model .
        if self . is_local_model :
            graph = tf . Graph ( )
            graph_default_context = graph . as_default ( )
            graph_default_context . __enter__ ( )
            # Now that the graph is created and entered - > deepcopoy ourselves and setup global model first ,
            # then continue .
            self . global_model = deepcopy ( self )
            # Switch on global construction / setup - mode for the pass to setup ( ) .
            self . global_model . is_local_model = False
            self . global_model . setup ( )
            self . graph = graph
            self . as_local_model ( )
            self . scope += '-worker' + str ( self . distributed_spec [ "task_index" ] )
        # The global _ model ( whose Variables are hosted by the ps ) .
        else :
            self . graph = tf . get_default_graph ( )
            # lives in the same graph as local model
            self . global_model = None
            self . device = tf . train . replica_device_setter ( # Place its Variables on the parameter server ( s ) ( round robin ) .
            # ps _ device = " / job : ps " , # default
            # Train - ops for the global _ model are hosted locally ( on this worker ' s node ) .
            worker_device = self . device , cluster = self . distributed_spec [ "cluster_spec" ] )
    else :
        raise TensorForceError ( "Unsupported job type: {}!" . format ( self . distributed_spec [ "job" ] ) )
else :
    raise TensorForceError ( "Unsupported distributed type: {}!" . format ( self . distributed_spec [ "type" ] ) )
return graph_default_context


signature:
def start_server ( self ) :
    """Creates and stores a tf server ( and optionally joins it if we are a parameter - server ) .
 Only relevant , if we are running in distributed mode ."""

implementation:
self . server = tf . train . Server ( server_or_cluster_def = self . distributed_spec [ "cluster_spec" ] , job_name = self . distributed_spec [ "job" ] , task_index = self . distributed_spec [ "task_index" ] , protocol = self . distributed_spec . get ( "protocol" ) , config = self . distributed_spec . get ( "session_config" ) , start = True )
if self . distributed_spec [ "job" ] == "ps" :
    self . server . join ( )
    # This is unreachable ?
    quit ( )