tessl/pypi-pysd

System Dynamics modeling library for Python that integrates with data science tools

—

Pending

Overview

Eval results

Files

Model Simulation and Execution

Name: tessl/pypi-pysd
Author: tessl

PySD's Model class provides comprehensive simulation capabilities, supporting both batch execution and step-by-step simulation with flexible parameter management and output configuration.

Capabilities

Batch Simulation

Execute complete model simulations with customizable time settings, parameter overrides, and output formatting.

def run(self, params=None, return_columns=None, return_timestamps=None,
        initial_condition='original', final_time=None, time_step=None,
        saveper=None, reload=False, progress=False, flatten_output=False,
        cache_output=True, output_file=None):
    """
    Execute model simulation.

    Parameters:
    - params: dict or None - Parameter values to override during simulation
    - return_columns: list or None - Specific variables to include in output
    - return_timestamps: list or None - Specific time points to include in output
    - initial_condition: str or tuple - Initial condition ('original', 'current', pickle file, or (time, values))
    - final_time: float or None - Simulation end time
    - time_step: float or None - Integration time step
    - saveper: float or None - Output sampling period
    - reload: bool - Whether to reload model from file before running
    - progress: bool - Whether to show progress bar during simulation
    - flatten_output: bool - Whether to flatten subscripted output
    - cache_output: bool - Whether to cache results for faster repeated access
    - output_file: str or None - File path to save results

    Returns:
        pandas.DataFrame: Simulation results with time as index and variables as columns

    Raises:
        ValueError: If time parameters are inconsistent
        KeyError: If parameter names in params don't exist in model
    """

Usage Examples

Basic simulation:

import pysd

model = pysd.read_vensim('population_model.mdl')

# Run with default settings
results = model.run()
print(results.head())

# Run with custom time settings
results = model.run(final_time=50, time_step=0.25, saveper=1)

Parameter override during simulation:

# Run with different parameter values
results = model.run(params={
    'birth_rate': 0.05,
    'death_rate': 0.02,
    'initial_population': 1000
})

# Run multiple scenarios
scenarios = [
    {'birth_rate': 0.03, 'death_rate': 0.01},
    {'birth_rate': 0.04, 'death_rate': 0.02},
    {'birth_rate': 0.05, 'death_rate': 0.03}
]

results = {}
for i, scenario in enumerate(scenarios):
    results[f'scenario_{i}'] = model.run(params=scenario)

Selective output:

# Return only specific variables
results = model.run(return_columns=['Population', 'Birth Rate', 'Death Rate'])

# Return specific time points
results = model.run(return_timestamps=[0, 10, 20, 30, 40, 50])

# Combine both
results = model.run(
    return_columns=['Population'],
    return_timestamps=list(range(0, 51, 5))  # Every 5 time units
)

Step-by-Step Simulation

Execute simulations one step at a time, enabling real-time parameter adjustments and interactive simulation control.

def set_stepper(self, output_obj, return_columns=None, **kwargs):
    """
    Configure model for step-by-step execution.

    Parameters:
    - output_obj: ModelOutput - Output handler for storing step results
    - return_columns: list or None - Variables to track during stepping
    - **kwargs: Additional configuration options

    Raises:
        TypeError: If output_obj is not a ModelOutput instance
    """

def step(self, num_steps=1, step_vars={}):
    """
    Execute one or more simulation steps.

    Parameters:
    - num_steps: int - Number of time steps to execute (default 1)
    - step_vars: dict - Variable values to set for this step

    Returns:
        dict: Current values of tracked variables

    Raises:
        RuntimeError: If stepper not configured via set_stepper()
        ValueError: If num_steps is not positive
    """

Usage Examples

from pysd.py_backend.output import ModelOutput

model = pysd.read_vensim('population_model.mdl')

# Configure stepping
output = ModelOutput()
model.set_stepper(output, return_columns=['Population', 'Birth Rate'])

# Execute single steps
step_result = model.step()
print(f"Time {model.time()}: Population = {step_result['Population']}")

# Execute multiple steps at once
results = model.step(num_steps=10)

# Adjust parameters during simulation
for i in range(20):
    if i == 10:
        # Change birth rate halfway through
        results = model.step(step_vars={'birth_rate': 0.06})
    else:
        results = model.step()
    
    print(f"Step {i}: Population = {results['Population']}")

Model State Management

Control model initialization, reloading, and state persistence.

def initialize(self):
    """Initialize the model for simulation."""

def reload(self):
    """Reload model from file, resetting all parameters."""
    
def copy(self, reload=False):
    """
    Create a copy of the current model.
    
    Parameters:
    - reload: bool - Whether to reload the copy from file
    
    Returns:
        Model: Independent copy of the model
    """

Usage Examples

model = pysd.read_vensim('model.mdl', initialize=False)

# Initialize when ready
model.initialize()

# Create independent copy for experimentation
model_copy = model.copy()
model_copy.set_components({'parameter': 100})

# Original model unchanged
original_results = model.run()
modified_results = model_copy.run()

# Reset model to file state
model.reload()

Initial Conditions

Set and manage initial conditions for simulation runs.

