Boilerplate code and refactoring

bin/bluebubbles_bot

@@ -5,72 +5,11 @@ import sys
 import yaml
 import logging
 import argparse
-import bluebubbles_bot
+import persona
-import inspect
+from typing import List
-import importlib
+from dataclasses import dataclass
-# Import your custom libraries here
+import datetime
-
+import socketio
-_plugin_map = {}
-_delegate_map = {}
-_initialized_plugins = []
-
-# Plugin loading code
-plugin_class = 'bluebubbles_bot'.strip('-').capitalize() + 'Plugin'
-def add_plugin(plugin,reload) :
-	"""Adds a given plugin and instance, reinitializing one if it already exists and such is specified."""
-	plugin_name = plugin.__module__.split('.')[-1]
-	if not reload and plugin_name in _plugin_map.keys():
-		pass
-	else :
-		# We can't startup the plugin here because it hasn't been configured. We'll handle that at runtime.
-		try:
-			# Remove any intialized objects of the same name, forcing a reinitialization
-			_initialized_plugins.remove(_plugin_map[plugin_name])
-		except:
-			pass
-		_plugin_map.update({plugin_name:plugin})
-
-def use_plugins(plugins,reload=False) :
-	"""Defines plugins that should be used in a lookup, optionally forcing them to reload."""
-	# Verify data
-	if type(plugins) != list :
-		raise ValueError('argument \'plugins\' should be of type list')
-	for plugin in plugins :
-		# Check if the plugin is a string or a descendent of a bluebubbles_bot-plugin class
-		if type(plugin) != str and plugin_class not in [x.__name__ for x in inspect.getmro(plugin)] :
-			raise ValueError('unkown type for plugin')
-		# Find plugins by name using a default path
-		if type(plugin) == str :
-			available_plugins = [y for x,y in search_plugins().items() if x == plugin and plugin_class in [z.__name__ for z in inspect.getmro(y)]]
-			if len(available_plugins) == 0 :
-				raise FileNotFoundError(plugin + '.py not found')
-			plugin = available_plugins[0]
-		if plugin_class in [x.__name__ for x in inspect.getmro(plugin)] :
-			add_plugin(plugin,reload)
-			continue
-
-def get_plugins() :
-	"""Returns a map of plugins configured and loaded."""
-	return _plugin_map
-
-def search_plugins(directory=None) :
-	"""Searches a given directory for compatible plugins and returns a map of available plugin names and classes."""
-	if not directory :
-		directory = '/'.join(os.path.realpath(__file__).split('/')[:-1]) + '/' + 'plugins'
-	directory = os.path.normpath(os.path.expanduser(os.path.expandvars(directory)))
-	name_map = {}
-	candidates = {x.split('.')[0]:x for x in os.listdir(directory) if x.endswith('.py')}
-	for name,filename in candidates.items() :
-		try :
-			spec = importlib.util.spec_from_file_location(name, directory + '/' + filename)
-			mod = importlib.util.module_from_spec(spec)
-			spec.loader.exec_module(mod)
-			instance = getattr(mod,plugin_class)
-			name_map.update({filename.split('.')[0]:instance})
-		except Exception as e :
-			# Handle plugin loading issues if desired
-			print("Unable to load plugin from " + filename + ": " + str(e))
-	return name_map
 
 class LoggingFormatter(logging.Formatter):
 	def format(self, record):
@@ -117,8 +56,15 @@ if __name__ == '__main__':
 	logging.getLogger().handlers[0].setFormatter(LoggingFormatter())
 	logging.propagate=True
 
-	# Load plugins
+	# Create persona instance
-	available_plugins = search_plugins(directory='/'.join(os.path.realpath(__file__).split('/')[:-2]) + '/bluebubbles_bot/' + 'plugins')
+	test = persona.Persona()
-	use_plugins([x for x in available_plugins.values()])
+
-	# Main functions
+	# Test Message
-	print('Hello World!')
+	message = persona.Message(text="Hello",sender_identifier="Daniel",chat_identifier=None,attachments=[],timestamp=datetime.datetime.now(), recipients=[], identifier=None)
+	responses = test.receive_message(message)
+	for response in responses :
+		print('Bot responded with:')
+		print(response.text)
+
+	# Create socket connection
+	sio = socketio.AsyncClient()

