ZnTrack

https://github.com/zincware/zntrack
Admin
ZnTrack is a lightweight Python package for converting existing Python code into reproducible...
Tokens:11,660
Snippets:52
Trust Score:7.4
License:Apache-2.0
Update:1 year ago
Show doc for...
Tokens:
Raw
### Initialize ZnTrack Project and DVC

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Commands to create a new project directory, initialize Git, and set up DVC for data pipeline management. This is the foundational step for any ZnTrack project.

```bash
mkdir my_project
cd my_project
git init
dvc init
```

--------------------------------

### Python: ZnTrack Project and Node Grouping

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Demonstrates how to create a ZnTrack project, add nodes, and organize them into groups and sub-groups. It also shows how names are generated based on grouping.

```python
project = zntrack.Project()

with project:
    add1 = Add(a=1, b=2)
    print(add1.name)
    >>> Add

with project.group("grp"):
    add2 = Add(a=1, b=2)
    print(add2.name)
    >>> grp_Add

with project.group("grp", "subgrp"):
    add3 = Add(a=3, b=4)
    print(add3.name)
    >>> grp_subgrp_Add

project.build()
```

--------------------------------

### JSON File: zntrack.json Configuration

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This JSON file contains metadata and configuration specific to ZnTrack, including project settings, node definitions, and potentially information about the tracked workflow history and reproducibility.

```JSON
{
  "project_name": "zntrack_example",
  "nodes": {
    "Smiles2Conformers": {
      "class": "pack_box.src.Smiles2Conformers",
      "parameters": {
        "smiles": "CCO",
        "n_conformers": 5
      }
    },
    "Pack": {
      "class": "pack_box.src.Pack",
      "parameters": {
        "input_file": "conformers.sdf"
      }
    },
    "StructureOptimization": {
      "class": "pack_box.src.StructureOptimization",
      "parameters": {
        "input_structure": "packed_structure.xyz"
      }
    },
    "MACE_MP": {
      "class": "pack_box.src.MACE_MP",
      "parameters": {
        "input_file": "packed_structure.xyz"
      }
    }
  },
  "dvc": {
    "outs": [
      "conformers.sdf",
      "packed_structure.xyz",
      "optimized_structure.xyz",
      "mace_result.json"
    ]
  }
}

```

--------------------------------

### Install ZnTrack using pip

Source: https://github.com/zincware/zntrack/blob/main/docs/source/index.rst

Installs the ZnTrack package using the pip package installer. This is the standard method for adding ZnTrack to your Python environment.

```bash
pip install zntrack
```

--------------------------------

### Bash: ZnTrack MLFlow Sync CLI Help

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Displays the help message for the zntrack mlflow-sync command-line interface. This command is used for integrating ZnTrack with MLFlow, allowing the upload of existing runs.

```bash
zntrack mlflow-sync --help
```

--------------------------------

### Basic Python function example

Source: https://github.com/zincware/zntrack/blob/main/docs/source/index.rst

A simple Python function that adds two integers and prints the result. This serves as a basic example before introducing ZnTrack's workflow capabilities.

```python
def add(a: int, b: int) -> int:
    return a + b

if __name__ == "__main__":
   x = add(1, 2)
   print(x)
   >>> 3
```

--------------------------------

### Execute ZnTrack Workflow with DVC

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Command to execute the defined ZnTrack workflow using DVC's reproduction command. This command triggers the execution of all nodes in the pipeline based on their dependencies.

```bash
dvc repro
```

--------------------------------

### Create Basic Project Structure

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Creates essential files and directories for a ZnTrack project, including the main workflow file 'main.py' and the 'src' directory for Node definitions. This structure helps organize workflow logic and custom node implementations.

```bash
touch main.py
mkdir src
touch src/__init__.py
```

--------------------------------

### Python Script: ZnTrack Main Workflow Execution

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This Python script orchestrates the ZnTrack workflow by instantiating and connecting the different Node objects. It defines the dependencies between nodes, creating the directed graph for execution.

```Python
from pathlib import Path

from pack_box.src import Smiles2Conformers, Pack, StructureOptimization, MACE_MP
from zntrack import Project

# Define output directory
project = Project()
project.dvc.add("dvc.yaml", "params.yaml", "zntrack.json")

# Define the workflow nodes
smiles = "CCO"
n_conformers = 5

# Node 1: Generate conformers from SMILES string
confs = Smiles2Conformers(smiles=smiles, n_conformers=n_conformers)

# Node 2: Pack molecules into a box
# Assuming confs.output('conformers.sdf') is the input file for Pack
pack = Pack(input_file=confs.output("conformers.sdf"))

# Node 3: Structure optimization using MACE-MP
# Assuming pack.output('packed_structure.xyz') is the input for optimization
opt = StructureOptimization(input_structure=pack.output("packed_structure.xyz"))

# Node 4: Direct MACE-MP calculation (alternative or complementary path)
# Assuming some initial structure file (e.g., from pack)
# This could also depend on the output of 'pack' or another source
mace_calc = MACE_MP(input_file=pack.output("packed_structure.xyz"))

# Define dependencies (implicitly done by passing outputs as inputs)
# The graph structure is defined by how nodes are connected:
# Smiles2Conformers --> Pack --> StructureOptimization
# MACE_MP --> StructureOptimization (This connection needs explicit definition if StructureOptimization takes MACE output)

# Example of running the workflow (typically done via dvc commands)
if __name__ == "__main__":
    # To run this, you would typically use dvc repro from the command line
    # after defining the graph and dependencies correctly.
    # For demonstration, we can manually trigger node runs (not recommended for production):
    print("Running workflow...")
    confs.run()
    pack.run()
    opt.run()
    mace_calc.run()
    print("Workflow finished.")

```

