Configuration Systems¶
Overview¶
This tutorial provides a comprehensive guide to the configuration systems in ssm-simulators. Understanding these configurations is essential for:
- Selecting and customizing models
- Controlling data generation pipelines
- Creating custom models (covered in dedicated Tutorial)
What you'll learn:
- The two configuration systems:
model_configandgenerator_config - How to access and modify existing configurations
- The nested structure of
generator_config - Best practices for configuration management
# Standard imports
import numpy as np
import matplotlib.pyplot as plt
from pprint import pprint
# ssms imports
from ssms.config import model_config
from ssms.config.generator_config.data_generator_config import get_default_generator_config
from ssms.config import ModelConfigBuilder
Introduction - Two Configuration Systems¶
The ssm-simulators package uses two complementary configuration systems:
| Configuration | Purpose | What it Specifies |
|---|---|---|
| model_config |
"What to simulate" Defines the model |
• Which SSM model (DDM, LBA, etc.) • Parameters and their bounds • Boundary functions • Drift functions • Parameter sampling constraints |
| generator_config |
"Simulation to Training Data" Controls data generation |
• How many parameter sets to sample • Simulation settings (time resolution, samples) • Estimator type (KDE vs PyDDM) • Training data format • Output options |
# Access the DDM config
ddm_config = model_config["ddm"]
print("DDM Configuration:")
print("=" * 50)
for key, value in ddm_config.items():
if callable(value):
print(f"{key}: <function {value.__name__}>")
else:
print(f"{key}: {value}")
DDM Configuration:
==================================================
name: ddm
params: ['v', 'a', 'z', 't']
param_bounds: [[-3.0, 0.3, 0.1, 0.0], [3.0, 2.5, 0.9, 2.0]]
boundary_name: constant
boundary: <function constant>
boundary_params: []
n_params: 4
default_params: [0.0, 1.0, 0.5, 0.001]
nchoices: 2
choices: [-1, 1]
n_particles: 1
simulator: <function ddm_flexbound>
parameter_transforms: {'sampling': [], 'simulation': []}
param_bounds_dict: {'v': (-3.0, 3.0), 'a': (0.3, 2.5), 'z': (0.1, 0.9), 't': (0.0, 2.0)}
Core fields explained:
name: Model identifierparams: List of parameter namesparam_bounds: Lower and upper bounds for each parameterparam_bounds_dict: Dictionary version (easier for sampling)nchoices: Number of choices the model can produceboundary: Boundary function (if not constant)n_params: Number of parametersdefault_params: Default parameter values
# Let's look at a few different models
models_to_examine = ["ddm", "angle", "ornstein"]
for model_name in models_to_examine:
config = model_config[model_name]
print(f"\n{model_name.upper()}:")
print(f" Parameters: {config['params']}")
print(f" Choices: {config['nchoices']}")
print(f" Has custom boundary: {'boundary' in config and config.get('boundary') is not None}")
DDM: Parameters: ['v', 'a', 'z', 't'] Choices: 2 Has custom boundary: True ANGLE: Parameters: ['v', 'a', 'z', 't', 'theta'] Choices: 2 Has custom boundary: True ORNSTEIN: Parameters: ['v', 'a', 'z', 'g', 't'] Choices: 2 Has custom boundary: True
Boundary Functions¶
Boundary functions define how decision thresholds behave over time. The package includes several built-in boundary types:
# Available boundary functions (from ssms.config._modelconfig.base)
from ssms.config._modelconfig.base import boundary_config as boundary_registry
print("Available Boundary Functions:")
print("=" * 60)
for name, config in boundary_registry.items():
params_str = ", ".join(config["params"]) if config["params"] else "none"
print(f"{name:25} | params: {params_str}")
Available Boundary Functions: ============================================================ constant | params: a angle | params: a, theta weibull_cdf | params: a, alpha, beta generalized_logistic | params: a, B, M, v conflict_gamma | params: a, theta, scale, alphaGamma, scaleGamma
# Visualize different boundary types
import ssms.basic_simulators.boundary_functions as bf
t = np.linspace(0, 5, 100)
a = 1.5 # Threshold parameter
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
# Constant boundary
constant_vals = bf.constant(t, a=a)
axes[0].plot(t, constant_vals, color='blue', linewidth=2, label='Upper bound')
axes[0].plot(t, -constant_vals, color='blue', linewidth=2, label='Lower bound')
axes[0].set_title('Constant Boundary', fontsize=14, fontweight='bold')
axes[0].set_xlabel('Time (s)')
axes[0].set_ylabel('Boundary')
axes[0].legend()
axes[0].grid(alpha=0.3)
# Angle boundary (collapsing)
angle_vals = bf.angle(t, a=a, theta=0.8)
axes[1].plot(t, angle_vals, color='red', linewidth=2, label='Upper bound')
axes[1].plot(t, -angle_vals, color='red', linewidth=2, label='Lower bound')
axes[1].set_title('Angle Boundary (Collapsing)', fontsize=14, fontweight='bold')
axes[1].set_xlabel('Time (s)')
axes[1].set_ylabel('Boundary')
axes[1].legend()
axes[1].grid(alpha=0.3)
# Weibull boundary
weibull_vals = bf.weibull_cdf(t, a=a, alpha=4.0, beta=0.5)
axes[2].plot(t, weibull_vals, color='green', linewidth=2, label='Upper bound')
axes[2].plot(t, -weibull_vals, color='green', linewidth=2, label='Lower bound')
axes[2].set_title('Weibull Boundary (Collapsing)', fontsize=14, fontweight='bold')
axes[2].set_xlabel('Time (s)')
axes[2].set_ylabel('Boundary')
axes[2].legend()
axes[2].grid(alpha=0.3)
plt.tight_layout()
plt.show()
print("\nNote: All boundary functions accept the 'a' (threshold) parameter and return the final boundary value.")
Note: All boundary functions accept the 'a' (threshold) parameter and return the final boundary value.
Drift Functions¶
Drift functions define how evidence accumulates over time. Most models use constant drift, but time-dependent drift is available.
Note: Drift functions can accept v (base drift rate) as a parameter, mirroring how boundary functions can accept a. Most built-in drift functions include v for consistency. For models like conflict and shrink_spot, v serves as an optional base drift that the time-varying component is added to.
# Available drift functions
from ssms.config._modelconfig.base import drift_config as drift_registry
print("Available Drift Functions:")
print("=" * 80)
for name, config in drift_registry.items():
params_str = ", ".join(config["params"]) if config["params"] else "none"
print(f"{name:40} | params: {params_str}")
Available Drift Functions: ================================================================================ constant | params: v gamma_drift | params: v, shape, scale, c conflict_ds_drift | params: v, tinit, dinit, tslope, dslope, tfixedp, tcoh, dcoh conflict_dsstimflex_drift | params: v, tinit, dinit, tslope, dslope, tfixedp, tcoh, dcoh, tonset, donset conflict_stimflex_drift | params: v, vt, vd, tcoh, dcoh, tonset, donset conflict_stimflexrel1_drift | params: v, vt, vd, tcoh, dcoh, tonset, donset conflict_stimflexrel1_dual_drift | params: vt, vd, tcoh, dcoh, tonset, donset, toffset, doffset attend_drift | params: v, ptarget, pouter, pinner, r, sda attend_drift_simple | params: v, ptarget, pouter, r, sda
# Visualize constant vs time-dependent drift
import ssms.basic_simulators.drift_functions as df
t = np.linspace(0, 5, 100)
fig, axes = plt.subplots(1, 2, figsize=(12, 4))
# Constant drift
constant_drift_val = 1.5
axes[0].axhline(y=constant_drift_val, color='blue', linewidth=2)
axes[0].set_title('Constant Drift', fontsize=14, fontweight='bold')
axes[0].set_xlabel('Time (s)')
axes[0].set_ylabel('Drift Rate')
axes[0].set_ylim([0, 3])
axes[0].grid(alpha=0.3)
# Gamma drift (time-dependent) - v is the base drift, gamma component is added
gamma_vals = df.gamma_drift(t, v=0.5, shape=2.0, scale=1.0, c=0.5)
axes[1].plot(t, gamma_vals, color='purple', linewidth=2)
axes[1].set_title('Gamma Drift (Time-Dependent)', fontsize=14, fontweight='bold')
axes[1].set_xlabel('Time (s)')
axes[1].set_ylabel('Drift Rate')
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
Parameter Transforms¶
Parameter transforms enforce constraints or apply transformations at different stages of the workflow:
Sampling transforms: Applied during the parameter sampling stage, which is relevant for the training data generation workflow. These enforce parameter relationships (e.g.,
a > z) when generating synthetic training data for likelihood approximation networks.Simulation transforms: Applied via ParameterSimulatorAdapters when running the basic Simulator. These prepare user-provided parameters for the low-level C/Cython simulators (e.g., stacking
v0, v1, v2into a singlevarray).
Both are specified in the unified parameter_transforms field of model_config.
# Example: DDM with a > z constraint
from ssms.transforms import SwapIfLessConstraint
ddm_with_constraint = model_config["ddm"].copy()
# Check if parameter_transforms exist
if "parameter_transforms" in ddm_with_constraint:
print("Existing parameter transforms:")
pprint(ddm_with_constraint["parameter_transforms"])
else:
print("No parameter transforms defined for standard DDM.")
print("\nWe can add them manually:")
# Add a swap transform to ensure a > z (using the new unified format)
ddm_with_constraint["parameter_transforms"] = {
"sampling": [SwapIfLessConstraint("a", "z")],
"simulation": [],
}
print("\nAdded transform:")
pprint(ddm_with_constraint["parameter_transforms"])
Existing parameter transforms:
{'sampling': [], 'simulation': []}
Added transform:
{'sampling': [SwapIfLessConstraint(param_a='a', param_b='z')], 'simulation': []}
Built-in transforms:
swap: Swaps two parameters if the first is less than the second- Use case: Enforce
a > zconstraint - Config:
{"type": "swap", "param_a": "a", "param_b": "z"}
- Use case: Enforce
normalize: Normalizes a list of parameters to sum to a value- Use case: Ensure drift rates sum to 1.0
- Config:
{"type": "normalize", "param_names": ["v1", "v2", "v3"]}
Custom transforms can be registered (see Tutorial 05 for details).
Creating custom model_config¶
The ModelConfigBuilder class provides convenient methods for creating and modifying model configurations:
# Method 1: Start from an existing model and override
custom_ddm = ModelConfigBuilder.from_model(
"ddm",
param_bounds=[[-2.0, 0.5, 0.2, 0.0], # Tighter bounds on v, a, z, t
[2.0, 2.0, 0.8, 1.5]]
)
print("Custom DDM with modified bounds:")
print(f" Parameters: {custom_ddm['params']}")
print(f" New bounds: {custom_ddm['param_bounds']}")
Custom DDM with modified bounds: Parameters: ['v', 'a', 'z', 't'] New bounds: [[-2.0, 0.5, 0.2, 0.0], [2.0, 2.0, 0.8, 1.5]]
# Method 2: Add a custom boundary to an existing model
ddm_with_angle = ModelConfigBuilder.from_model("ddm")
ddm_with_angle = ModelConfigBuilder.add_boundary(
ddm_with_angle,
"angle", # Boundary name from registry
)
print("\nDDM with angle (collapsing) boundary:")
print(f" Parameters: {ddm_with_angle['params']}")
print(f" Boundary: {ddm_with_angle.get('boundary_name', 'constant')}")
if 'boundary_params' in ddm_with_angle:
print(f" Boundary params: {ddm_with_angle['boundary_params']}")
DDM with angle (collapsing) boundary: Parameters: ['v', 'a', 'z', 't'] Boundary: angle Boundary params: ['a', 'theta']
# Method 3: Build from scratch (for completely custom models)
# Note: from_scratch requires a simulator_function (covered in Tutorial 05)
# For now, let's show an alternative: creating a custom variant of DDM
my_custom_model = ModelConfigBuilder.from_model(
"ddm",
param_bounds=[[-3.0, 0.5, 0.1, 0.0], # Custom bounds
[3.0, 2.5, 0.9, 1.0]],
)
# Rename it to make it clear it's custom
my_custom_model["name"] = "my_custom_ddm"
print("\nCustom DDM variant:")
print(f" Name: {my_custom_model['name']}")
print(f" Parameters: {my_custom_model['params']}")
print(f" Bounds: {my_custom_model['param_bounds']}")
print("\nNote: For completely new models with custom simulator functions,")
print("see Tutorial 05 for full details on using ModelConfigBuilder.from_scratch()")
Custom DDM variant: Name: my_custom_ddm Parameters: ['v', 'a', 'z', 't'] Bounds: [[-3.0, 0.5, 0.1, 0.0], [3.0, 2.5, 0.9, 1.0]] Note: For completely new models with custom simulator functions, see Tutorial 05 for full details on using ModelConfigBuilder.from_scratch()
# Validate a configuration
is_valid, errors = ModelConfigBuilder.validate_config(my_custom_model, strict=False)
if is_valid:
print("\n✓ Configuration is valid!")
else:
print("\n✗ Configuration has errors:")
for error in errors:
print(f" - {error}")
✓ Configuration is valid!
# Get default LAN config
lan_config = get_default_generator_config("lan")
print("generator_config structure:")
print("=" * 60)
for section in lan_config.keys():
print(f"\n[{section}]")
if isinstance(lan_config[section], dict):
for key, value in lan_config[section].items():
# Truncate long values for display
if isinstance(value, dict) and len(str(value)) > 50:
print(f" {key}: {{...}}")
else:
print(f" {key}: {value}")
else:
print(f" {lan_config[section]}")
generator_config structure:
============================================================
[pipeline]
n_parameter_sets: 1000
n_parameter_sets_rejected: 100
n_subruns: 10
n_cpus: all
simulation_filters: {...}
[estimator]
type: kde
kde_displace_t: False
pdf_interpolation: cubic
max_undecided_prob: 0.5
[training]
n_samples_per_param: 1000
mixture_probabilities: [0.8, 0.1, 0.1]
separate_response_channels: False
negative_rt_log_likelihood: -66.77497
[simulator]
n_samples: 1000
delta_t: 0.001
max_t: 20.0
smooth_unif: True
[output]
folder: data/lan_mlp/
nbins: 0
pickle_protocol: 4
bin_pointwise: False
[model]
ddm
Section-by-Section Breakdown¶
Let's examine each section in detail:
Pipeline Section¶
Controls the overall data generation workflow:
print("Pipeline Settings:")
print("=" * 60)
pprint(lan_config["pipeline"])
print("\n" + "=" * 60)
print("Key Parameters:")
print(" n_parameter_sets: How many parameter combinations to sample")
print(" n_subruns: Number of parallel batches (for large jobs)")
print(" n_cpus: CPU allocation ('all' or integer)")
print(" n_parameter_sets_rejected: Buffer for rejection sampling")
Pipeline Settings:
============================================================
{'n_cpus': 'all',
'n_parameter_sets': 1000,
'n_parameter_sets_rejected': 100,
'n_subruns': 10,
'simulation_filters': {'choice_cnt': 0,
'mean_rt': 17,
'mode': 20,
'mode_cnt_rel': 0.95,
'std': 0}}
============================================================
Key Parameters:
n_parameter_sets: How many parameter combinations to sample
n_subruns: Number of parallel batches (for large jobs)
n_cpus: CPU allocation ('all' or integer)
n_parameter_sets_rejected: Buffer for rejection sampling
Estimator Section¶
Controls how likelihoods are estimated:
print("Estimator Settings:")
print("=" * 60)
pprint(lan_config["estimator"])
print("\n" + "=" * 60)
print("Key Parameters:")
print(" type: 'kde' (simulation-based) or 'pyddm' (analytical)")
print(" bandwidth: KDE bandwidth (type='kde')")
print(" displace_t: Whether to displace time for KDE (type='kde'")
print(" use_pyddm_pdf: Use PyDDM for PDF computation (type='pyddm')")
print(" pdf_interpolation: 'linear' or 'cubic' (type='pyddm')")
print(" max_undecided_prob: Reject if P(undecided) > threshold (type='pyddm')")
Estimator Settings:
============================================================
{'kde_displace_t': False,
'max_undecided_prob': 0.5,
'pdf_interpolation': 'cubic',
'type': 'kde'}
============================================================
Key Parameters:
type: 'kde' (simulation-based) or 'pyddm' (analytical)
bandwidth: KDE bandwidth (type='kde')
displace_t: Whether to displace time for KDE (type='kde'
use_pyddm_pdf: Use PyDDM for PDF computation (type='pyddm')
pdf_interpolation: 'linear' or 'cubic' (type='pyddm')
max_undecided_prob: Reject if P(undecided) > threshold (type='pyddm')
Training Section¶
Controls training data generation:
print("Training Settings:")
print("=" * 60)
pprint(lan_config["training"])
print("\n" + "=" * 60)
print("Key Parameters:")
print(" mixture_probabilities: Data augmentation [full_trials, single_rt, single_choice]")
print(" n_samples_per_param: Training samples per parameter set")
print(" separate_response_channels: Choice-conditional formatting")
print(" n_subdatasets: Split into multiple datasets (optional)")
print(" n_trials_per_dataset: Trials per subdataset (optional)")
Training Settings:
============================================================
{'mixture_probabilities': [0.8, 0.1, 0.1],
'n_samples_per_param': 1000,
'negative_rt_log_likelihood': -66.77497,
'separate_response_channels': False}
============================================================
Key Parameters:
mixture_probabilities: Data augmentation [full_trials, single_rt, single_choice]
n_samples_per_param: Training samples per parameter set
separate_response_channels: Choice-conditional formatting
n_subdatasets: Split into multiple datasets (optional)
n_trials_per_dataset: Trials per subdataset (optional)
Simulator Section¶
Controls simulation behavior:
print("Simulator Settings:")
print("=" * 60)
pprint(lan_config["simulator"])
print("\n" + "=" * 60)
print("Key Parameters:")
print(" delta_t: Time step size (smaller = more accurate, slower)")
print(" max_t: Maximum simulation time")
print(" n_samples: Number of simulation trials per parameter set")
print(" smooth_unif: Uniform noise smoothing")
print(" filters: Quality filters (reject poor simulations)")
Simulator Settings:
============================================================
{'delta_t': 0.001, 'max_t': 20.0, 'n_samples': 1000, 'smooth_unif': True}
============================================================
Key Parameters:
delta_t: Time step size (smaller = more accurate, slower)
max_t: Maximum simulation time
n_samples: Number of simulation trials per parameter set
smooth_unif: Uniform noise smoothing
filters: Quality filters (reject poor simulations)
# Examine simulation filters
print("\nSimulation Filters:")
print("=" * 60)
if "filters" in lan_config["simulator"]:
for filter_name, threshold in lan_config["simulator"]["filters"].items():
print(f" {filter_name}: {threshold}")
print("\nFilters ensure simulation quality by rejecting parameter sets with:")
print(" - mode: RT mode at boundary")
print(" - choice_cnt: Too few samples for a choice")
print(" - mean_rt: Mean RT too high")
print(" - std: Non-positive standard deviation")
print(" - mode_cnt_rel: Mode contains too many samples")
Simulation Filters: ============================================================
Output Section¶
Controls output format and location:
print("Output Settings:")
print("=" * 60)
pprint(lan_config["output"])
print("\n" + "=" * 60)
print("Key Parameters:")
print(" folder: Output directory")
print(" pickle_protocol: Serialization version (4 is standard)")
print(" nbins: Number of bins for RT histograms (0 = no binning)")
print(" negative_rt_log_likelihood: Log-likelihood value assigned to negative RT samples")
Output Settings:
============================================================
{'bin_pointwise': False,
'folder': 'data/lan_mlp/',
'nbins': 0,
'pickle_protocol': 4}
============================================================
Key Parameters:
folder: Output directory
pickle_protocol: Serialization version (4 is standard)
nbins: Number of bins for RT histograms (0 = no binning)
negative_rt_log_likelihood: Log-likelihood value assigned to negative RT samples
Getting Default Configs¶
The package provides default configurations for different use cases:
# LAN config (simulation-based, KDE estimator)
lan_config = get_default_generator_config("lan")
print("LAN Config:")
print(f" Estimator type: {lan_config['estimator']['type']}")
print(f" Parameter sets: {lan_config['pipeline']['n_parameter_sets']}")
print(f" Samples per param: {lan_config['training']['n_samples_per_param']}")
# PyDDM config (analytical PDFs)
pyddm_config = get_default_generator_config("lan")
pyddm_config["estimator"]["type"] = "pyddm"
print("\nPyDDM Config:")
print(f" Estimator type: {pyddm_config['estimator']['type']}")
print(f" Parameter sets: {pyddm_config['pipeline']['n_parameter_sets']}")
print(f" Samples per param: {pyddm_config['training']['n_samples_per_param']}")
LAN Config: Estimator type: kde Parameter sets: 1000 Samples per param: 1000 PyDDM Config: Estimator type: pyddm Parameter sets: 1000 Samples per param: 1000
# Create a valid config
valid_config = ModelConfigBuilder.from_model("ddm")
is_valid, errors = ModelConfigBuilder.validate_config(valid_config, strict=False)
print("Valid Config Test:")
print(f" Is valid: {is_valid}")
print(f" Errors: {errors if errors else 'None'}")
Valid Config Test: Is valid: True Errors: None
# Create an invalid config (missing required field)
invalid_config = {
"name": "incomplete_model",
"params": ["v", "a"],
# Missing param_bounds!
}
is_valid, errors = ModelConfigBuilder.validate_config(invalid_config, strict=False)
print("\nInvalid Config Test:")
print(f" Is valid: {is_valid}")
if errors:
print(" Errors:")
for error in errors:
print(f" - {error}")
Invalid Config Test:
Is valid: False
Errors:
- Missing required field: 'nchoices'
- Missing required field: 'simulator'