rddl2tf package¶

Submodules¶

rddl2tf.compilers.compiler module¶

class rddl2tf.compiler.Compiler(rddl: pyrddl.rddl.RDDL, batch_mode: bool = False)¶

Bases: object

RDDL2TensorFlow compiler.

This is the core component of rddl2tf package.

Its API provides methods to compile RDDL fluents and expressions to TensorFlow tensors wrapped as rddl2tf.fluent.TensorFluent objects. It supports constants, random variables, functions and operators used in most RDDL expressions. Also, it can handle next state and intermediate fluent CPFs, and rewards and action constraints.

Parameters:	rddl (`pyrddl.rddl.RDDL`) – The RDDL model. batch_mode (bool) – The batch mode flag.

rddl¶

The RDDL model.

Type:	`pyrddl.rddl.RDDL`

batch_mode¶

The batch mode flag.

Type:	bool

graph¶

The computation graph.

Type:	`tensorflow.python.framework.ops.Graph`

_compile_aggregation_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile an aggregation expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL aggregation expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_arithmetic_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile an arithmetic expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL arithmetic expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_batch_fluents(fluents: List[Tuple[str, rddl2tf.fluent.TensorFluent]], batch_size: int) → Sequence[tensorflow.python.framework.ops.Tensor]¶

Compiles fluents into tensors with given batch_size.

Returns:	A tuple of tensors with first dimension corresponding to the batch size.
Return type:	Sequence[tf.Tensor]

_compile_boolean_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a boolean/logical expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL boolean expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_constant_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a constant expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL constant expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_control_flow_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a control flow expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL control flow expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile the expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_function_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a function expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL function expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_pvariable_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a pvariable expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL pvariable expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_random_variable_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a random variable expression expr into a TensorFluent in the given scope with optional batch size.

If reparam tensor is given, then it conditionally stops gradient backpropagation at the batch level where reparam is False.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL random variable expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_compile_relational_expression(expr: pyrddl.expr.Expression, scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[List[tensorflow.python.framework.ops.Tensor]] = None) → rddl2tf.fluent.TensorFluent¶

Compile a relational expression expr into a TensorFluent in the given scope with optional batch size.

Parameters:	expr (`rddl2tf.expr.Expression`) – A RDDL relational expression. scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – A fluent scope. batch_size (Optional[size]) – The batch size.
Returns:	The compiled expression as a TensorFluent.
Return type:	`rddl2tf.fluent.TensorFluent`

_initialize_default_action_fluents()¶: Returns the default action-fluents instantiated.

_initialize_initial_state_fluents()¶: Returns the initial state-fluents instantiated.

_initialize_non_fluents()¶: Returns the non-fluents instantiated.

_initialize_pvariables(pvariables: Dict[str, pyrddl.pvariable.PVariable], ordering: List[str], initializer: Optional[List[Tuple[Tuple[str, Optional[List[str]]], Union[int, float]]]] = None) → List[Tuple[str, rddl2tf.fluent.TensorFluent]]¶

Instantiates pvariables given an initialization list and returns a list of TensorFluents in the given ordering.

Returns:	A list of pairs of fluent name and fluent tensor.
Return type:	List[Tuple[str, TensorFluent]]

action_precondition_scope(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Optional[Sequence[tensorflow.python.framework.ops.Tensor]] = None) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns the action precondition fluent scope for the current state and action fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents.
Returns:	A mapping from fluent names to `rddl2tf.fluent.TensorFluent`.

action_scope(action_fluents: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns a partial scope with current action-fluents.

Parameters:	action_fluents (Sequence[tf.Tensor]) – The action fluents.
Returns:	A mapping from action fluent names to `rddl2tf.fluent.TensorFluent`.

batch_mode_off()¶: Sets off the batch mode flag.

batch_mode_on()¶: Sets on the batch mode flag.

compile_action_bound_constraints(state: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, Tuple[Optional[rddl2tf.fluent.TensorFluent], Optional[rddl2tf.fluent.TensorFluent]]]¶

Compiles all actions bounds for the given state.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents.
Returns:	A mapping from action names to a pair of `rddl2tf.fluent.TensorFluent` representing its lower and upper bounds.

compile_action_preconditions(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor]) → List[rddl2tf.fluent.TensorFluent]¶

Compiles the action preconditions given current state and action fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents.
Returns:	A list of `rddl2tf.fluent.TensorFluent`.

compile_action_preconditions_checking(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor]) → tensorflow.python.framework.ops.Tensor¶

Combines the action preconditions into an applicability checking op.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents.
Returns:	A boolean tensor for checking if action is application in state.

compile_cpfs(scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[Dict[str, Sequence[tensorflow.python.framework.ops.Tensor]]] = None) → Tuple[List[Tuple[str, rddl2tf.fluent.TensorFluent]], List[Tuple[str, rddl2tf.fluent.TensorFluent]]]¶

Compiles the intermediate and next state fluent CPFs given the current state and action scope.

Parameters:	scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – The fluent scope for CPF evaluation. batch_size (Optional[int]) – The batch size.
Returns:	A pair of lists of TensorFluent representing the intermediate and state CPFs.
Return type:	Tuple[List[CPFPair], List[CPFPair]]

compile_default_action(batch_size: Optional[int] = None) → Sequence[tensorflow.python.framework.ops.Tensor]¶

Returns a tuple of tensors representing the default action fluents.

Parameters:	batch_size (int) – The batch size.
Returns:	A tuple of tensors.
Return type:	Sequence[tf.Tensor]

compile_initial_state(batch_size: Optional[int] = None) → Sequence[tensorflow.python.framework.ops.Tensor]¶