--------------------------------

### YAML File: dvc.yaml Configuration

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This YAML file defines the Directed acyclic graph (DAG) for the DVC (Data Version Control) pipeline. It specifies the stages, their dependencies, and the commands to execute for each stage, integrating with ZnTrack nodes.

```YAML
stages:
  generate_conformers:
    cmd: python -m pack_box.src.Smiles2Conformers --smiles CCO --n_conformers 5
    deps:
      - pack_box/src/__init__.py
    outs:
      - conformers.sdf

  pack_molecules:
    cmd: python -m pack_box.src.Pack --input_file conformers.sdf
    deps:
      - pack_box/src/__init__.py
      - conformers.sdf
    outs:
      - packed_structure.xyz

  optimize_structure:
    cmd: python -m pack_box.src.StructureOptimization --input_structure packed_structure.xyz
    deps:
      - pack_box/src/__init__.py
      - packed_structure.xyz
    outs:
      - optimized_structure.xyz

  mace_calculation:
    cmd: python -m pack_box.src.MACE_MP --input_file packed_structure.xyz
    deps:
      - pack_box/src/__init__.py
      - packed_structure.xyz
    outs:
      - mace_result.json

```

--------------------------------

### YAML File: params.yaml Configuration

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This YAML file stores parameters used in the DVC pipeline. It allows for easy modification and versioning of hyperparameters, such as the number of conformers or specific settings for simulation steps.

```YAML
smiles: CCO
n_conformers: 5

```

--------------------------------

### Define Workflow with ZnTrack Project Context Manager

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Python code demonstrating how to define a workflow using ZnTrack's Project context manager. It instantiates 'Add' and 'Multiply' Nodes, connecting their outputs to inputs to create a computational graph. `project.build()` generates the necessary configuration files.

```python
import zntrack

from src import Add, Multiply

project = zntrack.Project()

with project:
    add1 = Add(a=1, b=2)
    add2 = Add(a=3, b=4)
    add3 = Multiply(a=add1.result, b=add2.result)

project.build()
```

--------------------------------

### Python Script: ZnTrack Converted Workflow - Initialization

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This Python script defines the initialization of the ZnTrack workflow, likely setting up the base Node structure and potentially defining common parameters or utilities used across the directed graph.

```Python
from zntrack import Node, dvc
from pathlib import Path

class Smiles2Conformers(Node):
    def __init__(self, smiles: str, n_conformers: int = 10, **kwargs):
        super().__init__(**kwargs)
        self.smiles = smiles
        self.n_conformers = n_conformers

    def run(self):
        # Placeholder for actual conformer generation logic
        print(f"Generating {self.n_conformers} conformers for {self.smiles}")
        # In a real scenario, this would involve RDKit or similar tools
        # and save the conformers to a file (e.g., SDF)
        self.output("conformers.sdf").write_text("# SDF data placeholder")

class Pack(Node):
    def __init__(self, input_file: Path, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file

    def run(self):
        # Placeholder for actual packing logic using Packmol or similar
        print(f"Packing molecules from {self.input_file}")
        # This would involve running an external tool like Packmol
        self.output("packed_structure.xyz").write_text("# XYZ data placeholder")

class StructureOptimization(Node):
    def __init__(self, input_structure: Path, **kwargs):
        super().__init__(**kwargs)
        self.input_structure = input_structure

    def run(self):
        # Placeholder for structure optimization logic (e.g., using MACE)
        print(f"Optimizing structure from {self.input_structure}")
        # This would involve running a computational chemistry code
        self.output("optimized_structure.xyz").write_text("# Optimized XYZ data placeholder")

class MACE_MP(Node):
    def __init__(self, input_file: Path, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file

    def run(self):
        # Placeholder for MACE-MP calculation logic
        print(f"Running MACE-MP calculation on {self.input_file}")
        # This would involve running the MACE-MP model
        self.output("mace_result.json").write_text("{}")

```

--------------------------------

### Original Scikit-learn Classifier Comparison Code

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

The original Python code for comparing scikit-learn classifiers, used as a basis for the ZnTrack adaptation. This snippet requires scikit-learn to be installed.

