import logging
from typing import Any, Dict, Hashable, List, Optional, Text

from dazu.config import DazuConfig, override_defaults
from dazu.typing import Message, Module, TrainingData
from dazu.typing.model import Metadata

logger = logging.getLogger(__name__)


class UnsupportedLanguageError(Exception):
    """Raised when a component is created but the language is not supported.

    :param component: component name
    :param language: language that component doesn't support
    """

    def __init__(self, component: Text, language: Text) -> None:
        self.component = component
        self.language = language

        super().__init__(component, language)

    def __str__(self) -> Text:
        return (
            f"component '{self.component}' does not support language '{self.language}'."
        )


# Component class inspired by RasaHQ/rasa
class Component(Module):
    """A component is a message processing unit in a pipeline.
    Components are collected sequentially in a pipeline. Each component
    is called one after another. This holds for
    initialization, training, persisting and loading the components.
    If a component comes first in a pipeline, its
    methods will be called first.
    E.g. to process an incoming message, the ``process`` method of
    each component will be called. During the processing
    (as well as the training, persisting and initialization)
    components can pass information to other components.
    The information is passed to other components by providing
    attributes to the so called pipeline context. The
    pipeline context contains all the information of the previous
    components a component can use to do its own
    processing. For example, a featurizer component can provide
    features that are used by another component down
    the pipeline to do intent classification."""

    # Defines what attributes the pipeline component will
    # provide when called. The listed attributes
    # should be set by the component on the message object
    # during test and train, e.g.
    # ```message.set("entities", [...])```
    provides = []

    # Which attributes on a message are required by this
    # component. E.g. if requires contains "tokens", than a
    # previous component in the pipeline needs to have "tokens"
    # within the above described `provides` property.
    # Use `any_of("option_1", "option_2")` to define that either
    # "option_1" or "option_2" needs to be present in the
    # provided properties from the previous components.
    requires = []

    # Defines the default configuration parameters of a component
    # these values can be overwritten in the pipeline configuration
    # of the model. The component should choose sensible defaults
    # and should be able to create reasonable results with the defaults.
    defaults = {}

    # Defines what language(s) this component can handle.
    # This attribute is designed for instance method: `can_handle_language`.
    # Default value is None which means it can handle all languages.
    # This is an important feature for backwards compatibility of components.
    language_list = None

    def __init__(self, component_config: Optional[Dict[Text, Any]] = None) -> None:

        if not component_config:
            component_config = {}

        # makes sure the name of the configuration is part of the config
        # this is important for e.g. persistence
        component_config["name"] = type(self).name()

        self.component_config = override_defaults(self.defaults, component_config)

        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) -> List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(
        cls,
        component_config: Dict[Text, Any],
        model_dir: Optional[Text] = None,
        model_metadata: Optional["Metadata"] = None,
        cached_component: Optional["Component"] = None,
        **kwargs: Any,
    ) -> "Component":
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(
        cls, component_config: Dict[Text, Any], config: DazuConfig
    ) -> "Component":
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""

        # Check language supporting
        language = config.language
        if not cls.can_handle_language(language):
            # check failed
            raise UnsupportedLanguageError(cls.name, language)

        return cls(component_config)

    def provide_context(self) -> Optional[Dict[Text, Any]]:
        """Initialize this component for a new pipeline
        This function will be called before the training
        is started and before the first message is processed using
        the interpreter. The component gets the opportunity to
        add information to the context that is passed through
        the pipeline during training and message parsing. Most
        components do not need to implement this method.
        It's mostly used to initialize framework environments
        like MITIE and spacy
        (e.g. loading word vectors for the pipeline)."""

    def train(
        self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
    ) -> None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) -> None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) -> Optional[Dict[Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(
        cls, component_meta: Dict[Text, Any], model_metadata: "Metadata"
    ) -> Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""

        return None

    def __getstate__(self) -> Any:
        d = self.__dict__.copy()
        # these properties should not be pickled
        if "partial_processing_context" in d:
            del d["partial_processing_context"]
        if "partial_processing_pipeline" in d:
            del d["partial_processing_pipeline"]
        return d

    def __eq__(self, other) -> bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(
        self, pipeline: List["Component"], context: Dict[Text, Any]
    ) -> None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""

        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) -> Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""

        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info("Failed to run partial processing due to missing pipeline.")
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) -> bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""

        # if language_list is set to `None` it means: support all languages
        if language is None or cls.language_list is None:
            return True

        return language in cls.language_list

