reacnetgenerator package

ReacNetGenerator is an automatic reaction network generator for reactive molecular dynamics simulation.[Rb972b5c58b34-1]_.

References

[1]

Jinzhe Zeng, Liqun Cao, Chih-Hao Chin, Haisheng Ren, John Z. H. Zhang, Tong Zhu, ReacNetGenerator: an automatic reaction network generator for reactive molecular dynamic simulations, Phys. Chem. Chem. Phys., 2020, 22 (2): 683-691, doi: 10.1039/C9CP05091D.

class reacnetgenerator.ReacNetGenerator(*args, **kwargs)

Bases: object

Factory class for reacnetgenerator.reacnetgen.ReacNetGenerator.

reacnetgenerator.run(*, input_path, input_type, atomname, output_dir=None, items=('species', 'reactions', 'network', 'report'), **kwargs)

Run ReacNetGenerator and return artifacts with parameter provenance.

Parameters:

input_pathstr or pathlib.Path or sequence

Input trajectory or bond file(s).

output_dirstr or pathlib.Path

Directory receiving all default-generated artifacts.

input_typestr

ReacNetGenerator input type, such as “dump” or “bond”.

atomnamesequence of str

Element names in the input trajectory.

itemssequence of str, optional

Requested stages. “species” and “reactions” are produced by the core run; “network” and “report” control the optional drawing/report stages.

**kwargs

Additional arguments forwarded to ReacNetGenerator.

Returns:

dict

artifacts maps semantic names to output path strings. provenance contains the JSON-serializable normalized parameters, explicitly supplied parameter names, and requested items.

Subpackages

• reacnetgenerator.schemas package

Submodules

reacnetgenerator.commandline module

Command line interface for reacnetgenerator.

reacnetgenerator.commandline.main_parser() → ArgumentParser

Return main parser.

Returns:

argparse.ArgumentParser

reacnetgenerator cli parser

reacnetgenerator.commandline.parm2cmd(pp: dict) → list[str]

Convert a parameter dictionary to command line arguments.

Parameters:

ppdict

Parameter dictionary

Returns:

List[str]

Command line arguments

reacnetgenerator.dps module

Connect molecule with Depth-First Search.

reacnetgenerator.dps.dps()

Connect molecule with Depth-First Search.

Parameters:

bondslist of list

The bonds of molecule.

levelslist of int

The levels of atoms.

Returns:

list of list

The connected atoms in each molecule.

list of list

The connected bonds in each molecule.

reacnetgenerator.dps.dps_reaction()

Find A+B->C+D reactions.

Parameters:

reactdictlist of dict of list

Two dictionaries of reactions.

Returns:

list of list of list

List of reactions. The secons axis matches the position of species (left or right).

reacnetgenerator.gui module

GUI version of ReacNetGenerator.

class reacnetgenerator.gui.GUI

Bases: object

GUI class.

gui()

Start the GUI.

property root

Return the root of the GUI.

reacnetgenerator.gui.gui()

Open GUI version of ReacNetGenerator.

reacnetgenerator.reacnetgen module

The main module of ReacNetGenerator.

reacnetgenerator.timedoutput module

Compact, read-only access to ReacNetGenerator timeline schema 1.0.

Iterators read numeric columns in blocks and retain one variable-size definition at a time. Closing an iterator releases its file handle. IDs are file-local; consumers should join on IDs rather than infer scientific equality from them.

class reacnetgenerator.timedoutput.Frame(frame: int, source_id: int, source_frame: int, timestep: int)

Bases: object

An analyzed frame and its zero-based source-file frame.

frame: int

source_frame: int

source_id: int

timestep: int

class reacnetgenerator.timedoutput.Molecule(molecule_id: int, species_id: int, atom_index: tuple[int, ...], bonds: tuple[tuple[int, int, int], ...])

Bases: object

One definition; atom indices use RNG’s zero-based canonical ordering.

atom_index: tuple[int, ...]

bonds: tuple[tuple[int, int, int], ...]

molecule_id: int

species_id: int

class reacnetgenerator.timedoutput.MoleculeRange(molecule_id: int, start_frame: int, end_frame: int)

Bases: object

One closed interval of a molecule’s effective analysis signal.

end_frame: int

molecule_id: int

start_frame: int

class reacnetgenerator.timedoutput.ReactionEvent(transition: int, reaction_type_id: int, count: int)

Bases: object

An aggregated reaction type at transition frame -> frame + 1.

count: int

reaction_type_id: int

transition: int

reacnetgenerator.timedoutput.compare_semantic_manifests(left, right)

Return JSON-pointer paths whose values differ between two manifests.