```Python
import matplotlib.pyplot as plt
import numpy as np
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier
from sklearn.gaussian_process import GaussianProcessClassifier
from sklearn.gaussian_process.kernels import RBF
from sklearn.model_selection import train_test_split
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.svm import SVC
from sklearn.neural_network import MLPClassifier
from sklearn.discriminant_analysis import QuadraticDiscriminantAnalysis


# Datasets
# Create datasets for classification tasks
def datasets():
    X, y = make_classification(
        n_samples=100, n_features=2, n_informative=2, n_redundant=0, random_state=42
    )
    return X, y


# Classifiers
# Define a list of classifiers to compare, including their configurations
class_names = [
    "Random Forest classifier",
    "SVC",
    "Decision Tree classifier",
    "KNeighbors classifier",
    "Gaussian Process classifier (RBF)",
    "Quadratic Discriminant Analysis",
    "MLP Classifier",
]

classifiers = [
    RandomForestClassifier(max_depth=5, n_estimators=10, max_features=1),
    SVC(gamma=2, C=1),
    DecisionTreeClassifier(max_depth=5),
    KNeighborsClassifier(3),
    GaussianProcessClassifier(1.0 * RBF(1.0)),
    QuadraticDiscriminantAnalysis(),
    MLPClassifier(alpha=1, max_iter=1000),
]


X, y = datasets()
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)


for name, clf in zip(class_names, classifiers):
    # Train the classifier
    clf.fit(X_train, y_train)
    # Score the classifier
    score = clf.score(X_test, y_test)
    print(f"{name}: {score:.2f}")

```

--------------------------------

### Handle Mutable Defaults with Zntrack default_factory (Python)

Source: https://context7.com/zincware/zntrack/llms.txt

This code example illustrates the use of `default_factory` in Zntrack for mutable default parameters, such as lists or dictionaries. The `DataCollector` node uses `default_factory=list` to ensure each instance gets a fresh list, avoiding shared mutable state issues.

```python
import zntrack

class DataCollector(zntrack.Node):
    # Use default_factory for mutable defaults
    initial_data: list = zntrack.params(default_factory=list)
    multiplier: int = zntrack.params(2)

    collected: list = zntrack.outs()

    def run(self):
        if not self.initial_data:
            self.collected = [0, 1, 2, 3, 4]
        else:
            self.collected = [x * self.multiplier for x in self.initial_data]

project = zntrack.Project()

with project:
    # Node with empty list default
    collector1 = DataCollector()

    # Node with custom initial data
    collector2 = DataCollector(
        initial_data=[10, 20, 30],
        multiplier=3
    )

project.repro()

print(collector1.collected)  # [0, 1, 2, 3, 4]
print(collector2.collected)  # [30, 60, 90]
```

--------------------------------

### Python Script: Original Packmol Workflow

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/pack_box.rst

This Python script represents the original atomistic simulation workflow before conversion to ZnTrack. It likely handles tasks such as generating molecular conformers and packing them into a simulation box using external tools.

```Python
import subprocess

def run_packmol(input_file, output_file):
    """Runs Packmol with the given input file and saves the output."""
    command = ["packmol"]
    # Add arguments for input and output files
    command.extend(["<", input_file, ">", output_file])
    subprocess.run(command, check=True)

if __name__ == "__main__":
    input_xyz = "input.xyz"
    output_xyz = "output.xyz"
    # Create a dummy input file for demonstration
    with open(input_xyz, "w") as f:
        f.write("3\n\n")
        f.write("ATOM C 0.0 0.0 0.0\n")
        f.write("ATOM C 1.0 0.0 0.0\n")
        f.write("ATOM C 0.0 1.0 0.0\n")

    run_packmol(input_xyz, output_xyz)
    print(f"Packmol simulation completed. Output saved to {output_xyz}")

```

--------------------------------

### ZnTrack Workflow for Classifier Comparison

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

Python code defining the main ZnTrack workflow for running the classifier comparison. This script orchestrates the execution of different classifier nodes and defines the overall pipeline. This requires ZnTrack to be installed.

```Python
from zntrack import Project

from classifier_comparison.src import get_classifiers


project = Project()
project.dvc.add("data/classification.csv")
project.dvc.add("params.yaml")

for clf in get_classifiers():
    project.add(clf)

project.run()

```

--------------------------------

### ZnTrack Nodes for Classifier Comparison

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

Python code defining ZnTrack Nodes for the classifier comparison workflow. These nodes encapsulate specific tasks like data loading, splitting, and classifier training. This requires ZnTrack to be installed.