bluebubbles_bot/__init__.py

@@ -1 +0,0 @@
-from bluebubbles_bot.bluebubbles_bot import *

bluebubbles_bot/bluebubbles_bot.py

@@ -1 +0,0 @@
-# Write your modular code (classes, functions, etc) here. They'll be automatically imported in bin/bluebubbles_bot

persona/__init__.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+from persona.persona import *
+import abc
+import os
+import inspect
+import importlib
+from typing import List
+from dataclasses import dataclass
+import datetime
+class PersonaException(Exception) :
+	"""The base Persona exception class."""
+
+class PersonaStartupException(PersonaException) :
+	"""The Persona exception class to raise during startup failures."""
+
+class PersonaIntentException(PersonaException) :
+	"""The Persona exception class to raise it cannot be determined if the skill should be matched."""
+
+class PersonaResponseException(PersonaException) :
+	"""The Persona exception class to raise when a response cannot be generated."""
+
+@dataclass
+class Attachment :
+    """Basic information about attachments in the messages the persona will receive."""
+    mime_type: str
+    data: any
+
+@dataclass
+class Message :
+	"""Basic information about the messages the persona will receive. Types are simple
+	   for simplicity and flexibility."""
+	sender_identifier: str
+	chat_identifier: str
+	text: str
+	attachments: List[Attachment]
+	timestamp: datetime.datetime
+	recipients: List[String]
+	identifier: str
+
+
+class PersonaBaseSkill() :
+	"""The base class for a Persona skill."""
+
+	def __init__(self) :
+		self.context = {}
+
+	def startup(self) :
+		"""Perform any initialization actions, such as searching the environment for config or logging into services.
+		   This action should raise PersonaStartupException on failure."""
+		pass
+
+	def shutdown(self) :
+		"""Perform any shutdown and cleanup actions, such as searching logging out of online services or removing
+		   temporary secrets. Exceptions raised here are ignored."""
+		pass
+
+	def match_intent(self,message: Message) -> Bool :
+		"""Receive text and determine if the skill should be used to respond to the message."""
+		raise PersonaIntentException('This skill does not know if it should respond.')
+
+	def respond(self, message: Message) -> Message :
+		"""Respond to a message by generating another message.
+		   This is called after the intent has been matched and the skill should produce a response.
+		   The Persona may modify the message, including its recipients, timestamp, or content."""
+		raise PersonaResponseException('This skill is not implemented to respond.')