ab0793b5a7e78699dd596469b083122e1623fd22
<mask token>


class Component(Module):
    <mask token>
    <mask token>
    <mask token>
    <mask token>
    <mask token>

    def __init__(self, component_config: Optional[Dict[Text, Any]]=None
        ) ->None:
        if not component_config:
            component_config = {}
        component_config['name'] = type(self).name()
        self.component_config = override_defaults(self.defaults,
            component_config)
        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) ->List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(cls, component_config: Dict[Text, Any], model_dir: Optional[
        Text]=None, model_metadata: Optional['Metadata']=None,
        cached_component: Optional['Component']=None, **kwargs: Any
        ) ->'Component':
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(cls, component_config: Dict[Text, Any], config: DazuConfig
        ) ->'Component':
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""
        language = config.language
        if not cls.can_handle_language(language):
            raise UnsupportedLanguageError(cls.name, language)
        return cls(component_config)
    <mask token>

    def train(self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
        ) ->None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) ->None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) ->Optional[Dict[
        Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(cls, component_meta: Dict[Text, Any], model_metadata:
        'Metadata') ->Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""
        return None

    def __getstate__(self) ->Any:
        d = self.__dict__.copy()
        if 'partial_processing_context' in d:
            del d['partial_processing_context']
        if 'partial_processing_pipeline' in d:
            del d['partial_processing_pipeline']
        return d

    def __eq__(self, other) ->bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(self, pipeline: List['Component'],
        context: Dict[Text, Any]) ->None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""
        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) ->Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""
        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info(
                'Failed to run partial processing due to missing pipeline.')
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) ->bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""
        if language is None or cls.language_list is None:
            return True
        return language in cls.language_list

<mask token>


class Component(Module):
    <mask token>
    <mask token>
    <mask token>
    <mask token>
    <mask token>

    def __init__(self, component_config: Optional[Dict[Text, Any]]=None
        ) ->None:
        if not component_config:
            component_config = {}
        component_config['name'] = type(self).name()
        self.component_config = override_defaults(self.defaults,
            component_config)
        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) ->List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(cls, component_config: Dict[Text, Any], model_dir: Optional[
        Text]=None, model_metadata: Optional['Metadata']=None,
        cached_component: Optional['Component']=None, **kwargs: Any
        ) ->'Component':
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(cls, component_config: Dict[Text, Any], config: DazuConfig
        ) ->'Component':
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""
        language = config.language
        if not cls.can_handle_language(language):
            raise UnsupportedLanguageError(cls.name, language)
        return cls(component_config)

    def provide_context(self) ->Optional[Dict[Text, Any]]:
        """Initialize this component for a new pipeline
        This function will be called before the training
        is started and before the first message is processed using
        the interpreter. The component gets the opportunity to
        add information to the context that is passed through
        the pipeline during training and message parsing. Most
        components do not need to implement this method.
        It's mostly used to initialize framework environments
        like MITIE and spacy
        (e.g. loading word vectors for the pipeline)."""

    def train(self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
        ) ->None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) ->None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) ->Optional[Dict[
        Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(cls, component_meta: Dict[Text, Any], model_metadata:
        'Metadata') ->Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""
        return None

    def __getstate__(self) ->Any:
        d = self.__dict__.copy()
        if 'partial_processing_context' in d:
            del d['partial_processing_context']
        if 'partial_processing_pipeline' in d:
            del d['partial_processing_pipeline']
        return d

    def __eq__(self, other) ->bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(self, pipeline: List['Component'],
        context: Dict[Text, Any]) ->None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""
        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) ->Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""
        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info(
                'Failed to run partial processing due to missing pipeline.')
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) ->bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""
        if language is None or cls.language_list is None:
            return True
        return language in cls.language_list

<mask token>


