The Command Scheduler

Architecture Diagram

Why use the command scheduler?

The command scheduler offers a unique way to run commands in a robot program. Commands can be run in parallel, sequentially, or in the main scheduler FIFO queue. This allows for more flexibility when compared with traditional linear programming. Commands may be scheduled using a trigger.

Examples

Basic Usage

from kevinbotlib.scheduler import Command, CommandScheduler


class PrintCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False

    def init(self):
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)
        self._finished = True

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return self._finished


scheduler = CommandScheduler()
scheduler.schedule(PrintCommand("Test"))
scheduler.schedule(PrintCommand("Test2"))
scheduler.iterate()

Parallel Commands

import time

from kevinbotlib.scheduler import Command, CommandScheduler, ParallelCommand


class PrintForOneSecondCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False
        self.start = time.time()

    def init(self):
        self.start = time.time()
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return time.time() > self.start + 1


start_time = time.time()


scheduler = CommandScheduler()
scheduler.schedule(ParallelCommand([PrintForOneSecondCommand("command 1"), PrintForOneSecondCommand("command 2")]))

while True:
    scheduler.iterate()
    time.sleep(0.1)

Sequential Commands

import time

from kevinbotlib.scheduler import Command, CommandScheduler, SequentialCommand


class PrintForOneSecondCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False
        self.start = time.time()

    def init(self):
        self.start = time.time()
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return time.time() > self.start + 1


start_time = time.time()


scheduler = CommandScheduler()
scheduler.schedule(SequentialCommand([PrintForOneSecondCommand("command 1"), PrintForOneSecondCommand("command 2")]))

while True:
    scheduler.iterate()
    time.sleep(0.1)

Conditionally Forked Commands

from kevinbotlib.scheduler import Command, CommandScheduler, ConditionallyForkedCommand


class PrintCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False

    def init(self):
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)
        self._finished = True

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return self._finished


if __name__ == "__main__":
    scheduler = CommandScheduler()
    scheduler.schedule(
        ConditionallyForkedCommand(lambda: True, PrintCommand("Condition Met"), PrintCommand("Condition Not Met"))
    )
    scheduler.schedule(
        ConditionallyForkedCommand(lambda: False, PrintCommand("Condition Met"), PrintCommand("Condition Not Met"))
    )

    for _ in range(2):
        scheduler.iterate()

Named Controller Command Trigger

import time

from kevinbotlib.joystick import LocalNamedController
from kevinbotlib.logger import Logger, LoggerConfiguration
from kevinbotlib.scheduler import Command, CommandScheduler

logger = Logger()
logger.configure(LoggerConfiguration())


class PrintCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False

    def init(self):
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)
        self._finished = True

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return self._finished


class PrintForOneSecondCommand(Command):
    def __init__(self, message: str):
        self.message = message
        self._finished = False
        self.start = time.time()

    def init(self):
        self.start = time.time()
        print(f"Initializing: {self.message}")

    def execute(self):
        print(self.message)

    def end(self):
        print(f"Ending: {self.message}")

    def finished(self):
        return time.time() > self.start + 1


start_time = time.time()


scheduler = CommandScheduler()

controller = LocalNamedController(0)
controller.start_polling()

controller.command.a().while_true(PrintForOneSecondCommand("A Button Command"))
controller.command.b().on_true(PrintForOneSecondCommand("B Button Command"))
controller.command.x().on_true(PrintCommand("X Button Command"))

while True:
    scheduler.iterate()
    time.sleep(0.1)