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