Source code for mevis._internal.io

import errno as _errno
import os as _os

import networkx as _nx
from opencog.atomspace import AtomSpace as _AtomSpace
from opencog.scheme import scheme_eval as _scheme_eval
from opencog.utilities import load_file as _load_file
from opencog.utilities import set_default_atomspace as _set_default_atomspace

from .args import check_arg as _check_arg


[docs]def create(set_as_default=True): """Create an empty AtomSpace and set it as default. Parameters ---------- set_as_default : bool If ``True``, the newly created AtomSpace is set as default in OpenCog. References ---------- - OpenCog Wiki: `Python <https://wiki.opencog.org/w/Python>`__ Returns ------- atomspace : AtomSpace """ atomspace = _AtomSpace() if set_as_default: _set_default_atomspace(atomspace) return atomspace
[docs]def load(filepath, method='primitive', verbose=False): """Load an Atomspace from a given filepath with a chosen method. Parameters ---------- filepath : str Path of a text file that contains Atomese code. method : str Possible values: - ``"basic"``: uses Scheme code ``(load "data.scm")`` - ``"primitive"``: uses Scheme code ``(primitive-load "data.scm")`` - ``"fast"``: uses Scheme code ``(load-file "data.scm")`` - ``"python"``: uses Python function ``opencog.utilities.load_file`` verbose : bool if ``True``, a short message is printed about the AtomSpace that was loaded. References ---------- - AtomSpace example: `persist-file.scm <https://github.com/opencog/atomspace/blob/master/examples/atomspace/persist-file.scm>`__ Returns ------- atomspace : AtomSpace """ # Argument processing _check_arg(filepath, 'filepath', str) _check_arg(method, 'method', str, ['basic', 'primitive', 'fast', 'python']) _check_arg(verbose, 'verbose', bool) if not _os.path.isfile(filepath): raise FileNotFoundError(_errno.ENOENT, _os.strerror(_errno.ENOENT), filepath) # Create Atomspace and load modules atomspace = _AtomSpace() _load_all_modules(atomspace) # Import from file if method == 'basic': _scheme_eval(atomspace, '(load "{}")'.format(filepath)) elif method == 'primitive': _scheme_eval(atomspace, '(primitive-load "{}")'.format(filepath)) elif method == 'fast': _try_scheme_eval(atomspace, '(use-modules (opencog persist-file))') _scheme_eval(atomspace, '(load-file "{}")'.format(filepath)) elif method == 'python': _load_file(filepath, atomspace) # Report if verbose: print('Imported an Atomspace with {} Atoms from filepath "{}".'.format( atomspace.size(), filepath)) return atomspace
[docs]def store(atomspace, filepath, method='basic', verbose=False, overwrite=False): """Store an Atomspace to a given filepath with a chosen method. Parameters ---------- filepath : str Path of a text file that contains Atomese code. method : str Possible values: - ``basic``: uses Scheme code ``(export-all-atoms "data.scm")`` - ``file-storage-node``: uses Scheme code ``(define fsn (FileStorageNode "data.scm"))`` Caution: This command adds a new ``FileStorageNode`` to the AtomSpace, which will also appear in the exported file. verbose : bool if ``True``, a short message is printed about the AtomSpace that was loaded. overwrite : bool If False, a ``FileExistsError`` will be raised in case there is already a file with the given filepath. if ``True``, the file will be overwritten. References ---------- - AtomSpace example: `persist-file.scm <https://github.com/opencog/atomspace/blob/master/examples/atomspace/persist-file.scm>`__ - OpenCog Wiki: `StorageNode <https://wiki.opencog.org/w/StorageNode>`__ """ # Argument processing _check_arg(filepath, 'filepath', str) _check_arg(method, 'method', str, ['basic', 'file-storage-node']) _check_arg(verbose, 'verbose', bool) # Checks if not overwrite: check_if_file_exists(filepath) # Load modules _load_storage_modules(atomspace) # Export to file if method == 'basic': _scheme_eval(atomspace, '(export-all-atoms "{}")'.format(filepath)) elif method == 'file-storage-node': _scheme_eval(atomspace, '(define fsn (FileStorageNode "{}"))'.format(filepath)) _scheme_eval(atomspace, '(cog-open fsn)') _scheme_eval(atomspace, '(store-atomspace fsn)') _scheme_eval(atomspace, '(cog-close fsn)') # Report if verbose: print('Exported an Atomspace with {} Atoms to filepath "{}".'.format( atomspace.size(), filepath))
[docs]def export(graph, filepath, overwrite=False): """Export a NetworkX graph in a chosen format. Parameters ---------- graph : NetworkX Graph, NetworkX DiGraph filepath : str Possible file extensions: - ``gml``: Export the graph as `Graph Modeling Language (GML) <https://en.wikipedia.org/wiki/Graph_Modelling_Language>`__ file. - ``gml.gz``: Export the graph as GML file compressed with `gzip <https://en.wikipedia.org/wiki/Gzip>`__. - ``gml.bz2``: Export the graph as GML file compressed with `bzip2 <https://en.wikipedia.org/wiki/Bzip2>`__. It takes longer than gzip but produces a bit smaller files. References ---------- - NetworkX documentation: `networkx.readwrite.gml.write_gml <https://networkx.org/documentation/stable/reference/readwrite/generated/networkx.readwrite.gml.write_gml.html>`__ """ # Argument processing _check_arg(graph, 'graph', (_nx.Graph, _nx.DiGraph)) _check_arg(filepath, 'filepath', str) _check_arg(overwrite, 'overwrite', bool) # Checks known_formats = ['gml', 'gml.gz', 'gml.bz2'] if not any(filepath.endswith(x) for x in known_formats): message = 'The filepath "{}" does not end in any known format: {}'.format( filepath, ', '.join(known_formats)) raise ValueError(message) if not overwrite: check_if_file_exists(filepath) # Export _nx.readwrite.gml.write_gml(graph, filepath)
def _try_scheme_eval(atomspace, code): """Try to run a given Scheme code fragment and ignore any RuntimeError.""" try: _scheme_eval(atomspace, code) except RuntimeError: pass def _load_all_modules(atomspace): """Load OpenCog modules that are often used and cleanup the Atomspace. References ---------- - https://github.com/opencog/agi-bio/tree/master/bioscience - https://github.com/opencog/rocca/blob/master/rocca/agents/core.py """ cmds = ( '(use-modules (opencog))', '(use-modules (opencog persist))', '(use-modules (opencog persist-file))', '(use-modules (opencog bioscience))', '(use-modules (opencog spacetime))', '(use-modules (opencog miner))', '(use-modules (opencog pln))', ) for cmd in cmds: _try_scheme_eval(atomspace, cmd) # Remove Atoms, e.g. miner introduces three (LambdaLink, PresentLink, VariableNode) atomspace.clear() def _load_storage_modules(atomspace): """Load OpenCog modules that might be required for storing an Atomspace to a file.""" cmds = ( '(use-modules (opencog))', '(use-modules (opencog persist))', '(use-modules (opencog persist-file))', ) for cmd in cmds: _try_scheme_eval(atomspace, cmd) def check_if_file_exists(filepath): """Check if a filepath already exists. Raises ------ FileExistsError : Raised if there is a file at the given filepath. """ if _os.path.isfile(filepath): raise FileExistsError(_errno.EEXIST, _os.strerror(_errno.EEXIST), filepath)