robosim.plugin

 1"""
 2plugin.py Class definition for Plugin, Events and Messages used in the Plugin System.
 3Import level: 3
 4"""
 5from __future__ import annotations
 6from abc import ABC, abstractmethod
 7from dataclasses import dataclass
 8from enum import StrEnum
 9from typing import Any
10from microecs import EntityId, World
11from microspec import Endpoint
12
13from robosim.traits import Drawable, Restorable
14from robosim.network import ObjToSimChannel
15
16@dataclass
17class Message:
18    """Message dataclass between the (robot/object) channel and (towards) a plugin"""
19    target: EntityId
20    cmd: str
21    payload: dict[str, Any]
22    channel: ObjToSimChannel
23
24# pylint: disable=invalid-name
25class ResponseKind(StrEnum):
26    """The kind of responses a plugin may respond with: 'ok' (msg4wire), 'error' (wire) or 'physics' (becomes event)"""
27    Ok      = "ok"
28    Error   = "error"
29    Physics = "physics"
30
31@dataclass
32class Response:
33    """The response class as returned by the 'plugin.on_message_receive(w, eid)' method for all plugins and entities"""
34    kind: ResponseKind
35    payload: dict[str, Any]
36
37    @staticmethod
38    def ok(status_message: str | dict[str, Any], **kwargs) -> Response:
39        """creates an ok/status response to a message"""
40        return Response(kind=ResponseKind.Ok, payload={"status": status_message, **kwargs})
41
42    @staticmethod
43    def err(error_message: str | dict[str, Any], **kwargs) -> Response:
44        """creates an error response to a message"""
45        return Response(kind=ResponseKind.Error, payload={"error": error_message, **kwargs})
46
47    @staticmethod
48    def physics(**payload) -> Response:
49        """creates a physics response to a message. The payload is read by the physics system and targets components"""
50        return Response(kind=ResponseKind.Physics, payload=payload)
51
52@dataclass
53class Event:
54    """Basic event with a source and a destination, a payload an a status (physics or error)"""
55    source: Plugin                                              # who created this
56    target: EntityId                                            # what this event influences (anything with .id)
57    message: Message | None = None                              # the message that created this event (for responses!)
58    payload: dict[str, Any] | None = None                       # what to update the ECS data with (all fields k=>v)
59    applied: bool = False                                       # whether this even was applied in the physics system
60    response: dict | None = None                                # the 'response' to be sent back to the source (channel)
61
62class Plugin(Drawable, Restorable, ABC):
63    """
64    Plugin for world-level logic customization. Each handles a subset of network-level endpoints (e.g. 'move')
65    A plugin can also update the physics or rendering logic through callbacks.
66    Order of execution:
67    [start tick] -> on_message_receive(w, eid) -> on_tick(w) -> [update] -> on_before_physics(w) ->
68                 -> [physics] -> on_after_physics(w) -> draw(w, eid?) -> on_message_response(w, eid, evt) -> [end tick]
69    """
70
71    @property
72    @abstractmethod
73    def endpoints(self) -> list[Endpoint]:
74        """the endpoints (commands) of this plugin"""
75
76    @abstractmethod
77    def on_message_receive(self, world: World, entity_id: int, message: Message) -> Response:
78        """Called for each message and the associated entity of this plugin. Messages influence the World via Events"""
79
80    @abstractmethod
81    def on_message_response(self, world: World, event: Event) -> Response:
82        """Called for each event generated by on_message_receive if they still need answering (.response not set)"""
83
84    def on_tick(self, world: World) -> list[Event]: # pylint: disable=unused-argument
85        """
86        Called on each tick. Returns a list of events, one or many each per entity. These are internal plugin events
87        (e.g. wind, or fixed trajectory) that will also influence the physics system.
88        """
89        return []
90
91    def on_before_physics(self, world: World):
92        """callback called in the main loop after state (motion) updates and before physics"""
93
94    def on_after_physics(self, world: World):
95        """callback called in the main loop after physics and before drawing"""
96
97    def draw(self, world: World, entity_id: int | None = None): # pylint: disable=arguments-differ
98        """callback called during the drawing phase. Called twice, one for global camera and one for FPV (id is set)"""