Events support

ankaios_sdk/_components/__init__.py

@@ -43,6 +43,7 @@
 from .response import *
 from .manifest import *
 from .log_campaign import *
+from .event_campaign import *
 from .control_interface import *
 from .file import *
 

ankaios_sdk/_components/complete_state.py

@@ -99,6 +99,7 @@ def __init__(
                 the complete state.
         """
         self._complete_state = _ank_base.CompleteState()
+        self._complete_state.desiredState.workloads.workloads.clear()
         self._set_api_version(SUPPORTED_API_VERSION)
         if _proto:
             self._complete_state = _proto

ankaios_sdk/_components/control_interface.py

@@ -101,7 +101,10 @@ class ControlInterface:
     "(str): The base path for the Ankaios control interface."
 
     def __init__(
-        self, add_response_callback: Callable, add_log_callback: Callable
+        self,
+        add_response_callback: Callable,
+        add_log_callback: Callable,
+        add_event_callback: Callable,
     ) -> None:
         """
         Initialize the ControlInterface object. This is used
@@ -124,6 +127,7 @@ def __init__(
 
         self._add_response_callback = add_response_callback
         self._add_log_callback = add_log_callback
+        self._add_event_callback = add_event_callback
 
         self._logger = get_logger()
 
@@ -397,6 +401,15 @@ def _handle_response(self, response: Response) -> None:
                 )
                 return
 
+            # Filter out the events
+            if response.content_type in [
+                ResponseType.EVENT_RESPONSE,
+            ]:
+                self._add_event_callback(
+                    response.get_request_id(), response.content
+                )
+                return
+
             # Send out the response to the Ankaios class
             self._add_response_callback(response)
 

ankaios_sdk/_components/event_campaign.py

@@ -0,0 +1,80 @@
+# Copyright (c) 2025 Elektrobit Automotive GmbH
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License, Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+
+"""
+This module defines the EventQueue class for handling events.
+
+Classes
+-------
+
+- EventQueue:
+    Represents a queue of events received from the Ankaios system.
+    Inherits from the standard Queue class.
+
+Usage
+-----
+
+- Check current state of the followed fields:
+    .. code-block:: python
+
+        event_queue: EventQueue
+        current_state = event_queue.complete_state
+
+- Get the events out of the queue:
+    .. code-block:: python
+
+        event_queue: EventQueue
+        event: EventEntry = event_queue.get()  # blocking call
+"""
+
+__all__ = ["EventQueue"]
+
+
+from queue import Queue
+from .request import EventsRequest
+from .response import EventEntry
+from .complete_state import CompleteState
+
+
+class EventQueue(Queue):
+    """
+    Represents a queue of updates through the event campaign.
+    Inherits from the standard Queue class.
+    All objects in this queue are of type :py:type:`EventEntry`.
+    """
+
+    def __init__(
+        self,
+        request_id: EventsRequest,
+    ) -> None:
+        """
+        Initializes the EventQueue with the given parameters.
+
+        Args:
+            request_id (str): The request id of the event campaign.
+        """
+        super().__init__()
+        self.complete_state: CompleteState = None
+        self._request_id = request_id
+
+    def add_event(self, event: EventEntry) -> None:
+        """
+        Adds an event to the queue and updates the known complete state.
+
+        Args:
+            event (EventEntry): The event to be added to the queue.
+        """
+        self.complete_state = event.complete_state
+        self.put(event)

ankaios_sdk/_components/log_campaign.py

@@ -44,7 +44,6 @@
 
 from queue import Queue
 from .workload_state import WorkloadInstanceName
-from .request import LogsRequest, LogsCancelRequest
 
 
 # pylint: disable=too-few-public-methods
@@ -79,35 +78,15 @@ class LogQueue(Queue):
     All objects in this queue are of type :py:type:`LogResponse`.
     """
 
-    # pylint: disable=too-many-arguments
     def __init__(
         self,
-        request: LogsRequest,
+        request_id: str,
     ) -> None:
         """
         Initializes the LogQueue with the given parameters.
 
         Args:
-            request (LogsRequest): The request object containing the log
-                campaign parameters.
+            request_id (str): The request id of the logs campaign.
         """
         super().__init__()
-        self._request = request
-
-    def _get_request(self) -> LogsRequest:
-        """
-        Returns the LogsRequest object.
-
-        Returns:
-            LogsRequest: The LogsRequest object.
-        """
-        return self._request
-
-    def _get_cancel_request(self) -> LogsCancelRequest:
-        """
-        Returns the LogsCancelRequest object.
-
-        Returns:
-            LogsCancelRequest: The LogsCancelRequest object.
-        """
-        return LogsCancelRequest(request_id=self._request.get_id())
+        self._request_id = request_id

ankaios_sdk/_components/request.py

@@ -31,6 +31,11 @@ class that should be subclassed for specific request types.
 - LogsCancelRequest:
     Represents a request to stop the real-time log stream from the
     Ankaios system.
+- EventsRequest:
+    Represents a request to register for events from the Ankaios system.
+- EventsCancelRequest:
+    Represents a request to unregister from the event stream of a specific
+    event campaign in the Ankaios system.
 
 Usage
 -----
@@ -77,6 +82,8 @@ class that should be subclassed for specific request types.
     "UpdateStateRequest",
     "LogsRequest",
     "LogsCancelRequest",
+    "EventsRequest",
+    "EventsCancelRequest",
 ]
 
 import uuid
@@ -163,6 +170,7 @@ def __init__(self, masks: list = []) -> None:
         """
         super().__init__()
         self._request.completeStateRequest.fieldMask[:] = masks
+        self._request.completeStateRequest.subscribeForEvents = False
 
         self.logger.debug(
             "Created request of type GetState with id %s",
@@ -282,3 +290,52 @@ def __init__(self, request_id: str) -> None:
             "Created request of type LogsCancelRequest with id %s",
             self._request.requestId,
         )
+
+
+# pylint: disable=too-few-public-methods, dangerous-default-value
+class EventsRequest(Request):
+    """
+    Represents a request for registering on events configured by the
+    given masks.
+    """
+
+    def __init__(self, masks: list = []) -> None:
+        """
+        Initializes a EventsRequest instance.
+
+        Args:
+            masks (list): The masks to subscribe to for the event.
+        """
+        super().__init__()
+        self._request.completeStateRequest.fieldMask[:] = masks
+        self._request.completeStateRequest.subscribeForEvents = True
+
+        self.logger.debug(
+            "Created request of type EventsRequest with id %s",
+            self._request.requestId,
+        )
+
+
+# pylint: disable=too-few-public-methods, dangerous-default-value
+class EventsCancelRequest(Request):
+    """
+    Represents a request for unregistering from the event stream
+    of a specific event campaign in the Ankaios system.
+    """
+
+    def __init__(self, request_id: str) -> None:
+        """
+        Initializes an EventsCancelRequest instance.
+
+        Args:
+            id (str): The request ID.
+        """
+        super().__init__(_id=request_id)
+        self._request.eventsCancelRequest.CopyFrom(
+            _ank_base.EventsCancelRequest()
+        )
+
+        self.logger.debug(
+            "Created request of type EventsCancelRequest with id %s",
+            self._request.requestId,
+        )

ankaios_sdk/_components/response.py

@@ -29,13 +29,15 @@
 - LogsStopResponse:
     Represents a response for marking the end of the log stream from a
     workload instance.
+- EventEntry:
+    Represents an event.
 
 Enums
 -----
 
 - ResponseType:
-    Enumeration for the different types of response. It includes
-    ERROR, COMPLETE_STATE, and UPDATE_STATE_SUCCESS and CONNECTION_CLOSED.
+    Enumeration for the different types of response. It includes responses
+    like ERROR, CONNECTION_CLOSED, COMPLETE_STATE and so on.
 
 Union Types
 -----------
@@ -73,6 +75,7 @@
     "LogEntry",
     "LogsStopResponse",
     "LogResponse",
+    "EventEntry",
 ]
 
 from dataclasses import dataclass
