clipspy example

import time
from queue import Queue, Empty

from clips import Environment, LoggingRouter


class FactsQueue:
    """Facts are coming from this queue."""
    def __init__(self):
        self.queue = Queue()  # this simulates your input

    def get_next_fact(self, timeout):
        """This function receives a fact already as CLIPS string.

        It blocks for maximum timeout seconds and returns None
        if no new fact is in the queue.

        """
        try:
            self.queue.get(timeout=timeout)
        except Empty:
            return None


def run(facts_queue, clips_file):
    tick_fact = None
    env = Environment()

    # The logging router will deal with CLIPS output messages itself
    router = LoggingRouter()
    router.add_to_environment(env)

    env.batch_star(clips_file)

    # Consume the next fact, wait max 3 seconds if not facts are available
    for fact in facts_queue.get_next_fact(3.0):
        # deal with time
        if tick_fact is not None:
            tick_fact.retract()
        tick_fact = env.assert_string("(current-time {})".format(time.time()))

        # deal with command messages
        if fact == 'ENABLE_FIRING_TRACING':
            enable_firings_tracing(env)
        elif fact == 'DISABLE_FIRING_TRACING':
            disable_firings_tracing(env)
        elif fact == 'STOP_ENGINE':
            break
        # assert the fact as a string in CLIPS
        elif fact is not None:
            env.assert_string(fact)

        # finally, fire the activations in the CLIPS agenda
        env.run()

    # print the currently asserted facts if the engine was stopped
    for fact in env.facts():
        print(fact)


def enable_firings_tracing(env):
    for rule in env.rules():
        rule.watch_firings = True


def disable_firings_tracing(env):
    for rule in env.rules():
        rule.watch_firings = False