```Python
from typing import Tuple

import numpy as np
from sklearn.model_selection import train_test_split
from zntrack import Node, dvc.deps, dvc.outs, zn


@zn.load(deps=dvc.deps.deps.from_paths("data/classification.csv"))
def load_data(data_path: str) -> Tuple[np.ndarray, np.ndarray]:
    """Loads data from a CSV file."""
    return np.loadtxt(data_path), np.loadtxt(data_path, dtype=int, usecols=-1)


@zn.load(deps=dvc.deps.deps.from_paths("params.yaml"))
def split_data(
    X: np.ndarray,
    y: np.ndarray,
    test_size: float,
    random_state: int,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Splits data into training and testing sets."""
    return train_test_split(X, y, test_size=test_size, random_state=random_state)


class Classifier(Node):
    def __init__(self, name: str, **kwargs):
        super().__init__(**kwargs)
        self.name = name

    def train(
        self, X_train, y_train, X_test, y_test, params: dict
    ) -> Tuple[float, dict]:
        """Trains the classifier and returns its score and model parameters."""
        raise NotImplementedError

    def run(self):
        X, y = load_data()
        X_train, X_test, y_train, y_test = split_data(
            X=X, y=y, test_size=self.params["test_size"], random_state=self.params["random_state"]
        )

        score, params = self.train(X_train, y_train, X_test, y_test, self.params)

        self.save_outputs(score=score, params=params)


def get_classifiers() -> list[Classifier]:
    """Returns a list of configured classifier nodes."""
    return [
        RandomForestClassifier(
            name="rf",
            params={
                "test_size": 0.3,
                "random_state": 42,
                "max_depth": 5,
                "n_estimators": 10,
                "max_features": 1,
            },
        ),
        SVC(
            name="svc",
            params={"test_size": 0.3, "random_state": 42, "gamma": 2, "C": 1},
        ),
        DecisionTreeClassifier(
            name="dt",
            params={"test_size": 0.3, "random_state": 42, "max_depth": 5},
        ),
        KNeighborsClassifier(
            name="knn",
            params={"test_size": 0.3, "random_state": 42, "n_neighbors": 3},
        ),
        GaussianProcessClassifier(
            name="gpc",
            params={"test_size": 0.3, "random_state": 42, "kernel": "RBF"},
        ),
        QuadraticDiscriminantAnalysis(
            name="qda", params={"test_size": 0.3, "random_state": 42}
        ),
        MLPClassifier(
            name="mlp",
            params={"test_size": 0.3, "random_state": 42, "alpha": 1, "max_iter": 1000},
        ),
    ]

```

--------------------------------

### Load ZnTrack Results with Python

Source: https://github.com/zincware/zntrack/blob/main/README.md

Demonstrates how to load the results of a ZnTrack workflow, specifically a StructureOptimization node, using the `zntrack.from_rev` function. It shows how to access the optimized trajectory data. This snippet requires the zntrack library to be installed.

```python
import zntrack

# Load the StructureOptimization Node
optm = zntrack.from_rev(name="StructureOptimization")
# you can pass `remote: str` and `rev: str` to access data from
# a different commit or a remote repository.

# Access the optimization trajectory
print(optm.frames)
```

--------------------------------

### Set Custom Node Name in ZnTrack

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Python code showing how to assign a custom name to a ZnTrack Node instance during its creation. This allows for more descriptive and unique identification of nodes within the workflow, especially when grouping.

```python
add1 = Add(a=1, b=2, name="custom_name")
```

--------------------------------

### Define MACE_MP Calculator Class in Python

Source: https://github.com/zincware/zntrack/blob/main/README.md

Defines a dataclass to represent the MACE-MP calculator. It serves as a dependency for other Nodes and provides a method to get the calculator instance with specified model parameters.

```Python
from dataclasses import dataclass
from pathlib import Path

import ase.io
from ase.optimize import LBFGS
from mace.calculators import mace_mp
from rdkit2ase import pack, smiles2conformers

import zntrack


@dataclass
class MACE_MP:
    model: str = "medium"  # Default model type

    def get_calculator(self, **kwargs):
        return mace_mp(model=self.model)
```

--------------------------------

### Use Custom Run Methods with Zntrack Apply (Python)

Source: https://context7.com/zincware/zntrack/llms.txt

This example demonstrates using alternative execution methods beyond the default `run` method in Zntrack nodes. It defines a `FlexibleNode` with both `run` and `alternative_run` methods and uses `zntrack.apply` to create a new node class that executes the alternative method.

```python
import zntrack

class FlexibleNode(zntrack.Node):
    value: int = zntrack.params(10)
    result: int = zntrack.outs()

    def run(self):
        """Default execution method"""
        self.result = self.value * 2

    def alternative_run(self):
        """Alternative execution method"""
        self.result = self.value ** 2

# Create nodes using different run methods
project = zntrack.Project()

with project:
    # Use default run method
    node1 = FlexibleNode(value=5)

    # Use alternative run method
    AlternativeNode = zntrack.apply(FlexibleNode, "alternative_run")
    node2 = AlternativeNode(value=5)

project.repro()

print(node1.result)  # 10 (5 * 2)
print(node2.result)  # 25 (5 ** 2)
```

--------------------------------

### Define Custom ZnTrack Nodes (Add, Multiply)

Source: https://github.com/zincware/zntrack/blob/main/docs/source/project.rst

