files: # Copyright 2019 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  """Decorators for use with MycroftSkill methods"""
  
  
  def intent_handler(intent_parser):
      """Decorator for adding a method as an intent handler."""
  
      def real_decorator(func):
          # Store the intent_parser inside the function
          # This will be used later to call register_intent
          if not hasattr(func, 'intents'):
              func.intents = []
          func.intents.append(intent_parser)
          return func
  
      return real_decorator
  
  
  def intent_file_handler(intent_file):
      """Decorator for adding a method as an intent file handler.
  
      This decorator is deprecated, use intent_handler for the same effect.
      """
  
      def real_decorator(func):
          # Store the intent_file inside the function
          # This will be used later to call register_intent_file
          if not hasattr(func, 'intent_files'):
              func.intent_files = []
          func.intent_files.append(intent_file)
          return func
  
      return real_decorator
  
  
  def resting_screen_handler(name):
      """Decorator for adding a method as an resting screen handler.
  
      If selected will be shown on screen when device enters idle mode.
      """
  
      def real_decorator(func):
          # Store the resting information inside the function
          # This will be used later in register_resting_screen
          if not hasattr(func, 'resting_handler'):
              func.resting_handler = name
          return func
  
      return real_decorator
  
  
  [... the rest is omitted ...]
files: mycroft/skills/mycroft_skill/decorators.py
files: # Copyright 2019 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  """Decorators for use with MycroftSkill methods"""
  
  
  def intent_handler(intent_parser):
      """Decorator for adding a method as an intent handler."""
  
      def real_decorator(func):
          # Store the intent_parser inside the function
          # This will be used later to call register_intent
          if not hasattr(func, 'intents'):
              func.intents = []
          func.intents.append(intent_parser)
          return func
  
      return real_decorator
  
  
  def intent_file_handler(intent_file):
      """Decorator for adding a method as an intent file handler."""
  
      def real_decorator(func):
          # Store the intent_file inside the function
          # This will be used later to call register_intent_file
          if not hasattr(func, 'intent_files'):
              func.intent_files = []
          func.intent_files.append(intent_file)
          return func
  
      return real_decorator
  
  
  def resting_screen_handler(name):
      """Decorator for adding a method as an resting screen handler.
  
      If selected will be shown on screen when device enters idle mode.
      """
  
      def real_decorator(func):
          # Store the resting information inside the function
          # This will be used later in register_resting_screen
          if not hasattr(func, 'resting_handler'):
              func.resting_handler = name
          return func
  
      return real_decorator
  
  
  [... the rest is omitted ...]