@@ -146,6 +149,7 @@ def _parse_response(self) -> None:
             self.get_request_id(),
         )
 
+    # pylint: disable=too-many-branches
     def _from_proto(self) -> None:
         """
         Converts the parsed protobuf message to a Response object.
@@ -158,9 +162,17 @@ def _from_proto(self) -> None:
         if self._response.HasField("error"):
             self.content_type = ResponseType.ERROR
             self.content = self._response.error.message
-        elif self._response.HasField("completeState"):
-            self.content_type = ResponseType.COMPLETE_STATE
-            self.content = CompleteState(_proto=self._response.completeState)
+        elif self._response.HasField("completeStateResponse"):
+            if self._response.completeStateResponse.HasField("alteredFields"):
+                self.content_type = ResponseType.EVENT_RESPONSE
+                self.content = EventEntry._from_response(
+                    self._response.completeStateResponse
+                )
+            else:
+                self.content_type = ResponseType.COMPLETE_STATE
+                self.content = CompleteState(
+                    _proto=self._response.completeStateResponse.completeState
+                )
         elif self._response.HasField("UpdateStateSuccess"):
             update_state_msg = self._response.UpdateStateSuccess
             self.content_type = ResponseType.UPDATE_STATE_SUCCESS
@@ -203,6 +215,9 @@ def _from_proto(self) -> None:
         elif self._response.HasField("logsCancelAccepted"):
             self.content_type = ResponseType.LOGS_CANCEL_ACCEPTED
             self.content = None
+        elif self._response.HasField("eventsCancelAccepted"):
+            self.content_type = ResponseType.EVENT_CANCEL_ACCEPTED
+            self.content = None
         else:
             raise ResponseException("Invalid response type.")
 
@@ -254,7 +269,11 @@ class ResponseType(Enum):
     "(int): Got logs stop response."
     LOGS_CANCEL_ACCEPTED = 8
     "(int): Logs cancel request accepted."
-    CONNECTION_CLOSED = 9
+    EVENT_RESPONSE = 9
+    "(int): Got an event response."
+    EVENT_CANCEL_ACCEPTED = 10
+    "(int): Event cancel request accepted."
+    CONNECTION_CLOSED = 20
     "(int): Connection closed by the server."
 
     def __str__(self) -> str:
@@ -396,3 +415,46 @@ def _from_stop_response(
 
 
 LogResponse = Union[LogEntry, LogsStopResponse]
+
+
+@dataclass
+class EventEntry:
+    """
+    Represents an event notification.
+    """
+
+    complete_state: CompleteState
+    """The complete state of the event indicating the state changes."""
+    added_fields: list[str]
+    """The list of added fields of the state."""
+    updated_fields: list[str]
+    """The list of updated fields of the state."""
+    removed_fields: list[str]
+    """The list of removed fields of the state."""
+
+    def __str__(self) -> str:
+        """
+        Converts the EventEntry to a string.
+
+        Returns:
+            str: The string representation of the EventEntry.
+        """
+        ret = "Event:\n"
+        if self.added_fields:
+            ret += f"  Added fields: {self.added_fields}\n"
+        if self.updated_fields:
+            ret += f"  Updated fields: {self.updated_fields}\n"
+        if self.removed_fields:
+            ret += f"  Deleted fields: {self.removed_fields}\n"
+        return ret
+
+    @staticmethod
+    def _from_response(
+        response: _ank_base.CompleteStateResponse,
+    ) -> "EventEntry":
+        return EventEntry(
+            CompleteState(_proto=response.completeState),
+            response.alteredFields.addedFields,
+            response.alteredFields.updatedFields,
+            response.alteredFields.removedFields,
+        )