Simulation Module

mmpp.simulation.upgrade_log_level(current_level, new_level)[source]

Pomocnicza funkcja do “promowania” poziomu logowania. Zwraca “najgorszy” (najwyższy priorytetem) z podanych. Priorytet: ERROR > WARNING > INFO

Parameters:

current_level (str)
new_level (str)

Return type:

str

class mmpp.simulation.SimulationManager(main_path, destination_path, prefix)[source]

Bases: object

Parameters:

main_path (str)
destination_path (str)
prefix (str)

__init__(main_path, destination_path, prefix)[source]

Parameters:

main_path (str)
destination_path (str)
prefix (str)

static create_path_if_not_exists(file_path)[source]

Ensure the directory for a given file path exists.

Parameters:: file_path (str)
Return type:: None

static verify_or_replace_file(new_file_path, existing_file_path)[source]

Check if a file needs replacing and replace if necessary.

Parameters:

new_file_path (str)
existing_file_path (str)

Return type:

bool

static find_status_file(path, sim_name, status)[source]

Locate a status file based on its name and type.

Parameters:

path (str)
sim_name (str)
status (str)

Return type:

Optional[str]

static get_file_status(file_path)[source]

Determine the status of a file based on its name.

Parameters:: file_path (str)
Return type:: str

static extract_sim_key(file_path)[source]

Extract a concise simulation key from a file path.

Parameters:: file_path (str)
Return type:: str

static check_simulation_completion(zarr_path)[source]

Check if a simulation represented by a .zarr directory is complete and return a tuple (is_complete, file_count).

is_complete: bool: True jeśli spełniony warunek >= 3000 plików w grupie ‘m’ oraz obecny ‘end_time’. False w przeciwnym razie, np. gdy brakuje ‘end_time’, ‘m’ albo jest < 3000 plików.
file_count: int: Rzeczywista liczba plików (kroków) w grupie ‘m’; 0 w razie błędu lub braku danych.

Parameters:: zarr_path (str)
Return type:: tuple[bool, int]

submit_python_code(code_to_execute, last_param_name=None, cleanup=False, sbatch=True, check=False, force=False, full_name=False, **kwargs)[source]

Submit a Python simulation based on provided parameters. Zmodyfikowana tak, aby wszystkie komunikaty na końcu łączyć w jeden raport.

Parameters:

code_to_execute (str)
last_param_name (Optional[str])
cleanup (bool)
sbatch (bool)
check (bool)
force (bool)
full_name (bool)
kwargs (Union[str, float, int])

Return type:

None

gen_sbatch_script(name, path)[source]

Generate an sbatch script for submitting jobs.

Parameters:

name (str)
path (str)

Return type:

str

static replace_variables_in_template(file_path, variables)[source]

Replace placeholders in a template file with actual values.

Parameters:

file_path (str)
variables (dict[str, Union[str, float, int]])

Return type:

str

static raw_code(*args, **kwargs)[source]

Generate the raw code by filling in a template with parameters.

Parameters:

args (Any)
kwargs (Union[str, float, int])

Return type:

str

submit_all_simulations(params, last_param_name, minsim=0, maxsim=None, sbatch=True, cleanup=False, template='template.mx3', check=False, force=False, pairs=False)[source]

Submit all simulations based on parameter combinations.

If pairs=False (default), generates all combinations using itertools.product. If pairs=True, generates paired values (par1[i], par2[i], …) where each parameter must have the same number of values.

Parameters:

params (dict[str, Any])
last_param_name (str)
minsim (int)
maxsim (Optional[int])
sbatch (bool)
cleanup (bool)
template (str)
check (bool)
force (bool)
pairs (bool)

Return type:

None

submit_all_simulations_with_progress(params, last_param_name, minsim=0, maxsim=None, sbatch=True, cleanup=False, template='template.mx3', check=False, force=False, pairs=False, progress_callback=None)[source]

Submit all simulations with progress tracking and organized folder structure.

Parameters:

params (dict[str, Any])
last_param_name (str)
minsim (int)
maxsim (Optional[int])
sbatch (bool)
cleanup (bool)
template (str)
check (bool)
force (bool)
pairs (bool)
progress_callback (Optional[callable])

Return type:

None

class mmpp.simulation.SimulationSwapper(config_file)[source]

Bases: object

Klasa do obsługi swapowania symulacji na podstawie plików konfiguracyjnych. Parsuje pliki YAML z parametrami i zarządza wykonaniem symulacji.

Parameters:: config_file (str)

__init__(config_file)[source]

Inicjalizacja SwappeParsera z plikiem konfiguracyjnym.

Parameters:: config_file (str) – Ścieżka do pliku konfiguracyjnego (YAML)

get_simulation_manager()[source]

Stwórz instancję SimulationManager na podstawie konfiguracji.

Return type:: SimulationManager

run_simulations()[source]

Uruchom wszystkie symulacje na podstawie konfiguracji.

Return type:: None

validate_config()[source]

Waliduj konfigurację i zwróć listę błędów/ostrzeżeń.

Return type:: list[str]
Returns:: Lista komunikatów o błędach/ostrzeżeniach

get_info()[source]

Zwróć informacje o konfiguracji.

Return type:: dict[str, Any]

class mmpp.simulation.TemplateParser(template_path)[source]

Bases: object

Klasa do parsowania plików template.mx3 i wyciągania parametrów do swapowania.

Parameters:: template_path (str)

__init__(template_path)[source]