reacnetgenerator.timedoutput.iter_frames(filename, *, block_rows=8192)

Yield frame mappings in analysis order, reading at most a block at once.

reacnetgenerator.timedoutput.iter_molecule_ranges(filename, *, block_rows=8192)

Yield compact closed ranges in molecule order, without frame expansion.

reacnetgenerator.timedoutput.iter_molecules(filename, *, block_rows=8192)

Yield definitions, retaining a block of IDs and one molecule’s graph.

A single molecule’s atom and bond payload can exceed block_rows. The component-size guard, not this reader’s block size, bounds that graph.

reacnetgenerator.timedoutput.iter_reaction_events(filename, *, block_rows=8192)

Yield aggregate events in transition order, without repeating counts.

reacnetgenerator.timedoutput.iter_reaction_types(filename)

Yield (type_id, reactant, product, total_count) dictionary entries.

reacnetgenerator.timedoutput.iter_species(filename)

Yield (species_id, name) pairs, reading one UTF-8 name at a time.

reacnetgenerator.timedoutput.read_metadata(filename)

Read configuration and format attributes without loading result tables.

This performs header checks only, not full semantic validation. Dataset IDs and offsets are public schema fields; a separate validator is planned.

reacnetgenerator.timedoutput.read_schema_descriptor()

Read the installed machine-readable descriptor for schema 1.0.

reacnetgenerator.timedoutput.semantic_manifest(filename, *, include_provenance=False, block_rows=8192)

Return a deterministic manifest for a validated timeline’s meaning.

reacnetgenerator.timedoutput.validate_timed_output(filename, *, block_rows=8192)

Validate one complete timeline and return its verified table sizes.

reacnetgenerator.timedoutputcheck module

Command-line validation and semantic comparison for timeline artifacts.

reacnetgenerator.timedoutputcheck.main(argv=None)

Run validation; return 0 for success and 1 for invalid or different data.

reacnetgenerator.tools module

Useful methods to futhur process ReacNetGenerator results.

reacnetgenerator.tools.calculate_rate(specfile: str | Path, reacfile: str | Path, cell: ndarray, timestep: float) → dict[str, float]

Calculate the rate constant of each reaction.

The rate constants are calculated by the method developed in [1]. The time interval of the trajectory is assumed to be uniform.

Parameters:

specfilestr

The species file.

reacfilestr

The reactions file.

cellnp.ndarray

The cell with the shape (3, 3). Unit: Angstrom.

timestepfloat

The time step. Unit: femtosecond.

Returns:

ratesDict[str, float]

The rate of each reaction. The dict key is the reaction SMILES. The value is in unit of [(cm^3/mol)^(n-1)s^(-1)], where n is the reaction order.

References

[1]

Yanze Wu, Huai Sun, Liang Wu, Joshua D. Deetz, Extracting the mechanisms and kinetic models of complex reactions from atomistic simulation data, J. Comput. Chem. 40, 16, 1586-1592.

Examples

>>> cell = np.eye(3) * 3.7601e1 # in unit Angstrom
>>> timestep = 0.1 # in unit fs
>>> rates = calculate_rate('methane.species', 'methane.reactionabcd', cell, timestep)

reacnetgenerator.tools.read_reactions(reacfile: str | Path) → list[tuple[int, Counter, str]]

Read reactions from the reactions file (ends with .reaction or .reactionsabcd).

For accuracy, HMM filter should be disabled.

Parameters:

reacfilestr or Path

The reactions file.

Returns:

occsList[Tuple[int, Counter, str]]

The number of occurences of each reaction. The tuple is (occurence, counter_reactants, reaction).

reacnetgenerator.tools.read_species(specfile: str | Path) → tuple[ndarray, dict[str, ndarray]]

Read species from the species file (ends with .species).

For accuracy, HMM filter should be disabled.

Parameters:

specfilestr or Path

The species file.

Returns:

step_idxnp.ndarray

The index of the step.

n_speciesDict[str, np.ndarray]

The number of species in each step. The dict key is the species SMILES.

Examples

Plot the number of methane in each step.

>>> from reacnetgenerator.tools import read_species
>>> import matplotlib.pyplot as plt
>>> step_idx, n_species = read_species('methane.species')
>>> plt.plot(step_idx, n_species['[H][C]([H])([H])[H]'])
>>> plt.savefig("methane.svg")

reacnetgenerator.utils module

Provide utils for ReacNetGenerator.

class reacnetgenerator.utils.SCOUROPTIONS

Bases: object

Scour (SVG optimization) options.

enable_viewboxing = True

newlines = False

remove_descriptions = True

remove_descriptive_elements = True

remove_metadata = True

remove_titles = True

shorten_ids = True