Python code defining two custom ZnTrack Nodes: 'Add' for simple addition and 'Multiply' for multiplication. 'Add' uses parameters 'a' and 'b' and outputs 'result'. 'Multiply' depends on other nodes ('zntrack.deps') for its inputs and also outputs 'result'.

```python
import zntrack

class Add(zntrack.Node):
    a: int = zntrack.params()
    b: int = zntrack.params()

    result: int = zntrack.outs()

    def run(self) -> None:
        self.result = self.a + self.b

# Multiply uses ``zntrack.deps`` to process data from other nodes
class Multiply(zntrack.Node):
    a: int = zntrack.deps()
    b: int = zntncack.deps()

    result: int = zntrack.outs()

    def run(self) -> None:
        self.result = self.a * self.b
```

--------------------------------

### Configure Node to Always Run in Python

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node_details.rst

Shows how to configure a Zntrack Node to always execute, regardless of input changes, by setting the `always_changed=True` parameter. This is particularly useful for debugging new nodes. The example uses `zntrack.examples.ParamsToOuts` with this setting.

```python
import zntrack.examples

project = zntrack.Project()

with project:
    node = zntrack.examples.ParamsToOuts(params=42, always_changed=True)

project.repro()
```

--------------------------------

### Working with Dataclass Dependencies

Source: https://context7.com/zincware/zntrack/llms.txt

Demonstrates how to use regular Python dataclasses as dependencies within ZnTrack Nodes. This allows for structured configuration management alongside computational steps. The example shows a `ModelConfig` dataclass used by a `ModelTrainer` Node.

```python
import zntrack
from dataclasses import dataclass
import ase

@dataclass
class ModelConfig:
    """Configuration dataclass"""
    name: str = "GPT2"
    layers: int = 12
    hidden_size: int = 768

class ModelTrainer(zntrack.Node):
    # Use dataclass as dependency
    config: ModelConfig = zntrack.deps()
    epochs: int = zntrack.params(10)

    model_path: str = zntrack.outs_path(zntrack.nwd / "model.pt")

    def run(self):
        # Access config attributes
        print(f"Training {self.config.name} with {self.config.layers} layers")
        # Simulate training
        with open(self.model_path, 'w') as f:
            f.write(f"Model: {self.config.name}\n")

project = zntrack.Project()

with project:
    config = ModelConfig(name="BERT", layers=24, hidden_size=1024)
    trainer = ModelTrainer(config=config, epochs=20)

project.repro()
```

--------------------------------

### Define Off-Graph Nodes with Parameters and Dependencies in Python

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node_details.rst

Illustrates how to define off-graph nodes using dataclasses that utilize `zntrack.params_path` and `zntrack.deps_path`. This allows off-graph nodes to reference configuration files or file dependencies, which is useful for methods that do not have a `run` method themselves. The example shows a `Calculator` dataclass configured via a file and a model path.

```python
from dataclasses import dataclass
import yaml
import zntrack

@dataclass
class Calculator:
    config_file: str = zntrack.params_path()
    model_path: str = zntrack.deps_path()

    def get_calculator(self):
        with open(self.config_file, "r") as f:
            config = yaml.safe_load(f)
        return func(model=self.model_path, **config)
```

--------------------------------

### Define Node Dependencies in ZnTrack

Source: https://context7.com/zincware/zntrack/llms.txt

Establishes workflow graphs by connecting nodes through dependencies in ZnTrack. Nodes declare dependencies using `zntrack.deps()`, allowing data to flow between them. The example shows a DataGenerator, DataProcessor, and DataAnalyzer where outputs of one node serve as inputs to the next.

```python
import zntrack

class DataGenerator(zntrack.Node):
    count: int = zntrack.params(100)
    data: list = zntrack.outs()

    def run(self):
        self.data = list(range(self.count))

class DataProcessor(zntrack.Node):
    # Declare dependency on another node's output
    input_data: list = zntrack.deps()
    squared: list = zntrack.outs()

    def run(self):
        self.squared = [x**2 for x in self.input_data]

class DataAnalyzer(zntrack.Node):
    processed_data: list = zntrack.deps()
    mean: float = zntrack.metrics()

    def run(self):
        self.mean = sum(self.processed_data) / len(self.processed_data)

# Build the workflow graph
project = zntrack.Project()

with project:
    generator = DataGenerator(count=50)
    processor = DataProcessor(input_data=generator.data)
    analyzer = DataAnalyzer(processed_data=processor.squared)

project.repro()

# Access results from any node
print(f"Mean of squared values: {analyzer.mean}")
```

--------------------------------

### Handle File Path Outputs in ZnTrack

Source: https://context7.com/zincware/zntrack/llms.txt

Utilizes file paths for managing large datasets or non-serializable objects in ZnTrack. `zntrack.outs_path` defines output file paths, and `zntrack.deps_path` declares dependencies on file paths from other nodes. The example demonstrates exporting data to a CSV file and then loading and summarizing it.

