"""Models for Controller Command Events."""
from __future__ import annotations
import json
from typing import Any, Self, override
import jsonschema
from emulator_engine.models.controller_command import ControllerCommand
from emulator_engine.schemas.controller_command_event import controller_command_event_schema
from ska_mid_cbf_emulators.common import EventSeverity, LoggerFactory, ManualEvent, ManualEventSubType
[docs]
class ControllerCommandEvent(ManualEvent):
"""Model for Controller command events (i.e. non-blocking/async API calls).
Subclasses :obj:`ManualEvent`.
Args:
command (:obj:`ControllerCommand`): The command to send to the controller.
value (:obj:`dict[Any, Any]`, optional): The value of this event/command, \
which can be any arbitrary dictionary. Default is `{}`.
severity (:obj:`EventSeverity`, optional): The severity of this event/command. \
Defaults to :obj:`EventSeverity.GENERAL`.
source_timestamp (:obj:`int`, optional): The timestamp, in milliseconds, for when the event/command \
was originally created. Defaults to the current system time.
update_timestamp (:obj:`int`, optional): The timestamp, in milliseconds, for when the event/command \
was created or last updated. Defaults to the value of source_timestamp.
Attributes:
command (:obj:`ControllerCommand`): The command sent to the controller.
value (:obj:`dict[Any, Any]`): The value of the event/command.
severity (:obj:`EventSeverity`): The severity of the event/command.
source_timestamp (:obj:`int`): The timestamp, in milliseconds, for when the event/command was originally created.
update_timestamp (:obj:`int`): The timestamp, in milliseconds, for when the event/command was created or last updated. \
For newly created events, update_timestamp == source_timestamp.
"""
def __init__(
self: Self,
command: ControllerCommand,
value: dict[str, Any] = {},
severity: EventSeverity = EventSeverity.GENERAL,
source_timestamp: int = None,
update_timestamp: int = None
) -> None:
super().__init__(
subtype=ManualEventSubType.GENERAL,
value=value,
severity=severity,
source_timestamp=source_timestamp,
update_timestamp=update_timestamp
)
self._command = command
[docs]
@override
@staticmethod
def decode(json_event_str: str) -> ControllerCommandEvent:
"""Decode a JSON-encoded controller command event.
Args:
encoded_event (:obj:`str`): The JSON string containing the event to decode
Returns:
:obj:`ControllerCommandEvent`: The decoded event
"""
logger = LoggerFactory.get_logger()
json_event = json.loads(json_event_str)
try:
jsonschema.validate(json_event, controller_command_event_schema)
except Exception as e:
logger.error(str(e))
raise e
logger.debug('Event schema validation was successful.')
return ControllerCommandEvent(
command=ControllerCommand(json_event.get('command').lower()),
value=json_event.get('value'),
severity=EventSeverity(json_event.get('severity').lower()),
source_timestamp=json_event.get('source_timestamp'),
update_timestamp=json_event.get('update_timestamp')
)
[docs]
@override
@staticmethod
def encode(event: ControllerCommandEvent) -> str:
"""Encode a controller command event to a JSON string.
Args:
event (:obj:`ControllerCommandEvent`): The event to encode
Returns:
:obj:`str`: The JSON-encoded event string
"""
e_dict = {
'type': event.type,
'subtype': event.subtype,
'command': event.command,
'severity': event.severity,
'source_timestamp': event.source_timestamp,
'update_timestamp': event.update_timestamp,
'value': event.value
}
return json.dumps(e_dict)
@property
def command(self: Self) -> ControllerCommand:
""":obj:`ControllerCommand`: The command to send to the controller."""
return self._command
@command.setter
def command(self: Self, new_command: ControllerCommand) -> None:
self._command = new_command