files: # 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 imp
  import operator
  import sys
  import time
  import csv
  from functools import wraps
  from inspect import getargspec
  import inspect
  
  import abc
  import re
  from adapt.intent import Intent, IntentBuilder
  from os import listdir
  from os.path import join, abspath, dirname, splitext, basename, exists
  from threading import Event
  
  from mycroft.api import DeviceApi
  from mycroft.client.enclosure.api import EnclosureAPI
  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.skills.settings import SkillSettings
  from mycroft.util import resolve_resource_file
  from mycroft.util.log import LOG
  # python 2+3 compatibility
  from past.builtins import basestring
  
  MainModule = '__init__'
  
  
  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 load_vocab_from_file(path, vocab_type, emitter):
      """
          Load mycroft vocabulary from file. and send it on the message bus for
          the intent handler.
  
          Args:
              path:       path to vocabulary file (*.voc)
              vocab_type: keyword name
              emitter:    emitter to access the message bus
      """
      if path.endswith('.voc'):
          with open(path, 'r') as voc_file:
              for line in voc_file.readlines():
                  parts = line.strip().split("|")
                  entity = parts[0]
  
                  emitter.emit(Message("register_vocab", {
                      'start': entity, 'end': vocab_type
                  }))
                  for alias in parts[1:]:
                      emitter.emit(Message("register_vocab", {
                          'start': alias, 'end': vocab_type, 'alias_of': entity
                      }))
  
  
  def load_regex_from_file(path, emitter):
      """
          Load regex from file and send it on the message bus for
          the intent handler.
  
          Args:
              path:       path to vocabulary file (*.voc)
              emitter:    emitter to access the message bus
      """
      if path.endswith('.rx'):
          with open(path, 'r') as reg_file:
              for line in reg_file.readlines():
                  re.compile(line.strip())
                  emitter.emit(
                      Message("register_vocab", {'regex': line.strip()}))
  
  
  def load_vocabulary(basedir, emitter):
      for vocab_type in listdir(basedir):
          if vocab_type.endswith(".voc"):
              load_vocab_from_file(
                  join(basedir, vocab_type), splitext(vocab_type)[0], emitter)
  
  
  def load_regex(basedir, emitter):
      for regex_type in listdir(basedir):
          if regex_type.endswith(".rx"):
              load_regex_from_file(
                  join(basedir, regex_type), emitter)
  
  
  def open_intent_envelope(message):
      """ Convert dictionary received over messagebus to Intent. """
      intent_dict = message.data
      return Intent(intent_dict.get('name'),
                    intent_dict.get('requires'),
                    intent_dict.get('at_least_one'),
                    intent_dict.get('optional'))
  
  
  def load_skill(skill_descriptor, emitter, skill_id, BLACKLISTED_SKILLS=None):
      """
          load skill from skill descriptor.
  
          Args:
              skill_descriptor: descriptor of skill to load
              emitter:          messagebus emitter
              skill_id:         id number for skill
          Returns:
              MycroftSkill: the loaded skill or None on failure
      """
      BLACKLISTED_SKILLS = BLACKLISTED_SKILLS or []
      try:
          LOG.info("ATTEMPTING TO LOAD SKILL: " + skill_descriptor["name"] +
                   " with ID " + str(skill_id))
          if skill_descriptor['name'] in BLACKLISTED_SKILLS:
              LOG.info("SKILL IS BLACKLISTED " + skill_descriptor["name"])
              return None
          skill_module = imp.load_module(
              skill_descriptor["name"] + MainModule, *skill_descriptor["info"])
          if (hasattr(skill_module, 'create_skill') and
                  callable(skill_module.create_skill)):
              # v2 skills framework
              skill = skill_module.create_skill()
              skill.settings.allow_overwrite = True
              skill.settings.load_skill_settings_from_file()
              skill.bind(emitter)
              skill.skill_id = skill_id
              skill.load_data_files(dirname(skill_descriptor['info'][1]))
              # Set up intent handlers
              skill.initialize()
              skill._register_decorated()
              LOG.info("Loaded " + skill_descriptor["name"])
  
              # The very first time a skill is run, speak the intro
              first_run = skill.settings.get("__mycroft_skill_firstrun", True)
              if first_run:
                  LOG.info("First run of "+skill_descriptor["name"])
                  skill.settings["__mycroft_skill_firstrun"] = False
                  skill.settings.store()
                  intro = skill.get_intro_message()
                  if intro:
                      skill.speak(intro)
              return skill
          else:
              LOG.warning(
                  "Module %s does not appear to be skill" % (
                      skill_descriptor["name"]))
      except:
          LOG.error(
              "Failed to load skill: " + skill_descriptor["name"],
              exc_info=True)
      return None
  
  
  def create_skill_descriptor(skill_folder):
      info = imp.find_module(MainModule, [skill_folder])
      return {"name": basename(skill_folder), "info": info}
  
  
  def get_handler_name(handler):
      """
          Return name (including class if available) of handler
          function.
  
          Args:
              handler (function): Function to be named
  
          Returns: handler name as string
      """
      name = ''
      if '__self__' in dir(handler) and 'name' in dir(handler.__self__):
          name += handler.__self__.name + '.'
      name += handler.__name__
      return name
  
  
  # Lists used when adding skill handlers using decorators
  _intent_list = []
  _intent_file_list = []
  
  
  def intent_handler(intent_parser):
      """ Decorator for adding a method as an intent handler. """
  
      def real_decorator(func):
          @wraps(func)
          def handler_method(*args, **kwargs):
              return func(*args, **kwargs)
  
          _intent_list.append((intent_parser, func))
          return handler_method
  
      return real_decorator
  
  
  def intent_file_handler(intent_file):
      """ Decorator for adding a method as an intent file handler. """
  
      def real_decorator(func):
          @wraps(func)
          def handler_method(*args, **kwargs):
              return func(*args, **kwargs)
  
          _intent_file_list.append((intent_file, func))
          return handler_method
  
      return real_decorator
  
  
  #######################################################################
  # MycroftSkill base class
  #######################################################################
  class MycroftSkill(object):
      """
      Abstract base class which provides common behaviour and parameters to all
      Skills implementation.
      """
  
      def __init__(self, name=None, emitter=None):
          self.name = name or self.__class__.__name__
          # Get directory of skill
          self._dir = dirname(abspath(sys.modules[self.__module__].__file__))
  
          self.bind(emitter)
          self.config_core = Configuration.get()
          self.config = self.config_core.get(self.name)
          self.dialog_renderer = None
          self.vocab_dir = None
          self.root_dir = None
          self.file_system = FileSystemAccess(join('skills', self.name))
          self.registered_intents = []
          self.log = LOG.create_logger(self.name)
          self.reload_skill = True
          self.events = []
          self.skill_id = 0
  
      @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
  
  [... the rest is omitted ...]
