text: import abc
  import json
  import logging
  import re
  from abc import ABC
  
  import jsonpickle
  import time
  import uuid
  from dateutil import parser
  from datetime import datetime
  from typing import (
      List,
      Dict,
      Text,
      Any,
      Type,
      Optional,
      TYPE_CHECKING,
      Iterable,
      cast,
      Tuple,
  )
  
  import rasa.shared.utils.common
  import rasa.shared.utils.io
  from typing import Union
  
  from rasa.shared.constants import DOCS_URL_TRAINING_DATA
  from rasa.shared.core.constants import (
      LOOP_NAME,
      EXTERNAL_MESSAGE_PREFIX,
      ACTION_NAME_SENDER_ID_CONNECTOR_STR,
      IS_EXTERNAL,
      USE_TEXT_FOR_FEATURIZATION,
      LOOP_INTERRUPTED,
      ENTITY_LABEL_SEPARATOR,
      ACTION_SESSION_START_NAME,
      ACTION_LISTEN_NAME,
  )
  from rasa.shared.exceptions import UnsupportedFeatureException
  from rasa.shared.nlu.constants import (
      ENTITY_ATTRIBUTE_TYPE,
      INTENT,
      TEXT,
      ENTITIES,
      ENTITY_ATTRIBUTE_VALUE,
      ACTION_TEXT,
      ACTION_NAME,
      INTENT_NAME_KEY,
      ENTITY_ATTRIBUTE_ROLE,
      ENTITY_ATTRIBUTE_GROUP,
      PREDICTED_CONFIDENCE_KEY,
      INTENT_RANKING_KEY,
      ENTITY_ATTRIBUTE_TEXT,
      ENTITY_ATTRIBUTE_START,
      ENTITY_ATTRIBUTE_CONFIDENCE,
      ENTITY_ATTRIBUTE_END,
  )
  
  if TYPE_CHECKING:
      from typing_extensions import TypedDict
  
      from rasa.shared.core.trackers import DialogueStateTracker
  
      EntityPrediction = TypedDict(
          "EntityPrediction",
          {
              ENTITY_ATTRIBUTE_TEXT: Text,
              ENTITY_ATTRIBUTE_START: Optional[float],
              ENTITY_ATTRIBUTE_END: Optional[float],
              ENTITY_ATTRIBUTE_VALUE: Text,
              ENTITY_ATTRIBUTE_CONFIDENCE: float,
              ENTITY_ATTRIBUTE_TYPE: Text,
              ENTITY_ATTRIBUTE_GROUP: Optional[Text],
              ENTITY_ATTRIBUTE_ROLE: Optional[Text],
              "additional_info": Any,
          },
          total=False,
      )
  
      IntentPrediction = TypedDict(
          "IntentPrediction", {INTENT_NAME_KEY: Text, PREDICTED_CONFIDENCE_KEY: float,},
      )
      NLUPredictionData = TypedDict(
          "NLUPredictionData",
          {
              INTENT: IntentPrediction,
              INTENT_RANKING_KEY: List[IntentPrediction],
              ENTITIES: List[EntityPrediction],
              "message_id": Optional[Text],
              "metadata": Dict,
          },
          total=False,
      )
  logger = logging.getLogger(__name__)
  
  
  def deserialise_events(serialized_events: List[Dict[Text, Any]]) -> List["Event"]:
      """Convert a list of dictionaries to a list of corresponding events.
  
      Example format:
          [{"event": "slot", "value": 5, "name": "my_slot"}]
      """
  
      deserialised = []
  
      for e in serialized_events:
          if "event" in e:
              event = Event.from_parameters(e)
              if event:
                  deserialised.append(event)
              else:
                  logger.warning(
                      f"Unable to parse event '{event}' while deserialising. The event"
                      " will be ignored."
                  )
  
      return deserialised
  
  
  def deserialise_entities(entities: Union[Text, List[Any]]) -> List[Dict[Text, Any]]:
      if isinstance(entities, str):
          entities = json.loads(entities)
  
      return [e for e in entities if isinstance(e, dict)]
  
  
  def format_message(
      text: Text, intent: Optional[Text], entities: Union[Text, List[Any]]
  ) -> Text:
      """Uses NLU parser information to generate a message with inline entity annotations.
  
      Arguments:
          text: text of the message
          intent: intent of the message
          entities: entities of the message
  
      Return:
          Message with entities annotated inline, e.g.
          `I am from [Berlin]{"entity": "city"}`.
      """
      from rasa.shared.nlu.training_data.formats.readerwriter import TrainingDataWriter
      from rasa.shared.nlu.training_data import entities_parser
  
      message_from_md = entities_parser.parse_training_example(text, intent)
      deserialised_entities = deserialise_entities(entities)
      return TrainingDataWriter.generate_message(
          {"text": message_from_md.get(TEXT), "entities": deserialised_entities}
      )
  
  
  def split_events(
      events: Iterable["Event"],
      event_type_to_split_on: Type["Event"],
      additional_splitting_conditions: Optional[Dict[Text, Any]] = None,
      include_splitting_event: bool = True,
  ) -> List[List["Event"]]:
      """Splits events according to an event type and condition.
  
      Examples:
          Splitting events according to the event type `ActionExecuted` and the
          `action_name` 'action_session_start' would look as follows:
  
          >> _events = split_events(
                          events,
                          ActionExecuted,
                          {"action_name": "action_session_start"},
                          True
                       )
  
      Args:
          events: Events to split.
          event_type_to_split_on: The event type to split on.
          additional_splitting_conditions: Additional event attributes to split on.
          include_splitting_event: Whether the events of the type on which the split
              is based should be included in the returned events.
  
      Returns:
          The split events.
      """
      sub_events = []
      current = []
  
      def event_fulfills_splitting_condition(evt: "Event") -> bool:
          # event does not have the correct type
          if not isinstance(evt, event_type_to_split_on):
              return False
  
          # the type is correct and there are no further conditions
          if not additional_splitting_conditions:
              return True
  
          # there are further conditions - check those
          return all(
              getattr(evt, k, None) == v
              for k, v in additional_splitting_conditions.items()
          )
  
      for event in events:
          if event_fulfills_splitting_condition(event):
              if current:
                  sub_events.append(current)
  
              current = []
              if include_splitting_event:
                  current.append(event)
          else:
              current.append(event)
  
      if current:
          sub_events.append(current)
  
      return sub_events
  
  
  def do_events_begin_with_session_start(events: List["Event"]) -> bool:
      """Determines whether `events` begins with a session start sequence.
  
      A session start sequence is a sequence of two events: an executed
      `action_session_start` as well as a logged `session_started`.
  
      Args:
          events: The events to inspect.
  
      Returns:
          Whether or not `events` begins with a session start sequence.
      """
      return len(events) > 1 and events[:2] == [
          ActionExecuted(ACTION_SESSION_START_NAME),
          SessionStarted(),
      ]
  
  
  class Event(ABC):
      """Describes events in conversation and how the affect the conversation state.
  
      Immutable representation of everything which happened during a conversation of the
      user with the assistant. Tells the `rasa.shared.core.trackers.DialogueStateTracker`
      how to update its state as the events occur.
      """
  
      type_name = "event"
  
      def __init__(
          self,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          self.timestamp = timestamp or time.time()
          self.metadata = metadata or {}
  
      def __ne__(self, other: Any) -> bool:
          # Not strictly necessary, but to avoid having both x==y and x!=y
          # True at the same time
          return not (self == other)
  
      @abc.abstractmethod
      def as_story_string(self) -> Optional[Text]:
          """Returns the event as story string.
  
          Returns:
              textual representation of the event or None.
          """
          # Every class should implement this
          raise NotImplementedError
  
      @staticmethod
      def from_story_string(
          event_name: Text,
          parameters: Dict[Text, Any],
          default: Optional[Type["Event"]] = None,
      ) -> Optional[List["Event"]]:
          event_class = Event.resolve_by_type(event_name, default)
  
          if not event_class:
              return None
  
          return event_class._from_story_string(parameters)
  
      @staticmethod
      def from_parameters(
          parameters: Dict[Text, Any], default: Optional[Type["Event"]] = None
      ) -> Optional["Event"]:
  
          event_name = parameters.get("event")
          if event_name is None:
              return None
  
          event_class: Optional[Type[Event]] = Event.resolve_by_type(event_name, default)
          if not event_class:
              return None
  
          return event_class._from_parameters(parameters)
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List["Event"]]:
          """Called to convert a parsed story line into an event."""
          return [cls(parameters.get("timestamp"), parameters.get("metadata"))]
  
      def as_dict(self) -> Dict[Text, Any]:
          d = {"event": self.type_name, "timestamp": self.timestamp}
  
          if self.metadata:
              d["metadata"] = self.metadata
  
          return d
  
      def fingerprint(self) -> Text:
          """Returns a unique hash for the event which is stable across python runs.
  
          Returns:
              fingerprint of the event
          """
          data = self.as_dict()
          del data["timestamp"]
          return rasa.shared.utils.io.get_dictionary_fingerprint(data)
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> Optional["Event"]:
          """Called to convert a dictionary of parameters to a single event.
  
          By default uses the same implementation as the story line
          conversation ``_from_story_string``. But the subclass might
          decide to handle parameters differently if the parsed parameters
          don't origin from a story file.
          """
          result = cls._from_story_string(parameters)
          if len(result) > 1:
              logger.warning(
                  f"Event from parameters called with parameters "
                  f"for multiple events. This is not supported, "
                  f"only the first event will be returned. "
                  f"Parameters: {parameters}"
              )
          return result[0] if result else None
  
      @staticmethod
      def resolve_by_type(
          type_name: Text, default: Optional[Type["Event"]] = None
      ) -> Optional[Type["Event"]]:
          """Returns a slots class by its type name."""
  
          for cls in rasa.shared.utils.common.all_subclasses(Event):
              if cls.type_name == type_name:
                  return cls
          if type_name == "topic":
              return None  # backwards compatibility to support old TopicSet evts
          elif default is not None:
              return default
          else:
              raise ValueError(f"Unknown event name '{type_name}'.")
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state.
  
          Args:
              tracker: The current conversation state.
          """
          pass
  
      @abc.abstractmethod
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          # Every class should implement this
          raise NotImplementedError()
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"{self.__class__.__name__}()"
  
  
  class AlwaysEqualEventMixin(Event, ABC):
      """Class to deduplicate common behavior for events without additional attributes."""
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, self.__class__):
              return NotImplemented
  
          return True
  
  
  class SkipEventInMDStoryMixin(Event, ABC):
      """Skips the visualization of an event in Markdown stories."""
  
      def as_story_string(self) -> None:
          """Returns the event as story string.
  
          Returns:
              None, as this event should not appear inside the story.
          """
          return
  
  
  class UserUttered(Event):
      """The user has said something to the bot.
  
      As a side effect a new `Turn` will be created in the `Tracker`.
      """
  
      type_name = "user"
  
      def __init__(
          self,
          text: Optional[Text] = None,
          intent: Optional[Dict] = None,
          entities: Optional[List[Dict]] = None,
          parse_data: Optional["NLUPredictionData"] = None,
          timestamp: Optional[float] = None,
          input_channel: Optional[Text] = None,
          message_id: Optional[Text] = None,
          metadata: Optional[Dict] = None,
          use_text_for_featurization: Optional[bool] = None,
      ) -> None:
          """Creates event for incoming user message.
  
          Args:
              text: Text of user message.
              intent: Intent prediction of user message.
              entities: Extracted entities.
              parse_data: Detailed NLU parsing result for message.
              timestamp: When the event was created.
              metadata: Additional event metadata.
              input_channel: Which channel the user used to send message.
              message_id: Unique ID for message.
              use_text_for_featurization: `True` if the message's text was used to predict
                  next action. `False` if the message's intent was used.
  
          """
          self.text = text
          self.intent = intent if intent else {}
          self.entities = entities if entities else []
          self.input_channel = input_channel
          self.message_id = message_id
  
          super().__init__(timestamp, metadata)
  
          # The featurization is set by the policies during prediction time using a
          # `DefinePrevUserUtteredFeaturization` event.
          self.use_text_for_featurization = use_text_for_featurization
          # define how this user utterance should be featurized
          if self.text and not self.intent_name:
              # happens during training
              self.use_text_for_featurization = True
          elif self.intent_name and not self.text:
              # happens during training
              self.use_text_for_featurization = False
  
          self.parse_data: "NLUPredictionData" = {
              INTENT: self.intent,
              # Copy entities so that changes to `self.entities` don't affect
              # `self.parse_data` and hence don't get persisted
              ENTITIES: self.entities.copy(),
              TEXT: self.text,
              "message_id": self.message_id,
              "metadata": self.metadata,
          }
  
          if parse_data:
              self.parse_data.update(**parse_data)
  
      @staticmethod
      def _from_parse_data(
          text: Text,
          parse_data: "NLUPredictionData",
          timestamp: Optional[float] = None,
          input_channel: Optional[Text] = None,
          message_id: Optional[Text] = None,
          metadata: Optional[Dict] = None,
      ) -> "UserUttered":
          return UserUttered(
              text,
              parse_data.get(INTENT),
              parse_data.get(ENTITIES, []),
              parse_data,
              timestamp,
              input_channel,
              message_id,
              metadata,
          )
  
      def __hash__(self) -> int:
          """Returns unique hash of object."""
          return hash(json.dumps(self.as_sub_state()))
  
      @property
      def intent_name(self) -> Optional[Text]:
          """Returns intent name or `None` if no intent."""
          return self.intent.get(INTENT_NAME_KEY)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, UserUttered):
              return NotImplemented
  
          return (
              self.text,
              self.intent_name,
              [jsonpickle.encode(ent) for ent in self.entities],
          ) == (
              other.text,
              other.intent_name,
              [jsonpickle.encode(ent) for ent in other.entities],
          )
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          entities = ""
          if self.entities:
              entities = [
                  f"{entity[ENTITY_ATTRIBUTE_VALUE]} "
                  f"(Type: {entity[ENTITY_ATTRIBUTE_TYPE]}, "
                  f"Role: {entity.get(ENTITY_ATTRIBUTE_ROLE)}, "
                  f"Group: {entity.get(ENTITY_ATTRIBUTE_GROUP)})"
                  for entity in self.entities
              ]
              entities = f", entities: {', '.join(entities)}"
  
          return (
              f"UserUttered(text: {self.text}, intent: {self.intent_name}"
              f"{entities}"
              f", use_text_for_featurization: {self.use_text_for_featurization})"
          )
  
      @staticmethod
      def empty() -> "UserUttered":
          return UserUttered(None)
  
      def is_empty(self) -> bool:
          return not self.text and not self.intent_name and not self.entities
  
      def as_dict(self) -> Dict[Text, Any]:
          _dict = super().as_dict()
          _dict.update(
              {
                  "text": self.text,
                  "parse_data": self.parse_data,
                  "input_channel": getattr(self, "input_channel", None),
                  "message_id": getattr(self, "message_id", None),
                  "metadata": self.metadata,
              }
          )
          return _dict
  
      def as_sub_state(self,) -> Dict[Text, Union[None, Text, List[Optional[Text]]]]:
          """Turns a UserUttered event into features.
  
          The substate contains information about entities, intent and text of the
          `UserUttered` event.
  
          Returns:
              a dictionary with intent name, text and entities
          """
          entities = [entity.get(ENTITY_ATTRIBUTE_TYPE) for entity in self.entities]
          entities.extend(
              (
                  f"{entity.get(ENTITY_ATTRIBUTE_TYPE)}{ENTITY_LABEL_SEPARATOR}"
                  f"{entity.get(ENTITY_ATTRIBUTE_ROLE)}"
              )
              for entity in self.entities
              if ENTITY_ATTRIBUTE_ROLE in entity
          )
          entities.extend(
              (
                  f"{entity.get(ENTITY_ATTRIBUTE_TYPE)}{ENTITY_LABEL_SEPARATOR}"
                  f"{entity.get(ENTITY_ATTRIBUTE_GROUP)}"
              )
              for entity in self.entities
              if ENTITY_ATTRIBUTE_GROUP in entity
          )
  
          out: Dict[Text, Union[None, Text, List[Optional[Text]]]] = {}
          # During training we expect either intent_name or text to be set.
          # During prediction both will be set.
          if self.text and (
              self.use_text_for_featurization or self.use_text_for_featurization is None
          ):
              out[TEXT] = self.text
          if self.intent_name and not self.use_text_for_featurization:
              out[INTENT] = self.intent_name
          # don't add entities for e2e utterances
          if entities and not self.use_text_for_featurization:
              out[ENTITIES] = entities
  
          return out
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
          try:
              return [
                  cls._from_parse_data(
                      parameters.get("text"),
                      parameters.get("parse_data"),
                      parameters.get("timestamp"),
                      parameters.get("input_channel"),
                      parameters.get("message_id"),
                      parameters.get("metadata"),
                  )
              ]
          except KeyError as e:
              raise ValueError(f"Failed to parse bot uttered event. {e}")
  
      def _entity_string(self) -> Text:
          if self.entities:
              return json.dumps(
                  {
                      entity[ENTITY_ATTRIBUTE_TYPE]: entity[ENTITY_ATTRIBUTE_VALUE]
                      for entity in self.entities
                  },
                  ensure_ascii=False,
              )
          return ""
  
      def as_story_string(self, e2e: bool = False) -> Text:
          """Return event as string for Markdown training format.
  
          Args:
              e2e: `True` if the the event should be printed in the format for
                  end-to-end conversation tests.
  
          Returns:
              Event as string.
          """
          if self.use_text_for_featurization and not e2e:
              raise UnsupportedFeatureException(
                  f"Printing end-to-end user utterances is not supported in the "
                  f"Markdown training format. Please use the YAML training data format "
                  f"instead. Please see {DOCS_URL_TRAINING_DATA} for more information."
              )
  
          if e2e:
              text_with_entities = format_message(
                  self.text or "", self.intent_name, self.entities
              )
  
              intent_prefix = f"{self.intent_name}: " if self.intent_name else ""
              return f"{intent_prefix}{text_with_entities}"
  
          return f"{self.intent_name or ''}{self._entity_string()}"
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to tracker. See docstring of `Event`."""
          tracker.latest_message = self
          tracker.clear_followup_action()
  
      @staticmethod
      def create_external(
          intent_name: Text,
          entity_list: Optional[List[Dict[Text, Any]]] = None,
          input_channel: Optional[Text] = None,
      ) -> "UserUttered":
          return UserUttered(
              text=f"{EXTERNAL_MESSAGE_PREFIX}{intent_name}",
              intent={INTENT_NAME_KEY: intent_name},
              metadata={IS_EXTERNAL: True},
              entities=entity_list or [],
              input_channel=input_channel,
          )
  
  
  class DefinePrevUserUtteredFeaturization(SkipEventInMDStoryMixin):
      """Stores information whether action was predicted based on text or intent."""
  
      type_name = "user_featurization"
  
      def __init__(
          self,
          use_text_for_featurization: bool,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates event.
  
          Args:
              use_text_for_featurization: `True` if message text was used to predict
                  action. `False` if intent was used.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          super().__init__(timestamp, metadata)
          self.use_text_for_featurization = use_text_for_featurization
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"DefinePrevUserUtteredFeaturization({self.use_text_for_featurization})"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.use_text_for_featurization)
  
      @classmethod
      def _from_parameters(
          cls, parameters: Dict[Text, Any]
      ) -> "DefinePrevUserUtteredFeaturization":
          return DefinePrevUserUtteredFeaturization(
              parameters.get(USE_TEXT_FOR_FEATURIZATION),
              parameters.get("timestamp"),
              parameters.get("metadata"),
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({USE_TEXT_FOR_FEATURIZATION: self.use_text_for_featurization})
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state.
  
          Args:
              tracker: The current conversation state.
          """
          if tracker.latest_action_name != ACTION_LISTEN_NAME:
              # featurization belong only to the last user message
              # a user message is always followed by action listen
              return
  
          # update previous user message's featurization based on this event
          tracker.latest_message.use_text_for_featurization = (
              self.use_text_for_featurization
          )
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, DefinePrevUserUtteredFeaturization):
              return NotImplemented
  
          return self.use_text_for_featurization == other.use_text_for_featurization
  
  
  class EntitiesAdded(SkipEventInMDStoryMixin):
      """Event that is used to add extracted entities to the tracker state."""
  
      type_name = "entities"
  
      def __init__(
          self,
          entities: List[Dict[Text, Any]],
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Initializes event.
  
          Args:
              entities: Entities extracted from previous user message. This can either
                  be done by NLU components or end-to-end policy predictions.
              timestamp: the timestamp
              metadata: some optional metadata
          """
          super().__init__(timestamp, metadata)
          self.entities = entities
  
      def __str__(self) -> Text:
          """Returns the string representation of the event."""
          entity_str = [e[ENTITY_ATTRIBUTE_TYPE] for e in self.entities]
          return f"{self.__class__.__name__}({entity_str})"
  
      def __hash__(self) -> int:
          """Returns the hash value of the event."""
          return hash(json.dumps(self.entities))
  
      def __eq__(self, other: Any) -> bool:
          """Compares this event with another event."""
          if not isinstance(other, EntitiesAdded):
              return NotImplemented
  
          return self.entities == other.entities
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "EntitiesAdded":
          return EntitiesAdded(
              parameters.get(ENTITIES),
              parameters.get("timestamp"),
              parameters.get("metadata"),
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Converts the event into a dict.
  
          Returns:
              A dict that represents this event.
          """
          d = super().as_dict()
          d.update({ENTITIES: self.entities})
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state.
  
          Args:
              tracker: The current conversation state.
          """
          if tracker.latest_action_name != ACTION_LISTEN_NAME:
              # entities belong only to the last user message
              # a user message always comes after action listen
              return
  
          for entity in self.entities:
              if entity not in tracker.latest_message.entities:
                  tracker.latest_message.entities.append(entity)
  
  
  class BotUttered(SkipEventInMDStoryMixin):
      """The bot has said something to the user.
  
      This class is not used in the story training as it is contained in the
  
      ``ActionExecuted`` class. An entry is made in the ``Tracker``.
      """
  
      type_name = "bot"
  
      def __init__(
          self,
          text: Optional[Text] = None,
          data: Optional[Dict] = None,
          metadata: Optional[Dict[Text, Any]] = None,
          timestamp: Optional[float] = None,
      ) -> None:
          """Creates event for a bot response.
  
          Args:
              text: Plain text which bot responded with.
              data: Additional data for more complex utterances (e.g. buttons).
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.text = text
          self.data = data or {}
          super().__init__(timestamp, metadata)
  
      def __members(self) -> Tuple[Optional[Text], Text, Text]:
          data_no_nones = {k: v for k, v in self.data.items() if v is not None}
          meta_no_nones = {k: v for k, v in self.metadata.items() if v is not None}
          return (
              self.text,
              jsonpickle.encode(data_no_nones),
              jsonpickle.encode(meta_no_nones),
          )
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.__members())
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, BotUttered):
              return NotImplemented
  
          return self.__members() == other.__members()
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return "BotUttered(text: {}, data: {}, metadata: {})".format(
              self.text, json.dumps(self.data), json.dumps(self.metadata)
          )
  
      def __repr__(self) -> Text:
          """Returns text representation of event for debugging."""
          return "BotUttered('{}', {}, {}, {})".format(
              self.text, json.dumps(self.data), json.dumps(self.metadata), self.timestamp
          )
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.latest_bot_utterance = self
  
      def message(self) -> Dict[Text, Any]:
          """Return the complete message as a dictionary."""
          m = self.data.copy()
          m["text"] = self.text
          m["timestamp"] = self.timestamp
          m.update(self.metadata)
  
          if m.get("image") == m.get("attachment"):
              # we need this as there is an oddity we introduced a while ago where
              # we automatically set the attachment to the image. to not break
              # any persisted events we kept that, but we need to make sure that
              # the message contains the image only once
              m["attachment"] = None
  
          return m
  
      @staticmethod
      def empty() -> "BotUttered":
          """Creates an empty bot utterance."""
          return BotUttered()
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({"text": self.text, "data": self.data, "metadata": self.metadata})
          return d
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "BotUttered":
          try:
              return BotUttered(
                  parameters.get("text"),
                  parameters.get("data"),
                  parameters.get("metadata"),
                  parameters.get("timestamp"),
              )
          except KeyError as e:
              raise ValueError(f"Failed to parse bot uttered event. {e}")
  
  
  class SlotSet(Event):
      """The user has specified their preference for the value of a `slot`.
  
      Every slot has a name and a value. This event can be used to set a
      value for a slot on a conversation.
  
      As a side effect the `Tracker`'s slots will be updated so
      that `tracker.slots[key]=value`.
      """
  
      type_name = "slot"
  
      def __init__(
          self,
          key: Text,
          value: Optional[Any] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates event to set slot.
  
          Args:
              key: Name of the slot which is set.
              value: Value to which slot is set.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.key = key
          self.value = value
          super().__init__(timestamp, metadata)
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"SlotSet(key: {self.key}, value: {self.value})"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash((self.key, jsonpickle.encode(self.value)))
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, SlotSet):
              return NotImplemented
  
          return (self.key, self.value) == (other.key, other.value)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          props = json.dumps({self.key: self.value}, ensure_ascii=False)
          return f"{self.type_name}{props}"
  
      @classmethod
      def _from_story_string(
          cls, parameters: Dict[Text, Any]
      ) -> Optional[List["SlotSet"]]:
  
          slots = []
          for slot_key, slot_val in parameters.items():
              slots.append(SlotSet(slot_key, slot_val))
  
          if slots:
              return slots
          else:
              return None
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({"name": self.key, "value": self.value})
          return d
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "SlotSet":
          try:
              return SlotSet(
                  parameters.get("name"),
                  parameters.get("value"),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
              )
          except KeyError as e:
              raise ValueError(f"Failed to parse set slot event. {e}")
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._set_slot(self.key, self.value)
  
  
  class Restarted(AlwaysEqualEventMixin):
      """Conversation should start over & history wiped.
  
      Instead of deleting all events, this event can be used to reset the
      trackers state (e.g. ignoring any past user messages & resetting all
      the slots).
      """
  
      type_name = "restart"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124312)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Resets the tracker and triggers a followup `ActionSessionStart`."""
          tracker._reset()
          tracker.trigger_followup_action(ACTION_SESSION_START_NAME)
  
  
  class UserUtteranceReverted(AlwaysEqualEventMixin):
      """Bot reverts everything until before the most recent user message.
  
      The bot will revert all events after the latest `UserUttered`, this
      also means that the last event on the tracker is usually `action_listen`
      and the bot is waiting for a new user message.
      """
  
      type_name = "rewind"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124315)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._reset()
          tracker.replay_events()
  
  
  class AllSlotsReset(AlwaysEqualEventMixin):
      """All Slots are reset to their initial values.
  
      If you want to keep the dialogue history and only want to reset the
      slots, you can use this event to set all the slots to their initial
      values.
      """
  
      type_name = "reset_slots"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124316)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._reset_slots()
  
  
  class ReminderScheduled(Event):
      """Schedules the asynchronous triggering of a user intent at a given time.
  
      The triggered intent can include entities if needed.
      """
  
      type_name = "reminder"
  
      def __init__(
          self,
          intent: Text,
          trigger_date_time: datetime,
          entities: Optional[List[Dict]] = None,
          name: Optional[Text] = None,
          kill_on_user_message: bool = True,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates the reminder.
  
          Args:
              intent: Name of the intent to be triggered.
              trigger_date_time: Date at which the execution of the action
                  should be triggered (either utc or with tz).
              name: ID of the reminder. If there are multiple reminders with
                   the same id only the last will be run.
              entities: Entities that should be supplied together with the
                   triggered intent.
              kill_on_user_message: ``True`` means a user message before the
                   trigger date will abort the reminder.
              timestamp: Creation date of the event.
              metadata: Optional event metadata.
          """
          self.intent = intent
          self.entities = entities
          self.trigger_date_time = trigger_date_time
          self.kill_on_user_message = kill_on_user_message
          self.name = name if name is not None else str(uuid.uuid1())
          super().__init__(timestamp, metadata)
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(
              (
                  self.intent,
                  self.entities,
                  self.trigger_date_time.isoformat(),
                  self.kill_on_user_message,
                  self.name,
              )
          )
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, ReminderScheduled):
              return NotImplemented
  
          return self.name == other.name
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return (
              f"ReminderScheduled(intent: {self.intent}, "
              f"trigger_date: {self.trigger_date_time}, "
              f"entities: {self.entities}, name: {self.name})"
          )
  
      def scheduled_job_name(self, sender_id: Text) -> Text:
          return (
              f"[{hash(self.name)},{hash(self.intent)},{hash(str(self.entities))}]"
              f"{ACTION_NAME_SENDER_ID_CONNECTOR_STR}"
              f"{sender_id}"
          )
  
      def _properties(self) -> Dict[Text, Any]:
          return {
              "intent": self.intent,
              "date_time": self.trigger_date_time.isoformat(),
              "entities": self.entities,
              "name": self.name,
              "kill_on_user_msg": self.kill_on_user_message,
          }
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          props = json.dumps(self._properties())
          return f"{self.type_name}{props}"
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update(self._properties())
          return d
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
  
          trigger_date_time = parser.parse(parameters.get("date_time"))
  
          return [
              ReminderScheduled(
                  parameters.get("intent"),
                  trigger_date_time,
                  parameters.get("entities"),
                  name=parameters.get("name"),
                  kill_on_user_message=parameters.get("kill_on_user_msg", True),
                  timestamp=parameters.get("timestamp"),
                  metadata=parameters.get("metadata"),
              )
          ]
  
  
  class ReminderCancelled(Event):
      """Cancel certain jobs."""
  
      type_name = "cancel_reminder"
  
      def __init__(
          self,
          name: Optional[Text] = None,
          intent: Optional[Text] = None,
          entities: Optional[List[Dict]] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates a ReminderCancelled event.
  
          If all arguments are `None`, this will cancel all reminders.
          are to be cancelled. If no arguments are supplied, this will cancel all
          reminders.
  
          Args:
              name: Name of the reminder to be cancelled.
              intent: Intent name that is to be used to identify the reminders to be
                  cancelled.
              entities: Entities that are to be used to identify the reminders to be
                  cancelled.
              timestamp: Optional timestamp.
              metadata: Optional event metadata.
          """
          self.name = name
          self.intent = intent
          self.entities = entities
          super().__init__(timestamp, metadata)
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash((self.name, self.intent, str(self.entities)))
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, ReminderCancelled):
              return NotImplemented
  
          return hash(self) == hash(other)
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return (
              f"ReminderCancelled(name: {self.name}, intent: {self.intent}, "
              f"entities: {self.entities})"
          )
  
      def cancels_job_with_name(self, job_name: Text, sender_id: Text) -> bool:
          """Determines if this event should cancel the job with the given name.
  
          Args:
              job_name: Name of the job to be tested.
              sender_id: The `sender_id` of the tracker.
  
          Returns:
              `True`, if this `ReminderCancelled` event should cancel the job with the
              given name, and `False` otherwise.
          """
          match = re.match(
              rf"^\[([\d\-]*),([\d\-]*),([\d\-]*)\]"
              rf"({re.escape(ACTION_NAME_SENDER_ID_CONNECTOR_STR)}"
              rf"{re.escape(sender_id)})",
              job_name,
          )
          if not match:
              return False
          name_hash, intent_hash, entities_hash = match.group(1, 2, 3)
  
          # Cancel everything unless names/intents/entities are given to
          # narrow it down.
          return (
              ((not self.name) or self._matches_name_hash(name_hash))
              and ((not self.intent) or self._matches_intent_hash(intent_hash))
              and ((not self.entities) or self._matches_entities_hash(entities_hash))
          )
  
      def _matches_name_hash(self, name_hash: Text) -> bool:
          return str(hash(self.name)) == name_hash
  
      def _matches_intent_hash(self, intent_hash: Text) -> bool:
          return str(hash(self.intent)) == intent_hash
  
      def _matches_entities_hash(self, entities_hash: Text) -> bool:
          return str(hash(str(self.entities))) == entities_hash
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          props = json.dumps(
              {"name": self.name, "intent": self.intent, "entities": self.entities}
          )
          return f"{self.type_name}{props}"
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
          return [
              ReminderCancelled(
                  parameters.get("name"),
                  parameters.get("intent"),
                  parameters.get("entities"),
                  timestamp=parameters.get("timestamp"),
                  metadata=parameters.get("metadata"),
              )
          ]
  
  
  class ActionReverted(AlwaysEqualEventMixin):
      """Bot undoes its last action.
  
      The bot reverts everything until before the most recent action.
      This includes the action itself, as well as any events that
      action created, like set slot events - the bot will now
      predict a new action using the state before the most recent
      action.
      """
  
      type_name = "undo"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124318)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._reset()
          tracker.replay_events()
  
  
  class StoryExported(Event):
      """Story should get dumped to a file."""
  
      type_name = "export"
  
      def __init__(
          self,
          path: Optional[Text] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates event about story exporting.
  
          Args:
              path: Path to which story was exported to.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.path = path
          super().__init__(timestamp, metadata)
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124319)
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
          return [
              StoryExported(
                  parameters.get("path"),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
              )
          ]
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          if self.path:
              tracker.export_stories_to_file(self.path)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, StoryExported):
              return NotImplemented
  
          return self.path == other.path
  
  
  class FollowupAction(Event):
      """Enqueue a followup action."""
  
      type_name = "followup"
  
      def __init__(
          self,
          name: Text,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates an event which forces the model to run a certain action next.
  
          Args:
              name: Name of the action to run.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.action_name = name
          super().__init__(timestamp, metadata)
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.action_name)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, FollowupAction):
              return NotImplemented
  
          return self.action_name == other.action_name
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"FollowupAction(action: {self.action_name})"
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          props = json.dumps({"name": self.action_name})
          return f"{self.type_name}{props}"
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
  
          return [
              FollowupAction(
                  parameters.get("name"),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
              )
          ]
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({"name": self.action_name})
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.trigger_followup_action(self.action_name)
  
  
  class ConversationPaused(AlwaysEqualEventMixin):
      """Ignore messages from the user to let a human take over.
  
      As a side effect the `Tracker`'s `paused` attribute will
      be set to `True`.
      """
  
      type_name = "pause"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124313)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return str(self)
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._paused = True
  
  
  class ConversationResumed(AlwaysEqualEventMixin):
      """Bot takes over conversation.
  
      Inverse of `PauseConversation`. As a side effect the `Tracker`'s
      `paused` attribute will be set to `False`.
      """
  
      type_name = "resume"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124314)
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          return self.type_name
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker._paused = False
  
  
  class ActionExecuted(Event):
      """An operation describes an action taken + its result.
  
      It comprises an action and a list of events. operations will be appended
      to the latest `Turn`` in `Tracker.turns`.
      """
  
      type_name = "action"
  
      def __init__(
          self,
          action_name: Optional[Text] = None,
          policy: Optional[Text] = None,
          confidence: Optional[float] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict] = None,
          action_text: Optional[Text] = None,
          hide_rule_turn: bool = False,
      ) -> None:
          """Creates event for a successful event execution.
  
          Args:
              action_name: Name of the action which was executed. `None` if it was an
                  end-to-end prediction.
              policy: Policy which predicted action.
              confidence: Confidence with which policy predicted action.
              timestamp: When the event was created.
              metadata: Additional event metadata.
              action_text: In case it's an end-to-end action prediction, the text which
                  was predicted.
              hide_rule_turn: If `True`, this action should be hidden in the dialogue
                  history created for ML-based policies.
          """
          self.action_name = action_name
          self.policy = policy
          self.confidence = confidence
          self.unpredictable = False
          self.action_text = action_text
          self.hide_rule_turn = hide_rule_turn
  
          super().__init__(timestamp, metadata)
  
      def __members__(self) -> Tuple[Optional[Text], Optional[Text], Text]:
          meta_no_nones = {k: v for k, v in self.metadata.items() if v is not None}
          return (
              self.action_name,
              self.action_text,
              jsonpickle.encode(meta_no_nones),
          )
  
      def __repr__(self) -> Text:
          """Returns event as string for debugging."""
          return "ActionExecuted(action: {}, policy: {}, confidence: {})".format(
              self.action_name, self.policy, self.confidence
          )
  
      def __str__(self) -> Optional[Text]:
          """Returns event as human readable string."""
          return self.action_name or self.action_text
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.__members__())
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, ActionExecuted):
              return NotImplemented
  
          return self.__members__() == other.__members__()
  
      def as_story_string(self) -> Optional[Text]:
          """Returns event in Markdown format."""
          if self.action_text:
              raise UnsupportedFeatureException(
                  f"Printing end-to-end bot utterances is not supported in the "
                  f"Markdown training format. Please use the YAML training data format "
                  f"instead. Please see {DOCS_URL_TRAINING_DATA} for more information."
              )
  
          return self.action_name
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> Optional[List[Event]]:
          return [
              ActionExecuted(
                  parameters.get("name"),
                  parameters.get("policy"),
                  parameters.get("confidence"),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
                  parameters.get("action_text"),
                  parameters.get("hide_rule_turn", False),
              )
          ]
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update(
              {
                  "name": self.action_name,
                  "policy": self.policy,
                  "confidence": self.confidence,
                  "action_text": self.action_text,
                  "hide_rule_turn": self.hide_rule_turn,
              }
          )
          return d
  
      def as_sub_state(self) -> Dict[Text, Text]:
          """Turns ActionExecuted into a dictionary containing action name or action text.
  
          One action cannot have both set at the same time
  
          Returns:
              a dictionary containing action name or action text with the corresponding
              key.
          """
          if self.action_name:
              return {ACTION_NAME: self.action_name}
          else:
              # FIXME: we should define the type better here, and require either
              #        `action_name` or `action_text`
              return {ACTION_TEXT: cast(Text, self.action_text)}
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.set_latest_action(self.as_sub_state())
          tracker.clear_followup_action()
  
  
  class AgentUttered(SkipEventInMDStoryMixin):
      """The agent has said something to the user.
  
      This class is not used in the story training as it is contained in the
      ``ActionExecuted`` class. An entry is made in the ``Tracker``.
      """
  
      type_name = "agent"
  
      def __init__(
          self,
          text: Optional[Text] = None,
          data: Optional[Any] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """See docstring of `BotUttered`."""
          self.text = text
          self.data = data
          super().__init__(timestamp, metadata)
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash((self.text, jsonpickle.encode(self.data)))
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, AgentUttered):
              return NotImplemented
  
          return (self.text, jsonpickle.encode(self.data)) == (
              other.text,
              jsonpickle.encode(other.data),
          )
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return "AgentUttered(text: {}, data: {})".format(
              self.text, json.dumps(self.data)
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({"text": self.text, "data": self.data})
          return d
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "AgentUttered":
          try:
              return AgentUttered(
                  parameters.get("text"),
                  parameters.get("data"),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
              )
          except KeyError as e:
              raise ValueError(f"Failed to parse agent uttered event. {e}")
  
  
  class ActiveLoop(Event):
      """If `name` is given: activates a loop with `name` else deactivates active loop."""
  
      type_name = "active_loop"
  
      def __init__(
          self,
          name: Optional[Text],
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates event for active loop.
  
          Args:
              name: Name of activated loop or `None` if current loop is deactivated.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.name = name
          super().__init__(timestamp, metadata)
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"Loop({self.name})"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.name)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, ActiveLoop):
              return NotImplemented
  
          return self.name == other.name
  
      def as_story_string(self) -> Text:
          """Returns text representation of event."""
          props = json.dumps({LOOP_NAME: self.name})
          return f"{ActiveLoop.type_name}{props}"
  
      @classmethod
      def _from_story_string(cls, parameters: Dict[Text, Any]) -> List["ActiveLoop"]:
          """Called to convert a parsed story line into an event."""
          return [
              ActiveLoop(
                  parameters.get(LOOP_NAME),
                  parameters.get("timestamp"),
                  parameters.get("metadata"),
              )
          ]
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({LOOP_NAME: self.name})
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.change_loop_to(self.name)
  
  
  class LegacyForm(ActiveLoop):
      """Legacy handler of old `Form` events.
  
      The `ActiveLoop` event used to be called `Form`. This class is there to handle old
      legacy events which were stored with the old type name `form`.
      """
  
      type_name = "form"
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          # Dump old `Form` events as `ActiveLoop` events instead of keeping the old
          # event type.
          d["event"] = ActiveLoop.type_name
          return d
  
      def fingerprint(self) -> Text:
          """Returns the hash of the event."""
          d = self.as_dict()
          # Revert event name to legacy subclass name to avoid different event types
          # having the same fingerprint.
          d["event"] = self.type_name
          del d["timestamp"]
          return rasa.shared.utils.io.get_dictionary_fingerprint(d)
  
  
  class LoopInterrupted(SkipEventInMDStoryMixin):
      """Event added by FormPolicy and RulePolicy.
  
      Notifies form action whether or not to validate the user input.
      """
  
      type_name = "loop_interrupted"
  
      def __init__(
          self,
          is_interrupted: bool,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Event to notify that loop was interrupted.
  
          This e.g. happens when a user is within a form, and is de-railing the
          form-filling by asking FAQs.
  
          Args:
              is_interrupted: `True` if the loop execution was interrupted, and ML
                  policies had to take over the last prediction.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          super().__init__(timestamp, metadata)
          self.is_interrupted = is_interrupted
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return f"{LoopInterrupted.__name__}({self.is_interrupted})"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.is_interrupted)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, LoopInterrupted):
              return NotImplemented
  
          return self.is_interrupted == other.is_interrupted
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "LoopInterrupted":
          return LoopInterrupted(
              parameters.get(LOOP_INTERRUPTED, False),
              parameters.get("timestamp"),
              parameters.get("metadata"),
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update({LOOP_INTERRUPTED: self.is_interrupted})
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.interrupt_loop(self.is_interrupted)
  
  
  class LegacyFormValidation(LoopInterrupted):
      """Legacy handler of old `FormValidation` events.
  
      The `LoopInterrupted` event used to be called `FormValidation`. This class is there
      to handle old legacy events which were stored with the old type name
      `form_validation`.
      """
  
      type_name = "form_validation"
  
      def __init__(
          self,
          validate: bool,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """See parent class docstring."""
          # `validate = True` is the same as `interrupted = False`
          super().__init__(not validate, timestamp, metadata)
  
      @classmethod
      def _from_parameters(cls, parameters: Dict) -> "LoopInterrupted":
          return LoopInterrupted(
              # `validate = True` means `is_interrupted = False`
              not parameters.get("validate", True),
              parameters.get("timestamp"),
              parameters.get("metadata"),
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          # Dump old `Form` events as `ActiveLoop` events instead of keeping the old
          # event type.
          d["event"] = LoopInterrupted.type_name
          return d
  
      def fingerprint(self) -> Text:
          """Returns hash of the event."""
          d = self.as_dict()
          # Revert event name to legacy subclass name to avoid different event types
          # having the same fingerprint.
          d["event"] = self.type_name
          del d["timestamp"]
          return rasa.shared.utils.io.get_dictionary_fingerprint(d)
  
  
  class ActionExecutionRejected(SkipEventInMDStoryMixin):
      """Notify Core that the execution of the action has been rejected."""
  
      type_name = "action_execution_rejected"
  
      def __init__(
          self,
          action_name: Text,
          policy: Optional[Text] = None,
          confidence: Optional[float] = None,
          timestamp: Optional[float] = None,
          metadata: Optional[Dict[Text, Any]] = None,
      ) -> None:
          """Creates event.
  
          Args:
              action_name: Action which was rejected.
              policy: Policy which predicted the rejected action.
              confidence: Confidence with which the reject action was predicted.
              timestamp: When the event was created.
              metadata: Additional event metadata.
          """
          self.action_name = action_name
          self.policy = policy
          self.confidence = confidence
          super().__init__(timestamp, metadata)
  
      def __str__(self) -> Text:
          """Returns text representation of event."""
          return (
              "ActionExecutionRejected("
              "action: {}, policy: {}, confidence: {})"
              "".format(self.action_name, self.policy, self.confidence)
          )
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(self.action_name)
  
      def __eq__(self, other: Any) -> bool:
          """Compares object with other object."""
          if not isinstance(other, ActionExecutionRejected):
              return NotImplemented
  
          return self.action_name == other.action_name
  
      @classmethod
      def _from_parameters(cls, parameters: Dict[Text, Any]) -> "ActionExecutionRejected":
          return ActionExecutionRejected(
              parameters.get("name"),
              parameters.get("policy"),
              parameters.get("confidence"),
              parameters.get("timestamp"),
              parameters.get("metadata"),
          )
  
      def as_dict(self) -> Dict[Text, Any]:
          """Returns serialized event."""
          d = super().as_dict()
          d.update(
              {
                  "name": self.action_name,
                  "policy": self.policy,
                  "confidence": self.confidence,
              }
          )
          return d
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          tracker.reject_action(self.action_name)
  
  
  class SessionStarted(AlwaysEqualEventMixin):
      """Mark the beginning of a new conversation session."""
  
      type_name = "session_started"
  
      def __hash__(self) -> int:
          """Returns unique hash for event."""
          return hash(32143124320)
  
      def as_story_string(self) -> None:
          """Skips representing event in stories."""
          logger.warning(
              f"'{self.type_name}' events cannot be serialised as story strings."
          )
  
      def apply_to(self, tracker: "DialogueStateTracker") -> None:
          """Applies event to current conversation state."""
          # noinspection PyProtectedMember
          tracker._reset()
  