botbuilder-python/libraries/botbuilder-dialogs/botbuilder/dialogs/dialog.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from abc import ABC, abstractmethod

from botbuilder.core import TurnContext, NullTelemetryClient, BotTelemetryClient
from .dialog_reason import DialogReason
from .dialog_event import DialogEvent
from .dialog_turn_status import DialogTurnStatus
from .dialog_turn_result import DialogTurnResult
from .dialog_instance import DialogInstance


class Dialog(ABC):
    end_of_turn = DialogTurnResult(DialogTurnStatus.Waiting)

    def __init__(self, dialog_id: str):
        if dialog_id is None or not dialog_id.strip():
            raise TypeError("Dialog(): dialogId cannot be None.")

        self._telemetry_client = NullTelemetryClient()
        self._id = dialog_id

    @property
    def id(self) -> str:  # pylint: disable=invalid-name
        return self._id

    @property
    def telemetry_client(self) -> BotTelemetryClient:
        """
        Gets the telemetry client for logging events.
        """
        return self._telemetry_client

    @telemetry_client.setter
    def telemetry_client(self, value: BotTelemetryClient) -> None:
        """
        Sets the telemetry client for logging events.
        """
        if value is None:
            self._telemetry_client = NullTelemetryClient()
        else:
            self._telemetry_client = value

    @abstractmethod
    async def begin_dialog(
        self, dialog_context: "DialogContext", options: object = None
    ):
        """
        Method called when a new dialog has been pushed onto the stack and is being activated.
        :param dialog_context: The dialog context for the current turn of conversation.
        :param options: (Optional) additional argument(s) to pass to the dialog being started.
        """
        raise NotImplementedError()

    async def continue_dialog(self, dialog_context: "DialogContext"):
        """
        Method called when an instance of the dialog is the "current" dialog and the
        user replies with a new activity. The dialog will generally continue to receive the user's
        replies until it calls either `end_dialog()` or `begin_dialog()`.
        If this method is NOT implemented then the dialog will automatically be ended when the user replies.
        :param dialog_context: The dialog context for the current turn of conversation.
        :return:
        """
        # By default just end the current dialog.
        return await dialog_context.end_dialog(None)

    async def resume_dialog(  # pylint: disable=unused-argument
        self, dialog_context: "DialogContext", reason: DialogReason, result: object
    ):
        """
        Method called when an instance of the dialog is being returned to from another
        dialog that was started by the current instance using `begin_dialog()`.
        If this method is NOT implemented then the dialog will be automatically ended with a call
        to `end_dialog()`. Any result passed from the called dialog will be passed
        to the current dialog's parent.
        :param dialog_context: The dialog context for the current turn of conversation.
        :param reason: Reason why the dialog resumed.
        :param result: (Optional) value returned from the dialog that was called. The type of the value returned is
        dependent on the dialog that was called.
        :return:
        """
        # By default just end the current dialog and return result to parent.
        return await dialog_context.end_dialog(result)

    # TODO: instance is DialogInstance
    async def reprompt_dialog(  # pylint: disable=unused-argument
        self, context: TurnContext, instance: DialogInstance
    ):
        """
        :param context:
        :param instance:
        :return:
        """
        # No-op by default
        return

    # TODO: instance is DialogInstance
    async def end_dialog(  # pylint: disable=unused-argument
        self, context: TurnContext, instance: DialogInstance, reason: DialogReason
    ):
        """
        :param context:
        :param instance:
        :param reason:
        :return:
        """
        # No-op by default
        return

    def get_version(self) -> str:
        return self.id

    async def on_dialog_event(
        self, dialog_context: "DialogContext", dialog_event: DialogEvent
    ) -> bool:
        """
        Called when an event has been raised, using `DialogContext.emitEvent()`, by either the current dialog or a
         dialog that the current dialog started.
        :param dialog_context: The dialog context for the current turn of conversation.
        :param dialog_event: The event being raised.
        :return: True if the event is handled by the current dialog and bubbling should stop.
        """
        # Before bubble
        handled = await self._on_pre_bubble_event(dialog_context, dialog_event)

        # Bubble as needed
        if (not handled) and dialog_event.bubble and dialog_context.parent:
            handled = await dialog_context.parent.emit(
                dialog_event.name, dialog_event.value, True, False
            )

        # Post bubble
        if not handled:
            handled = await self._on_post_bubble_event(dialog_context, dialog_event)

        return handled

    async def _on_pre_bubble_event(  # pylint: disable=unused-argument
        self, dialog_context: "DialogContext", dialog_event: DialogEvent
    ) -> bool:
        """
        Called before an event is bubbled to its parent.
        This is a good place to perform interception of an event as returning `true` will prevent
        any further bubbling of the event to the dialogs parents and will also prevent any child
        dialogs from performing their default processing.
        :param dialog_context: The dialog context for the current turn of conversation.
        :param dialog_event: The event being raised.
        :return: Whether the event is handled by the current dialog and further processing should stop.
        """
        return False

    async def _on_post_bubble_event(  # pylint: disable=unused-argument
        self, dialog_context: "DialogContext", dialog_event: DialogEvent
    ) -> bool:
        """
        Called after an event was bubbled to all parents and wasn't handled.
        This is a good place to perform default processing logic for an event. Returning `true` will
        prevent any processing of the event by child dialogs.
        :param dialog_context: The dialog context for the current turn of conversation.
        :param dialog_event: The event being raised.
        :return: Whether the event is handled by the current dialog and further processing should stop.
        """
        return False

    def _on_compute_id(self) -> str:
        """
        Computes an unique ID for a dialog.
        :return: An unique ID for a dialog
        """
        return self.__class__.__name__

    def _register_source_location(
        self, path: str, line_number: int
    ):  # pylint: disable=unused-argument
        """
        Registers a SourceRange in the provided location.
        :param path: The path to the source file.
        :param line_number: The line number where the source will be located on the file.
        :return:
        """
        if path:
            # This will be added when debbuging support is ported.
            # DebugSupport.source_map.add(self, SourceRange(
            #     path = path,
            #     start_point = SourcePoint(line_index = line_number, char_index = 0 ),
            #     end_point = SourcePoint(line_index = line_number + 1, char_index = 0 ),
            # )
            return