Source code for graphbrain.hypergraph

from graphbrain.hyperedge import *


[docs]class Hypergraph(object): """Hypergraph interface.""" # ================================================================ # Interface abstract methods, to be implemented in derived classes # ================================================================
[docs] def close(self): """Closes the hypergraph.""" raise NotImplementedError()
[docs] def name(self): """Returns name of the hypergraph.""" raise NotImplementedError()
[docs] def destroy(self): """Erase the entire hypergraph.""" raise NotImplementedError()
[docs] def all(self): """Returns a generator of all the edges.""" raise NotImplementedError()
[docs] def all_attributes(self): """Returns a generator with a tuple for each edge. The first element of the tuple is the edge itself, the second is a dictionary of attribute values (as strings).""" raise NotImplementedError()
[docs] def atom_count(self): """Returns total number of atoms.""" raise NotImplementedError()
[docs] def edge_count(self): """Returns total number of edges.""" raise NotImplementedError()
[docs] def primary_atom_count(self): """Returns number of primary atoms.""" raise NotImplementedError()
[docs] def primary_edge_count(self): """Returns number of primary edges.""" raise NotImplementedError()
# ============================ # High-level interface methods # ============================
[docs] def all_atoms(self): """Returns a generator of all the atoms.""" for edge in self.all(): if edge.is_atom(): yield edge
[docs] def all_non_atoms(self): """Returns a generator of all the edges that are not atoms.""" for edge in self.all(): if not edge.is_atom(): yield edge
[docs] def exists(self, edge): """Checks if the given edge exists.""" return self._exists(hedge(edge))
[docs] def add(self, edge, primary=True): """Adds an edge if it does not exist yet, returns same edge. All children are recursively added as non-primary edge, for indexing purposes. Edges can be passed in both Hyperedge or string format. Keyword argument: primary -- edge is primary, meaning, for example, that it counts towards degrees. Non-primary edges are used for indexing purposes, for example to make it easy to find the subedges contained in primary edges when performing queries. """ if isinstance(edge, Hyperedge): if edge.is_atom(): return edge else: # recursively add all sub-edges as non-primary edges. for child in edge: self.add(child, primary=False) # add entity itself return self._add(edge, primary=primary) else: return self.add(hedge(edge), primary=primary)
[docs] def remove(self, edge, deep=False): """Removes an edge. Keyword argument: deep -- recursively remove all subedges (default False) """ self._remove(hedge(edge), deep=deep)
[docs] def is_primary(self, edge): """Check if an edge is primary.""" return self._is_primary(hedge(edge))
[docs] def set_primary(self, edge, value): """Make edge primary if value is True, make it non-primary otherwise. """ self._set_primary(hedge(edge), value)
[docs] def search(self, pattern): """Returns generator for all the edges that match a pattern. Patterns are themselves edges. They can match families of edges by employing special atoms: -> '*' represents a general wildcard (matches any edge) -> '@' represents an atomic wildcard (matches any atom) -> '&' represents a non-atomic wildcard (matches any non-atom) -> '...' at the end indicates an open-ended pattern. The pattern can be a string, that must represent an edge. Examples: '(is/pd graphbrain/c @)' '(says/pd * ...)' Atomic patterns can also be used to match all edges in the hypergraph (*), all atoms (@), and all non-atoms (&). """ pattern = hedge(pattern) if pattern.is_atom() and len(pattern.parts()) == 1: if pattern[0][0] == '*': return self.all() elif pattern[0][0] == '@': return self.all_atoms() elif pattern[0][0] == '&': return self.all_non_atoms() if pattern.is_full_pattern(): return self.all() else: return self._search(pattern)
[docs] def star(self, center, limit=None): """Returns generator of the edges that contain the center. Keyword argument: limit -- maximum number of results to return, infinite if None """ return self._star(hedge(center), limit=limit)
[docs] def atoms_with_root(self, root): """Returns generator of all atoms with the given root.""" if len(root) == 0: return {} return self._atoms_with_root(root)
[docs] def edges_with_edges(self, edges, root=None): """Returns generator of all edges containing the given edges, and optionally a given root. Keyword argument: root -- edge must also contain an atom with this root (default None) """ return self._edges_with_edges(edges, root)
[docs] def set_attribute(self, edge, attribute, value): """Sets the value of an attribute.""" return self._set_attribute(hedge(edge), attribute, value)
[docs] def inc_attribute(self, edge, attribute): """Increments an attribute of an entity.""" return self._inc_attribute(hedge(edge), attribute)
[docs] def dec_attribute(self, edge, attribute): """Increments an attribute of an entity.""" return self._dec_attribute(hedge(edge), attribute)
[docs] def get_str_attribute(self, edge, attribute, or_else=None): """Returns attribute as string. Keyword argument: or_else -- value to return if the entity does not have the give attribute. (default None) """ return self._get_str_attribute(hedge(edge), attribute, or_else=or_else)
[docs] def get_int_attribute(self, edge, attribute, or_else=None): """Returns attribute as integer value. or_else -- value to return if the entity does not have the give attribute. (default None) """ return self._get_int_attribute(hedge(edge), attribute, or_else=or_else)
[docs] def get_float_attribute(self, edge, attribute, or_else=None): """Returns attribute as float value. or_else -- value to return if the entity does not have the give attribute. (default None) """ return self._get_float_attribute(hedge(edge), attribute, or_else=or_else)
[docs] def degree(self, edge): """Returns the degree of an entity.""" return self._degree(hedge(edge))
[docs] def deep_degree(self, edge): """Returns the deep degree of an entity.""" return self._deep_degree(hedge(edge))
[docs] def ego(self, center): """Returns all atoms directly connected to center by hyperedges. """ edges = self.star(center) atom_set = set() for edge in edges: for atom in edge.atoms(): atom_set.add(atom) return atom_set
[docs] def remove_by_pattern(self, pattern): """Removes all edges that match the pattern.""" edges = self.search(pattern) for edge in edges: self.remove(edge)
[docs] def lemma_degrees(self, edge): """TODO: document and test""" if edge.is_atom(): roots = {edge.root()} # find lemma for edge in self.search((const.lemma_pred, edge, '*')): roots.add(edge[2].root()) # compute degrees d = 0 dd = 0 for r in roots: atoms = set(self.atoms_with_root(r)) d += sum([self.degree(atom) for atom in atoms]) dd += sum([self.deep_degree(atom) for atom in atoms]) return d, dd else: return self.degree(edge), self.deep_degree(edge)
[docs] def root_degrees(self, edge): """TODO: document and test""" if edge.is_atom(): atoms = self.atoms_with_root(edge.root()) d = sum([self.degree(atom) for atom in atoms]) dd = sum([self.deep_degree(atom) for atom in atoms]) return d, dd else: return self.degree(edge), self.deep_degree(edge)
# ============================================================== # Private abstract methods, to be implemented in derived classes # ============================================================== def _exists(self, edge): raise NotImplementedError() def _add(self, edge, primary): raise NotImplementedError() def _remove(self, edge, deep): raise NotImplementedError() def _is_primary(self, edge): raise NotImplementedError() def _set_primary(self, edge, value): raise NotImplementedError() def _search(self, pattern): raise NotImplementedError() def _star(self, center, limit=None): raise NotImplementedError() def _atoms_with_root(self, root): raise NotImplementedError() def _edges_with_edges(self, edges, root): raise NotImplementedError() def _set_attribute(self, edge, attribute, value): raise NotImplementedError() def _inc_attribute(self, edge, attribute): raise NotImplementedError() def _dec_attribute(self, edge, attribute): raise NotImplementedError() def _get_str_attribute(self, edge, attribute, or_else=None): raise NotImplementedError() def _get_int_attribute(self, edge, attribute, or_else=None): raise NotImplementedError() def _get_float_attribute(self, edge, attribute, or_else=None): raise NotImplementedError() def _degree(self, edge): raise NotImplementedError() def _deep_degree(self, edge): raise NotImplementedError()