dedebenui/scgenerator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

New branch-generator system

Browse Source
This commit is contained in:
Benoît Sierro
2021-09-02 10:53:28 +02:00
parent 7746dae8d4
commit f1fd9c6c39
6 changed files with 345 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
README.md
View File

@@ -21,7 +21,7 @@ SCGENERATOR_PBAR_POLICY : "none", "file", "print", "both", optional
# Configuration
You can load parameters by simply passing the path to a toml file to the appropriate simulation function. Each possible key of this dictionary is described below. Every value must be given in standard SI units (m, s, W, J, ...)
The configuration file can have a ```name``` parameter at the root and must otherwise contain the following sections with the specified parameters. Every section ("fiber", "gas", "pulse", "simulation") can have a "variable" subsection where parameters are specified in a list INSTEAD of being specified as a single value in the main section. This has the effect of running one simulation per value in the list. If many parameters are variable, all possible combinations are ran.
the root of the file has information concerning the whole simulation : name, grid information, input pulse, ...
Examples :
```
@@ -40,6 +40,48 @@ n2 = [2.1e-20, 2.4e-20, 2.6e-20]
```
NOT ALLOWED
Here is an example of a configuration file
```
name = "Test/Compound 1"
field_file = "Toptica/init_field.npz"
repetition_rate = 40e6
wavelength = 1535.6e-9
dt = 1e-15
t_num = 16384
tolerated_error = 1e-6
quantum_noise = true
z_num = 32
mean_power = 200e-3
repeat = 3
# will be applied to the whole simulation fiber PM1550_1 only
# fiber parameters specified here would apply to the whole simulation as well
# unless overridden in one of the individual fiber
[variable]
behaviors = [["spm", "ss", "raman"], ["spm", "ss"]]
raman_type = ["agrawal", "stolen"]
[[Fiber]]
name = "PM1550_1"
n2 = 2.2e-20
dispersion_file = "PM1550/Dispersion/PM1550XP extrapolated 1.npz"
length = 0.01
effective_mode_diameter = 10.1e-6
[[Fiber]]
name = "PM2000D_2"
length = 0.01
n2 = 3.4e-20
A_eff_file = "PM2000D/PM2000D_A_eff_marcuse.npz"
dispersion_file = "PM2000D/Dispersion/PM2000D_1 extrapolated 0 4.npz"
[Fiber.variable] # this variable parameter will be applied to PM2000D_2
input_transmission = [0.9, 0.95]
```
note : internally, another structure with a flattened dictionary is used
@@ -129,10 +171,13 @@ n2 : float, optional
A_eff : float, optional
effective mode field area
A_eff_file : str, optional
file containing an A_eff array (in m^2) as function of a wavelength array (in m)
length: float, optional
length of the fiber in m. default : 1
input_transmission : float
input_transmission : float, optional
number between 0 and 1 indicating how much light enters the fiber, useful when chaining many fibers together, default : 1

52
play.py
View File

@@ -1,15 +1,47 @@
from enum import Enum, auto
from typing import Any, Generator
import scgenerator as sc
import itertools
import numpy as np
class Test:
class State(Enum):
complete = auto()
partial = auto()
absent = auto()
class DataPather:
def __init__(self, dl: list[dict[str, Any]]):
self.dict_list = dl
self.n = len(self.dict_list)
self.final_list = list(self.dico_iterator(self.n))
def state(self):
return self.State.complete
def dico_iterator(self, index: int) -> Generator[list[list[tuple[str, Any]]], None, None]:
d_tem_list = [el for d in self.dict_list[: index + 1] for el in d.items()]
dict_pos = np.cumsum([0] + [len(d) for d in self.dict_list[: index + 1]])
ranges = [range(len(l)) for _, l in d_tem_list]
for r in itertools.product(*ranges):
flat = [(d_tem_list[i][0], d_tem_list[i][1][j]) for i, j in enumerate(r)]
out = [flat[left:right] for left, right in zip(dict_pos[:-1], dict_pos[1:])]
yield out
def all_vary_list(self, index):
for l in self.dico_iterator(index):
yield sc.utils.parameter.format_variable_list(
sc.utils.parameter.reduce_all_variable(l[:index])
), sc.utils.parameter.format_variable_list(
sc.utils.parameter.reduce_all_variable(l)
), l[
index
]
a = Test()
print(a.state() == Test.State.complete)
configs, name = sc.utils.load_config_sequence(
"/Users/benoitsierro/Nextcloud/PhD/Supercontinuum/PCF Simulations/Test/NewStyle.toml"
)
dp = DataPather([config["variable"] for config in configs])
# pprint(list(dp.dico_iterator(1)))
for i in range(3):
for prev_path, this_path, this_vary in dp.all_vary_list(i):
print(prev_path)
print(this_path)
print(this_vary)
print()
print()