files: mycroft/skills/core.py
files: # 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 imp
  import operator
  import sys
  import time
  import csv
  from functools import wraps
  from inspect import getargspec
  import inspect
  
  import abc
  import re
  from adapt.intent import Intent, IntentBuilder
  from os import listdir
  from os.path import join, abspath, dirname, splitext, basename, exists
  from threading import Event
  
  from mycroft.api import DeviceApi
  from mycroft.client.enclosure.api import EnclosureAPI
  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.skills.settings import SkillSettings
  from mycroft.util import resolve_resource_file
  from mycroft.util.log import LOG
  # python 2+3 compatibility
  from past.builtins import basestring
  
  MainModule = '__init__'
  
  
  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 load_vocab_from_file(path, vocab_type, emitter):
      """
          Load mycroft vocabulary from file. and send it on the message bus for
          the intent handler.
  
          Args:
              path:       path to vocabulary file (*.voc)
              vocab_type: keyword name
              emitter:    emitter to access the message bus
      """
      if path.endswith('.voc'):
          with open(path, 'r') as voc_file:
              for line in voc_file.readlines():
                  parts = line.strip().split("|")
                  entity = parts[0]
  
                  emitter.emit(Message("register_vocab", {
                      'start': entity, 'end': vocab_type
                  }))
                  for alias in parts[1:]:
                      emitter.emit(Message("register_vocab", {
                          'start': alias, 'end': vocab_type, 'alias_of': entity
                      }))
  
  
  def load_regex_from_file(path, emitter):
      """
          Load regex from file and send it on the message bus for
          the intent handler.
  
          Args:
              path:       path to vocabulary file (*.voc)
              emitter:    emitter to access the message bus
      """
      if path.endswith('.rx'):
          with open(path, 'r') as reg_file:
              for line in reg_file.readlines():
                  re.compile(line.strip())
                  emitter.emit(
                      Message("register_vocab", {'regex': line.strip()}))
  
  
  def load_vocabulary(basedir, emitter):
      for vocab_type in listdir(basedir):
          if vocab_type.endswith(".voc"):
              load_vocab_from_file(
                  join(basedir, vocab_type), splitext(vocab_type)[0], emitter)
  
  
  def load_regex(basedir, emitter):
      for regex_type in listdir(basedir):
          if regex_type.endswith(".rx"):
              load_regex_from_file(
                  join(basedir, regex_type), emitter)
  
  
  def open_intent_envelope(message):
      """ Convert dictionary received over messagebus to Intent. """
      intent_dict = message.data
      return Intent(intent_dict.get('name'),
                    intent_dict.get('requires'),
                    intent_dict.get('at_least_one'),
                    intent_dict.get('optional'))
  
  
  def load_skill(skill_descriptor, emitter, skill_id, BLACKLISTED_SKILLS=None):
      """
          load skill from skill descriptor.
  
          Args:
              skill_descriptor: descriptor of skill to load
              emitter:          messagebus emitter
              skill_id:         id number for skill
          Returns:
              MycroftSkill: the loaded skill or None on failure
      """
      BLACKLISTED_SKILLS = BLACKLISTED_SKILLS or []
      try:
          LOG.info("ATTEMPTING TO LOAD SKILL: " + skill_descriptor["name"] +
                   " with ID " + str(skill_id))
          if skill_descriptor['name'] in BLACKLISTED_SKILLS:
              LOG.info("SKILL IS BLACKLISTED " + skill_descriptor["name"])
              return None
          skill_module = imp.load_module(
              skill_descriptor["name"] + MainModule, *skill_descriptor["info"])
          if (hasattr(skill_module, 'create_skill') and
                  callable(skill_module.create_skill)):
              # v2 skills framework
              skill = skill_module.create_skill()
              skill.settings.allow_overwrite = True
              skill.settings.load_skill_settings_from_file()
              skill.bind(emitter)
              skill.skill_id = skill_id
              skill.load_data_files(dirname(skill_descriptor['info'][1]))
              # Set up intent handlers
              skill.initialize()
              skill._register_decorated()
              LOG.info("Loaded " + skill_descriptor["name"])
  
              # The very first time a skill is run, speak the intro
              first_run = skill.settings.get("__mycroft_skill_firstrun", True)
              if first_run:
                  LOG.info("First run of "+skill_descriptor["name"])
                  skill.settings["__mycroft_skill_firstrun"] = False
                  skill.settings.store()
                  intro = skill.get_intro_message()
                  if intro:
                      skill.speak(intro)
              return skill
          else:
              LOG.warning(
                  "Module %s does not appear to be skill" % (
                      skill_descriptor["name"]))
      except:
          LOG.error(
              "Failed to load skill: " + skill_descriptor["name"],
              exc_info=True)
      return None
  
  
  def create_skill_descriptor(skill_folder):
      info = imp.find_module(MainModule, [skill_folder])
      return {"name": basename(skill_folder), "info": info}
  
  
  def get_handler_name(handler):
      """
          Return name (including class if available) of handler
          function.
  
          Args:
              handler (function): Function to be named
  
          Returns: handler name as string
      """
      name = ''
      if '__self__' in dir(handler) and 'name' in dir(handler.__self__):
          name += handler.__self__.name + '.'
      name += handler.__name__
      return name
  
  
  # Lists used when adding skill handlers using decorators
  _intent_list = []
  _intent_file_list = []
  
  
  def intent_handler(intent_parser):
      """ Decorator for adding a method as an intent handler. """
  
      def real_decorator(func):
          @wraps(func)
          def handler_method(*args, **kwargs):
              return func(*args, **kwargs)
  
          _intent_list.append((intent_parser, func))
          return handler_method
  
      return real_decorator
  
  
  def intent_file_handler(intent_file):
      """ Decorator for adding a method as an intent file handler. """
  
      def real_decorator(func):
          @wraps(func)
          def handler_method(*args, **kwargs):
              return func(*args, **kwargs)
  
          _intent_file_list.append((intent_file, func))
          return handler_method
  
      return real_decorator
  
  
  #######################################################################
  # MycroftSkill base class
  #######################################################################
  class MycroftSkill(object):
      """
      Abstract base class which provides common behaviour and parameters to all
      Skills implementation.
      """
  
      def __init__(self, name=None, emitter=None):
          self.name = name or self.__class__.__name__
          # Get directory of skill
          self._dir = dirname(abspath(sys.modules[self.__module__].__file__))
  
          self.bind(emitter)
          self.config_core = Configuration.get()
          self.config = self.config_core.get(self.name)
          self.dialog_renderer = None
          self.vocab_dir = None
          self.root_dir = None
          self.file_system = FileSystemAccess(join('skills', self.name))
          self.registered_intents = []
          self.log = LOG.create_logger(self.name)
          self.reload_skill = True
          self.events = []
          self.skill_id = 0
  
      @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
  
  [... the rest is omitted ...]