Inicjalizacja parsera z ścieżką do pliku template.

Parameters:: template_path (str) – Ścieżka do pliku template.mx3

get_parameters()[source]

Zwróć listę parametrów w kolejności alfabetycznej.

Return type:: list[str]

generate_yaml_template(last_param=None, prefix='v1', template_name='template.mx3')[source]

Generuj szablon YAML na podstawie wykrytych parametrów.

Parameters:

last_param (Optional[str]) – Nazwa ostatniego parametru (do iteracji)
prefix (str) – Prefix symulacji
template_name (str) – Nazwa pliku template

Return type:

str

Returns:

String z zawartością YAML

Simulation Management

SimulationManager

class mmpp.simulation.SimulationManager(main_path, destination_path, prefix)[source]

Bases: object

Parameters:

main_path (str)
destination_path (str)
prefix (str)

__init__(main_path, destination_path, prefix)[source]

Parameters:

main_path (str)
destination_path (str)
prefix (str)

static create_path_if_not_exists(file_path)[source]

Ensure the directory for a given file path exists.

Parameters:: file_path (str)
Return type:: None

static verify_or_replace_file(new_file_path, existing_file_path)[source]

Check if a file needs replacing and replace if necessary.

Parameters:

new_file_path (str)
existing_file_path (str)

Return type:

bool

static find_status_file(path, sim_name, status)[source]

Locate a status file based on its name and type.

Parameters:

path (str)
sim_name (str)
status (str)

Return type:

Optional[str]

static get_file_status(file_path)[source]

Determine the status of a file based on its name.

Parameters:: file_path (str)
Return type:: str

static extract_sim_key(file_path)[source]

Extract a concise simulation key from a file path.

Parameters:: file_path (str)
Return type:: str

static check_simulation_completion(zarr_path)[source]

Check if a simulation represented by a .zarr directory is complete and return a tuple (is_complete, file_count).

is_complete: bool: True jeśli spełniony warunek >= 3000 plików w grupie ‘m’ oraz obecny ‘end_time’. False w przeciwnym razie, np. gdy brakuje ‘end_time’, ‘m’ albo jest < 3000 plików.
file_count: int: Rzeczywista liczba plików (kroków) w grupie ‘m’; 0 w razie błędu lub braku danych.

Parameters:: zarr_path (str)
Return type:: tuple[bool, int]

submit_python_code(code_to_execute, last_param_name=None, cleanup=False, sbatch=True, check=False, force=False, full_name=False, **kwargs)[source]

Submit a Python simulation based on provided parameters. Zmodyfikowana tak, aby wszystkie komunikaty na końcu łączyć w jeden raport.

Parameters:

code_to_execute (str)
last_param_name (Optional[str])
cleanup (bool)
sbatch (bool)
check (bool)
force (bool)
full_name (bool)
kwargs (Union[str, float, int])

Return type:

None

gen_sbatch_script(name, path)[source]

Generate an sbatch script for submitting jobs.

Parameters:

name (str)
path (str)

Return type:

str

static replace_variables_in_template(file_path, variables)[source]

Replace placeholders in a template file with actual values.

Parameters:

file_path (str)
variables (dict[str, Union[str, float, int]])

Return type:

str

static raw_code(*args, **kwargs)[source]

Generate the raw code by filling in a template with parameters.

Parameters:

args (Any)
kwargs (Union[str, float, int])

Return type:

str

submit_all_simulations(params, last_param_name, minsim=0, maxsim=None, sbatch=True, cleanup=False, template='template.mx3', check=False, force=False, pairs=False)[source]

Submit all simulations based on parameter combinations.

If pairs=False (default), generates all combinations using itertools.product. If pairs=True, generates paired values (par1[i], par2[i], …) where each parameter must have the same number of values.

Parameters:

params (dict[str, Any])
last_param_name (str)
minsim (int)
maxsim (Optional[int])
sbatch (bool)
cleanup (bool)
template (str)
check (bool)
force (bool)
pairs (bool)

Return type:

None

submit_all_simulations_with_progress(params, last_param_name, minsim=0, maxsim=None, sbatch=True, cleanup=False, template='template.mx3', check=False, force=False, pairs=False, progress_callback=None)[source]

Submit all simulations with progress tracking and organized folder structure.

Parameters:

params (dict[str, Any])
last_param_name (str)
minsim (int)
maxsim (Optional[int])
sbatch (bool)
cleanup (bool)
template (str)
check (bool)
force (bool)
pairs (bool)
progress_callback (Optional[callable])

Return type:

None

Usage Examples

Parameter Sweeps

from mmpp.simulation import SimulationManager
import numpy as np

# Initialize simulation manager
manager = SimulationManager(
    main_path="/path/to/simulations/",
    destination_path="/path/to/results/",
    prefix="sweep_v1"
)

# Define parameter space
params = {
    "b0": np.linspace(0.0001, 0.01, 10),
    "fcut": np.linspace(2.6, 2.8, 20),
}

# Submit all simulations
manager.submit_all_simulations(
    params=params,
    last_param_name="fcut",
    sbatch=True,
    template="template.mx3"
)

Paired Parameter Studies

# For paired parameters (not Cartesian product)
params = {
    "field": [0.1, 0.2, 0.3],
    "freq": [2.0, 2.5, 3.0],  # Must have same length
}

manager.submit_all_simulations(
    params=params,
    last_param_name="freq",
    pairs=True  # Use paired values instead of product
)