Adapter Development

Create A New Adapter

Create a messaging instance and pass in a service name that will uniquely identify it.

from vexbot.adapters.messaging import Messaging

messaging = Messaging('unique_service_name', run_control_loop=True)
messaging.run(blocking=False)
# Your messaging code here

# Some sort of loop, using a `while` loop for illustration purposes
while True:
    author = ''
    message = ''
    # optional
    # channel = ''

    messaging.send_chatter(author=author,
                           message=message)

    # NOTE: Alternate implementation
    """
    messaging.send_chatter(author=author,
                           message=message,
                           channel=channel)
    """

Dope. But what about something that sends commands to the robot?

from vexbot.adapters.messaging import Messaging

messaging = Messaging('unique_service_name', run_control_loop=True)
messaging.run(blocking=False)

# Your code here. You would probably want this in a loop as well.
command = ''
args = []
kwargs = {}

messaging.send_command(command, *args, **kwargs)

You probably want a response back out of that command, huh?

from vexbot.observer import Observer
from vexbot.adapters.messaging import Messaging

class MyObserver(Observer):
    def on_next(self, request):
        result = request.kwargs.get('result')
        # your code here

    def on_error(self, *args, **kwargs):
        pass

    def on_completed(*args, **kwargs):
        pass

messaging = Messaging('unique_service_name', run_control_loop=True)
messaging.run(blocking=False)

my_observer = MyObserver()

messaging.command.subscribe(my_observer)
# You can also pass in methods to the `subscribe` method
messaging.command.subscribe(your_custom_method_here)

Actually you probably want the ability to dynamically load commands, persist your dynamic commands, and see all the installed commands available.

import shelve
from os import path
from vexbot.observer import Observer
from vexbot.extensions import extensions

from vexbot.util.get_cache_filepath import get_cache
from vexbot.util.get_cache_filepath import get_cache_filepath as get_cache_dir

class MyObserver(Observer):
    extensions = (extensions.add_extensions,
                  extensions.remove_extension,
                  # NOTE: you can pass in dict's here to change the behavior
                  {'method': your_method_here,
                   'hidden': True,
                   'name': 'some_alternate_method_name',
                   'alias': ['method_name2',
                             'method_name3']},

                  extensions.get_extensions,
                  extensions.get_installed_extensions)

    def __init__(self):
        super().__init__()
        self._commands = {}
        cache_dir = get_cache_dir()
        mkdir = not path.isdir(cache_dir)
        if mkdir:
            os.makedirs(cache_dir, exist_ok=True)

        filepath = get_cache(__name__ + '.pickle')
        init = not path.isfile(filepath)

        self._config = shelve.open(filepath, flag='c', writeback=True)

        if init:
            self._config['extensions'] = {}
            self._config['disabled'] = {}
            self._config['modules'] = {}

    # NOTE: Here's our command handeling
    def handle_command(self, command: str, *args, **kwargs):
        callback = self._commands.get(command)
        if callback is None:
            return

        # Wrap our callback to catch errors
        try:
             result = callback(*args, **kwargs)
        except Exception as e:
             self.on_error(command, e, args, kwargs)

        print(result)

    def on_next(self, request):
        # NOTE: Here's our responses back from the bot
        result = request.kwargs.get('result')
        # your code here

    def on_error(self, *args, **kwargs):
        pass

    def on_completed(*args, **kwargs):
        pass

>> observer = MyObserver()
>> observer.handle_command('get_extensions')
>> []
>> observer.handle_command('add_extensions', 'log_level')
>> observer.handle_command('get_extensions')
>> ['log_level']