base.py

import hyperonpy as hp

from .atoms import Atom, BindingsSet
from hyperonpy import SyntaxNodeType

class AbstractSpace:
    """
    A virtual base class upon which Spaces can be implemented in Python
    """
    def __init__(self):
        """Initialiize the AbstractSpace. Does nothing in the base class"""
        return

    def query(self, query_atom):
        """
        Performs the specified query on the Space.
        Should be overridden to return a BindingsSet as the result of the query.
        """
        raise RuntimeError("Space::query() is not implemented")

    # TODO (INTERNAL): Currently unimplemented.  We may do this differently depending on lazy / comprehensions
    # so I don't want to write throw-away code unless it's needed downstream.
    # def subst(self, pattern, templ):
    #     None

    def add(self, atom):
        """
        Adds an Atom to the atom space. Must be implemented in derived classes.
        """
        raise RuntimeError("Space::add() is not implemented")

    def remove(self, atom):
        """
        Removes an Atom from the atom space. Must be implemented in derived classes.
        """
        raise RuntimeError("Space::remove() is not implemented")

    def replace(self, atom, replacement):
        """
        Replaces an Atom from the atom space. Must be implemented in derived classes.
        """
        raise RuntimeError("Space::replace() is not implemented")

    def atom_count(self):
        """
        Counts the number of atoms in the atom space. Optional for derived classes.
        """
        None

    def atoms_iter(self):
        """
        Returns an iterator over atoms in the Space. Optional for derived classes.
        """
        None

class GroundingSpace(AbstractSpace):
    """
    A wrapper over the native GroundingSpace implementation, which can be subclassed
    and extended within Python
    """
    def __init__(self):
        """Initialize GroundingSpace and its underlying native implementation."""
        super().__init__()
        # self.cspace = hp.space_new_grounding()
        self.gspace = GroundingSpaceRef()

    def query(self, query_atom):
        """
        Delegates the query to the underlying native GroundingSpace
        and returns the result BindingsSet
        """
        return self.gspace.query(query_atom)

    # TODO (INTERNAL): Currently unimplemented.
    # def subst(self, pattern, templ):

    def add(self, atom):
        """
        Adds an Atom to the atom space.
        """
        self.gspace.add_atom(atom)

    def remove(self, atom):
        """
        Removes an Atom from the atom space.
        """
        return self.gspace.remove_atom(atom)

    def replace(self, from_atom, to_atom):
        """
        Replaces an Atom in the atom space.
        """
        return self.gspace.replace_atom(from_atom, to_atom)

    def atom_count(self):
        """
        Counts the number of Atoms in the atom space.
        """
        return self.gspace.atom_count()

    def atoms_iter(self):
        """
        Returns an iterator over atoms in the atom space.
        """
        return iter(self.gspace.get_atoms())

def _priv_call_query_on_python_space(space, query_catom):
    """
    Private glue for Hyperonpy implementation.
    Translates a native 'catom' into an Atom object, and then delegates the query
    to the provided 'space' object.
    """
    query_atom = Atom._from_catom(query_catom)
    return space.query(query_atom)

def _priv_call_add_on_python_space(space, catom):
    """
    Private glue for Hyperonpy implementation.
    Translates a native 'catom' into an Atom object, and then adds it
    to the provided 'space' object.
    """
    atom = Atom._from_catom(catom)
    space.add(atom)

def _priv_call_remove_on_python_space(space, catom):
    """
    Private glue for Hyperonpy implementation.
    Translates a native 'catom' into an Atom object, and then removes it
    from the provided 'space' object.
    """
    atom = Atom._from_catom(catom)
    return space.remove(atom)

def _priv_call_replace_on_python_space(space, cfrom, cto):
    """
    Private glue for Hyperonpy implementation.
    Translates native 'catom' objects into Atom objects, and then replaces
    the first with the second in the provided 'space' object.
    """
    from_atom = Atom._from_catom(cfrom)
    to_atom = Atom._from_catom(cto)
    return space.replace(from_atom, to_atom)

def _priv_call_atom_count_on_python_space(space):
    """
    Private glue for Hyperonpy implementation.
    Returns the number of Atoms in the provided 'space' object.
    """
    if hasattr(space, "atom_count"):
        count = space.atom_count()
        if count is not None:
            return count
        else:
            return -1
    else:
        return -1

def _priv_call_new_iter_state_on_python_space(space):
    """
    Private glue for Hyperonpy implementation.
    Returns an iterator over Atoms in the provided 'space' object.
    """
    if hasattr(space, "atoms_iter"):
        return space.atoms_iter()
    else:
        return None