files: mycroft/skills/mycroft_skill/decorators.py
files: # Copyright 2019 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  """ Collection of functions relating to the implementation of mycroft skills.
  """
  import sys
  import re
  import traceback
  import inspect
  from inspect import ismethod, signature
  import collections
  import csv
  from itertools import chain
  from adapt.intent import Intent, IntentBuilder
  from os import walk
  from os.path import join, abspath, dirname, basename, exists
  from threading import Event, Timer
  
  from mycroft import dialog
  from mycroft.api import DeviceApi
  from mycroft.audio import wait_while_speaking
  from mycroft.enclosure.api import EnclosureAPI
  from mycroft.enclosure.gui import SkillGUI
  from mycroft.configuration import Configuration
  from mycroft.dialog import DialogLoader
  from mycroft.filesystem import FileSystemAccess
  from mycroft.messagebus.message import Message
  from mycroft.metrics import report_metric, report_timing, Stopwatch
  from mycroft.util import (camel_case_split,
                            resolve_resource_file,
                            play_audio_file)
  from mycroft.util.log import LOG
  
  from ..settings import SkillSettings
  from ..skill_data import (load_vocabulary, load_regex, to_alnum,
                            munge_regex, munge_intent_parser, read_vocab_file)
  from ..event_scheduler import EventSchedulerInterface
  
  
  def simple_trace(stack_trace):
      """Generate a simplified traceback.
  
      Arguments:
          stack_trace: Stack trace to simplify
  
      Returns: (str) Simplified stack trace.
      """
      stack_trace = stack_trace[:-1]
      tb = "Traceback:\n"
      for line in stack_trace:
          if line.strip():
              tb += line
      return tb
  
  
  def get_non_properties(obj):
      """Get attibutes that are not properties from object.
  
      Will return members of object class along with bases down to MycroftSkill.
  
      Arguments:
          obj:    object to scan
  
      Returns:
          Set of attributes that are not a property.
      """
  
      def check_class(cls):
          """Find all non-properties in a class."""
          # Current class
          d = cls.__dict__
          np = [k for k in d if not isinstance(d[k], property)]
          # Recurse through base classes excluding MycroftSkill and object
          for b in [b for b in cls.__bases__ if b not in (object, MycroftSkill)]:
              np += check_class(b)
          return np
  
      return set(check_class(obj.__class__))
  
  
  def dig_for_message():
      """Dig Through the stack for message."""
      stack = inspect.stack()
      # Limit search to 10 frames back
      stack = stack if len(stack) < 10 else stack[:10]
      local_vars = [frame[0].f_locals for frame in stack]
      for l in local_vars:
          if 'message' in l and isinstance(l['message'], Message):
              return l['message']
  
  
  def unmunge_message(message, skill_id):
      """Restore message keywords by removing the Letterified skill ID.
  
      Arguments:
          message (Message): Intent result message
          skill_id (str): skill identifier
  
      Returns:
          Message without clear keywords
      """
      if isinstance(message, Message) and isinstance(message.data, dict):
          skill_id = to_alnum(skill_id)
          for key in list(message.data.keys()):
              if key.startswith(skill_id):
                  # replace the munged key with the real one
                  new_key = key[len(skill_id):]
                  message.data[new_key] = message.data.pop(key)
  
      return message
  
  
  def get_handler_name(handler):
      """Name (including class if available) of handler function.
  
      Arguments:
          handler (function): Function to be named
  
      Returns:
          string: handler name as string
      """
      if '__self__' in dir(handler) and 'name' in dir(handler.__self__):
          return handler.__self__.name + '.' + handler.__name__
      else:
          return handler.__name__
  
  
  class MycroftSkill:
      """Base class for mycroft skills providing common behaviour and parameters
      to all Skill implementations.
  
      For information on how to get started with creating mycroft skills see
      https://https://mycroft.ai/documentation/skills/introduction-developing-skills/
  
      Arguments:
          name (str): skill name
          bus (MycroftWebsocketClient): Optional bus connection
          use_settings (bool): Set to false to not use skill settings at all
      """
      def __init__(self, name=None, bus=None, use_settings=True):
          self.name = name or self.__class__.__name__
          self.resting_name = None
          # Get directory of skill
          self._dir = dirname(abspath(sys.modules[self.__module__].__file__))
          if use_settings:
              self.settings = SkillSettings(self._dir, self.name)
          else:
              self.settings = None
  
          self.gui = SkillGUI(self)
  
          self._bus = None
          self._enclosure = None
          self.bind(bus)
          #: Mycroft global configuration. (dict)
          self.config_core = Configuration.get()
          # TODO: 19.08 - Remove
          self._config = self.config_core.get(self.name) or {}
          self.dialog_renderer = None
          self.root_dir = self._dir  #: skill root directory
  
          #: Filesystem access to skill specific folder.
          #: See mycroft.filesystem for details.
          self.file_system = FileSystemAccess(join('skills', self.name))
          self.registered_intents = []
          self.log = LOG.create_logger(self.name)  #: Skill logger instance
          self.reload_skill = True  #: allow reloading (default True)
          self.events = []
          self.skill_id = ''  # will be set from the path, so guaranteed unique
          self.voc_match_cache = {}
  
          # Delegator classes
          self.event_scheduler = EventSchedulerInterface(self)
  
      @property
      def enclosure(self):
          if self._enclosure:
              return self._enclosure
          else:
              LOG.error("Skill not fully initialized. Move code " +
                        "from  __init__() to initialize() to correct this.")
              LOG.error(simple_trace(traceback.format_stack()))
              raise Exception("Accessed MycroftSkill.enclosure in __init__")
  
      @property
      def bus(self):
          if self._bus:
              return self._bus
          else:
              LOG.error("Skill not fully initialized. Move code " +
                        "from __init__() to initialize() to correct this.")
              LOG.error(simple_trace(traceback.format_stack()))
              raise Exception("Accessed MycroftSkill.bus in __init__")
  
      @property
      def config(self):
          """Provide deprecation warning when accessing config.
          TODO: Remove in 19.08
          """
          LOG.warning('self.config is deprecated.  Switch to using '
                      'self.setting["whatever"] within your skill.')
          LOG.warning(simple_trace(traceback.format_stack()))
          return self._config
  
      @property
      def location(self):
          """Get the JSON data struction holding location information."""
          # TODO: Allow Enclosure to override this for devices that
          # contain a GPS.
          return self.config_core.get('location')
  
      @property
      def location_pretty(self):
          """Get a more 'human' version of the location as a string."""
          loc = self.location
          if type(loc) is dict and loc["city"]:
              return loc["city"]["name"]
          return None
  
      @property
      def location_timezone(self):
          """Get the timezone code, such as 'America/Los_Angeles'"""
          loc = self.location
          if type(loc) is dict and loc["timezone"]:
              return loc["timezone"]["code"]
          return None
  
      @property
      def lang(self):
          """Get the configured language."""
          return self.config_core.get('lang')
  
      def bind(self, bus):
          """Register messagebus emitter with skill.
  
          Arguments:
              bus: Mycroft messagebus connection
          """
          if bus:
              self._bus = bus
              self._enclosure = EnclosureAPI(bus, self.name)
              self.add_event('mycroft.stop', self.__handle_stop)
              self.add_event('mycroft.skill.enable_intent',
                             self.handle_enable_intent)
              self.add_event('mycroft.skill.disable_intent',
                             self.handle_disable_intent)
              self.add_event("mycroft.skill.set_cross_context",
                             self.handle_set_cross_context)
              self.add_event("mycroft.skill.remove_cross_context",
                             self.handle_remove_cross_context)
  
              # Trigger settings update if requested
              self._add_light_event('mycroft.skills.settings.update',
                                    self.settings.run_poll)
              # Trigger Settings meta upload on pairing complete
              self._add_light_event('mycroft.paired', self.settings.run_poll)
  
              # Intialize the SkillGui
              self.gui.setup_default_handlers()
  
      def _add_light_event(self, msg_type, func):
          """This adds an event handler that will automatically be unregistered
          when the skill shutsdown but none of the metrics or error feedback will
          be triggered.
  
          Arguments:
              msg_tupe (str): Message type
              func (Function): function to be invoked
          """
          self.bus.on(msg_type, func)
          self.events.append((msg_type, func))
  
      def detach(self):
          for (name, intent) in self.registered_intents:
              name = str(self.skill_id) + ':' + name
              self.bus.emit(Message("detach_intent", {"intent_name": name}))
  
      def initialize(self):
          """Perform any final setup needed for the skill.
  
          Invoked after the skill is fully constructed and registered with the
          system.
          """
          pass
  
      def get_intro_message(self):
          """Get a message to speak on first load of the skill.
  
          Useful for post-install setup instructions.
  
          Returns:
              str: message that will be spoken to the user
          """
          return None
  
      def converse(self, utterances, lang=None):
          """Handle conversation.
  
          This method gets a peek at utterances before the normal intent
          handling process after a skill has been invoked once.
  
          To use, override the converse() method and return True to
          indicate that the utterance has been handled.
  
          Arguments:
              utterances (list): The utterances from the user.  If there are
                                 multiple utterances, consider them all to be
                                 transcription possibilities.  Commonly, the
                                 first entry is the user utt and the second
                                 is normalized() version of the first utterance
              lang:       language the utterance is in, None for default
  
          Returns:
              bool: True if an utterance was handled, otherwise False
          """
          return False
  
      def __get_response(self):
          """Helper to get a reponse from the user
  
          Returns:
              str: user's response or None on a timeout
          """
          event = Event()
  
          def converse(utterances, lang=None):
              converse.response = utterances[0] if utterances else None
              event.set()
              return True
  
          # install a temporary conversation handler
          self.make_active()
          converse.response = None
          default_converse = self.converse
          self.converse = converse
          event.wait(15)  # 10 for listener, 5 for SST, then timeout
          self.converse = default_converse
          return converse.response
  
      def get_response(self, dialog='', data=None, validator=None
  
  [... the rest is omitted ...]
files: mycroft/skills/mycroft_skill/mycroft_skill.py
files: # Copyright 2019 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  """ Collection of functions relating to the implementation of mycroft skills.
  """
  import sys
  import re
  import traceback
  import inspect
  from inspect import ismethod, signature
  import collections
  import csv
  from itertools import chain
  from adapt.intent import Intent, IntentBuilder
  from os import walk
  from os.path import join, abspath, dirname, basename, exists
  from threading import Event, Timer
  
  from mycroft import dialog
  from mycroft.api import DeviceApi
  from mycroft.audio import wait_while_speaking
  from mycroft.enclosure.api import EnclosureAPI
  from mycroft.enclosure.gui import SkillGUI
  from mycroft.configuration import Configuration
  from mycroft.dialog import DialogLoader
  from mycroft.filesystem import FileSystemAccess
  from mycroft.messagebus.message import Message
  from mycroft.metrics import report_metric, report_timing, Stopwatch
  from mycroft.util import (camel_case_split,
                            resolve_resource_file,
                            play_audio_file)
  from mycroft.util.log import LOG
  
  from ..settings import SkillSettings
  from ..skill_data import (load_vocabulary, load_regex, to_alnum,
                            munge_regex, munge_intent_parser, read_vocab_file)
  from ..event_scheduler import EventSchedulerInterface
  
  
  def simple_trace(stack_trace):
      """Generate a simplified traceback.
  
      Arguments:
          stack_trace: Stack trace to simplify
  
      Returns: (str) Simplified stack trace.
      """
      stack_trace = stack_trace[:-1]
      tb = "Traceback:\n"
      for line in stack_trace:
          if line.strip():
              tb += line
      return tb
  
  
  def get_non_properties(obj):
      """Get attibutes that are not properties from object.
  
      Will return members of object class along with bases down to MycroftSkill.
  
      Arguments:
          obj:    object to scan
  
      Returns:
          Set of attributes that are not a property.
      """
  
      def check_class(cls):
          """Find all non-properties in a class."""
          # Current class
          d = cls.__dict__
          np = [k for k in d if not isinstance(d[k], property)]
          # Recurse through base classes excluding MycroftSkill and object
          for b in [b for b in cls.__bases__ if b not in (object, MycroftSkill)]:
              np += check_class(b)
          return np
  
      return set(check_class(obj.__class__))
  
  
  def dig_for_message():
      """Dig Through the stack for message."""
      stack = inspect.stack()
      # Limit search to 10 frames back
      stack = stack if len(stack) < 10 else stack[:10]
      local_vars = [frame[0].f_locals for frame in stack]
      for l in local_vars:
          if 'message' in l and isinstance(l['message'], Message):
              return l['message']
  
  
  def unmunge_message(message, skill_id):
      """Restore message keywords by removing the Letterified skill ID.
  
      Arguments:
          message (Message): Intent result message
          skill_id (str): skill identifier
  
      Returns:
          Message without clear keywords
      """
      if isinstance(message, Message) and isinstance(message.data, dict):
          skill_id = to_alnum(skill_id)
          for key in list(message.data.keys()):
              if key.startswith(skill_id):
                  # replace the munged key with the real one
                  new_key = key[len(skill_id):]
                  message.data[new_key] = message.data.pop(key)
  
      return message
  
  
  def get_handler_name(handler):
      """Name (including class if available) of handler function.
  
      Arguments:
          handler (function): Function to be named
  
      Returns:
          string: handler name as string
      """
      if '__self__' in dir(handler) and 'name' in dir(handler.__self__):
          return handler.__self__.name + '.' + handler.__name__
      else:
          return handler.__name__
  
  
  class MycroftSkill:
      """Base class for mycroft skills providing common behaviour and parameters
      to all Skill implementations.
  
      For information on how to get started with creating mycroft skills see
      https://https://mycroft.ai/documentation/skills/introduction-developing-skills/
  
      Arguments:
          name (str): skill name
          bus (MycroftWebsocketClient): Optional bus connection
          use_settings (bool): Set to false to not use skill settings at all
      """
      def __init__(self, name=None, bus=None, use_settings=True):
          self.name = name or self.__class__.__name__
          self.resting_name = None
          # Get directory of skill
          self._dir = dirname(abspath(sys.modules[self.__module__].__file__))
          if use_settings:
              self.settings = SkillSettings(self._dir, self.name)
          else:
              self.settings = None
  
          self.gui = SkillGUI(self)
  
          self._bus = None
          self._enclosure = None
          self.bind(bus)
          #: Mycroft global configuration. (dict)
          self.config_core = Configuration.get()
          # TODO: 19.08 - Remove
          self._config = self.config_core.get(self.name) or {}
          self.dialog_renderer = None
          self.root_dir = self._dir  #: skill root directory
  
          #: Filesystem access to skill specific folder.
          #: See mycroft.filesystem for details.
          self.file_system = FileSystemAccess(join('skills', self.name))
          self.registered_intents = []
          self.log = LOG.create_logger(self.name)  #: Skill logger instance
          self.reload_skill = True  #: allow reloading (default True)
          self.events = []
          self.skill_id = ''  # will be set from the path, so guaranteed unique
          self.voc_match_cache = {}
  
          # Delegator classes
          self.event_scheduler = EventSchedulerInterface(self)
  
      @property
      def enclosure(self):
          if self._enclosure:
              return self._enclosure
          else:
              LOG.error("Skill not fully initialized. Move code " +
                        "from  __init__() to initialize() to correct this.")
              LOG.error(simple_trace(traceback.format_stack()))
              raise Exception("Accessed MycroftSkill.enclosure in __init__")
  
      @property
      def bus(self):
          if self._bus:
              return self._bus
          else:
              LOG.error("Skill not fully initialized. Move code " +
                        "from __init__() to initialize() to correct this.")
              LOG.error(simple_trace(traceback.format_stack()))
              raise Exception("Accessed MycroftSkill.bus in __init__")
  
      @property
      def config(self):
          """Provide deprecation warning when accessing config.
          TODO: Remove in 19.08
          """
          LOG.warning('self.config is deprecated.  Switch to using '
                      'self.setting["whatever"] within your skill.')
          LOG.warning(simple_trace(traceback.format_stack()))
          return self._config
  
      @property
      def location(self):
          """Get the JSON data struction holding location information."""
          # TODO: Allow Enclosure to override this for devices that
          # contain a GPS.
          return self.config_core.get('location')
  
      @property
      def location_pretty(self):
          """Get a more 'human' version of the location as a string."""
          loc = self.location
          if type(loc) is dict and loc["city"]:
              return loc["city"]["name"]
          return None
  
      @property
      def location_timezone(self):
          """Get the timezone code, such as 'America/Los_Angeles'"""
          loc = self.location
          if type(loc) is dict and loc["timezone"]:
              return loc["timezone"]["code"]
          return None
  
      @property
      def lang(self):
          """Get the configured language."""
          return self.config_core.get('lang')
  
      def bind(self, bus):
          """Register messagebus emitter with skill.
  
          Arguments:
              bus: Mycroft messagebus connection
          """
          if bus:
              self._bus = bus
              self._enclosure = EnclosureAPI(bus, self.name)
              self.add_event('mycroft.stop', self.__handle_stop)
              self.add_event('mycroft.skill.enable_intent',
                             self.handle_enable_intent)
              self.add_event('mycroft.skill.disable_intent',
                             self.handle_disable_intent)
              self.add_event("mycroft.skill.set_cross_context",
                             self.handle_set_cross_context)
              self.add_event("mycroft.skill.remove_cross_context",
                             self.handle_remove_cross_context)
  
              # Trigger settings update if requested
              self._add_light_event('mycroft.skills.settings.update',
                                    self.settings.run_poll)
              # Trigger Settings meta upload on pairing complete
              self._add_light_event('mycroft.paired', self.settings.run_poll)
  
              # Intialize the SkillGui
              self.gui.setup_default_handlers()
  
      def _add_light_event(self, msg_type, func):
          """This adds an event handler that will automatically be unregistered
          when the skill shutsdown but none of the metrics or error feedback will
          be triggered.
  
          Arguments:
              msg_tupe (str): Message type
              func (Function): function to be invoked
          """
          self.bus.on(msg_type, func)
          self.events.append((msg_type, func))
  
      def detach(self):
          for (name, intent) in self.registered_intents:
              name = str(self.skill_id) + ':' + name
              self.bus.emit(Message("detach_intent", {"intent_name": name}))
  
      def initialize(self):
          """Perform any final setup needed for the skill.
  
          Invoked after the skill is fully constructed and registered with the
          system.
          """
          pass
  
      def get_intro_message(self):
          """Get a message to speak on first load of the skill.
  
          Useful for post-install setup instructions.
  
          Returns:
              str: message that will be spoken to the user
          """
          return None
  
      def converse(self, utterances, lang=None):
          """Handle conversation.
  
          This method gets a peek at utterances before the normal intent
          handling process after a skill has been invoked once.
  
          To use, override the converse() method and return True to
          indicate that the utterance has been handled.
  
          Arguments:
              utterances (list): The utterances from the user.  If there are
                                 multiple utterances, consider them all to be
                                 transcription possibilities.  Commonly, the
                                 first entry is the user utt and the second
                                 is normalized() version of the first utterance
              lang:       language the utterance is in, None for default
  
          Returns:
              bool: True if an utterance was handled, otherwise False
          """
          return False
  
      def __get_response(self):
          """Helper to get a reponse from the user
  
          Returns:
              str: user's response or None on a timeout
          """
          event = Event()
  
          def converse(utterances, lang=None):
              converse.response = utterances[0] if utterances else None
              event.set()
              return True
  
          # install a temporary conversation handler
          self.make_active()
          converse.response = None
          default_converse = self.converse
          self.converse = converse
          event.wait(15)  # 10 for listener, 5 for SST, then timeout
          self.converse = default_converse
          return converse.response
  
      def get_response(self, dialog='', data=None, validator=None
  
  [... the rest is omitted ...]
files: mycroft/skills/mycroft_skill/mycroft_skill.py
files: # -*- coding: utf-8 -*-
  # Copyright 2017 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  import sys
  import unittest
  
  from unittest.mock import MagicMock, patch
  from adapt.intent import IntentBuilder
  from os.path import join, dirname, abspath
  from re import error
  from datetime import datetime
  
  from mycroft.configuration import Configuration
  from mycroft.messagebus.message import Message
  from mycroft.skills.skill_data import (load_regex_from_file, load_regex,
                                         load_vocab_from_file, load_vocabulary)
  from mycroft.skills.core import (MycroftSkill, load_skill,
                                   create_skill_descriptor,
                                   resting_screen_handler)
  from mycroft.skills.intent_service import open_intent_envelope
  
  from test.util import base_config
  
  BASE_CONF = base_config()
  
  
  class MockEmitter(object):
      def __init__(self):
          self.reset()
  
      def emit(self, message):
          self.types.append(message.type)
          self.results.append(message.data)
  
      def get_types(self):
          return self.types
  
      def get_results(self):
          return self.results
  
      def on(self, event, f):
          pass
  
      def reset(self):
          self.types = []
          self.results = []
  
  
  class FunctionTest(unittest.TestCase):
      def test_resting_screen_handler(self):
          class T(MycroftSkill):
              def __init__(self):
                  self.name = 'TestObject'
  
              @resting_screen_handler('humbug')
              def f(self):
                  pass
  
          test_class = T()
          self.assertTrue('resting_handler' in dir(test_class.f))
          self.assertEquals(test_class.f.resting_handler, 'humbug')
  
  
  class MycroftSkillTest(unittest.TestCase):
      emitter = MockEmitter()
      regex_path = abspath(join(dirname(__file__), '../regex_test'))
      vocab_path = abspath(join(dirname(__file__), '../vocab_test'))
  
      def setUp(self):
          self.emitter.reset()
  
      def check_vocab_from_file(self, filename, vocab_type=None,
                                result_list=None):
          result_list = result_list or []
          load_vocab_from_file(join(self.vocab_path, filename), vocab_type,
                               self.emitter)
          self.check_emitter(result_list)
  
      def check_regex_from_file(self, filename, result_list=None):
          result_list = result_list or []
          regex_file = join(self.regex_path, filename)
          load_regex_from_file(regex_file, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_vocab(self, path, result_list=None):
          result_list = result_list or []
          load_vocabulary(path, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_regex(self, path, result_list=None):
          result_list = result_list or []
          load_regex(path, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_emitter(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_vocab')
          self.assertEqual(sorted(self.emitter.get_results(),
                                  key=lambda d: sorted(d.items())),
                           sorted(result_list, key=lambda d: sorted(d.items())))
          self.emitter.reset()
  
      def test_load_regex_from_file_single(self):
          self.check_regex_from_file('valid/single.rx',
                                     [{'regex': '(?P<ASingleTest>.*)'}])
  
      def test_load_regex_from_file_multiple(self):
          self.check_regex_from_file('valid/multiple.rx',
                                     [{'regex': '(?P<AMultipleTest1>.*)'},
                                      {'regex': '(?P<AMultipleTest2>.*)'}])
  
      def test_load_regex_from_file_none(self):
          self.check_regex_from_file('invalid/none.rx')
  
      def test_load_regex_from_file_invalid(self):
          with self.assertRaises(error):
              self.check_regex_from_file('invalid/invalid.rx')
  
      def test_load_regex_from_file_does_not_exist(self):
          with self.assertRaises(IOError):
              self.check_regex_from_file('does_not_exist.rx')
  
      def test_load_regex_full(self):
          self.check_regex(join(self.regex_path, 'valid'),
                           [{'regex': '(?P<AMultipleTest1>.*)'},
                            {'regex': '(?P<AMultipleTest2>.*)'},
                            {'regex': '(?P<ASingleTest>.*)'}])
  
      def test_load_regex_empty(self):
          self.check_regex(join(dirname(__file__),
                                'empty_dir'))
  
      def test_load_regex_fail(self):
          try:
              self.check_regex(join(dirname(__file__),
                                    'regex_test_fail'))
          except OSError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_load_vocab_from_file_single(self):
          self.check_vocab_from_file('valid/single.voc', 'test_type',
                                     [{'start': 'test', 'end': 'test_type'}])
  
      def test_load_vocab_from_file_single_alias(self):
          self.check_vocab_from_file('valid/singlealias.voc', 'test_type',
                                     [{'start': 'water', 'end': 'test_type'},
                                      {'start': 'watering', 'end': 'test_type',
                                       'alias_of': 'water'}])
  
      def test_load_vocab_from_file_multiple(self):
          self.check_vocab_from_file('valid/multiple.voc', 'test_type',
                                     [{'start': 'animal', 'end': 'test_type'},
                                      {'start': 'animals', 'end': 'test_type'}])
  
      def test_load_vocab_from_file_multiple_alias(self):
          self.check_vocab_from_file('valid/multiplealias.voc', 'test_type',
                                     [{'start': 'chair', 'end': 'test_type'},
                                      {'start': 'chairs', 'end': 'test_type',
                                       'alias_of': 'chair'},
                                      {'start': 'table', 'end': 'test_type'},
                                      {'start': 'tables', 'end': 'test_type',
                                       'alias_of': 'table'}])
  
      def test_load_vocab_from_file_none(self):
          self.check_vocab_from_file('none.voc')
  
      def test_load_vocab_from_file_does_not_exist(self):
          try:
              self.check_vocab_from_file('does_not_exist.voc')
          except IOError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_load_vocab_full(self):
          self.check_vocab(join(self.vocab_path, 'valid'),
                           [{'start': 'test', 'end': 'Asingle'},
                            {'start': 'water', 'end': 'Asinglealias'},
                            {'start': 'watering', 'end': 'Asinglealias',
                             'alias_of': 'water'},
                            {'start': 'animal', 'end': 'Amultiple'},
                            {'start': 'animals', 'end': 'Amultiple'},
                            {'start': 'chair', 'end': 'Amultiplealias'},
                            {'start': 'chairs', 'end': 'Amultiplealias',
                             'alias_of': 'chair'},
                            {'start': 'table', 'end': 'Amultiplealias'},
                            {'start': 'tables', 'end': 'Amultiplealias',
                             'alias_of': 'table'}])
  
      def test_load_vocab_empty(self):
          self.check_vocab(join(dirname(__file__), 'empty_dir'))
  
      def test_load_vocab_fail(self):
          try:
              self.check_regex(join(dirname(__file__),
                                    'vocab_test_fail'))
          except OSError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_open_envelope(self):
          name = 'Jerome'
          intent = IntentBuilder(name).require('Keyword')
          intent.name = name
          m = Message("register_intent", intent.__dict__)
          unpacked_intent = open_intent_envelope(m)
          self.assertEqual(intent.__dict__, unpacked_intent.__dict__)
  
      def check_detach_intent(self):
          self.assertTrue(len(self.emitter.get_types()) > 0)
          for msg_type in self.emitter.get_types():
              self.assertEqual(msg_type, 'detach_intent')
          self.emitter.reset()
  
      def check_register_intent(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_intent')
          self.assertEqual(sorted(self.emitter.get_results()),
                           sorted(result_list))
          self.emitter.reset()
  
      def check_register_vocabulary(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_vocab')
          self.assertEqual(sorted(self.emitter.get_results()),
                           sorted(result_list))
          self.emitter.reset()
  
      def test_register_intent(self):
          # Test register Intent object
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          s = SimpleSkill2()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
  
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          with self.assertRaises(ValueError):
              s = SimpleSkill3()
              s.bind(self.emitter)
              s.initialize()
  
      def test_enable_disable_intent(self):
          """Test disable/enable intent."""
          # Setup basic test
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
          self.check_register_intent(expected)
  
          # Test disable/enable cycle
          s.disable_intent('a')
          self.check_detach_intent()
          s.enable_intent('a')
          self.check_register_intent(expected)
  
      def test_register_vocab(self):
          """Test disable/enable intent."""
          # Setup basic test
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
  
          # Normal vocaubulary
          self.emitter.reset()
          expected = [{'start': 'hello', 'end': 'AHelloKeyword'}]
          s.register_vocabulary('hello', 'HelloKeyword')
          self.check_register_vocabulary(expected)
          # Regex
          s.register_regex('weird (?P<Weird>.+) stuff')
          expected = [{'regex': 'weird (?P<AWeird>.+) stuff'}]
          self.check_register_vocabulary(expected)
  
      def check_register_object_file(self, types_list, result_list):
          self.assertEqual(sorted(self.emitter.get_types()),
                           sorted(types_list))
          self.assertEqual(sorted(self.emitter.get_results(),
                                  key=lambda d: sorted(d.items())),
                           sorted(result_list, key=lambda d: sorted(d.items())))
          self.emitter.reset()
  
      def test_register_intent_file(self):
          self._test_intent_file(SimpleSkill4())
  
      def test_register_intent_intent_file
  
  [... the rest is omitted ...]
files: test/unittests/skills/test_core.py
files: # -*- coding: utf-8 -*-
  # Copyright 2017 Mycroft AI Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  import sys
  import unittest
  
  from unittest.mock import MagicMock, patch
  from adapt.intent import IntentBuilder
  from os.path import join, dirname, abspath
  from re import error
  from datetime import datetime
  
  from mycroft.configuration import Configuration
  from mycroft.messagebus.message import Message
  from mycroft.skills.skill_data import (load_regex_from_file, load_regex,
                                         load_vocab_from_file, load_vocabulary)
  from mycroft.skills.core import (MycroftSkill, load_skill,
                                   create_skill_descriptor,
                                   resting_screen_handler)
  from mycroft.skills.intent_service import open_intent_envelope
  
  from test.util import base_config
  
  BASE_CONF = base_config()
  
  
  class MockEmitter(object):
      def __init__(self):
          self.reset()
  
      def emit(self, message):
          self.types.append(message.type)
          self.results.append(message.data)
  
      def get_types(self):
          return self.types
  
      def get_results(self):
          return self.results
  
      def on(self, event, f):
          pass
  
      def reset(self):
          self.types = []
          self.results = []
  
  
  class FunctionTest(unittest.TestCase):
      def test_resting_screen_handler(self):
          class T(MycroftSkill):
              def __init__(self):
                  self.name = 'TestObject'
  
              @resting_screen_handler('humbug')
              def f(self):
                  pass
  
          test_class = T()
          self.assertTrue('resting_handler' in dir(test_class.f))
          self.assertEquals(test_class.f.resting_handler, 'humbug')
  
  
  class MycroftSkillTest(unittest.TestCase):
      emitter = MockEmitter()
      regex_path = abspath(join(dirname(__file__), '../regex_test'))
      vocab_path = abspath(join(dirname(__file__), '../vocab_test'))
  
      def setUp(self):
          self.emitter.reset()
  
      def check_vocab_from_file(self, filename, vocab_type=None,
                                result_list=None):
          result_list = result_list or []
          load_vocab_from_file(join(self.vocab_path, filename), vocab_type,
                               self.emitter)
          self.check_emitter(result_list)
  
      def check_regex_from_file(self, filename, result_list=None):
          result_list = result_list or []
          regex_file = join(self.regex_path, filename)
          load_regex_from_file(regex_file, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_vocab(self, path, result_list=None):
          result_list = result_list or []
          load_vocabulary(path, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_regex(self, path, result_list=None):
          result_list = result_list or []
          load_regex(path, self.emitter, 'A')
          self.check_emitter(result_list)
  
      def check_emitter(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_vocab')
          self.assertEqual(sorted(self.emitter.get_results(),
                                  key=lambda d: sorted(d.items())),
                           sorted(result_list, key=lambda d: sorted(d.items())))
          self.emitter.reset()
  
      def test_load_regex_from_file_single(self):
          self.check_regex_from_file('valid/single.rx',
                                     [{'regex': '(?P<ASingleTest>.*)'}])
  
      def test_load_regex_from_file_multiple(self):
          self.check_regex_from_file('valid/multiple.rx',
                                     [{'regex': '(?P<AMultipleTest1>.*)'},
                                      {'regex': '(?P<AMultipleTest2>.*)'}])
  
      def test_load_regex_from_file_none(self):
          self.check_regex_from_file('invalid/none.rx')
  
      def test_load_regex_from_file_invalid(self):
          with self.assertRaises(error):
              self.check_regex_from_file('invalid/invalid.rx')
  
      def test_load_regex_from_file_does_not_exist(self):
          with self.assertRaises(IOError):
              self.check_regex_from_file('does_not_exist.rx')
  
      def test_load_regex_full(self):
          self.check_regex(join(self.regex_path, 'valid'),
                           [{'regex': '(?P<AMultipleTest1>.*)'},
                            {'regex': '(?P<AMultipleTest2>.*)'},
                            {'regex': '(?P<ASingleTest>.*)'}])
  
      def test_load_regex_empty(self):
          self.check_regex(join(dirname(__file__),
                                'empty_dir'))
  
      def test_load_regex_fail(self):
          try:
              self.check_regex(join(dirname(__file__),
                                    'regex_test_fail'))
          except OSError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_load_vocab_from_file_single(self):
          self.check_vocab_from_file('valid/single.voc', 'test_type',
                                     [{'start': 'test', 'end': 'test_type'}])
  
      def test_load_vocab_from_file_single_alias(self):
          self.check_vocab_from_file('valid/singlealias.voc', 'test_type',
                                     [{'start': 'water', 'end': 'test_type'},
                                      {'start': 'watering', 'end': 'test_type',
                                       'alias_of': 'water'}])
  
      def test_load_vocab_from_file_multiple(self):
          self.check_vocab_from_file('valid/multiple.voc', 'test_type',
                                     [{'start': 'animal', 'end': 'test_type'},
                                      {'start': 'animals', 'end': 'test_type'}])
  
      def test_load_vocab_from_file_multiple_alias(self):
          self.check_vocab_from_file('valid/multiplealias.voc', 'test_type',
                                     [{'start': 'chair', 'end': 'test_type'},
                                      {'start': 'chairs', 'end': 'test_type',
                                       'alias_of': 'chair'},
                                      {'start': 'table', 'end': 'test_type'},
                                      {'start': 'tables', 'end': 'test_type',
                                       'alias_of': 'table'}])
  
      def test_load_vocab_from_file_none(self):
          self.check_vocab_from_file('none.voc')
  
      def test_load_vocab_from_file_does_not_exist(self):
          try:
              self.check_vocab_from_file('does_not_exist.voc')
          except IOError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_load_vocab_full(self):
          self.check_vocab(join(self.vocab_path, 'valid'),
                           [{'start': 'test', 'end': 'Asingle'},
                            {'start': 'water', 'end': 'Asinglealias'},
                            {'start': 'watering', 'end': 'Asinglealias',
                             'alias_of': 'water'},
                            {'start': 'animal', 'end': 'Amultiple'},
                            {'start': 'animals', 'end': 'Amultiple'},
                            {'start': 'chair', 'end': 'Amultiplealias'},
                            {'start': 'chairs', 'end': 'Amultiplealias',
                             'alias_of': 'chair'},
                            {'start': 'table', 'end': 'Amultiplealias'},
                            {'start': 'tables', 'end': 'Amultiplealias',
                             'alias_of': 'table'}])
  
      def test_load_vocab_empty(self):
          self.check_vocab(join(dirname(__file__), 'empty_dir'))
  
      def test_load_vocab_fail(self):
          try:
              self.check_regex(join(dirname(__file__),
                                    'vocab_test_fail'))
          except OSError as e:
              self.assertEqual(e.strerror, 'No such file or directory')
  
      def test_open_envelope(self):
          name = 'Jerome'
          intent = IntentBuilder(name).require('Keyword')
          intent.name = name
          m = Message("register_intent", intent.__dict__)
          unpacked_intent = open_intent_envelope(m)
          self.assertEqual(intent.__dict__, unpacked_intent.__dict__)
  
      def check_detach_intent(self):
          self.assertTrue(len(self.emitter.get_types()) > 0)
          for msg_type in self.emitter.get_types():
              self.assertEqual(msg_type, 'detach_intent')
          self.emitter.reset()
  
      def check_register_intent(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_intent')
          self.assertEqual(sorted(self.emitter.get_results()),
                           sorted(result_list))
          self.emitter.reset()
  
      def check_register_vocabulary(self, result_list):
          for type in self.emitter.get_types():
              self.assertEqual(type, 'register_vocab')
          self.assertEqual(sorted(self.emitter.get_results()),
                           sorted(result_list))
          self.emitter.reset()
  
      def test_register_intent(self):
          # Test register Intent object
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          s = SimpleSkill2()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
  
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          with self.assertRaises(ValueError):
              s = SimpleSkill3()
              s.bind(self.emitter)
              s.initialize()
  
      def test_enable_disable_intent(self):
          """Test disable/enable intent."""
          # Setup basic test
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': 'A:a',
                       'optional': [],
                       'requires': [('AKeyword', 'AKeyword')]}]
          self.check_register_intent(expected)
  
          # Test disable/enable cycle
          s.disable_intent('a')
          self.check_detach_intent()
          s.enable_intent('a')
          self.check_register_intent(expected)
  
      def test_register_vocab(self):
          """Test disable/enable intent."""
          # Setup basic test
          s = SimpleSkill1()
          s.bind(self.emitter)
          s.initialize()
  
          # Normal vocaubulary
          self.emitter.reset()
          expected = [{'start': 'hello', 'end': 'AHelloKeyword'}]
          s.register_vocabulary('hello', 'HelloKeyword')
          self.check_register_vocabulary(expected)
          # Regex
          s.register_regex('weird (?P<Weird>.+) stuff')
          expected = [{'regex': 'weird (?P<AWeird>.+) stuff'}]
          self.check_register_vocabulary(expected)
  
      def check_register_object_file(self, types_list, result_list):
          self.assertEqual(sorted(self.emitter.get_types()),
                           sorted(types_list))
          self.assertEqual(sorted(self.emitter.get_results(),
                                  key=lambda d: sorted(d.items())),
                           sorted(result_list, key=lambda d: sorted(d.items())))
          self.emitter.reset()
  
      def test_register_intent_file(self):
          s = SimpleSkill4()
          s.root_dir = abspath(join(dirname(__file__
  
  [... the rest is omitted ...]
files: test/unittests/skills/test_core.py