strip_comments = True

strip_ids = True

strip_xml_prolog = True

strip_xml_space_attribute = True

class reacnetgenerator.utils.SharedRNGData(rng: reacnetgenerator.ReacNetGenerator, usedRNGKeys: list[str], returnedRNGKeys: list[str], extraNoneKeys: list[str] | None = None)

Bases: object

Share ReacNetGenerator data with a class of the submodule.

Parameters:

rng: reacnetgenerator.ReacNetGenerator

The centered ReacNetGenerator class.

usedRNGKeys: list of strs

Keys that needs to pass from ReacNetGenerator class to the submodule.

returnedRNGKeys: list of strs

Keys that needs to pass from the submodule to ReacNetGenerator class.

extraNoneKeys: list of strs, optional, default: None

Set keys to None, which will be used in the submodule.

returnkeys() → None

Return back keys to ReacNetGenerator class.

class reacnetgenerator.utils.WriteBuffer(f: IO, linenumber: int = 1200, sep: AnyStr | None = None)

Bases: Generic

Store a buffer for writing files.

It is expensive to write to a file, so we need to make a buffer.

Parameters:

f: fileObject

The file object to write.

linenumber: int, default: 1200

The number of contents to store in the buffer. The buffer will be flushed if it exceeds the set number.

sep: str or bytes, default: None

The separator for contents. If None (default), there will be no separator.

append(text: AnyStr) → None

Append a text.

Parameters:

textstr or bytes

The text to be appended.

buff: list[AnyStr]

check() → None

Check if the number of stored contents exceeds.

If so, the buffer will be flushed.

extend(text: Iterable) → None

Extend texts.

Parameters:

textlist of strs or bytes

Texts to be extended.

flush() → None

Flush the buffer.

reacnetgenerator.utils.appendIfNotNone(f: WriteBuffer[str] | ExitStack, wbytes: str | None) → None

reacnetgenerator.utils.appendIfNotNone(f: WriteBuffer[bytes] | ExitStack, wbytes: bytes | None) → None

Append a line to a file if the line is not None.

Parameters:

fWriteBuffer

The file to write.

wbytesstr or bytes

The line to write.

reacnetgenerator.utils.bytestolist(x: bytes) → Any

Convert a compressed line to an object.

Parameters:

xbytes

The compressed line.

Returns:

object

The decompressed object.

reacnetgenerator.utils.check_zero_signal(signal: ndarray) → bool

Check if the given signal contains only zeros.

Parameters:

signal1D array of bool

The signal to check. The dtype should be bool.

Returns:

bool

False if the signal contains only zeros, True otherwise.

reacnetgenerator.utils.checksha256(filename: str, sha256_check: str | list[str])

Check sha256 of a file is correct.

Parameters:

filenamestr

The filename.

sha256_checkstr or list of strs

The sha256 to be checked.

Returns:

bool

Indicate whether sha256 is correct.

reacnetgenerator.utils.compress(x: str | bytes) → bytes

Compress the line.

This function reduces IO overhead to speed up the program. The functions will use lz4 to compress, since lz4 has better performance that any others.

The compressed format is size + data + size + data + …, where size is a 64-bit little-endian integer.

Parameters:

xstr or bytes

The line to compress.

Returns:

bytes

The compressed line, with a linebreak in the end.

reacnetgenerator.utils.decompress(x: bytes, isbytes: bool = False) → str | bytes

Decompress the line.

Parameters:

xbytes

The line to decompress.

isbytesbool, optional, default: False

If the decompressed content is bytes. If not, the line will be decoded.

Returns:

str or bytes

The decompressed line.

async reacnetgenerator.utils.download_file(urls: str | list[str], pathfilename: str, sha256: str | None) → str

Download files from remote urls if not exists.

Parameters:

urls: str or list of strs

The url(s) that is available to download.

pathfilename: str

The downloading path of the file.

sha256: str

Sha256 of the file. If not None and match the file, the download will be skiped.

Returns:

pathfilename: str

The downloading path of the file.

reacnetgenerator.utils.download_multifiles(urls: list[dict]) → None

Download multiple files from dicts.

Parameters:

urlslist of dicts

The information of download files. Each dict should contain the following key:

• url: str or list of strs

  The url(s) that is available to download.

• pathfilename: str

  The downloading path of the file.

• sha256: str, optional, default: None

  Sha256 of the file. If not None and match the file, the download will be skiped.

async reacnetgenerator.utils.gather_download_files(urls: list[dict]) → None

Asynchronously download files from remote urls if not exists.

See download_multifiles function for details.

See also

download_multifiles

reacnetgenerator.utils.get_timestep_value(timestep: Any) → Any