files: mycroft/skills/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
  
  import mock
  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.core import load_regex_from_file, load_regex, \
      load_vocab_from_file, load_vocabulary, MycroftSkill, \
      load_skill, create_skill_descriptor, open_intent_envelope
  
  
  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 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 []
          load_regex_from_file(join(self.regex_path, filename), self.emitter)
          self.check_emitter(result_list)
  
      def check_vocab(self, path, result_list=None):
          result_list = result_list or []
          load_vocabulary(path, self.emitter)
          self.check_emitter(result_list)
  
      def check_regex(self, path, result_list=None):
          result_list = result_list or []
          load_regex(path, self.emitter)
          self.check_emitter(result_list)
  
      def check_emitter(self, result_list):
          for type in self.emitter.get_types():
              self.assertEquals(type, 'register_vocab')
          self.assertEquals(sorted(self.emitter.get_results()),
                            sorted(result_list))
          self.emitter.reset()
  
      def test_load_regex_from_file_single(self):
          self.check_regex_from_file('valid/single.rx',
                                     [{'regex': '(?P<SingleTest>.*)'}])
  
      def test_load_regex_from_file_multiple(self):
          self.check_regex_from_file('valid/multiple.rx',
                                     [{'regex': '(?P<MultipleTest1>.*)'},
                                      {'regex': '(?P<MultipleTest2>.*)'}])
  
      def test_load_regex_from_file_none(self):
          self.check_regex_from_file('invalid/none.rx')
  
      def test_load_regex_from_file_invalid(self):
          try:
              self.check_regex_from_file('invalid/invalid.rx')
          except error as e:
              self.assertEquals(e.__str__(),
                                'unexpected end of regular expression')
  
      def test_load_regex_from_file_does_not_exist(self):
          try:
              self.check_regex_from_file('does_not_exist.rx')
          except IOError as e:
              self.assertEquals(e.strerror, 'No such file or directory')
  
      def test_load_regex_full(self):
          self.check_regex(join(self.regex_path, 'valid'),
                           [{'regex': '(?P<MultipleTest1>.*)'},
                            {'regex': '(?P<MultipleTest2>.*)'},
                            {'regex': '(?P<SingleTest>.*)'}])
  
      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.assertEquals(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.assertEquals(e.strerror, 'No such file or directory')
  
      def test_load_vocab_full(self):
          self.check_vocab(join(self.vocab_path, 'valid'),
                           [{'start': 'test', 'end': 'single'},
                            {'start': 'water', 'end': 'singlealias'},
                            {'start': 'watering', 'end': 'singlealias',
                             'alias_of': 'water'},
                            {'start': 'animal', 'end': 'multiple'},
                            {'start': 'animals', 'end': 'multiple'},
                            {'start': 'chair', 'end': 'multiplealias'},
                            {'start': 'chairs', 'end': 'multiplealias',
                             'alias_of': 'chair'},
                            {'start': 'table', 'end': 'multiplealias'},
                            {'start': 'tables', 'end': 'multiplealias',
                             '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.assertEquals(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 test_load_skill(self):
          """ Verify skill load function. """
          e_path = join(dirname(__file__), 'test_skill')
          s = load_skill(create_skill_descriptor(e_path), MockEmitter(), 847)
          self.assertEquals(s._dir, e_path)
          self.assertEquals(s.skill_id, 847)
          self.assertEquals(s.name, 'LoadTestSkill')
  
      def check_register_intent(self, result_list):
          for type in self.emitter.get_types():
              self.assertEquals(type, 'register_intent')
          self.assertEquals(sorted(self.emitter.get_results()),
                            sorted(result_list))
          self.emitter.reset()
  
      def test_register_intent(self):
          # Test register Intent object
          s = TestSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': '0:a',
                       'optional': [],
                       'requires': [('Keyword', 'Keyword')]}]
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          s = TestSkill2()
          s.bind(self.emitter
  
  [... the rest is omitted ...]
files: test/unittests/skills/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
  
  import mock
  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.core import load_regex_from_file, load_regex, \
      load_vocab_from_file, load_vocabulary, MycroftSkill, \
      load_skill, create_skill_descriptor, open_intent_envelope
  
  
  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 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 []
          load_regex_from_file(join(self.regex_path, filename), self.emitter)
          self.check_emitter(result_list)
  
      def check_vocab(self, path, result_list=None):
          result_list = result_list or []
          load_vocabulary(path, self.emitter)
          self.check_emitter(result_list)
  
      def check_regex(self, path, result_list=None):
          result_list = result_list or []
          load_regex(path, self.emitter)
          self.check_emitter(result_list)
  
      def check_emitter(self, result_list):
          for type in self.emitter.get_types():
              self.assertEquals(type, 'register_vocab')
          self.assertEquals(sorted(self.emitter.get_results()),
                            sorted(result_list))
          self.emitter.reset()
  
      def test_load_regex_from_file_single(self):
          self.check_regex_from_file('valid/single.rx',
                                     [{'regex': '(?P<SingleTest>.*)'}])
  
      def test_load_regex_from_file_multiple(self):
          self.check_regex_from_file('valid/multiple.rx',
                                     [{'regex': '(?P<MultipleTest1>.*)'},
                                      {'regex': '(?P<MultipleTest2>.*)'}])
  
      def test_load_regex_from_file_none(self):
          self.check_regex_from_file('invalid/none.rx')
  
      def test_load_regex_from_file_invalid(self):
          try:
              self.check_regex_from_file('invalid/invalid.rx')
          except error as e:
              self.assertEquals(e.__str__(),
                                'unexpected end of regular expression')
  
      def test_load_regex_from_file_does_not_exist(self):
          try:
              self.check_regex_from_file('does_not_exist.rx')
          except IOError as e:
              self.assertEquals(e.strerror, 'No such file or directory')
  
      def test_load_regex_full(self):
          self.check_regex(join(self.regex_path, 'valid'),
                           [{'regex': '(?P<MultipleTest1>.*)'},
                            {'regex': '(?P<MultipleTest2>.*)'},
                            {'regex': '(?P<SingleTest>.*)'}])
  
      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.assertEquals(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.assertEquals(e.strerror, 'No such file or directory')
  
      def test_load_vocab_full(self):
          self.check_vocab(join(self.vocab_path, 'valid'),
                           [{'start': 'test', 'end': 'single'},
                            {'start': 'water', 'end': 'singlealias'},
                            {'start': 'watering', 'end': 'singlealias',
                             'alias_of': 'water'},
                            {'start': 'animal', 'end': 'multiple'},
                            {'start': 'animals', 'end': 'multiple'},
                            {'start': 'chair', 'end': 'multiplealias'},
                            {'start': 'chairs', 'end': 'multiplealias',
                             'alias_of': 'chair'},
                            {'start': 'table', 'end': 'multiplealias'},
                            {'start': 'tables', 'end': 'multiplealias',
                             '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.assertEquals(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 test_load_skill(self):
          """ Verify skill load function. """
          e_path = join(dirname(__file__), 'test_skill')
          s = load_skill(create_skill_descriptor(e_path), MockEmitter(), 847)
          self.assertEquals(s._dir, e_path)
          self.assertEquals(s.skill_id, 847)
          self.assertEquals(s.name, 'LoadTestSkill')
  
      def check_register_intent(self, result_list):
          for type in self.emitter.get_types():
              self.assertEquals(type, 'register_intent')
          self.assertEquals(sorted(self.emitter.get_results()),
                            sorted(result_list))
          self.emitter.reset()
  
      def test_register_intent(self):
          # Test register Intent object
          s = TestSkill1()
          s.bind(self.emitter)
          s.initialize()
          expected = [{'at_least_one': [],
                       'name': '0:a',
                       'optional': [],
                       'requires': [('Keyword', 'Keyword')]}]
          self.check_register_intent(expected)
  
          # Test register IntentBuilder object
          s = TestSkill2()
          s.bind(self.emitter
  
  [... the rest is omitted ...]
files: test/unittests/skills/core.py