A simple python script for demonstrating calling the atomspace from different processes

parent_child_proc.py

__author__ = 'Abdulrahman Semrie<[email protected]>'

import os
from opencog.scheme_wrapper import scheme_eval, scheme_eval_h
from opencog.atomspace import AtomSpace
from multiprocessing import Pool

PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))

OPENCOG_DEPS_PATH = os.path.join(PROJECT_ROOT, "opencog_deps")

DATASET_FOLDER = os.path.join(PROJECT_ROOT, "sample_dataset.scm")

FUNCTIONS_FOLDER = os.path.join(PROJECT_ROOT, "functions")
FUNCTION_PATHs = [os.path.join(FUNCTIONS_FOLDER, fn) for fn in os.listdir(FUNCTIONS_FOLDER) if
                  os.path.isfile(os.path.join(FUNCTIONS_FOLDER, fn))]


def load_atomspace(atomspace):
    """
    loads atomspace with knowledge bases and annotation scheme functions found in scm directory.
    :return: atomspace instance
    """
    print("Loading Atoms")
    scheme_eval_h(atomspace, '(primitive-load "{}")'.format(OPENCOG_DEPS_PATH))
    atomspace = load_functions(atomspace)
    atomspace = load_datasets(atomspace)
    print("Atoms loaded!")

    # snapshot = tracemalloc.take_snapshot()
    # top_stats = snapshot.statistics('lineno')
    #
    # logger.warning("[ Top 10 ], Trace Malloc")
    # for stat in top_stats[:10]:
    #     logger.warning(stat)

    return atomspace


def load_datasets(atomspace):
    """
    loads datasets from scm/datasets directory to atomspace
    :param atomspace: atomspace instance that will be loaded with datasets.
    :return: a loaded atomspace instance
    """
    print("Loading datasets")
    scheme_eval_h(atomspace, '(primitive-load "{}")'.format(DATASET_FOLDER))
    return atomspace


def load_functions(atomspace):
    """
    loads annotation functions from scm/functions directory to atomspace
    :param atomspace: atomspace instance taht will be loaded with functions
    :return: a loaded atomspace instance
    """
    print("loading functions")
    for fn in FUNCTION_PATHs:
        scheme_eval_h(atomspace, '(primitive-load "{}")'.format(fn))

    return atomspace


atomspace = AtomSpace()
load_atomspace(atomspace)


def count_atoms():
    res = scheme_eval(atomspace, "(count-all)").decode("utf-8")
    return res


def child_proc():
    res = scheme_eval(atomspace, "(count-all)").decode("utf-8")
    print("Child Count: " + res)
    os._exit(0)


def parent_proc():
    pid = os.fork()
    if pid == 0:
        child_proc()
    else:
        res = scheme_eval(atomspace, "(count-all)").decode("utf-8")
        print("Parent Count: " + res)


if __name__ == "__main__":
    # with Pool(2) as p:
    #     res = p.apply_async(count_atoms, ())
    #     print(res.get(timeout=20))
    parent_proc()

worker_process.py

__author__ = 'Abdulrahman Semrie<[email protected]>'

import os
from opencog.scheme_wrapper import scheme_eval, scheme_eval_h
from opencog.atomspace import AtomSpace
from multiprocessing import current_process, Process, Lock

PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))

OPENCOG_DEPS_PATH = os.path.join(PROJECT_ROOT, "opencog_deps")

DATASET_FOLDER = os.path.join(PROJECT_ROOT, "sample_dataset.scm")

FUNCTIONS_FOLDER = os.path.join(PROJECT_ROOT, "functions")
FUNCTION_PATHs = [os.path.join(FUNCTIONS_FOLDER, fn) for fn in os.listdir(FUNCTIONS_FOLDER) if
                  os.path.isfile(os.path.join(FUNCTIONS_FOLDER, fn))]


def load_atomspace(atomspace):
    """
    loads atomspace with knowledge bases and annotation scheme functions found in scm directory.
    :return: atomspace instance
    """
    print("Loading Atoms")
    scheme_eval_h(atomspace, '(primitive-load "{}")'.format(OPENCOG_DEPS_PATH))
    atomspace = load_functions(atomspace)
    atomspace = load_datasets(atomspace)
    print("Atoms loaded!")

    # snapshot = tracemalloc.take_snapshot()
    # top_stats = snapshot.statistics('lineno')
    #
    # logger.warning("[ Top 10 ], Trace Malloc")
    # for stat in top_stats[:10]:
    #     logger.warning(stat)

    return atomspace


def load_datasets(atomspace):
    """
    loads datasets from scm/datasets directory to atomspace
    :param atomspace: atomspace instance that will be loaded with datasets.
    :return: a loaded atomspace instance
    """
    print("Loading datasets")
    scheme_eval_h(atomspace, '(primitive-load "{}")'.format(DATASET_FOLDER))
    return atomspace


def load_functions(atomspace):
    """
    loads annotation functions from scm/functions directory to atomspace
    :param atomspace: atomspace instance taht will be loaded with functions
    :return: a loaded atomspace instance
    """
    print("loading functions")
    for fn in FUNCTION_PATHs:
        scheme_eval_h(atomspace, '(primitive-load "{}")'.format(fn))

    return atomspace


atomspace = AtomSpace()
load_atomspace(atomspace)


def count_atoms(lock):
    try:
        lock.acquire()
        print("In Process: " + current_process().name)
        res = scheme_eval(atomspace, "(count-all)").decode("utf-8")
        print(res)

    finally:
        lock.release()


if __name__ == "__main__":
    lock = Lock()
    procs = []
    for i in range(2):
        proc = Process(target=count_atoms, args=(lock,))
        procs.append(proc)
        proc.start()

    for proc in procs:
        proc.join()