2
src/scgenerator/physics/simulate.py
View File

@@ -703,7 +703,7 @@ def run_simulation(
config_file: os.PathLike,
method=None,
):
config = Configuration.load(config_file)
config = Configuration(config_file)
sim = new_simulation(config, method)
sim.run()

2
src/scgenerator/scripts/slurm_submit.py
View File

@@ -126,7 +126,7 @@ def main():
"time format must be an integer number of minute or must match the pattern hh:mm:ss"
)
config = Configuration.load(args.config)
config = Configuration(args.config)
final_name = config.name
sim_num = config.num_sim

129
src/scgenerator/utils/__init__.py
View File

@@ -14,9 +14,11 @@ import re
import shutil
import threading
from collections import abc
from copy import deepcopy
from io import StringIO
from pathlib import Path
from typing import Any, Callable, Generator, Iterable, Sequence, TypeVar, Union
from string import printable as str_printable
from typing import Any, Callable, Generator, Iterable, MutableMapping, Sequence, TypeVar, Union
import numpy as np
import pkg_resources as pkg
@@ -28,7 +30,6 @@ from ..env import TMP_FOLDER_KEY_BASE, data_folder, pbar_policy
from ..errors import IncompleteDataFolderError
from ..logger import get_logger
T_ = TypeVar("T_")
PathTree = list[tuple[Path, ...]]
@@ -119,6 +120,28 @@ def save_toml(path: os.PathLike, dico):
return dico
def load_config_sequence(final_config_path: os.PathLike) -> tuple[list[dict[str, Any]], str]:
loaded_config = load_toml(final_config_path)
final_name = loaded_config.get("name")
fiber_list = loaded_config.pop("Fiber")
configs = []
if fiber_list is not None:
for i, params in enumerate(fiber_list):
params.setdefault("variable", loaded_config.get("variable", {}) if i == 0 else {})
configs.append(loaded_config | params)
else:
configs.append(loaded_config)
while "previous_config_file" in configs[0]:
configs.insert(0, load_toml(configs[0]["previous_config_file"]))
configs[0].setdefault("variable", {})
for pre, nex in zip(configs[:-1], configs[1:]):
variable = nex.pop("variable", {})
nex.update({k: v for k, v in pre.items() if k not in nex})
nex["variable"] = variable
return configs, final_name
def save_parameters(
params: dict[str, Any], destination_dir: Path, file_name: str = PARAM_FN
) -> Path:
@@ -163,23 +186,6 @@ def load_material_dico(name: str) -> dict[str, Any]:
return toml.loads(Paths.gets("gas"))[name]
def get_data_dirs(sim_dir: Path) -> list[Path]:
"""returns a list of absolute paths corresponding to a particular run
Parameters
----------
sim_dir : Path
path to directory containing the initial config file and the spectra sub folders
Returns
-------
list[Path]
paths to sub folders
"""
return [p.resolve() for p in sim_dir.glob("*") if p.is_dir()]
def update_appended_params(source: Path, destination: Path, z: Sequence):
z_num = len(z)
params = load_toml(source)
@@ -195,10 +201,21 @@ def update_appended_params(source: Path, destination: Path, z: Sequence):
save_toml(destination, params)
def to_62(i: int) -> str:
arr = []
if i == 0:
return "0"
i = abs(i)
while i:
i, value = divmod(i, 62)
arr.append(str_printable[value])
return "".join(reversed(arr))
def build_path_trees(sim_dir: Path) -> list[PathTree]:
sim_dir = sim_dir.resolve()
path_branches: list[tuple[Path, ...]] = []
to_check = list(sim_dir.glob("id*num*"))
to_check = list(sim_dir.glob("*fiber*num*"))
with PBars(len(to_check), desc="Building path trees") as pbar:
for branch in map(build_path_branch, to_check):
if branch is not None:
@@ -260,7 +277,7 @@ def group_path_branches(path_branches: list[tuple[Path, ...]]) -> list[PathTree]
b_id = branch_id(branch)
out_trees_map.setdefault(b_id, {i: {} for i in range(size)})
for sim_part, data_dir in enumerate(branch):
*_, num = data_dir.name.split()
num = re.search(r"(?<=num )[0-9]+", data_dir.name)[0]
out_trees_map[b_id][sim_part][int(num)] = data_dir
return [
@@ -335,7 +352,7 @@ def merge(destination: os.PathLike, path_trees: list[PathTree] = None):
)
for path_tree in path_trees:
pbars.reset(1)
iden_items = path_tree[-1][0].name.split()[2:-2]
iden_items = path_tree[-1][0].name.split()[2:]
for i, p_name in list(enumerate(iden_items))[-2::-2]:
if p_name == "num":
del iden_items[i + 1]
@@ -572,62 +589,7 @@ def progress_worker(
def branch_id(branch: tuple[Path, ...]) -> str:
return "".join("".join(re.sub(r"id\d+\S*num\d+", "", b.name).split()[2:-2]) for b in branch)
def check_data_integrity(sub_folders: list[Path], init_z_num: int):
"""checks the integrity and completeness of a simulation data folder
Parameters
----------
path : str
path to the data folder
init_z_num : int
z_num as specified by the initial configuration file
Raises
------
IncompleteDataFolderError
raised if not all spectra are present in any folder
"""
for sub_folder in PBars(sub_folders, "Checking integrity"):
if num_left_to_propagate(sub_folder, init_z_num) != 0:
raise IncompleteDataFolderError(
f"not enough spectra of the specified {init_z_num} found in {sub_folder}"
)
def num_left_to_propagate(sub_folder: Path, init_z_num: int) -> int:
"""checks if a propagation has completed
Parameters
----------
sub_folder : Path
path to the sub folder containing the spectra
init_z_num : int
number of z position to store as specified in the master config file
Returns
-------
bool
True if the propagation has completed
Raises
------
IncompleteDataFolderError
raised if init_z_num doesn't match that specified in the individual parameter file
"""
z_num = load_toml(sub_folder / PARAM_FN)["z_num"]
num_spectra = find_last_spectrum_num(sub_folder) + 1 # because of zero-indexing
if z_num != init_z_num:
raise IncompleteDataFolderError(
f"initial config specifies {init_z_num} spectra per"
+ f" but the parameter file in {sub_folder} specifies {z_num}"
)
return z_num - num_spectra
return branch[-1].name.split()[1]
def find_last_spectrum_num(data_dir: Path):
@@ -653,3 +615,14 @@ def auto_crop(x: np.ndarray, y: np.ndarray, rel_thr: float = 0.01) -> np.ndarray
np.arange(ind_above[-1] + 1, min(len(y), ind_above[-1] + width)),
)
)
def translate_parameters(d: dict[str, Any]) -> dict[str, Any]:
old_names = dict(interp_degree="interpolation_degree")
new = {}
for k, v in d.items():
if isinstance(v, MutableMapping):
new[k] = translate_parameters(v)
else:
new[old_names.get(k, k)] = v
return new