```python
import zntrack
import pandas as pd
from pathlib import Path

class DataExporter(zntrack.Node):
    rows: int = zntrack.params(1000)

    # Define output file path using nwd (node working directory)
    output_file: Path = zntrack.outs_path(zntrack.nwd / "data.csv")

    def run(self):
        # Generate and save data
        df = pd.DataFrame({
            'x': range(self.rows),
            'y': [i**2 for i in range(self.rows)]
        })
        df.to_csv(self.output_file, index=False)

class DataLoader(zntrack.Node):
    # Depend on file path from another node
    input_file: Path = zntrack.deps_path()
    summary: dict = zntrack.outs()

    def run(self):
        df = pd.read_csv(self.input_file)
        self.summary = {
            'rows': len(df),
            'mean_y': df['y'].mean()
        }

project = zntrack.Project()

with project:
    exporter = DataExporter(rows=500)
    loader = DataLoader(input_file=exporter.output_file)

project.repro()
print(loader.summary)  # {'rows': 500, 'mean_y': 41541.5}
```

--------------------------------

### Define Off-Graph Nodes Using Dataclasses in Python

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node_details.rst

Demonstrates how to define off-graph nodes using Python dataclasses for modularity and reusability. These nodes provide methods but do not produce output files directly when the graph is executed, serving as dependencies for other nodes. This example shows a 'ManipulateNumber' node that uses either a 'Shift' or 'Scale' off-graph node.

```python
from dataclasses import dataclass
import zntrack

@dataclass
class Shift:
    shift: float

    def compute(self, input: float) -> float:
        return input + self.shift

@dataclass
class Scale:
    scale: float

    def compute(self, input: float) -> float:
        return input * self.scale

class ManipulateNumber(zntrack.Node):
    number: float = zntrack.params()
    method: Shift | Scale = zntrack.deps()

    result: float = zntrack.outs()

    def run(self) -> None:
        self.result = self.method.compute(self.number)

if __name__ == "__main__":
    project = zntrack.Project()

    # You can define these Nodes anywhere, but
    # to avoid confusion, they should be placed outside the Project context
    shift = Shift(shift=1.0)
    scale = Scale(scale=2.0)

    with project:
        shifted_number = ManipulateNumber(number=1.0, method=shift)
        scaled_number = ManipulateNumber(number=1.0, method=scale)
    project.repro()
```

--------------------------------

### Initialize ZnTrack Project

Source: https://github.com/zincware/zntrack/blob/main/docs/source/index.rst

Commands to initialize a new ZnTrack project. This involves creating a directory, initializing Git and DVC repositories, and creating the main Python file.

```bash
mkdir my_project
cd my_project
git init
dvc init
touch main.py # Add the code below to main.py
```

--------------------------------

### Initialize Project in Bash

Source: https://github.com/zincware/zntrack/blob/main/README.md

Initializes a new project directory using Git and DVC for version control, which are prerequisites for using ZnTrack.

```Bash
git init
dvc init
```

--------------------------------

### Define and Run a ZnTrack Node

Source: https://github.com/zincware/zntrack/blob/main/docs/source/index.rst

Demonstrates how to define a ZnTrack Node with parameters and outputs, and then execute it within a ZnTrack project. This code converts a simple function into a stateful Node.

```python
import zntrack

class Add(zntrack.Node):
   a: int = zntrack.params()
   b: int = zntrack.params()
   result: int = zntrack.outs()

   def run(self) -> None:
      self.result = self.a + self.b

if __name__ == "__main__":
   project = zntrack.Project()

   with project:
      x = Add(a=1, b=2)

   project.repro()

   print(x.result)
   >>> 3
```

--------------------------------

### Initialize and Build ZnTrack Workflow in Python

Source: https://github.com/zincware/zntrack/blob/main/README.md

Demonstrates initializing a ZnTrack project, defining Nodes (Smiles2Conformers, Pack, StructureOptimization), and building the workflow graph. The `project.repro()` command executes the defined workflow.

```Python
from src import MACE_MP, Pack, Smiles2Conformers, StructureOptimization

import zntrack

# Initialize the ZnTrack project
project = zntrack.Project()

# Define the MACE-MP model
model = MACE_MP()

# Build the workflow graph
with project:
    etoh = Smiles2Conformers(smiles="CCO", numConfs=32)
    box = Pack(data=[etoh.frames], counts=[32], density=789)
    optm = StructureOptimization(model=model, data=box.frames, data_id=-1, fmax=0.5)

# Execute the workflow
project.repro()
```

--------------------------------

### Configuring Project Entry Points for ZnTrack Nodes (TOML)

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node_details.rst

This TOML configuration sets up a project entry point for ZnTrack, allowing your defined Nodes to be discovered. The 'mypackage' alias points to the 'nodes' function within the 'mypackage.nodes' module.

```toml
[project.entry-points."zntrack.nodes"]
mypackage = "mypackage.nodes:nodes"
```

--------------------------------

