System Dynamics modeling library for Python that integrates with data science tools
—
PySD provides command-line tools for model translation, batch execution, benchmarking, and file processing, enabling automated workflows and integration with larger data analysis pipelines.
Main CLI entry point for PySD operations including model translation and batch simulation.
def main(args):
"""
Main CLI entry point.
Parameters:
- args: list - Command line arguments
Available commands:
- translate: Convert Vensim/XMILE models to Python
- run: Execute model simulations
- help: Display command help
Examples:
- pysd translate model.mdl
- pysd run model.py --final-time 100
- pysd --help
"""Command line model translation:
# Translate Vensim model
pysd translate population_model.mdl
# Translate with options
pysd translate large_model.mdl --split-views --encoding utf-8
# Translate XMILE model
pysd translate stella_model.xmlCommand line simulation:
# Run translated model
pysd run population_model.py
# Run with custom parameters
pysd run model.py --final-time 50 --time-step 0.25
# Run with parameter file
pysd run model.py --params-file config.jsonLower-level functions used by the CLI interface.
def load(model_file, data_files, missing_values, split_views, **kwargs):
"""
CLI model loading function.
Parameters:
- model_file: str - Path to model file (.mdl, .xml, or .py)
- data_files: list - External data files
- missing_values: str - Missing value handling strategy
- split_views: bool - Whether to split views for large models
- **kwargs: Additional loading options
Returns:
Model: Loaded PySD model object
"""
def create_configuration(model, options):
"""
Create run configuration from CLI options.
Parameters:
- model: Model - PySD model object
- options: dict - CLI configuration options
Returns:
dict: Simulation configuration parameters
"""Tools for model validation, performance testing, and output comparison.
def runner(model_file, canonical_file=None, data_files=None,
missing_values="warning", split_views=False, **kwargs):
"""
Run model and compare with canonical output.
Parameters:
- model_file: str - Path to model file
- canonical_file: str or None - Reference output file for comparison
- data_files: list or None - External data files
- missing_values: str - Missing value handling
- split_views: bool - Split model views
- **kwargs: Additional run parameters (final_time, time_step, etc.)
Returns:
pandas.DataFrame: Simulation results
Raises:
AssertionError: If output doesn't match canonical within tolerance
"""
def assert_frames_close(actual, expected, rtol=1e-3, atol=1e-6,
check_names=True, check_dtype=False):
"""
Compare simulation outputs with specified tolerance.
Parameters:
- actual: pandas.DataFrame - Actual simulation results
- expected: pandas.DataFrame - Expected reference results
- rtol: float - Relative tolerance for numerical comparison
- atol: float - Absolute tolerance for numerical comparison
- check_names: bool - Whether to check column names match
- check_dtype: bool - Whether to check data types match
Raises:
AssertionError: If DataFrames don't match within tolerance
"""
def assert_allclose(x, y, rtol=1e-5, atol=1e-5):
"""
Compare arrays with tolerance.
Parameters:
- x: array-like - First array
- y: array-like - Second array
- rtol: float - Relative tolerance
- atol: float - Absolute tolerance
Raises:
AssertionError: If arrays don't match within tolerance
"""from pysd.tools.benchmarking import runner, assert_frames_close
import pandas as pd
# Run model and compare with reference
results = runner(
'test_model.mdl',
canonical_file='reference_output.csv',
final_time=50,
time_step=0.25
)
# Manual comparison of results
actual_results = pd.read_csv('actual_output.csv')
expected_results = pd.read_csv('expected_output.csv')
assert_frames_close(
actual_results,
expected_results,
rtol=1e-3, # 0.1% relative tolerance
atol=1e-6 # Absolute tolerance
)Tools for processing netCDF simulation outputs and converting between formats.
class NCFile:
"""
NetCDF file processing class.
Handles reading, processing, and converting PySD simulation outputs
stored in netCDF format. Supports various output formats and
data extraction operations.
Methods:
- __init__(filename, parallel=False) - Initialize with netCDF file
- to_text_file(outfile="result.tab", sep="\t") - Convert to text format
- get_varnames() - Get list of variable names in file
- get_coords() - Get coordinate information
- get_data(varname) - Extract specific variable data
- close() - Close file and release resources
"""from pysd.tools.ncfiles import NCFile
# Open netCDF simulation output
nc_file = NCFile('simulation_results.nc')
# Convert to tab-delimited text file
nc_file.to_text_file('results.tab', sep='\t')
# Convert to CSV format
nc_file.to_text_file('results.csv', sep=',')
# Extract specific variable data
population_data = nc_file.get_data('Population')
time_coords = nc_file.get_coords()['time']
# Get available variables
variables = nc_file.get_varnames()
print(f"Available variables: {variables}")
# Clean up
nc_file.close()Integration with testing frameworks for model validation.
import pytest
from pysd.tools.benchmarking import runner, assert_frames_close
def test_population_model():
"""Test population model against known output."""
results = runner(
'population_model.mdl',
canonical_file='population_reference.csv',
final_time=100
)
# Additional custom checks
final_population = results['Population'].iloc[-1]
assert 9000 < final_population < 11000, "Population should stabilize around 10000"
def test_economic_model_sensitivity():
"""Test model sensitivity to parameter changes."""
base_results = runner('economic_model.mdl', final_time=50)
# Test with different parameter
modified_results = runner(
'economic_model.mdl',
final_time=50,
params={'growth_rate': 0.05}
)
# Compare outcomes
base_gdp = base_results['GDP'].iloc[-1]
modified_gdp = modified_results['GDP'].iloc[-1]
assert modified_gdp > base_gdp, "Higher growth rate should increase GDP"Scripts and utilities for processing multiple models or parameter sets.
import pysd
from pathlib import Path
def batch_translate_models(input_dir, output_dir):
"""Translate all models in directory."""
input_path = Path(input_dir)
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
for model_file in input_path.glob('*.mdl'):
try:
print(f"Translating {model_file.name}...")
model = pysd.read_vensim(str(model_file))
print(f"Successfully translated {model_file.name}")
except Exception as e:
print(f"Error translating {model_file.name}: {e}")
def batch_run_scenarios(model_file, scenarios, output_dir):
"""Run model with multiple parameter scenarios."""
model = pysd.load(model_file)
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
for i, scenario in enumerate(scenarios):
print(f"Running scenario {i+1}/{len(scenarios)}...")
results = model.run(params=scenario)
results.to_csv(output_path / f'scenario_{i+1}.csv')
model.reload() # Reset for next scenarioPySD CLI tools integrate with external data analysis and workflow tools:
# Makefile for automated model processing
translate: model.mdl
pysd translate model.mdl
run: model.py
pysd run model.py --final-time 100 --output results.csv
test: model.py reference.csv
python -m pytest test_model.py#!/bin/bash
# Batch process multiple models
for model in models/*.mdl; do
echo "Processing $model..."
pysd translate "$model"
base=$(basename "$model" .mdl)
pysd run "models/${base}.py" --output "results/${base}_results.csv"
done#!/usr/bin/env python
"""Automated model validation pipeline."""
import sys
from pathlib import Path
from pysd.tools.benchmarking import runner, assert_frames_close
def validate_model(model_path, reference_path):
"""Validate single model against reference."""
try:
results = runner(str(model_path), canonical_file=str(reference_path))
print(f"✓ {model_path.name} passed validation")
return True
except AssertionError as e:
print(f"✗ {model_path.name} failed validation: {e}")
return False
if __name__ == "__main__":
models_dir = Path("models")
references_dir = Path("references")
success_count = 0
total_count = 0
for model_file in models_dir.glob("*.mdl"):
reference_file = references_dir / f"{model_file.stem}_reference.csv"
if reference_file.exists():
total_count += 1
if validate_model(model_file, reference_file):
success_count += 1
print(f"\nValidation Results: {success_count}/{total_count} models passed")
sys.exit(0 if success_count == total_count else 1)CLI tools provide comprehensive error reporting:
import logging
# Configure logging for detailed diagnostics
logging.basicConfig(level=logging.DEBUG)
try:
results = runner('problematic_model.mdl', canonical_file='reference.csv')
except Exception as e:
logging.error(f"Model execution failed: {e}")
# Additional diagnostic information available in logsTools for monitoring and optimizing model performance:
import time
from pysd.tools.benchmarking import runner
def benchmark_model(model_file, iterations=10):
"""Benchmark model execution time."""
times = []
for i in range(iterations):
start_time = time.time()
runner(model_file, final_time=100)
end_time = time.time()
times.append(end_time - start_time)
avg_time = sum(times) / len(times)
print(f"Average execution time: {avg_time:.2f} seconds")
print(f"Min: {min(times):.2f}s, Max: {max(times):.2f}s")
return timesInstall with Tessl CLI
npx tessl i tessl/pypi-pysd