tessl/pypi-py-pde
Python package for solving partial differential equations with a focus on ease of use and performance
—
Pending
trackers.mddocs/
Trackers
Tracker classes monitor simulation progress, collect data, create visualizations, and implement custom analysis during PDE time evolution with flexible interrupt scheduling.
Capabilities
Data Collection Trackers
Collect and store field data during simulations for analysis and visualization.
class DataTracker:
def __init__(self, interrupts=1, *, filename=None):
"""
Track field data throughout simulation.
Parameters:
- interrupts: interrupt schedule for data collection
- filename: str, optional file for data storage
"""
@property
def data(self):
"""MemoryStorage: Collected field data"""
def get_time_series(self, *, field_index=0):
"""
Extract time series data for analysis.
Parameters:
- field_index: int, field index for FieldCollection
Returns:
dict: Time series data with 't' and 'data' arrays
"""
class CallbackTracker:
def __init__(self, func, interrupts=1):
"""
Call custom function during simulation.
Parameters:
- func: callable, function to call with (state, time)
- interrupts: interrupt schedule
"""
def handle(self, field, t):
"""
Handle data by calling user function.
Parameters:
- field: FieldBase, current field state
- t: float, current time
"""Visualization Trackers
Create plots and visualizations during simulation execution.
class PlotTracker:
def __init__(self, interrupts=1, *, filename=None, **kwargs):
"""
Create static plots during simulation.
Parameters:
- interrupts: interrupt schedule for plotting
- filename: str, filename pattern (supports time formatting)
- kwargs: additional plotting parameters
"""
def finalize(self):
"""Finalize and save any pending plots"""
class LivePlotTracker:
def __init__(self, interrupts=1, *, show=True, **kwargs):
"""
Create live updating plots during simulation.
Parameters:
- interrupts: interrupt schedule for plot updates
- show: bool, whether to display plots
- kwargs: additional plotting parameters
"""
class InteractivePlotTracker:
def __init__(self, interrupts=1, *, **kwargs):
"""
Interactive plotting with napari integration.
Parameters:
- interrupts: interrupt schedule
- kwargs: napari viewer parameters
Note:
Requires napari installation
"""Progress and Status Trackers
Monitor simulation progress and provide status updates.
class ProgressTracker:
def __init__(self, interrupts="0:10"):
"""
Display simulation progress bar.
Parameters:
- interrupts: interrupt schedule (default: every 10 real-time seconds)
"""
class PrintTracker:
def __init__(self, interrupts=1):
"""
Print field information during simulation.
Parameters:
- interrupts: interrupt schedule for printing
"""
class RuntimeTracker:
def __init__(self, interrupts=1):
"""
Track simulation runtime and performance.
Parameters:
- interrupts: interrupt schedule for timing updates
"""
@property
def runtime(self):
"""float: Total runtime in seconds"""
def get_performance_data(self):
"""
Get detailed performance statistics.
Returns:
dict: Performance metrics
"""Analysis Trackers
Analyze field properties and detect special conditions during simulation.
class SteadyStateTracker:
def __init__(self, atol=1e-8, rtol=1e-5, interrupts=1):
"""
Detect when simulation reaches steady state.
Parameters:
- atol: float, absolute tolerance for steady state
- rtol: float, relative tolerance for steady state
- interrupts: interrupt schedule for checking
"""
@property
def is_steady_state(self):
"""bool: Whether steady state has been reached"""
class ConsistencyTracker:
def __init__(self, interrupts=1):
"""
Check field consistency and detect numerical issues.
Parameters:
- interrupts: interrupt schedule for consistency checks
"""
def check_finite(self, field):
"""
Check if field contains only finite values.
Parameters:
- field: FieldBase, field to check
Returns:
bool: True if all values are finite
"""
class MaterialConservationTracker:
def __init__(self, interrupts=1):
"""
Monitor conservation of integrated quantities.
Parameters:
- interrupts: interrupt schedule for conservation checks
"""
@property
def conserved_quantity(self):
"""float: Current value of conserved quantity"""Interrupt Scheduling
Classes for defining when trackers should be called during simulation.
class ConstantInterrupts:
def __init__(self, dt):
"""
Constant time intervals.
Parameters:
- dt: float, time interval between interrupts
"""
class FixedInterrupts:
def __init__(self, times):
"""
Fixed time points.
Parameters:
- times: array-like, specific times for interrupts
"""
class LogarithmicInterrupts:
def __init__(self, t_start, t_end, *, factor=2):
"""
Logarithmically spaced intervals.
Parameters:
- t_start: float, first interrupt time
- t_end: float, final time
- factor: float, spacing factor between interrupts
"""
class RealtimeInterrupts:
def __init__(self, duration):
"""
Real-time based intervals.
Parameters:
- duration: float, real-time seconds between interrupts
"""
def parse_interrupt(interrupts):
"""
Parse interrupt specification from various formats.
Parameters:
- interrupts: int, float, str, or InterruptData, interrupt specification
Returns:
InterruptData: Parsed interrupt schedule
"""Tracker Management
Functions and classes for organizing multiple trackers.
def get_named_trackers():
"""
Get dictionary of available named trackers.
Returns:
dict: Mapping of names to tracker classes
"""
class TrackerCollection:
def __init__(self, trackers=None):
"""
Collection of multiple trackers.
Parameters:
- trackers: list of TrackerBase, trackers to collect
"""
def append(self, tracker):
"""
Add tracker to collection.
Parameters:
- tracker: TrackerBase, tracker to add
"""
def handle(self, field, t):
"""
Handle data with all trackers in collection.
Parameters:
- field: FieldBase, current field state
- t: float, current time
"""Usage Examples
Basic Data Collection
import pde
# Set up simulation
grid = pde.UnitGrid([64], periodic=True)
state = pde.ScalarField.random_uniform(grid)
eq = pde.DiffusionPDE(diffusivity=0.1)
# Track data every 0.5 time units
tracker = pde.DataTracker(interrupts=0.5)
# Run simulation
result = eq.solve(state, t_range=10.0, tracker=tracker)
# Analyze collected data
times = tracker.data.times
print(f"Collected {len(times)} time points")
print(f"Time range: {times[0]:.2f} to {times[-1]:.2f}")Visualization During Simulation
import pde
grid = pde.CartesianGrid([[0, 10], [0, 10]], [64, 64])
state = pde.ScalarField.random_uniform(grid)
eq = pde.AllenCahnPDE()
# Create movie during simulation
movie_tracker = pde.PlotTracker(
interrupts=0.1,
filename="evolution_{time:06.2f}.png"
)
# Live plot updates
live_tracker = pde.LivePlotTracker(interrupts=0.5)
# Combine trackers
trackers = [movie_tracker, live_tracker]
result = eq.solve(state, t_range=5.0, tracker=trackers)
print("Simulation complete - check generated images")Progress Monitoring
import pde
# Long-running simulation setup
grid = pde.UnitGrid([128, 128], periodic=True)
eq = pde.SwiftHohenbergPDE()
state = eq.get_initial_condition(grid)
# Multiple progress trackers
trackers = [
pde.ProgressTracker(), # Progress bar
pde.RuntimeTracker(interrupts=1.0), # Performance monitoring
pde.PrintTracker(interrupts=5.0) # Periodic status prints
]
result = eq.solve(state, t_range=100.0, tracker=trackers)Steady State Detection
import pde
grid = pde.UnitGrid([64], periodic=False)
eq = pde.DiffusionPDE(diffusivity=1.0)
state = pde.ScalarField.random_uniform(grid, 0.4, 0.6)
# Detect when simulation reaches steady state
steady_tracker = pde.SteadyStateTracker(
atol=1e-6,
rtol=1e-4,
interrupts=0.1
)
try:
result = eq.solve(state, t_range=float('inf'), tracker=steady_tracker)
print(f"Reached steady state at t={steady_tracker.time:.3f}")
except StopIteration:
print("Steady state detected - simulation stopped")Custom Analysis Tracker
import pde
import numpy as np
class EnergyTracker(pde.TrackerBase):
"""Custom tracker for monitoring system energy"""
def __init__(self, interrupts=1):
super().__init__(interrupts=interrupts)
self.energies = []
self.times = []
def handle(self, field, t):
# Calculate energy as field variance
energy = np.var(field.data)
self.energies.append(energy)
self.times.append(t)
# Stop if energy becomes too high
if energy > 10.0:
raise StopIteration("Energy exceeded threshold")
# Use custom tracker
grid = pde.UnitGrid([32], periodic=True)
eq = pde.KuramotoSivashinskyPDE()
state = eq.get_initial_condition(grid)
energy_tracker = EnergyTracker(interrupts=0.1)
result = eq.solve(state, t_range=50.0, tracker=energy_tracker)
print(f"Final energy: {energy_tracker.energies[-1]:.3f}")Real-time Interrupt Scheduling
import pde
grid = pde.UnitGrid([64, 64], periodic=True)
eq = pde.AllenCahnPDE()
state = eq.get_initial_condition(grid)
# Different interrupt schedules
trackers = [
pde.DataTracker("0:30"), # Every 30 real-time seconds
pde.PlotTracker("1.0"), # Every 1.0 simulation time units
pde.ProgressTracker("0:5") # Every 5 real-time seconds
]
result = eq.solve(state, t_range=20.0, tracker=trackers)Additional Interrupt Classes
Additional interrupt scheduling options for specialized use cases.
class GeometricInterrupts:
def __init__(self, initial_dt, factor, *, max_dt=None):
"""
Geometric sequence of interrupt times.
Parameters:
- initial_dt: float, initial time interval
- factor: float, multiplication factor for each step
- max_dt: float, optional maximum time interval
"""Tracker Registry Functions
Functions for discovering and managing available tracker types.
def get_named_trackers():
"""
Get dictionary of all named tracker classes.
Returns:
dict: Mapping of names to tracker classes
"""
def parse_interrupt(interrupts, t_start=0, t_end=None):
"""
Parse interrupt specification into interrupt object.
Parameters:
- interrupts: interrupt specification (float, str, or object)
- t_start: float, start time for simulation
- t_end: float, end time for simulation
Returns:
InterruptBase: Parsed interrupt object
"""Install with Tessl CLI
npx tessl i tessl/pypi-py-pde