class Component(Module):
    """A component is a message processing unit in a pipeline.
    Components are collected sequentially in a pipeline. Each component
    is called one after another. This holds for
    initialization, training, persisting and loading the components.
    If a component comes first in a pipeline, its
    methods will be called first.
    E.g. to process an incoming message, the ``process`` method of
    each component will be called. During the processing
    (as well as the training, persisting and initialization)
    components can pass information to other components.
    The information is passed to other components by providing
    attributes to the so called pipeline context. The
    pipeline context contains all the information of the previous
    components a component can use to do its own
    processing. For example, a featurizer component can provide
    features that are used by another component down
    the pipeline to do intent classification."""
    provides = []
    requires = []
    defaults = {}
    language_list = None

    def __init__(self, component_config: Optional[Dict[Text, Any]]=None
        ) ->None:
        if not component_config:
            component_config = {}
        component_config['name'] = type(self).name()
        self.component_config = override_defaults(self.defaults,
            component_config)
        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) ->List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(cls, component_config: Dict[Text, Any], model_dir: Optional[
        Text]=None, model_metadata: Optional['Metadata']=None,
        cached_component: Optional['Component']=None, **kwargs: Any
        ) ->'Component':
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(cls, component_config: Dict[Text, Any], config: DazuConfig
        ) ->'Component':
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""
        language = config.language
        if not cls.can_handle_language(language):
            raise UnsupportedLanguageError(cls.name, language)
        return cls(component_config)

    def provide_context(self) ->Optional[Dict[Text, Any]]:
        """Initialize this component for a new pipeline
        This function will be called before the training
        is started and before the first message is processed using
        the interpreter. The component gets the opportunity to
        add information to the context that is passed through
        the pipeline during training and message parsing. Most
        components do not need to implement this method.
        It's mostly used to initialize framework environments
        like MITIE and spacy
        (e.g. loading word vectors for the pipeline)."""

    def train(self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
        ) ->None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) ->None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) ->Optional[Dict[
        Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(cls, component_meta: Dict[Text, Any], model_metadata:
        'Metadata') ->Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""
        return None

    def __getstate__(self) ->Any:
        d = self.__dict__.copy()
        if 'partial_processing_context' in d:
            del d['partial_processing_context']
        if 'partial_processing_pipeline' in d:
            del d['partial_processing_pipeline']
        return d

    def __eq__(self, other) ->bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(self, pipeline: List['Component'],
        context: Dict[Text, Any]) ->None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""
        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) ->Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""
        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info(
                'Failed to run partial processing due to missing pipeline.')
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) ->bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""
        if language is None or cls.language_list is None:
            return True
        return language in cls.language_list

<mask token>


class UnsupportedLanguageError(Exception):
    <mask token>

    def __init__(self, component: Text, language: Text) ->None:
        self.component = component
        self.language = language
        super().__init__(component, language)

    def __str__(self) ->Text:
        return (
            f"component '{self.component}' does not support language '{self.language}'."
            )


class Component(Module):
    """A component is a message processing unit in a pipeline.
    Components are collected sequentially in a pipeline. Each component
    is called one after another. This holds for
    initialization, training, persisting and loading the components.
    If a component comes first in a pipeline, its
    methods will be called first.
    E.g. to process an incoming message, the ``process`` method of
    each component will be called. During the processing
    (as well as the training, persisting and initialization)
    components can pass information to other components.
    The information is passed to other components by providing
    attributes to the so called pipeline context. The
    pipeline context contains all the information of the previous
    components a component can use to do its own
    processing. For example, a featurizer component can provide
    features that are used by another component down
    the pipeline to do intent classification."""
    provides = []
    requires = []
    defaults = {}
    language_list = None

    def __init__(self, component_config: Optional[Dict[Text, Any]]=None
        ) ->None:
        if not component_config:
            component_config = {}
        component_config['name'] = type(self).name()
        self.component_config = override_defaults(self.defaults,
            component_config)
        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) ->List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(cls, component_config: Dict[Text, Any], model_dir: Optional[
        Text]=None, model_metadata: Optional['Metadata']=None,
        cached_component: Optional['Component']=None, **kwargs: Any
        ) ->'Component':
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(cls, component_config: Dict[Text, Any], config: DazuConfig
        ) ->'Component':
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""
        language = config.language
        if not cls.can_handle_language(language):
            raise UnsupportedLanguageError(cls.name, language)
        return cls(component_config)

    def provide_context(self) ->Optional[Dict[Text, Any]]:
        """Initialize this component for a new pipeline
        This function will be called before the training
        is started and before the first message is processed using
        the interpreter. The component gets the opportunity to
        add information to the context that is passed through
        the pipeline during training and message parsing. Most
        components do not need to implement this method.
        It's mostly used to initialize framework environments
        like MITIE and spacy
        (e.g. loading word vectors for the pipeline)."""

    def train(self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
        ) ->None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) ->None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) ->Optional[Dict[
        Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(cls, component_meta: Dict[Text, Any], model_metadata:
        'Metadata') ->Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""
        return None

    def __getstate__(self) ->Any:
        d = self.__dict__.copy()
        if 'partial_processing_context' in d:
            del d['partial_processing_context']
        if 'partial_processing_pipeline' in d:
            del d['partial_processing_pipeline']
        return d

    def __eq__(self, other) ->bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(self, pipeline: List['Component'],
        context: Dict[Text, Any]) ->None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""
        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) ->Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""
        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info(
                'Failed to run partial processing due to missing pipeline.')
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) ->bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""
        if language is None or cls.language_list is None:
            return True
        return language in cls.language_list