persona/persona.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+import os
+import inspect
+import importlib
+from typing import List
+from dataclasses import dataclass
+import datetime
+
+class Persona :
+	"""A class that represents state data for the chatbot. This is the personality for the bot.
+	   Personas load skills that allow them to execute different commands. The personas can receive
+	   state data from the skills to affect their current context."""
+
+	_skill_map = {}
+	_delegate_map = {}
+	_initialized_skills = []
+	skill_class = 'Persona'.strip('-').capitalize() + 'Skill'
+	skill_class = 'PersonaSkill'
+	ready_skills = {}
+	def __init__(self) :
+		# Load skills
+		available_skills = self.search_skills(directory='/'.join(os.path.realpath(__file__).split('/')[:-2]) + '/persona' + '/skills')
+		self.use_skills([x for x in available_skills.values()])
+		self.startup()
+
+	# loading code
+	def add_skill(self,skill,reload) :
+		"""Adds a given skill and instance, reinitializing one if it already exists and such is specified."""
+		skill_name = skill.__module__.split('.')[-1]
+		if not reload and skill_name in self._skill_map.keys():
+			pass
+		else :
+			# We can't startup the skill here because it hasn't been configured. We'll handle that at runtime.
+			try:
+				# Remove any intialized objects of the same name, forcing a reinitialization
+				_initialized_skills.remove(self._skill_map[skill_name])
+			except:
+				pass
+			self._skill_map.update({skill_name:skill})
+
+	def use_skills(self,skills,reload=False) :
+		"""Defines skills that should be used in a lookup, optionally forcing them to reload."""
+		# Verify data
+		if type(skills) != list :
+			raise ValueError('argument \'skills\' should be of type list')
+		for skill in skills :
+			# Check if the skill is a string or a descendent of a persona-skill class
+			if type(skill) != str and self.skill_class not in [x.__name__ for x in inspect.getmro(skill)] :
+				raise ValueError('unkown type for skill')
+			# Find skills by name using a default path
+			if type(skill) == str :
+				available_skills = [y for x,y in search_skills().items() if x == skill and self.skill_class in [z.__name__ for z in inspect.getmro(y)]]
+				if len(available_skills) == 0 :
+					raise FileNotFoundError(skill + '.py not found')
+				skill = available_skills[0]
+			if self.skill_class in [x.__name__ for x in inspect.getmro(skill)] :
+				skill_name = skill.__module__.split('.')[-1]
+				self.add_skill(skill,reload)
+				continue
+
+	def get_skills(self) :
+		"""Returns a map of skills configured and loaded."""
+		return self._skill_map
+
+	def search_skills(self,directory=None) :
+		"""Searches a given directory for compatible skills and returns a map of available skill names and classes."""
+		if not directory :
+			directory = '/'.join(os.path.realpath(__file__).split('/')[:-1]) + '/' + 'skills'
+		directory = os.path.normpath(os.path.expanduser(os.path.expandvars(directory)))
+		name_map = {}
+		candidates = {x.split('.')[0]:x for x in os.listdir(directory) if x.endswith('.py')}
+		for name,filename in candidates.items() :
+			try :
+				spec = importlib.util.spec_from_file_location(name, directory + '/' + filename)
+				mod = importlib.util.module_from_spec(spec)
+				spec.loader.exec_module(mod)
+				instance = getattr(mod,self.skill_class)
+				name_map.update({filename.split('.')[0]:instance})
+			except Exception as e :
+				# Handle skill loading issues if desired
+				print("Unable to load skill from " + filename + ": " + str(e))
+		return name_map
+
+	def startup(self) :
+		"""Startup all message skill processors"""
+		for skill in self._skill_map.keys() :
+			ClassInstance = self._skill_map[skill]
+			skillinstance = ClassInstance()
+			skillinstance.startup()
+			self.ready_skills.update({skill:skillinstance})
+
+
+	def receive_message(self,message: Message) :
+		"""Process the receipt of a message."""
+		generated_messages = []
+		for skill in self.ready_skills.values() :
+			should_respond = skill.match_intent(message)
+			if should_respond :
+				response = skill.respond(message=message)
+				generated_messages.append(response)
+		return generated_messages

persona/skills/greeter.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+from typing import List
+import persona
+from persona import PersonaBaseSkill
+from dataclasses import dataclass
+import datetime
+import random
+import re
+
+class PersonaSkill(PersonaBaseSkill) :
+	"""A simple test skill that responds to the message 'Hello' with 'Hello!'"""
+	def match_intent(self,message: Message) -> Bool :
+		print(message.text)
+		matches = re.search('^Hello', message.text)
+		if matches :
+			return True
+		return False
+
+	def respond(self, message: Message) -> Message :
+		"""Respond to a message by generating another message."""
+		response_options = ['Hello!','Howdy!','Hello there!','What\'s up!','Hi there!']
+		response_text = random.choice(response_options)
+		return persona.Message(text=response_text,sender_identifier=message.sender_identifier,chat_identifier=message.chat_identifier,attachments=[],timestamp=datetime.datetime.now(), recipients=[message.sender_identifier], identifier=None)

setup.py

@@ -53,10 +53,9 @@ setup(name='bluebubbles_bot',
 	author_email='github@cronocide.com',
 	description='A chatbot for a local BlueBlubbles server.',
 	packages=find_packages(exclude=['tests']),
-	package_data={"": ['plugins/*.py']},
+	package_data={"": ['skills/*.py']},
-	install_requires=['pyyaml',],
+	install_requires=['pyyaml','datetime','python-socketio[asyncio_client]'],
 	scripts=['bin/bluebubbles_bot'],
 	long_description=open('README.md').read(),
-	
 	zip_safe=True
 )