### Registering ZnTrack Nodes as Entry Points (Python)

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node_details.rst

This Python function defines a list of ZnTrack Node classes to be exposed. It's intended to be registered as a project entry point.

```python
import zntrack

def nodes() -> list[zntrack.Node]:
    return [
        mypackage.MyNode1,
        mypackage.MyNode2,
    ]
```

--------------------------------

### Create Basic Node in ZnTrack

Source: https://context7.com/zincware/zntrack/llms.txt

Defines a computational task as a Node class with parameters and outputs in ZnTrack. The `run` method contains the core logic. The workflow is initialized, nodes are created within a project context, and then executed using `project.repro()`. Results can be accessed after execution.

```python
import zntrack

class AddNumbers(zntrack.Node):
    # Define parameters with default values
    a: int = zntrack.params()
    b: int = zntrack.params(10)

    # Define output field
    result: int = zntrack.outs()

    def run(self):
        """Execute the node's computation"""
        self.result = self.a + self.b

# Initialize a project and create nodes
project = zntrack.Project()

with project:
    node = AddNumbers(a=5, b=3)

# Build and execute the workflow
project.repro()

# Access results
print(node.result)  # Output: 8
```

--------------------------------

### Define a Simple ZnTrack Node for Addition

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node.rst

Demonstrates how to define a basic ZnTrack Node that inherits from zntrack.Node. It takes two integer parameters 'a' and 'b', and defines an integer output 'result'. The 'run' method performs the addition and assigns it to the output.

```python
import zntrack

class Add(zntrack.Node):
    # Define parameters similar to dataclass.Field
    a: int = zntrack.params()
    b: int = zntrack.params()

    # Define an output
    result: int = zntrack.outs()

    def run(self) -> None:
        # Core computation of the Node
        self.result = self.a + self.b
```

--------------------------------

### ZnTrack JSON Configuration File

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

The zntrack.json file contains metadata and information about the ZnTrack project, including node definitions and their relationships. This file is managed by ZnTrack.

```JSON
{
  "project_name": "zntrack",
  "version": "0.1.0",
  "nodes": [
    {
      "name": "rf",
      "class": "classifier_comparison.src.RandomForestClassifier",
      "version": "unknown",
      "uuid": "7281181519642724512",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./rf/score.json",
          "cache": false
        },
        "params": {
          "path": "./rf/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "max_depth": 5,
        "n_estimators": 10,
        "max_features": 1
      }
    },
    {
      "name": "svc",
      "class": "classifier_comparison.src.SVC",
      "version": "unknown",
      "uuid": "-1955735027868396195",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./svc/score.json",
          "cache": false
        },
        "params": {
          "path": "./svc/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "gamma": 2,
        "C": 1
      }
    },
    {
      "name": "dt",
      "class": "classifier_comparison.src.DecisionTreeClassifier",
      "version": "unknown",
      "uuid": "-7416211033180818131",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./dt/score.json",
          "cache": false
        },
        "params": {
          "path": "./dt/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "max_depth": 5
      }
    },
    {
      "name": "knn",
      "class": "classifier_comparison.src.KNeighborsClassifier",
      "version": "unknown",
      "uuid": "2970738473944956783",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./knn/score.json",
          "cache": false
        },
        "params": {
          "path": "./knn/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "n_neighbors": 3
      }
    },
    {
      "name": "gpc",
      "class": "classifier_comparison.src.GaussianProcessClassifier",
      "version": "unknown",
      "uuid": "3648002496916818541",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./gpc/score.json",
          "cache": false
        },
        "params": {
          "path": "./gpc/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "kernel": "RBF"
      }
    },
    {
      "name": "qda",
      "class": "classifier_comparison.src.QuadraticDiscriminantAnalysis",
      "version": "unknown",
      "uuid": "-1600281830869554388",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./qda/score.json",
          "cache": false
        },
        "params": {
          "path": "./qda/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42
      }
    },
    {
      "name": "mlp",
      "class": "classifier_comparison.src.MLPClassifier",
      "version": "unknown",
      "uuid": "-5947216813815836816",
      "deps": {
        "data/classification.csv": "167a120b23394e1b898a11348f6a1b75"
      },
      "outs": {
        "score": {
          "path": "./mlp/score.json",
          "cache": false
        },
        "params": {
          "path": "./mlp/params.json",
          "cache": false
        }
      },
      "params": {
        "test_size": 0.3,
        "random_state": 42,
        "alpha": 1,
        "max_iter": 1000
      }
    }
  ]
}

```

--------------------------------

### Molecular Dynamics Workflow Original Python Script

Source: https://github.com/zincware/zntrack/blob/main/README.md

This Python script sets up a molecular dynamics simulation for Ethanol using ASE, MACE-MP-0, and RDKit2ASE. It requires specific dependencies like MACE, packmol, and rdkit2ase to run.