class SpaceRef:
    """
    A reference to a Space, which may be accessed directly, wrapped in a grounded atom,
    or passed to a MeTTa interpreter.
    """

    def __init__(self, space_obj):
        """
        Initialize a new SpaceRef based on the given space object, either a CSpace 
        or a custom Python object.
        """
        if type(space_obj) is hp.CSpace:
            self.cspace = space_obj
        else:
            self.cspace = hp.space_new_custom(space_obj)

    def __del__(self):
        """Free the underlying CSpace object """
        hp.space_free(self.cspace)

    def __eq__(self, other):
        """Compare two SpaceRef objects for equality, based on their underlying spaces."""
        return hp.space_eq(self.cspace, other.cspace)

    @staticmethod
    def _from_cspace(cspace):
        """
        Create a new SpaceRef based on the given CSpace object.
        """
        return SpaceRef(cspace)

    def copy(self):
        """
        Returns a new copy of the SpaceRef, referencing the same underlying Space.
        """
        return self

    def add_atom(self, atom):
        """
        Add an Atom to the Space.
        """
        hp.space_add(self.cspace, atom.catom)

    def remove_atom(self, atom):
        """
        Delete the specified Atom from the Space.
        """
        return hp.space_remove(self.cspace, atom.catom)

    def replace_atom(self, atom, replacement):
        """
        Replaces the specified Atom, if it exists in the Space, with the supplied replacement.
        """
        return hp.space_replace(self.cspace, atom.catom, replacement.catom)

    def atom_count(self):
        """
        Returns the number of Atoms in the Space, or -1 if it cannot be readily computed.
        """
        return hp.space_atom_count(self.cspace)

    def get_atoms(self):
        """
        Returns a list of all Atoms in the Space, or None if that is impossible.
        """
        res = hp.space_list(self.cspace)
        if res == None:
            return None
        result = []
        for r in res:
            result.append(Atom._from_catom(r))
        return result

    def get_payload(self):
        """
        Returns the Space object referenced by the SpaceRef, or None if the object does not have a
        direct Python interface.
        """
        return hp.space_get_payload(self.cspace)

    def query(self, pattern):
        """
        Performs the specified query on the Space, and returns the result as a BindingsSet.
        """
        result = hp.space_query(self.cspace, pattern.catom)
        return BindingsSet(result)

    def subst(self, pattern, templ):
        """
        Performs a substitution within the Space, based on a pattern and a template.
        """
        return [Atom._from_catom(catom) for catom in
                hp.space_subst(self.cspace, pattern.catom,
                                         templ.catom)]

class GroundingSpaceRef(SpaceRef):
    """
    A reference to a native GroundingSpace, implemented by the MeTTa core library.
    """

    def __init__(self, cspace = None):
        """
        Initialize a new GroundingSpaceRef.
        If a CSpace object is provided, use it; otherwise create a new GroundingSpace.
        """
        if cspace is None:
            self.cspace = hp.space_new_grounding()
        else:
            self.cspace = cspace

    @staticmethod
    def _from_cspace(cspace):
        """
        Creates a GroundingSpaceRef from a CSpace object.
        """
        return GroundingSpaceRef(cspace)

class Tokenizer:
    """
    A class responsible for text tokenization in the context of Hyperon.
    This class wraps around a Tokenizer object from the core library.
    """

    def __init__(self, ctokenizer = None):
        """
        Initialize a new Tokenizer.
        """
        if ctokenizer is None:
            self.ctokenizer = hp.tokenizer_new()
        else:
            self.ctokenizer = ctokenizer

    @staticmethod
    def _from_ctokenizer(ctokenizer):
        """
        Creates a Tokenizer from a CTokenizer object.
        """
        return Tokenizer(ctokenizer)

    def __del__(self):
        """
        Destructor that frees the underlying resources when the Tokenizer instance is destroyed.
        """
        hp.tokenizer_free(self.ctokenizer)

    def register_token(self, regex, constr):
        """
        Registers a new custom Token in the Tokenizer based on a regular expression.

        Parameters:
        ----------
        regex:
           A string representing the regular expression to match incoming text.
           Hyperon uses the Rust RegEx engine and syntax.
       constr:
           A constructor function for generating a new atom when the regex is triggered.
       """
        hp.tokenizer_register_token(self.ctokenizer, regex, constr)

