server.plugins.manual_move_plugin

manual_move_plugin.py - ECS variant of the ManualMovePlugin

 1"""manual_move_plugin.py - ECS variant of the ManualMovePlugin"""
 2from overrides import overrides
 3from microecs import World
 4from robosim.plugin import Plugin, Event, Response, Message
 5from microspec import Endpoint, Field
 6
 7class ManualMovePlugin(Plugin):
 8    @property
 9    @overrides
10    def endpoints(self) -> list[Endpoint]:
11        return [
12            Endpoint("move",
13                     input={"control_input": Field("float32", shape=(6, ), range=[-1, 1])},
14                     output={"move_applied": Field("dict")}
15            )
16        ]
17
18    @overrides
19    def on_message_receive(self, world: World, entity_id: int, message: Message) -> Response:
20        # all syntax validation happens earlier thanks to microspec and self.endpoints schema
21        return Response.physics(motion_input=message.payload["control_input"])
22
23    @overrides
24    def on_message_response(self, world: World, event: Event) -> Response:
25        if not event.applied:
26            return Response.err("move_not_applied")
27
28        entity = world.get_entity(event.target)
29        if entity.is_colliding.item():
30            robot_ix = event.message.channel.index
31            return Response.err({"collision_error": f"robot {robot_ix} collided (eid: {event.target})"})
32
33        return Response.ok("move_applied", new_state=entity.to_dict(serialization_field="serializable"))
34
35    @overrides
36    def to_dict(self) -> dict:
37        return {}
38
39    @overrides
40    def load_state_dict(self, state: dict):
41        pass
class ManualMovePlugin(robosim.plugin.Plugin): 
 8class ManualMovePlugin(Plugin):
 9    @property
10    @overrides
11    def endpoints(self) -> list[Endpoint]:
12        return [
13            Endpoint("move",
14                     input={"control_input": Field("float32", shape=(6, ), range=[-1, 1])},
15                     output={"move_applied": Field("dict")}
16            )
17        ]
18
19    @overrides
20    def on_message_receive(self, world: World, entity_id: int, message: Message) -> Response:
21        # all syntax validation happens earlier thanks to microspec and self.endpoints schema
22        return Response.physics(motion_input=message.payload["control_input"])
23
24    @overrides
25    def on_message_response(self, world: World, event: Event) -> Response:
26        if not event.applied:
27            return Response.err("move_not_applied")
28
29        entity = world.get_entity(event.target)
30        if entity.is_colliding.item():
31            robot_ix = event.message.channel.index
32            return Response.err({"collision_error": f"robot {robot_ix} collided (eid: {event.target})"})
33
34        return Response.ok("move_applied", new_state=entity.to_dict(serialization_field="serializable"))
35
36    @overrides
37    def to_dict(self) -> dict:
38        return {}
39
40    @overrides
41    def load_state_dict(self, state: dict):
42        pass

Plugin for world-level logic customization. Each handles a subset of network-level endpoints (e.g. 'move') A plugin can also update the physics or rendering logic through callbacks. Order of execution: [start tick] -> on_message_receive(w, eid) -> on_tick(w) -> [update] -> on_before_physics(w) -> -> [physics] -> on_after_physics(w) -> draw(w, eid?) -> on_message_response(w, eid, evt) -> [end tick]

endpoints: list[microspec.microspec.Endpoint] 
 9    @property
10    @overrides
11    def endpoints(self) -> list[Endpoint]:
12        return [
13            Endpoint("move",
14                     input={"control_input": Field("float32", shape=(6, ), range=[-1, 1])},
15                     output={"move_applied": Field("dict")}
16            )
17        ]

the endpoints (commands) of this plugin

@overrides
def on_message_receive( self, world: microecs.world.World, entity_id: int, message: robosim.plugin.Message) -> robosim.plugin.Response: 
19    @overrides
20    def on_message_receive(self, world: World, entity_id: int, message: Message) -> Response:
21        # all syntax validation happens earlier thanks to microspec and self.endpoints schema
22        return Response.physics(motion_input=message.payload["control_input"])

Called for each message and the associated entity of this plugin. Messages influence the World via Events

@overrides
def on_message_response( self, world: microecs.world.World, event: robosim.plugin.Event) -> robosim.plugin.Response: 
24    @overrides
25    def on_message_response(self, world: World, event: Event) -> Response:
26        if not event.applied:
27            return Response.err("move_not_applied")
28
29        entity = world.get_entity(event.target)
30        if entity.is_colliding.item():
31            robot_ix = event.message.channel.index
32            return Response.err({"collision_error": f"robot {robot_ix} collided (eid: {event.target})"})
33
34        return Response.ok("move_applied", new_state=entity.to_dict(serialization_field="serializable"))

Called for each event generated by on_message_receive if they still need answering (.response not set)

@overrides
def to_dict(self) -> dict: 
36    @overrides
37    def to_dict(self) -> dict:
38        return {}

the dict representation of this object for serialization purposes

@overrides
def load_state_dict(self, state: dict): 
40    @overrides
41    def load_state_dict(self, state: dict):
42        pass

Loads in place this object from a serialized dict representation

Inherited Members
robosim.plugin.Plugin
on_tick
on_before_physics
on_after_physics
draw