def set_initial_condition(self, initial_condition):
    """
    Set initial simulation conditions.
    
    Parameters:
    - initial_condition: str or tuple - Initial condition specification
      - 'original': Use model's original initial values
      - 'current': Use current model state as initial condition
      - str path: Load initial condition from pickle file
      - (time, dict): Set specific time and variable values
    """

def set_initial_value(self, time, initial_value):
    """
    Set initial values for specific stocks.
    
    Parameters:
    - time: float - Time at which to set initial values
    - initial_value: dict - Variable names and initial values
    """

Usage Examples

# Use original model initial conditions
model.set_initial_condition('original')
results1 = model.run()

# Use current state as initial condition for next run
model.set_initial_condition('current')
results2 = model.run()

# Set specific initial values
model.set_initial_condition((0, {
    'Population': 1500,
    'Economic Development Index': 0.75
}))
results3 = model.run()

# Load initial condition from saved state
model.export('initial_state.pkl')
# ... later ...
model.set_initial_condition('initial_state.pkl')

Output Configuration

Control simulation output format, content, and storage.

Progress Monitoring

# Show progress bar for long simulations
results = model.run(progress=True)

Output File Writing

# Save results directly to file
model.run(output_file='simulation_results.csv')

# Save with custom format (supported: .csv, .tab, .nc)
model.run(output_file='results.nc')  # NetCDF format

Subscripted Variable Handling

# Flatten subscripted variables to separate columns
results = model.run(flatten_output=True)

# Keep subscripted variables as multi-dimensional arrays (default)
results = model.run(flatten_output=False)

Error Handling

Common simulation errors and their resolution:

ValueError: Time parameters conflict (e.g., time_step > final_time)
KeyError: Parameter name in params doesn't exist in model
RuntimeError: Step-by-step simulation not properly configured
MemoryError: Large models with high resolution time settings

try:
    results = model.run(params={'nonexistent_param': 100})
except KeyError as e:
    print(f"Parameter not found: {e}")
    # Check available parameters
    print("Available parameters:", list(model.namespace.keys()))

Model Introspection and Metadata

Access model structure, documentation, and metadata for analysis and debugging.

def doc(self):
    """
    Get comprehensive model documentation.
    
    Returns:
        pandas.DataFrame: Documentation with variables, units, limits, and descriptions
    """

def namespace(self):
    """
    Get model variable namespace mapping.
    
    Returns:
        dict: Mapping of model variable names to Python function names
    """

def dependencies(self):
    """
    Get model variable dependencies.
    
    Returns:
        dict: Dictionary of variable dependencies for each model element
    """

def subscripts(self):
    """
    Get model subscript definitions.
    
    Returns:
        dict: Dictionary describing possible dimensions of model subscripts
    """

def modules(self):
    """
    Get model module structure (for split-view models).
    
    Returns:
        dict or None: Dictionary of module names and their variables, or None for single-file models
    """

def get_dependencies(self, vars=[], modules=[]):
    """
    Get dependencies for specific variables or modules.
    
    Parameters:
    - vars: list - List of variable names to get dependencies for
    - modules: list - List of module names to get dependencies for
    
    Returns:
        Dependencies: Object containing variable and stateful dependencies
    """

Model State Management

Initialize, reset, and manage model state and caches.

def initialize(self):
    """
    Initialize model state and prepare for simulation.
    
    Sets up initial values, external data, and stateful components.
    """

def clean_caches(self):
    """
    Clear all model caches to free memory.
    
    Useful for large models or when running multiple simulations.
    """

def initialize_external_data(self, externals=None):
    """
    Initialize external data sources.
    
    Parameters:
    - externals: list or None - Specific external objects to initialize
    """

def export(self):
    """
    Export current model state.
    
    Returns:
        dict: Dictionary containing current values of all model variables
    """

def serialize_externals(self, export_path="externals.nc", encoding="zlib"):
    """
    Serialize external data to NetCDF file.
    
    Parameters:
    - export_path: str - Path for exported NetCDF file
    - encoding: str - Compression encoding for NetCDF file
    """

Submodel Selection

Extract and work with subsets of large models.

def select_submodel(self, vars=[], modules=[], exogenous_components={}):
    """
    Create a submodel containing only specified variables and modules.
    
    Parameters:
    - vars: list - Variables to include in submodel
    - modules: list - Modules to include in submodel  
    - exogenous_components: dict - External values for excluded variables
    
    Returns:
        Model: New Model object containing only selected components
    """

Usage Examples

Model introspection:

# Get model documentation
doc_df = model.doc()
print(doc_df[['Variable', 'Units', 'Type']])

# Check variable namespace
namespace = model.namespace()
print("Python names:", list(namespace.values()))

# Examine dependencies
deps = model.dependencies()
print("Variables that depend on time:", [var for var, dep in deps.items() if 'time' in dep])

# View subscripts
subs = model.subscripts()
print("Available subscripts:", list(subs.keys()))

Model state management:

# Clear caches to free memory
model.clean_caches()

# Re-initialize model
model.initialize()

# Export current state
current_state = model.export()
print("Current population:", current_state.get('Population', 'Not found'))

# Create submodel for testing
submodel = model.select_submodel(
    vars=['Population', 'Birth Rate', 'Death Rate'],
    exogenous_components={'Economic Growth': 0.03}
)
results = submodel.run()

Install with Tessl CLI