Python Scheduler

Occasionally one needs to schedule tasks. Here’s a couple of classes I wrote to make life easier. This is a little one I whipped up recently. It has the interesting feature of being able to quit the game on the first error it encounters. This is helpful in isolating the cause of issues, as messages don’t get lost in the terminal scrollback. If you want it to report errors, set EXIT_ON_ERROR to false, in which case the failure of one task does not effect the scheduling of others.


''' This module creates a generic schedular that can be used to schedule
events based on time'''

import traceback
import time
import logging
import bge

EXIT_ON_ERROR = True

LOG = logging.getLogger()


class Scheduler():
    '''A scheduler to manage the execution of events'''
    events = []
    has_run = False

    def __init__(self):
        '''Sets up an empty scheduler'''
        LOG.debug("Started Scheduler")
        self.update()


    def update(self):
        '''Runs any evens registered to the scheduler'''
        for event in self.events:
            if event.time_until_next_run <= 0:
                event.run()


    def register_function(self, function, args=None, time_between=0):
        '''Registers a function as something that needs to be run'''
        event = Event(function, args, time_between)
        self._add_event(event)
        return event

    def _add_event(self, event):
        '''Adds an even to be run'''
        self.events.append(event)

    def remove_event(self, event):
        '''Cancels any future events'''
        self.events.remove(event)


class Event():
    '''An event that needs to be scheduled. Basically a container
    for a function and parameters specifying how often it should
    run'''

    def __init__(self, function, args=None, time_between=0, num_runs=-1):
        '''Creates an event to be run on a specified interval'''
        self.funct = function
        self.time_between = time_between
        self.remaining_runs = num_runs
        self.last_run = time.time()
        self.args = args

    def run(self):
        '''Executes the event and updates the internal timing properties'''
        try:
            if self.args is None:
                self.funct()
            else:
                self.funct(*self.args)
        except Exception as e:
            LOG.error(traceback.format_exc())
            if EXIT_ON_ERROR:
                raise(e)
             bge.logic.endGame()

        if self.time_between == 0:
            self.last_run = time.time()
        else:
            self.last_run = self.last_run + self.time_between

    @property
    def time_since_last_run(self):
        '''Returns the time since the event was last run.

        If the schedular is not keeping up with this event, this
        may not be accurate'''
        return self.last_run - time.time()

    @property
    def time_until_next_run(self):
        '''Returns the (ideal) time until the next execution.

        If the schedular is not keeping up with this event, it may be
        negative'''
        return (self.last_run + self.time_between) - time.time()

    def __str__(self):
        return "Event({}.{})".format(self.funct.__module__,
                                     self.funct.__name__)


You’ll need another script to instance it and update it, I reccomend something like this:


import bge
import scheduler

def init(cont):
    bge.logic.scheduler = scheduler.Scheduler()
    cont.script = __name__ + '.update'

def update(cont):
    bge.logic.scheduler.update()

Then to add a task to the scheduler, you use:


import bge

def init_something():
    bge.logic.scheduler.register_function(self, do_every_second, time_between=1)
    

def do_every_second():
   print("I run every second")