cleave.core.client package

Submodules

cleave.core.client.actuator module

exception cleave.core.client.actuator.RegisteredActuatorWarning

Bases: Warning

exception cleave.core.client.actuator.UnregisteredPropertyWarning

Bases: Warning

class cleave.core.client.actuator.ActuatorArray(actuators: Collection[Actuator], control: BaseControllerInterface)

Bases: Recordable

Internal utility class to manage a collection of Actuators attached to a Plant.

get_actuation_inputs() Mapping[str, Union[int, float, bool]]

Fetches raw commands from the controller, processes them and returns.

Returns

A mapping from actuated property names to output values from the corresponding actuators.

Return type

PhyPropMapping

cleave.core.client.control module

class cleave.core.client.control.BaseControllerInterface

Bases: ABC

Defines the core interface for interacting with controllers.

abstract put_sensor_values(prop_values: Mapping[str, Union[int, float, bool]]) None

Send a sample of sensor values to the controller.

Parameters

prop_values – Mapping from property names to sensor values.

abstract get_actuator_values() Mapping[str, Union[int, float, bool]]

Waits for incoming data from the controller and returns a mapping from actuated property names to values.

Returns

Mapping from actuated property names to values.

Return type

Mapping

class cleave.core.client.control.DummyControllerInterface

Bases: BaseControllerInterface

put_sensor_values(prop_values: Mapping[str, Union[int, float, bool]]) None

Send a sample of sensor values to the controller.

Parameters

prop_values – Mapping from property names to sensor values.

get_actuator_values() Mapping[str, Union[int, float, bool]]

Waits for incoming data from the controller and returns a mapping from actuated property names to values.

Returns

Mapping from actuated property names to values.

Return type

Mapping

cleave.core.client.physicalsim module

class cleave.core.client.physicalsim.PhysicalSimulation(state: State, tick_rate: int)

Bases: Recordable

advance_state(input_values: Mapping[str, Union[int, float, bool]]) Mapping[str, Union[int, float, bool]]

Performs a single step update using the given actuation values as inputs and returns the updated values for the sensed variables.

Parameters

input_values – Actuation inputs.

Returns

Mapping from sensed property names to values.

Return type

PhyPropMapping

cleave.core.client.plant module

class cleave.core.client.plant.Plant(tick_dt: float)

Bases: ABC

Interface for all plants.

class cleave.core.client.plant.BasePlant(physim: PhysicalSimulation, sensors: Collection[Sensor], actuators: Collection[Actuator], control_interface: BaseControllerInterface)

Bases: Plant

on_init() None

Sets up the simulation of this plant

tick(missed_count: int) None

Executes the emulation timestep. Intended use is inside a Twisted LoopingCall, hence why it takes a single integer parameter which specifies the number of calls queued up in a time interval (should be 1).

Parameters

missed_count

on_shutdown() None

Called on shutdown of the plant.

class cleave.core.client.plant.CSVRecordingPlant(physim: PhysicalSimulation, sensors: Collection[Sensor], actuators: Collection[Actuator], control_interface: BaseControllerInterface, recording_output_dir: Path = PosixPath('.'))

Bases: BasePlant

Plant with built-in CSV recording capabilities of metrics from the the physical properties and the network connection.

on_init() None

Sets up the simulation of this plant

on_shutdown() None

Called on shutdown of the plant.

cleave.core.client.sensor module

class cleave.core.client.sensor.SensorArray(plant_tick_rate: int, sensors: Collection[Sensor], control: BaseControllerInterface)

Bases: Recordable

Internal utility class to manage a collection of Sensors attached to a Plant.

process_and_send_samples(prop_values: Mapping[str, Union[int, float, bool]]) None

Processes measured properties by passing them to the internal collection of sensors and returns the processed values.

Parameters

prop_values – Dictionary containing mappings from property names to measured values.

Returns

A dictionary containing mappings from property names to processed sensor values.

Return type

Dict

exception cleave.core.client.sensor.IncompatibleFrequenciesError

Bases: Exception

exception cleave.core.client.sensor.MissingPropertyError

Bases: Exception

cleave.core.client.statebase module

class cleave.core.client.statebase.BaseSemanticVariable(value: Any, record: bool = True, sanity_check: Optional[Callable[[Any], bool]] = None)

Bases: SupportsFloat, SupportsInt, SupportsBytes, ABC

Base class for semantically significant variables in a State.

Parameters
  • value – The initial value for this variable.

  • record – Whether this variable should be recorded or not (WIP) TODO: record variables

  • sanity_check – A callable which takes a single input value and returns a boolean. If this value is not None, the callable give will be invoked with the updated value of the semantic variable after every tick of the plant. If the callable returns false, the simulation will fail with an UnrecoverableState error.

class cleave.core.client.statebase.StateBase

Bases: ABC

Defines the underlying structure for the State interface without having to resort to hacks.

final get_sensed_prop_names() Set[str]
Returns

Set containing the identifiers of the sensed variables.

Return type

Set

final get_actuated_prop_names() Set[str]
Returns

Set containing the identifiers of the actuated variables.

Return type

Set

final get_record_variables() Mapping[str, Union[int, float, bool]]
Returns

A mapping containing the values of the recorded variables in this state.

Return type

PhyPropMapping

final get_controller_parameters() Mapping[str, Union[int, float, bool]]
Returns

A mapping from strings to values containing the initialization parameters for the controller associated with this physical simulation.

Return type

PhyPropMapping

final check_semantic_sanity() Dict[str, Any]

Checks that the semantic variables in this state have acceptable values.

Returns

Returns a mapping containing the properties for which the check failed and their associated values.

Return type

Dict[str, Any]

cleave.core.client.timing module

class cleave.core.client.timing.TimingResult(start, end, duration, results)

Bases: tuple

Create new instance of TimingResult(start, end, duration, results)

start: float

Alias for field number 0

end: float

Alias for field number 1

duration: float

Alias for field number 2

results: Any

Alias for field number 3

class cleave.core.client.timing.SimClock

Bases: object

This class provides consistent simulation timing measurements.

get_sim_time() float
Returns

A float representing the elapsed time in seconds since the initialization of this clock.

Return type

float

get_adjusted_realtime() float
Returns

A float representing the elapsed time since the UNIX Epoch, adjusted for monotonicity.

Return type

float

time_subroutine(fn: Callable[[Any], Any], *args, **kwargs) TimingResult

Times the execution of a function with respect to this clock.

Parameters
  • fn – Function to be timed.

  • args – Args to pass to the timed function.

  • kwargs – Kwargs to pass to the timed function.

Returns

A TimingResult object containing the start and end timestamps of the function call, the duration of the function call in seconds and the results of the call.

Return type

TimingResult

class cleave.core.client.timing.Rate(tick_count, interval_s)

Bases: tuple

Create new instance of Rate(tick_count, interval_s)

tick_count: int

Alias for field number 0

interval_s: float

Alias for field number 1

class cleave.core.client.timing.SimTicker

Bases: object

Utility class to measure the rate of the plant and provide time deltas between ticks.