class SyntaxNode:
    """
    A class representing a node in a parsed syntax tree
    """

    def __init__(self, cnode):
        """
        Initialize a new Tokenizer.
        """
        self.cnode = cnode

    def __del__(self):
        """
        Destructor for the SyntaxNode
        """
        hp.syntax_node_free(self.cnode)

    def get_type(self):
        """
        Returns the type of a SyntaxNode
        """
        return hp.syntax_node_type(self.cnode)

    def src_range(self):
        """
        Returns the range of offsets into the source code of the text represented by the SyntaxNode
        """
        range_tuple = hp.syntax_node_src_range(self.cnode)
        return range(range_tuple[0], range_tuple[1])

    def unroll(self):
        """
        Returns a list of all leaf nodes recursively contained within a SyntaxNode
        """
        syntax_nodes = []
        for cnode in hp.syntax_node_unroll(self.cnode):
            syntax_nodes.append(SyntaxNode(cnode))
        return syntax_nodes

class SExprParser:
    """
    A class responsible for parsing S-expressions (Symbolic Expressions).
    This class wraps around a SExprParser object from the core library.
    """

    def __init__(self, text):
        """Initialize a new SExprParser object."""
        self.cparser = hp.CSExprParser(text)

    def parse(self, tokenizer):
        """
        Parses the S-expression using the provided Tokenizer.
        """
        catom = self.cparser.parse(tokenizer.ctokenizer)
        if (catom is None):
            err_str = self.cparser.sexpr_parser_err_str()
            if (err_str is None):
                return None
            else:
                raise SyntaxError(err_str)
        else:
            return Atom._from_catom(catom)

    def parse_to_syntax_tree(self):
        """
        Parses the S-expression into a SyntaxNode representing the top-level of a syntax tree.
        """
        cnode = self.cparser.parse_to_syntax_tree()
        return SyntaxNode(cnode) if cnode is not None else None

class Interpreter:
    """
    A wrapper class for the MeTTa interpreter that handles the interpretation of expressions in a given grounding space.

    NOTE: This is a low-level API, and most applications would be better served by a `MeTTa` runner object
    """

    def __init__(self, gnd_space, expr):
        """
        Initializes the interpreter with the given grounding space and expression.
        """
        self.step_result = hp.interpret_init(gnd_space.cspace, expr.catom)

    def has_next(self):
        """
        Checks if there are more steps to execute in the interpretation plan.
        """
        return hp.step_has_next(self.step_result)

    def next(self):
        """
        Executes the next step in the interpretation plan.
        """
        if not self.has_next():
            raise StopIteration()
        self.step_result = hp.interpret_step(self.step_result)

    def get_result(self):
        """
        Retrieves the final outcome of the interpretation plan.
        """
        if self.has_next():
            raise RuntimeError("Plan execution is not finished")
        return hp.step_get_result(self.step_result)

    def get_step_result(self):
        """
        Gets the current result of the interpretation plan.
        """
        return self.step_result


def interpret(gnd_space, expr):
    """
    Parses the given expression in the specified grounding space.
    """
    interpreter = Interpreter(gnd_space, expr)
    while interpreter.has_next():
        interpreter.next()
    return [Atom._from_catom(catom) for catom in interpreter.get_result()]

def check_type(gnd_space, atom, type):
    """
    Checks whether the given Atom has the specified type in the given space context.

    Parameters
    ----------
    gnd_space:
        A pointer to the space_t representing the space context in which to perform
        the check
    atom:
        A pointer to the atom_t or atom_ref_t representing the atom whose Type the
        function will check
    type:
        A pointer to the atom_t or atom_ref_t representing the type to check against
    """

    return hp.check_type(gnd_space.cspace, atom.catom, type.catom)

def validate_atom(gnd_space, atom):
    """
    Checks whether the given Atom is correctly typed.

    Parameters
    ----------
    gnd_space:
        A pointer to the space_t representing the space context in which to perform
        the check
    atom:
        A pointer to the atom_t or atom_ref_t representing the atom whose Type the
        function will check

    Returns
    -------
    True if the Atom is correctly typed, otherwise false
    """
    return hp.validate_atom(gnd_space.cspace, atom.catom)

def get_atom_types(gnd_space, atom):
    """Provides all types for the given Atom in the context of the given Space."""
    result = hp.get_atom_types(gnd_space.cspace, atom.catom)
    return [Atom._from_catom(catom) for catom in result]

def atom_is_error(atom):
    """Checks whether an Atom is an error expression"""
    return hp.atom_is_error(atom.catom)