Returns a tuple of tensors representing the initial state fluents.

Parameters:	batch_size (Optional[int]) – The batch size.
Returns:	A tuple of tensors.
Return type:	Sequence[tf.Tensor]

compile_intermediate_cpfs(scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[Dict[str, Sequence[tensorflow.python.framework.ops.Tensor]]] = None) → List[Tuple[str, rddl2tf.fluent.TensorFluent]]¶

Compiles the intermediate fluent CPFs given the current state and action scope.

Parameters:	scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – The fluent scope for CPF evaluation. batch_size (Optional[int]) – The batch size.
Returns:	A list of intermediate fluent CPFs compiled to `rddl2tf.fluent.TensorFluent`.

compile_non_fluents()¶

Returns a tuple of tensors representing the non fluents.

Returns:	A tuple of tensors.
Return type:	Sequence[tf.Tensor]

compile_reward(scope: Dict[str, rddl2tf.fluent.TensorFluent]) → rddl2tf.fluent.TensorFluent¶

Compiles the reward function given the fluent scope.

Parameters:	scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – The fluent scope for reward evaluation.
Returns:	A `rddl2tf.fluent.TensorFluent` representing the reward function.

compile_state_action_constraints(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor]) → List[rddl2tf.fluent.TensorFluent]¶

Compiles the state-action constraints given current state and action fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents.
Returns:	A list of `rddl2tf.fluent.TensorFluent`.

compile_state_cpfs(scope: Dict[str, rddl2tf.fluent.TensorFluent], batch_size: Optional[int] = None, noise: Optional[Dict[str, Sequence[tensorflow.python.framework.ops.Tensor]]] = None) → List[Tuple[str, rddl2tf.fluent.TensorFluent]]¶

Compiles the next state fluent CPFs given the current state and action scope.

Parameters:	scope (Dict[str, `rddl2tf.fluent.TensorFluent`]) – The fluent scope for CPF evaluation. batch_size (Optional[int]) – The batch size.
Returns:	A list of state fluent CPFs compiled to `rddl2tf.fluent.TensorFluent`.

compile_state_invariants(state: Sequence[tensorflow.python.framework.ops.Tensor]) → List[rddl2tf.fluent.TensorFluent]¶

Compiles the state invarints given current state fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents.
Returns:	A list of `rddl2tf.fluent.TensorFluent`.

cpfs(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor], noise: Optional[Dict[str, Sequence[tensorflow.python.framework.ops.Tensor]]] = None) → Tuple[List[rddl2tf.fluent.TensorFluent], List[rddl2tf.fluent.TensorFluent]]¶

Compiles the intermediate and next state fluent CPFs given the current state and action.

Parameters:	state (Sequence[tf.Tensor]) – A tuple of state tensors. action (Sequence[tf.Tensor]) – A tuple of action tensors.
Returns:	A pair of lists of TensorFluent representing the intermediate and state CPFs.
Return type:	Tuple[List[TensorFluent], List[TensorFluent]]

next_state_scope(next_state_fluents: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns a partial scope with current next state-fluents.

Parameters:	next_state_fluents (Sequence[tf.Tensor]) – The next state fluents.
Returns:	A mapping from next state fluent names to `rddl2tf.fluent.TensorFluent`.

non_fluents_scope() → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns a partial scope with non-fluents.

Returns:	A mapping from non-fluent names to `rddl2tf.fluent.TensorFluent`.

reward(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor], next_state: Sequence[tensorflow.python.framework.ops.Tensor]) → tensorflow.python.framework.ops.Tensor¶

Compiles the reward function given the current state, action and next_state.

Parameters:	state (Sequence[tf.Tensor]) – A tuple of current state tensors. action (Sequence[tf.Tensor]) – A tuple of action tensors. next_state (Sequence[tf.Tensor]) – A tuple of next state tensors.
Returns:	A tensor representing the reward function.
Return type:	(`tf.Tensor`)

reward_scope(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor], next_state: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns the complete reward fluent scope for the current state, action fluents, and next_state fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents. next_state (Sequence[tf.Tensor]) – The next state fluents.
Returns:	A mapping from fluent names to `rddl2tf.fluent.TensorFluent`.

state_invariant_scope(state: Sequence[tensorflow.python.framework.ops.Tensor])¶

Returns the state invariant fluent scope for the current state.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents.
Returns:	A mapping from fluent names to `rddl2tf.fluent.TensorFluent`.

state_scope(state_fluents: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns a partial scope with current state-fluents.

Parameters:	state_fluents (Sequence[tf.Tensor]) – The current state fluents.
Returns:	A mapping from state fluent names to `rddl2tf.fluent.TensorFluent`.

transition_scope(state: Sequence[tensorflow.python.framework.ops.Tensor], action: Sequence[tensorflow.python.framework.ops.Tensor]) → Dict[str, rddl2tf.fluent.TensorFluent]¶

Returns the complete transition fluent scope for the current state and action fluents.

Parameters:	state (Sequence[tf.Tensor]) – The current state fluents. action (Sequence[tf.Tensor]) – The action fluents.
Returns:	A mapping from fluent names to `rddl2tf.fluent.TensorFluent`.

rddl2tf.core.fluent module¶

rddl2tf.core.fluentscope module¶

rddl2tf.core.fluentshape module¶

rddl2tf.utils module¶

rddl2tf.utils.identifier(name)¶

rddl2tf.utils.python_type_to_dtype(python_type: type) → Optional[tensorflow.python.framework.dtypes.DType]¶: Maps python types to TensorFlow dtypes.

rddl2tf.utils.range_type_to_dtype(range_type: str) → Optional[tensorflow.python.framework.dtypes.DType]¶: Maps RDDL range types to TensorFlow dtypes.