Normalize stored timestep metadata to the timestep value.

reacnetgenerator.utils.idx_to_signal(idx: ndarray, step: int)

Convert an index array to a signal array.

Parameters:

idxarray_like

Index array.

stepint

Step size.

Returns:

signalndarray

Signal array in int8.

reacnetgenerator.utils.listtobytes(x: Any) → bytes

Convert an object to a compressed line.

Parameters:

xobject

The object to convert, such as numpy.ndarray.

Returns:

bytes

The compressed line.

reacnetgenerator.utils.listtostirng(l: str | list | tuple | ndarray, sep: list[str] | tuple[str, ...]) → str

Convert a list to string, that is easier to store.

Parameters:

lstr or array-like

The list to convert, which can contain any number of dimensions.

seplist of strs

The seperators for each dimension.

Returns:

str

The converted string.

reacnetgenerator.utils.multiopen(pool: multiprocessing.pool.Pool, func: Callable, l: IO, semaphore: multiprocessing.synchronize.Semaphore | None = None, nlines: int | None = None, unordered: bool = True, return_num: bool = False, start: int = 0, extra: Any | None = None, interval: int | None = None, bar: bool = True, desc: str | None = None, unit: str = 'it', total: int | None = None) → Iterable

Return an interated object for process a file with multiple processors.

Parameters:

poolmultiprocessing.Pool

The pool for multiprocessing.

funcfunction

The function to process lines.

lFile object

The file object.

semaphoremultiprocessing.Semaphore, optional, default: None

The semaphore to acquire. If None (default), the object will be passed without control.

nlinesint, optional, default: None

The number of lines to pass to the function each time. If None (default), only one line will be passed to the function.

unorderedbool, optional, default: True

Whether the process can be unordered.

return_numbool, optional, default: False

If True, adds a counter to an iterable.

startint, optional, default: 0

The start number of the counter.

extraobject, optional, default: None

The extra object passed to the item.

intervalint, optional, default: None

The interval of items that will be passed to the function. For example, if set to 10, a item will be passed once every 10 items and others will be dropped.

barbool, optional, default: True

If True, show a tqdm bar for the iteration.

descstr, optional, default: None

The description of the iteration shown in the bar.

unitstr, optional, default: it

The unit of the iteration shown in the bar.

totalint, optional, default: None

The total number of the iteration shown in the bar.

Returns:

object

An object that can be iterated.

reacnetgenerator.utils.must_be_list(obj: Any | list[Any]) → list[Any]

Convert a object to a list if the object is not a list.

Parameters:

objObject

The object to convert.

Returns:

obj: list

If the input object is not a list, returns a list that only contains that object. Otherwise, returns that object.

reacnetgenerator.utils.produce(semaphore: multiprocessing.synchronize.Semaphore, plist: Iterable[Any], parameter: Any) → Generator[tuple[Any, Any], None, None]

Item producer with a semaphore.

Prevent large memory usage due to slow IO.

Parameters:

semaphoremultiprocessing.Semaphore

The semaphore to acquire.

plistlist of objects

The list of items to be passed.

parameterobject

The parameter yielded with each item.

Yields:

item: object

The item to be yielded.

parameter: object

The parameter yielded with each item.

reacnetgenerator.utils.read_compressed_block(f: BinaryIO) → Generator[bytes, None, None]

Read compressed binary file, assuming the format is size + data + size + data + …

Parameters:

ffileObject

The file object to read.

Yields:

data: bytes

The compressed block.

reacnetgenerator.utils.run_mp(nproc: int, *, max_inflight: int | None = None, disk_ordered: bool = False, ordered_spool_dir: str | None = None, initializer: Callable | None = None, initargs: tuple = (), **kwargs: Any) → Generator[Any, None, None]

Process a file with multiple processors.

Parameters:

nprocint

The number of processors to be used.

max_inflightint, optional

Maximum number of submitted inputs not yet delivered to the consumer. If omitted, the existing nproc * 150 semaphore path is unchanged.

disk_orderedbool, optional, default: False

Run workers to completion out of order, spool early results to disk, and yield them in input order. This requires unordered=False and an exact total.

ordered_spool_dirstr, optional

Parent directory for temporary ordered-result files.

initializercallable, optional

Called once in each worker before it receives tasks. Use this to attach shared data so individual tasks need only carry indices. Requires bounded execution (max_inflight or disk_ordered), whose worker-exit detection prevents failed initializers from hanging the consumer.

initargstuple, optional

Arguments passed to initializer.

**kwargsdict, optional

Other parameters can be found in the multiopen method.

Yields:

object

The yielded object from the multiopen method.

See also

multiopen

reacnetgenerator.utils_np module