```python
from ase.optimize import LBFGS
from mace.calculators import mace_mp
from rdkit2ase import pack, smiles2conformers

model = mace_mp()

frames = smiles2conformers(smiles="CCO", numConfs=32)
box = pack(data=[frames], counts=[32], density=789)

box.calc = model

dyn = LBFGS(box, trajectory="optim.traj")
dyn.run(fmax=0.5)
```

--------------------------------

### DVC Configuration File

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

The dvc.yaml file defines the DVC pipeline stages for the ZnTrack workflow. It specifies the dependencies, outputs, and commands for each stage. This file is generated by DVC.

```YAML
stages:
  default:
    cmd: python classifier_comparison/main.py
    deps:
    - data/classification.csv
    - params.yaml
    outs:
    - zntrack.json

```

--------------------------------

### Parameters Configuration File

Source: https://github.com/zincware/zntrack/blob/main/docs/source/examples/classifier_comparison.rst

The params.yaml file stores hyperparameters used in the classifier comparison workflow. These parameters are loaded by the ZnTrack nodes. This file is typically managed by DVC.

```YAML
test_size: 0.3
random_state: 42

```

--------------------------------

### Track Metrics and Generate Plots with Zntrack (Python)

Source: https://context7.com/zincware/zntrack/llms.txt

This snippet demonstrates how to track performance metrics (like accuracy and loss) and generate plots (using pandas DataFrames) within Zntrack experiments. It defines a `ModelTrainer` node that simulates training and stores results in `zntrack.metrics` and `zntrack.plots`.

```python
import zntrack
import pandas as pd

class ModelTrainer(zntrack.Node):
    learning_rate: float = zntrack.params(0.01)
    epochs: int = zntrack.params(10)

    # Metrics for DVC experiment tracking
    final_accuracy: dict = zntrack.metrics()

    # Plots stored as pandas DataFrames
    training_history: pd.DataFrame = zntrack.plots(y=["loss", "accuracy"])

    def run(self):
        history = []
        for epoch in range(self.epochs):
            # Simulate training
            loss = 1.0 / (epoch + 1)
            accuracy = 1.0 - loss
            history.append({
                'epoch': epoch,
                'loss': loss,
                'accuracy': accuracy
            })

        self.training_history = pd.DataFrame(history)
        self.final_accuracy = {
            'accuracy': history[-1]['accuracy'],
            'loss': history[-1]['loss']
        }

project = zntrack.Project()

with project:
    trainer = ModelTrainer(learning_rate=0.001, epochs=20)

project.repro()

# View metrics
print(trainer.final_accuracy)  # {'accuracy': 0.95, 'loss': 0.05}

# Access plot data
print(trainer.training_history.head())
```

--------------------------------

### Import External Data with Zntrack (Python)

Source: https://context7.com/zincware/zntrack/llms.txt

This snippet shows how to import external data from URLs or remote DVC repositories into a Zntrack pipeline. The `DataPipeline` node uses `zntrack.deps` with a `default_factory` to specify the data source (a URL in this case) and its local path.

```python
import zntrack
from pathlib import Path

class DataPipeline(zntrack.Node):
    # Import data from external source
    external_data = zntrack.deps(default_factory=lambda: zntrack.add(
        url="https://example.com/dataset.csv",
        path="data/external.csv",
        rev="v1.0.0"  # Optional: specific version
    ))

    processed_data: Path = zntrack.outs_path(zntrack.nwd / "processed.csv")

    def run(self):
        # Process the imported data
        import pandas as pd
        df = pd.read_csv(self.external_data)
        # Process and save
        df_processed = df.dropna()
        df_processed.to_csv(self.processed_data, index=False)

project = zntrack.Project()

with project:
    pipeline = DataPipeline()

project.repro()
```

--------------------------------

### Load ZnTrack Node from Git Repository

Source: https://github.com/zincware/zntrack/blob/main/docs/source/index.rst

Loads a previously executed ZnTrack Node from a Git repository using its name and commit hash. This allows accessing historical results and reproducing previous states.

```python
import zntrack

x = zntrack.from_rev(name="Add", rev="HEAD", remote="https://github.com/user/repo")
print(x.result)
>>> 3
```

--------------------------------

### Define ZnTrack Node with Explicit File Handling

Source: https://github.com/zincware/zntrack/blob/main/docs/source/node.rst

Illustrates an alternative way to define a ZnTrack Node where inputs and outputs are explicitly handled via file paths. It reads parameters from a JSON file, performs addition, and saves the result to a JSON file in the node's working directory.

```python
from pathlib import Path

class AddViaFile(zntrack.Node):
    params_file: str = zntrack.params_path()
    results_file: Path = zntrack.outs_path(zntrack.nwd / "results.json")

    def run(self) -> None:
        import json

        with open(self.params_file, "r") as f:
            params = json.load(f)

        result = params[self.name]["a"] + params[self.name]["b"]

        self.results_file.parent.mkdir(parents=True, exist_ok=True)
        with open(self.results_file, "w") as f:
            json.dump({"result": result}, f)
```
Tokens:
Raw