semantic_digital_twin.input_synchronization

Classes

InputSynchronizer	Writes an external source of truth, e.g. a robot's joint states, into the world
WorldStateInputs	All inputs that one loop applies before it computes anything.

Module Contents

class semantic_digital_twin.input_synchronization.InputSynchronizer

Bases: abc.ABC

Writes an external source of truth, e.g. a robot’s joint states, into the world state.

world: semantic_digital_twin.world.World

The world whose state is kept in sync with the external source.

abstract apply() → bool

Write the most recent input into the world state.

Returns:

Whether anything was written.

rewriting_every_cycle() → InputSynchronizer

Returns:

An input that writes what it last read in every cycle, for a loop that moves the world state away from it between cycles.

Defaults to this input, which already writes whatever it holds whenever it is applied.

close() → None

Release the resources used to receive inputs.

Does nothing by default, as an input that receives nothing holds none.

class semantic_digital_twin.input_synchronization.WorldStateInputs

All inputs that one loop applies before it computes anything.

world: semantic_digital_twin.world.World

The world whose state is written and whose observers are notified.

synchronizers: list[InputSynchronizer] = []

The inputs, applied in the order they were added.

reapplies_inputs: bool = False

Whether this loop needs its inputs written again in every cycle, as a loop does that moves the world state away from what it last read.

added_inputs: list[InputSynchronizer] = []

The inputs added to this loop, as they were handed to it rather than as they are applied.

add_robot_inputs(robot: semantic_digital_twin.robots.robot_parts.AbstractRobot) → None

Apply everything the parts of a robot are read from in this loop from now on.

Parts that are read from the world they stand in need nothing applied and are therefore left out.

Parameters:

robot – The robot whose inputs are added.

add_robot_part_input(robot_part: semantic_digital_twin.robots.robot_part_mixins.HasInputSource) → None

Apply what one part of a robot is read from in this loop from now on.

A part that is read from the world it stands in needs nothing applied and is therefore left out.

Parameters:

robot_part – The part whose input is added.

add_input(synchronizer: InputSynchronizer) → None

Apply the given input in this loop from now on, in the way this loop needs it.

Parameters:

synchronizer – The input to add.

Raises:

InputAlreadyAddedError – If the input was already added to this loop.

apply_inputs() → bool

Write all inputs into the world state, in the order they were added.

Returns:

Whether any of them wrote something.

synchronize() → None

Write all inputs into the world state and announce the change.

Nothing is announced when no input wrote, because announcing recomputes the forward kinematics and reaches every observer of the world.

synchronize_and_announce() → None

Write all inputs into the world state and announce the state even if no input wrote.

Use this where nothing else announces, so that the observers of the world do not go stale.

announce_state() → None

Hand the current world state to the observers of the world.

Nothing is announced while the world model is being modified, because the observers would see an inconsistent model.