runner ¶

def config_dir():
    """Returns the config dir in the common environment"""
    path = hp.environment_config_dir()
    if (len(path) > 0):
        return path
    else:
        return None

def custom_env(working_dir = None, config_dir = None, create_config = None, is_test = None, include_paths = []):
    """Returns an EnvBuilder object that can be used to init a MeTTa runner, if you need multiple environments to coexist in the same process

    Keyword arguments:
        working_dir -- working directory for the environment (default None)
        config_dir -- path to the configuration directory, None - no
        configuration directory, "" - default configuration directory
        (default None)
        create_config -- create configuration directory if not found
        (default None)
        is_test -- is environment used in unit-test flag (default None)
        include_paths -- additional search paths to search for MeTTa
        modules in the file system (default [])
    """
    builder = hp.env_builder_start()
    if working_dir is not None:
        hp.env_builder_set_working_dir(builder, working_dir)
    if config_dir is not None:
        if config_dir == "":
            hp.env_builder_set_default_config_dir(builder)
        else:
            hp.env_builder_set_config_dir(builder, config_dir)
    if create_config is not None:
        hp.env_builder_create_config_dir(builder, create_config)
    if is_test is not None:
        hp.env_builder_set_is_test(builder, is_test)
    for path in include_paths:
        hp.env_builder_push_include_path(builder, path)
    return builder

def init_common_env(working_dir = None, config_dir = None, create_config = None, is_test = None, include_paths = []):
    """Initialize the common environment with the supplied args

    Keyword arguments:
        working_dir -- working directory for the environment (default None)
        config_dir -- path to the configuration directory, None - no
        configuration directory, "" - default configuration directory
        (default None)
        create_config -- create configuration directory if it doesn't exist
        (default None)
        is_test -- is environment used in unit-test flag (default None)
        include_paths -- additional search paths to search for MeTTa
        modules in the file system (default [])
    """
    builder = Environment.custom_env(working_dir, config_dir, create_config, is_test, include_paths)
    return hp.env_builder_init_common_env(builder)

def test_env():
    """Returns an EnvBuilder object specifying a unit-test environment, that can be used to init a MeTTa runner"""
    return hp.env_builder_use_test_env()

def __eq__(self, other):
    """Checks if two MeTTa runner handles point to the same runner."""
    return (hp.metta_eq(self.cmetta, other.cmetta))

def load_module_at_path(self, path, mod_name=None):
    """
    Loads a module into the runner directly from resource at a file system path, trying the formats
    from the runner's environment in succession
    """
    mod_id = hp.metta_load_module_at_path(self.cmetta, path, mod_name)
    err_str = hp.metta_err_str(self.cmetta)
    if (err_str is not None):
        raise RuntimeError(err_str)
    return mod_id

def load_module_direct_from_func(self, mod_name, loader_func):
    """Loads a module into the runner using a loader function, with the specified name and scope"""
    mod_id = hp.metta_load_module_direct(self.cmetta, mod_name, loader_func)
    err_str = hp.metta_err_str(self.cmetta)
    if (err_str is not None):
        raise RuntimeError(err_str)
    return mod_id

def load_module_direct_from_pymod(self, mod_name, pymod_name):
    """Loads a module into the runner directly from a Python module, with the specified name and scope"""
    if not isinstance(pymod_name, str):
        pymod_name = repr(pymod_name)
    def loader_func(tokenizer, metta):
        _priv_register_module_tokens(pymod_name, tokenizer, metta)
    return self.load_module_direct_from_func(mod_name, loader_func)

def parse_all(self, program):
    """Parse an entire program from text into atoms, using the Tokenizer of the runner's top module"""
    return list(self._parse_all(program))

def parse_single(self, program):
    """Parse the next single token from the text program"""
    return next(self._parse_all(program))

def register_atom(self, name, symbol):
    """Registers an Atom"""
    self.register_token(name, lambda _: symbol)

def register_token(self, regexp, constr):
    """Registers a token"""
    self.tokenizer().register_token(regexp, constr)

def run(self, program, flat=False):
    """Runs the MeTTa code from the program string containing S-Expression MeTTa syntax"""
    parser = SExprParser(program)
    results = hp.metta_run(self.cmetta, parser.cparser)
    self._run_check_for_error()
    if flat:
        return [Atom._from_catom(catom) for result in results for catom in result]
    else:
        return [[Atom._from_catom(catom) for catom in result] for result in results]

def space(self):
    """Gets the space for the runner's top-level module"""
    return GroundingSpaceRef._from_cspace(hp.metta_space(self.cmetta))

def tokenizer(self):
    """Gets the tokenizer for the runner's top-level module"""
    return Tokenizer._from_ctokenizer(hp.metta_tokenizer(self.cmetta))

def working_dir(self):
    """Returns the working dir from the environment associated with the runner"""
    return hp.metta_working_dir(self.cmetta)

def import_dependency(self, mod_id):
    """Imports a loaded module as a dependency of the running module"""
    if mod_id.is_valid():
        hp.run_context_import_dependency(self.c_run_context, mod_id)
    else:
        raise RuntimeError("Invalid ModuleId")

def init_self_module(self, space, resource_dir):
    """Must be called exactly once from within a module loader to initialize the module being loaded"""
    hp.run_context_init_self_module(self.c_run_context, space.cspace, resource_dir)

def load_module(self, mod_name):
    """Resolves a module by name in the context of the running module, and loads it into the runner"""
    return hp.run_context_load_module(self.c_run_context, mod_name)

def metta(self):
    """Access the MeTTa runner that the RunContext is running within"""
    return MeTTa(cmetta = hp.run_context_get_metta(self.c_run_context))

def register_atom(self, name, symbol):
    """Registers an Atom with a name in the currently running module's Tokenizer"""
    self.register_token(name, lambda _: symbol)

def register_token(self, regexp, constr):
    """Registers a token in the currently running module's Tokenizer"""
    self.tokenizer().register_token(regexp, constr)

def space(self):
    """Access the space for the currently running module"""
    return GroundingSpaceRef._from_cspace(hp.run_context_get_space(self.c_run_context))

def tokenizer(self):
    """Access the tokenizer for the currently running module"""
    return Tokenizer._from_ctokenizer(hp.run_context_get_tokenizer(self.c_run_context))

def __del__(self):
    """Frees a RunnerState and all associated resources."""
    hp.runner_state_free(self.cstate)

def current_results(self, flat=False):
    """
    Returns the current in-progress results from an in-flight program evaluation
    """
    results = hp.runner_state_current_results(self.cstate)
    if flat:
        return [Atom._from_catom(catom) for result in results for catom in result]
    else:
        return [[Atom._from_catom(catom) for catom in result] for result in results]

def is_complete(self):
    """
    Returns True if the runner has concluded, or False if there are more steps remaining to execute
    """
    return hp.runner_state_is_complete(self.cstate)

def run_step(self):
    """
    Executes the next step in the interpretation plan, or begins interpretation of the next atom in the stream of MeTTa code.
    """
    hp.runner_state_step(self.cstate)
    err_str = hp.runner_state_err_str(self.cstate)
    if (err_str is not None):
        raise RuntimeError(err_str)