364
src/scgenerator/utils/parameter.py
View File

@@ -6,12 +6,11 @@ import os
import re
import time
from collections import defaultdict
from copy import deepcopy
from copy import copy, deepcopy
from dataclasses import asdict, dataclass, fields
from functools import cache, lru_cache
from pathlib import Path
from typing import Any, Callable, Generator, Iterable, Literal, Optional, TypeVar, Union
import numpy as np
from .. import math, utils
@@ -25,6 +24,76 @@ T = TypeVar("T")
# Validator
VALID_VARIABLE = {
"dispersion_file",
"prev_data_dir",
"field_file",
"loss_file",
"A_eff_file",
"beta2_coefficients",
"gamma",
"pitch",
"pitch_ratio",
"effective_mode_diameter",
"core_radius",
"capillary_num",
"capillary_outer_d",
"capillary_thickness",
"capillary_spacing",
"capillary_resonance_strengths",
"capillary_nested",
"he_mode",
"fit_parameters",
"input_transmission",
"n2",
"pressure",
"temperature",
"gas_name",
"plasma_density",
"peak_power",
"mean_power",
"peak_power",
"energy",
"quantum_noise",
"shape",
"wavelength",
"intensity_noise",
"width",
"t0",
"soliton_num",
"behaviors",
"raman_type",
"tolerated_error",
"step_size",
"interpolation_degree",
"ideal_gas",
}
MANDATORY_PARAMETERS = [
"name",
"w_c",
"w",
"w0",
"w_power_fact",
"alpha",
"spec_0",
"field_0",
"input_transmission",
"z_targets",
"length",
"beta2_coefficients",
"gamma_arr",
"behaviors",
"raman_type",
"hr_w",
"adapt_step_size",
"tolerated_error",
"dynamic_dispersion",
"recovery_last_stored",
"output_path",
]
@lru_cache
def type_checker(*types):
def _type_checker_wrapper(validator, n=None):
@@ -173,28 +242,6 @@ def func_validator(name, n):
raise TypeError(f"{name!r} must be callable")
# other
def translate(p_name: str, p_value: T) -> tuple[str, T]:
"""translates old parameters
Parameters
----------
p_name : str
parameter name
p_value : T
parameter value
Returns
-------
tuple[str, T]
translated pair
"""
old_names = dict(interp_degree="interpolation_degree")
return old_names.get(p_name, p_name), p_value
# classes
@@ -259,76 +306,6 @@ class Parameter:
return f"{num_str} {unit}"
valid_variable = {
"dispersion_file",
"prev_data_dir",
"field_file",
"loss_file",
"A_eff_file",
"beta2_coefficients",
"gamma",
"pitch",
"pitch_ratio",
"effective_mode_diameter",
"core_radius",
"capillary_num",
"capillary_outer_d",
"capillary_thickness",
"capillary_spacing",
"capillary_resonance_strengths",
"capillary_nested",
"he_mode",
"fit_parameters",
"input_transmission",
"n2",
"pressure",
"temperature",
"gas_name",
"plasma_density",
"peak_power",
"mean_power",
"peak_power",
"energy",
"quantum_noise",
"shape",
"wavelength",
"intensity_noise",
"width",
"t0",
"soliton_num",
"behaviors",
"raman_type",
"tolerated_error",
"step_size",
"interpolation_degree",
"ideal_gas",
}
mandatory_parameters = [
"name",
"w_c",
"w",
"w0",
"w_power_fact",
"alpha",
"spec_0",
"field_0",
"input_transmission",
"z_targets",
"length",
"beta2_coefficients",
"gamma_arr",
"behaviors",
"raman_type",
"hr_w",
"adapt_step_size",
"tolerated_error",
"dynamic_dispersion",
"recovery_last_stored",
"output_path",
]
@dataclass
class Parameters:
"""
@@ -396,7 +373,9 @@ class Parameters:
t0: float = Parameter(in_range_excl(0, 1e-9), display_info=(1e15, "fs"))
# simulation
behaviors: str = Parameter(validator_list(literal("spm", "raman", "ss")), default=["spm", "ss"])
behaviors: tuple[str] = Parameter(
validator_list(literal("spm", "raman", "ss")), converter=tuple, default=("spm", "ss")
)
parallel: bool = Parameter(boolean, default=True)
raman_type: str = Parameter(
literal("measured", "agrawal", "stolen"), converter=str.lower, default="agrawal"
@@ -453,7 +432,7 @@ class Parameters:
param_dict = {k: v for k, v in asdict(self).items() if v is not None}
evaluator = Evaluator.default()
evaluator.set(**param_dict)
for p_name in mandatory_parameters:
for p_name in MANDATORY_PARAMETERS:
evaluator.compute(p_name)
valid_fields = self.all_parameters()
for k, v in evaluator.params.items():
@@ -603,7 +582,7 @@ class Evaluator:
def evaluate_default(cls, params: dict[str, Any], check_only=False) -> dict[str, Any]:
evaluator = cls.default()
evaluator.set(**params)
for target in mandatory_parameters:
for target in MANDATORY_PARAMETERS:
evaluator.compute(target, check_only=check_only)
return evaluator.params
@@ -798,19 +777,17 @@ class Configuration:
WAIT = enum.auto()
SKIP = enum.auto()
@classmethod
def load(cls, path: os.PathLike) -> "Configuration":
return cls(utils.load_toml(path))
def __init__(
self,
final_config: dict[str, Any],
final_config_path: os.PathLike,
overwrite: bool = True,
skip_callback: Callable[[int], None] = None,
):
self.logger = get_logger(__name__)
self.configs = [final_config]
self.configs, self.name = utils.load_config_sequence(final_config_path)
if self.name is None:
self.name = Parameters.name.default
self.z_num = 0
self.total_length = 0.0
self.total_num_steps = 0
@@ -818,20 +795,23 @@ class Configuration:
self.overwrite = overwrite
self.skip_callback = skip_callback
self.worker_num = self.configs[0].get("worker_num", max(1, os.cpu_count() // 2))
while "previous_config_file" in self.configs[0]:
self.configs.insert(0, utils.load_toml(self.configs[0]["previous_config_file"]))
self.override_configs()
self.name = self.configs[-1].get("name", Parameters.name.default)
self.repeat = self.configs[0].get("repeat", 1)
names = set()
for i, config in enumerate(self.configs):
self.z_num += config["z_num"]
self.total_length += config["length"]
config.setdefault("name", f"{Parameters.name.default} {i}")
given_name = config["name"]
i = 0
while config["name"] in names:
config["name"] = given_name + f"_{i}"
i += 1
names.add(config["name"])
self.sim_dirs.append(
utils.ensure_folder(
Path(config["name"] + PARAM_SEPARATOR + "sc_tmp"),
Path("__" + config["name"] + "__"),
mkdir=False,
prevent_overwrite=not self.overwrite,
)
@@ -855,53 +835,38 @@ class Configuration:
)
self.parallel = self.configs[0].get("parallel", Parameters.parallel.default)
def override_configs(self):
self.configs[0].setdefault("variable", {})
for pre, nex in zip(self.configs[:-1], self.configs[1:]):
variable = nex.pop("variable", {})
nex.update({k: v for k, v in pre.items() if k not in nex})
nex["variable"] = variable
def __validate_variable(self, config: dict[str, Any]):
for k, v in config.get("variable", {}).items():
p = getattr(Parameters, k)
validator_list(p.validator)("variable " + k, v)
if k not in valid_variable:
if k not in VALID_VARIABLE:
raise TypeError(f"{k!r} is not a valid variable parameter")
if len(v) == 0:
raise ValueError(f"variable parameter {k!r} must not be empty")
def __compute_sim_dirs(self):
self.data_dirs: list[list[Path]] = [None] * len(self.configs)
self.all_required: list[list[tuple[list[tuple[str, Any]], dict[str, Any]]]] = [None] * len(
self.configs
)
self.all_required[0], self.data_dirs[0] = self.__prepare_1_fiber(
0, self.configs[0], first=True
)
self.all_required = []
self.data_dirs = []
self.configs[0]["variable"]["num"] = list(range(self.configs[0].get("repeat", 1)))
dp = DataPather([c["variable"] for c in self.configs])
for i, conf in enumerate(self.configs):
self.all_required.append([])
self.data_dirs.append([])
for prev_path, this_path, this_vary in dp.all_vary_list(i):
this_conf = conf.copy()
if i > 0:
prev_path = utils.ensure_folder(
self.sim_dirs[i - 1] / prev_path, not self.overwrite, False
)
this_conf["prev_data_dir"] = str(prev_path)
for i, config in enumerate(self.configs[1:]):
config["variable"]["prev_data_dir"] = [str(p.resolve()) for p in self.data_dirs[i]]
self.all_required[i + 1], self.data_dirs[i + 1] = self.__prepare_1_fiber(i + 1, config)
def __prepare_1_fiber(
self, index: int, config: dict[str, Any], first=False
) -> tuple[list[tuple[list[tuple[str, Any]], dict[str, Any]]], list[Path]]:
required: list[tuple[list[tuple[str, Any]], dict[str, Any]]] = list(
variable_iterator(config, first)
)
for vary_list, _ in required:
vary_list.insert(
1, ("Fiber", "ABCDEFGHIJKLMNOPQRSTUVWXYZ"[index % 26] * (index // 26 + 1))
)
return required, [
utils.ensure_folder(
self.sim_dirs[index] / format_variable_list(vary_list),
mkdir=False,
prevent_overwrite=not self.overwrite,
)
for vary_list, c in required
]
this_path = utils.ensure_folder(
self.sim_dirs[i] / this_path, not self.overwrite, False
)
self.data_dirs[i].append(this_path)
this_conf.pop("variable")
this_conf.update({k: v for k, v in this_vary if k != "num"})
self.all_required[i].append((this_vary, this_conf))
def __iter__(self) -> Generator[tuple[list[tuple[str, Any]], Parameters], None, None]:
for sim_paths, fiber in zip(self.data_dirs, self.all_required):
@@ -912,6 +877,25 @@ class Configuration:
def __iter_1_sim(
self, sim_paths: list[Path], fiber: list[tuple[list[tuple[str, Any]], dict[str, Any]]]
) -> Generator[tuple[list[tuple[str, Any]], Path, Parameters], None, None]:
"""iterates through the parameters of only one fiber. It takes care of recovery partially completed
simulations, skipping complete ones and waiting for the previous fiber to finish
Parameters
----------
sim_paths : list[Path]
output_paths of the desired simulations
fiber : list[tuple[list[tuple[str, Any]], dict[str, Any]]]
list of variable list and config dict as yielded by variable_iterator
Yields
-------
list[tuple[str, Any]]
list of variable paramters
Path
desired output path
Parameters
computed Parameters obj
"""
sim_dict: dict[Path, tuple[list[tuple[str, Any]], dict[str, Any]]] = dict(
zip(sim_paths, fiber)
)
@@ -1009,6 +993,56 @@ class Configuration:
utils.save_toml(sim_dir / f"initial_config.toml", config)
class DataPather:
def __init__(self, dl: list[dict[str, Any]]):
self.dict_list = dl
self.n = len(self.dict_list)
self.final_list = list(self.dico_iterator(self.n))
def dico_iterator(self, index: int) -> Generator[list[list[tuple[str, Any]]], None, None]:
"""iterates through every possible combination of a list of dict of lists
Parameters
----------
index : int
up to where in the stored dict_list to go
Yields
-------
list[list[tuple[str, Any]]]
list of list of (key, value) pairs
Example
-------
self.dict_list = [{a:[56, 57], b:["?", "!"]}, {c:[0, -1]}] ->
[
[[(a, 56), (b, "?")], [(c, 0)]],
[[(a, 56), (b, "?")], [(c, 1)]],
[[(a, 56), (b, "!")], [(c, 0)]],
[[(a, 56), (b, "!")], [(c, 1)]],
[[(a, 57), (b, "?")], [(c, 0)]],
[[(a, 57), (b, "?")], [(c, 1)]],
[[(a, 57), (b, "!")], [(c, 0)]],
[[(a, 57), (b, "!")], [(c, 1)]],
]
"""
d_tem_list = [el for d in self.dict_list[: index + 1] for el in d.items()]
dict_pos = np.cumsum([0] + [len(d) for d in self.dict_list[: index + 1]])
ranges = [range(len(l)) for _, l in d_tem_list]
for r in itertools.product(*ranges):
flat = [(d_tem_list[i][0], d_tem_list[i][1][j]) for i, j in enumerate(r)]
out = [flat[left:right] for left, right in zip(dict_pos[:-1], dict_pos[1:])]
yield out
def all_vary_list(self, index):
for l in self.dico_iterator(index):
yield format_variable_list(reduce_all_variable(l[:index])), format_variable_list(
reduce_all_variable(l)
), l[index]
@dataclass
class PlotRange:
left: float = Parameter(type_checker(int, float))
@@ -1120,22 +1154,29 @@ def _mock_function(num_args: int, num_returns: int) -> Callable:
def format_variable_list(l: list[tuple[str, Any]]) -> str:
"""formats a variable list into a str such that each simulation has a unique
directory name. A u_XXX unique identifier and b_XXX (ignoring repeat simulations)
branch identifier are added at the beginning.
Parameters
----------
l : list[tuple[str, Any]]
list of variable parameters
Returns
-------
str
directory name
"""
str_list = []
previous_fibers = []
num = None
for p_name, p_value in l:
if p_name == "prev_data_dir":
prev_dir_items = Path(p_value).name.split()[2:]
prev_dir_dic = dict(zip(prev_dir_items[::2], prev_dir_items[1::2]))
num = prev_dir_dic.pop("num")
previous_fibers += sum(([k, v] for k, v in prev_dir_dic.items()), [])
elif p_name == "num" and num is None:
num = str(p_value)
else:
ps = p_name.replace("/", "").replace(PARAM_SEPARATOR, "")
vs = format_value(p_name, p_value).replace("/", "").replace(PARAM_SEPARATOR, "")
str_list.append(ps + PARAM_SEPARATOR + vs)
return PARAM_SEPARATOR.join(str_list[:1] + previous_fibers + str_list[1:] + ["num", num])
ps = p_name.replace("/", "").replace(PARAM_SEPARATOR, "")
vs = format_value(p_name, p_value).replace("/", "").replace(PARAM_SEPARATOR, "")
str_list.append(ps + PARAM_SEPARATOR + vs)
tmp_name = PARAM_SEPARATOR.join(str_list)
unique_id = "u_" + utils.to_62(hash(str(l)))
branch_id = "b_" + utils.to_62(hash(str([el for el in l if el[0] != "num"])))
return unique_id + PARAM_SEPARATOR + branch_id + PARAM_SEPARATOR + tmp_name
def format_value(name: str, value) -> str:
@@ -1227,7 +1268,8 @@ def variable_iterator(
param_dict.pop("variable")
param_dict.update(indiv_config)
for repeat_index in range(repeat):
variable_ind = [("id", master_index)] + variable_list
# variable_ind = [("id", master_index)] + variable_list
variable_ind = variable_list
if first:
variable_ind += [("num", repeat_index)]
yield variable_ind, param_dict