import logging
from typing import Any, Dict, Hashable, List, Optional, Text

from dazu.config import DazuConfig, override_defaults
from dazu.typing import Message, Module, TrainingData
from dazu.typing.model import Metadata

logger = logging.getLogger(__name__)


class UnsupportedLanguageError(Exception):
    """Raised when a component is created but the language is not supported.

    :param component: component name
    :param language: language that component doesn't support
    """

    def __init__(self, component: Text, language: Text) -> None:
        self.component = component
        self.language = language

        super().__init__(component, language)

    def __str__(self) -> Text:
        return (
            f"component '{self.component}' does not support language '{self.language}'."
        )


# Component class inspired by RasaHQ/rasa
class Component(Module):
    """A component is a message processing unit in a pipeline.
    Components are collected sequentially in a pipeline. Each component
    is called one after another. This holds for
    initialization, training, persisting and loading the components.
    If a component comes first in a pipeline, its
    methods will be called first.
    E.g. to process an incoming message, the ``process`` method of
    each component will be called. During the processing
    (as well as the training, persisting and initialization)
    components can pass information to other components.
    The information is passed to other components by providing
    attributes to the so called pipeline context. The
    pipeline context contains all the information of the previous
    components a component can use to do its own
    processing. For example, a featurizer component can provide
    features that are used by another component down
    the pipeline to do intent classification."""

    # Defines what attributes the pipeline component will
    # provide when called. The listed attributes
    # should be set by the component on the message object
    # during test and train, e.g.
    # ```message.set("entities", [...])```
    provides = []

    # Which attributes on a message are required by this
    # component. E.g. if requires contains "tokens", than a
    # previous component in the pipeline needs to have "tokens"
    # within the above described `provides` property.
    # Use `any_of("option_1", "option_2")` to define that either
    # "option_1" or "option_2" needs to be present in the
    # provided properties from the previous components.
    requires = []

    # Defines the default configuration parameters of a component
    # these values can be overwritten in the pipeline configuration
    # of the model. The component should choose sensible defaults
    # and should be able to create reasonable results with the defaults.
    defaults = {}

    # Defines what language(s) this component can handle.
    # This attribute is designed for instance method: `can_handle_language`.
    # Default value is None which means it can handle all languages.
    # This is an important feature for backwards compatibility of components.
    language_list = None

    def __init__(self, component_config: Optional[Dict[Text, Any]] = None) -> None:

        if not component_config:
            component_config = {}

        # makes sure the name of the configuration is part of the config
        # this is important for e.g. persistence
        component_config["name"] = type(self).name()

        self.component_config = override_defaults(self.defaults, component_config)

        self.partial_processing_pipeline = None
        self.partial_processing_context = None

    @classmethod
    def required_packages(cls) -> List[Text]:
        """Specify which python packages need to be installed to use this
        component, e.g. ``["spacy"]``. More specifically, these should be
        importable python package names e.g. `sklearn` and not package
        names in the dependencies sense e.g. `scikit-learn`
        This list of requirements allows us to fail early during training
        if a required package is not installed."""
        return []

    @classmethod
    def load(
        cls,
        component_config: Dict[Text, Any],
        model_dir: Optional[Text] = None,
        model_metadata: Optional["Metadata"] = None,
        cached_component: Optional["Component"] = None,
        **kwargs: Any,
    ) -> "Component":
        """Load this component from file.
        After a component has been trained, it will be persisted by
        calling `persist`. When the pipeline gets loaded again,
        this component needs to be able to restore itself.
        Components can rely on any context attributes that are
        created by :meth:`components.Component.create`
        calls to components previous
        to this one."""
        if cached_component:
            return cached_component
        else:
            return cls(component_config)

    @classmethod
    def create(
        cls, component_config: Dict[Text, Any], config: DazuConfig
    ) -> "Component":
        """Creates this component (e.g. before a training is started).
        Method can access all configuration parameters."""

        # Check language supporting
        language = config.language
        if not cls.can_handle_language(language):
            # check failed
            raise UnsupportedLanguageError(cls.name, language)

        return cls(component_config)

    def provide_context(self) -> Optional[Dict[Text, Any]]:
        """Initialize this component for a new pipeline
        This function will be called before the training
        is started and before the first message is processed using
        the interpreter. The component gets the opportunity to
        add information to the context that is passed through
        the pipeline during training and message parsing. Most
        components do not need to implement this method.
        It's mostly used to initialize framework environments
        like MITIE and spacy
        (e.g. loading word vectors for the pipeline)."""

    def train(
        self, training_data: TrainingData, cfg: DazuConfig, **kwargs: Any
    ) -> None:
        """Train this component.
        This is the components chance to train itself provided
        with the training data. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.train`
        of components previous to this one."""

    def process(self, message: Message, **kwargs: Any) -> None:
        """Process an incoming message.
        This is the components chance to process an incoming
        message. The component can rely on
        any context attribute to be present, that gets created
        by a call to :meth:`rasa.nlu.components.Component.create`
        of ANY component and
        on any context attributes created by a call to
        :meth:`rasa.nlu.components.Component.process`
        of components previous to this one."""

    def persist(self, file_name: Text, model_dir: Text) -> Optional[Dict[Text, Any]]:
        """Persist this component to disk for future loading."""

    @classmethod
    def cache_key(
        cls, component_meta: Dict[Text, Any], model_metadata: "Metadata"
    ) -> Optional[Text]:
        """This key is used to cache components.
        If a component is unique to a model it should return None.
        Otherwise, an instantiation of the
        component will be reused for all models where the
        metadata creates the same key."""

        return None

    def __getstate__(self) -> Any:
        d = self.__dict__.copy()
        # these properties should not be pickled
        if "partial_processing_context" in d:
            del d["partial_processing_context"]
        if "partial_processing_pipeline" in d:
            del d["partial_processing_pipeline"]
        return d

    def __eq__(self, other) -> bool:
        return self.__dict__ == other.__dict__

    def prepare_partial_processing(
        self, pipeline: List["Component"], context: Dict[Text, Any]
    ) -> None:
        """Sets the pipeline and context used for partial processing.
        The pipeline should be a list of components that are
        previous to this one in the pipeline and
        have already finished their training (and can therefore
        be safely used to process messages)."""

        self.partial_processing_pipeline = pipeline
        self.partial_processing_context = context

    def partially_process(self, message: Message) -> Message:
        """Allows the component to process messages during
        training (e.g. external training data).
        The passed message will be processed by all components
        previous to this one in the pipeline."""

        if self.partial_processing_context is not None:
            for component in self.partial_processing_pipeline:
                component.process(message, **self.partial_processing_context)
        else:
            logger.info("Failed to run partial processing due to missing pipeline.")
        return message

    @classmethod
    def can_handle_language(cls, language: Hashable) -> bool:
        """Check if component supports a specific language.
        This method can be overwritten when needed. (e.g. dynamically
        determine which language is supported.)"""

        # if language_list is set to `None` it means: support all languages
        if language is None or cls.language_list is None:
            return True

        return language in cls.language_list
