Source code for ska_ser_skallop.subscribing.event_item

import re
from datetime import datetime
from typing import Any, Tuple, Union

from ska_ser_skallop.subscribing import base, helpers


class EventItem(base.EventItemBase):
    """
    Class for tying an incoming event from a subscription to the subscription itself as
    well as a registered event handler and allow the extraction of these items
    """

    def __init__(
        self,
        event: base.EventDataInt,
        subscription: base.SubscriptionBase,
        handler: Union[base.MessageHandlerBase, None],
    ) -> None:
        """

        :param event: the event produced as a result of a subscription
        :param subscription: the subscription that is causing the events to be produced
        :param handler: an handler that is registered with this subscription to handle
            events coming from it
        """
        self.event = event
        self.subscription = subscription
        self.handler = handler
        self.init_date = datetime.now()

    def get_date_lodged(self) -> datetime:
        """Returns the date the event was created by the producer

        :return: the date when the event was lodged
        """
        if self.event.attr_value is None:
            return self.init_date
        return self.event.attr_value.time.todatetime()

    def get_date_init(self) -> datetime:
        """Returns the date the event item (not the event it self) was created

        :return: the date of event item creation
        """
        return self.init_date

    def get_producer_name(self) -> str:
        """Returns the name of the producer (or device) that generatted the evnt

        :return: the name of the producer
        """
        return self.event.device.name()

    def get_attr_name(self) -> str:
        """Returns the attribute of the producer upon which events were raised

        :return: the name of the attribute
        """
        if self.event.attr_value is None:
            return re.search(r"\w*(?<=$)", self.event.attr_name).group(0)
        return self.event.attr_value.name

    def get_attr_value_str(self) -> str:
        """Returns the current value of the attribute rendered as a string

        :return: the current value of the attribute
        """
        if self.event.attr_value is None:
            if self.event.err:
                return str(self.event.errors)
            return ""
        return helpers.get_attr_value_as_str(self.event.attr_value)

    def get_date_lodged_isoformat(self) -> str:
        """Returns the date the event was generated, but rendered as a string in
        isoformat (usefull for logging purposes)

        :return: the isoformatted date
        """
        if self.event.attr_value is None:
            return self.init_date.isoformat()
        return self.event.attr_value.time.isoformat()

    def describe(self) -> Tuple[str, str, str, str]:
        """Returns a description of the event data in the form of a tuple
        of strings in the following order:

        1. The name of the producer (device)
        2. The name of the attribute for which the subscription was based
        3. The value of the attribute
        4. The date at which the event was generated

        :return: [description]
        """
        return helpers.describe_event(self.event, self.init_date)

    def handle_event(self):
        if self.handler:
            self.handler.handle_event(self.event, self.subscription)
            self.handler.print_event(self.event)

    def event_value(self) -> Any:
        return self.event.attr_value.value