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

mostly working with this weird file handling system

Browse Source
This commit is contained in:
Benoît Sierro
2021-02-03 09:09:10 +01:00
parent 3c2e98d1e9
commit d2a85713b9
44 changed files with 674 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
README.md
View File

@@ -200,3 +200,17 @@ upper_wavelength_interp_limit: float
dispersion coefficients are computed over a certain wavelength range. This parameter
sets the lowest end of this range. If the set value is higher than the higher end of the
wavelength window, it is lowered down to that point. default : 1900e-9
## Environment parameters
path_prefixes : dict[str, str]
key : hostname (as returned by `socket.gethostname()`)
value : path to the head's current working directory
When running the simulations on multiple instances, the head's working directory needs to be mounted as a network drive on every other node, with its path specified with this parameter
Example:
[environment.path_prefixes]
Excellent_node = "Z:\\simulations\\"
this means that if I'm working on Average_node (i.e. Average_node is the head of the ray cluster) in `/Users/username/simulations/` and connecting Excellent_node (Windows) to the ray cluster, I need to be able to access Average_node's `simulations` directory by mounting it as a network drive. In this example, `username` is shared on the network by Average_node and Excellent_node is mounting it as a network share with the same credentials as Average_node's (to avoid permission problems). This means that `Z:\\simulations\` on Excellent_node points to the same directory as `/Users/username/simulations/` on Average_nodes. Jobs sent by the head's scgenerator module to Excellent_node will have an environment variable set so that Average_node's cwd so that files are all saved in the same place.

4
developement_help.md
View File

@@ -1,6 +1,6 @@
## add parameter
- add it to ```const.valid_param_types```
- add it to README.md
- add the necessary logic in the appropriate ```initialize.ensure_consistency``` subfunction
- add the necessary logic in the appropriate ```initialize._ensure_consistency``` subfunction
- optional : add a default value
- optional : add to valid varying
- optional : add to valid_variable

6
requirements.txt Normal file
View File

@@ -0,0 +1,6 @@
numpy
matplotlib
scipy
ray
send2trash
toml

31
scgenerator.log
View File

@@ -0,0 +1,31 @@
scgenerator.initialize: computed initial N = 8.66
scgenerator.initialize: computed initial N = 8.66
scgenerator.initialize: computed initial N = 8.66
scgenerator.initialize: computed initial N = 8.66
scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: scgenerator.initialize: computed initial N = 8.66
INFO: id 0 wavelength 8.35e-07 num 0: energy conserved
INFO: id 2 wavelength 8.35e-07 num 2: energy conserved
INFO: id 3 wavelength 8.35e-07 num 3: energy conserved
INFO: id 4 wavelength 8.35e-07 num 4: energy conserved
INFO: id 5 wavelength 8.35e-07 num 5: energy conserved
INFO: id 9 wavelength 8.35e-07 num 9: energy conserved
INFO: id 10 wavelength 8.3375e-07 num 0: energy conserved
INFO: id 14 wavelength 8.3375e-07 num 4: energy conserved
INFO: id 16 wavelength 8.3375e-07 num 6: energy conserved
INFO: id 19 wavelength 8.3375e-07 num 9: energy conserved
INFO: id 21 wavelength 8.325e-07 num 1: energy conserved

2
src/scgenerator/__init__.py
View File

@@ -1,5 +1,5 @@
from .initialize import compute_init_parameters
from .io import Paths, iter_load_sim_data, load_toml, load_sim_data
from .io import Paths, load_toml
from .math import abs2, argclosest, span
from .physics import fiber, materials, pulse, simulate, units
from .physics.simulate import RK4IP, new_simulations

16
src/scgenerator/const.py
View File

@@ -11,6 +11,11 @@ def integer(n):
return isinstance(n, int) and n > 0
def generic_dict(d):
"""must be a dictionary"""
return isinstance(d, dict)
def boolean(b):
"""must be a boolean"""
return type(b) == bool
@@ -146,6 +151,9 @@ valid_param_types = dict(
upper_wavelength_interp_limit=num,
frep=num,
),
environment=dict(
path_prefixes=generic_dict,
),
)
hc_model_specific_parameters = dict(
@@ -163,7 +171,7 @@ hc_model_specific_parameters = dict(
)
"""dependecy map only includes actual fiber parameters and exclude gas parameters"""
valid_varying = dict(
valid_variable = dict(
fiber=[
"beta",
"gamma",
@@ -190,8 +198,10 @@ valid_varying = dict(
"soliton_num",
],
simulation=["behaviors", "raman_type", "tolerated_error", "step_size", "ideal_gas"],
environment=[],
)
TMP_FOLDER_KEY_BASE = "SCGENERATOR_TMP"
ENVIRON_KEY_BASE = "SCGENERATOR_"
TMP_FOLDER_KEY_BASE = ENVIRON_KEY_BASE + "TMP_"
PREFIX_KEY_BASE = ENVIRON_KEY_BASE + "PREFIX_"
PARAM_SEPARATOR = " "

2
src/scgenerator/defaults.py
View File

@@ -82,7 +82,7 @@ def get(section_dict, param, **kwargs):
# whether the parameter is in the right place and valid is checked elsewhere,
# here, we just make sure it is present.
if param not in section_dict and param not in section_dict.get("varying", {}):
if param not in section_dict and param not in section_dict.get("variable", {}):
try:
section_dict[param] = default_parameters[param]
# LOG

85
src/scgenerator/initialize.py
View File

@@ -1,15 +1,17 @@
import os
from typing import Any, Iterator, List, Mapping, Tuple
from collections.abc import Mapping
from typing import Any, Iterator, List, Tuple
import numpy as np
from numpy import pi
from . import defaults, io, utils
from .math import length, power_fact
from .physics import fiber, pulse, units
from .const import valid_param_types, valid_varying, hc_model_specific_parameters
from .const import hc_model_specific_parameters, valid_param_types, valid_variable
from .errors import *
from .logger import get_logger
from .utils import varying_iterator, count_variations
from .math import length, power_fact
from .physics import fiber, pulse, units
from .utils import count_variations, variable_iterator
class ParamSequence(Mapping):
@@ -17,16 +19,32 @@ class ParamSequence(Mapping):
self.config = validate(config)
self.name = self.config["name"]
self.num_sim, self.num_varying = count_variations(self.config)
self.num_sim, self.num_variable = count_variations(self.config)
self.single_sim = self.num_sim == 1
def __iter__(self) -> Iterator[Tuple[list, dict]]:
def iterate_without_computing(self) -> Iterator[Tuple[List[Tuple[str, Any]], dict]]:
"""takes the output of `scgenerator.utils.variable_iterator` which is a new dict per different
parameter set and iterates through every single necessary simulation
Yields
-------
Iterator[Tuple[List[Tuple[str, Any]], dict]]
variable_ind : a list of (name, value) tuple of parameter name and value that are variable. The parameter
"num" (how many times this specific parameter set has been yielded already) and "id" (how many parameter sets
have been exhausted already) are added to the list to make sure every yielded list is unique.
"""
i = 0 # unique sim id
for variable_only, full_config in variable_iterator(self.config):
for j in range(self["simulation", "repeat"]):
variable_ind = [("id", i)] + variable_only + [("num", j)]
i += 1
yield variable_ind, full_config
def __iter__(self) -> Iterator[Tuple[List[Tuple[str, Any]], dict]]:
"""iterates through all possible parameters, yielding a config as welle as a flattened
computed parameters set each time"""
for varying_only, full_config in varying_iterator(self.config):
for i in range(self["simulation", "repeat"]):
varying = varying_only + [("num", i)]
yield varying, compute_init_parameters(full_config)
for variable_list, full_config in self.iterate_without_computing():
yield variable_list, compute_init_parameters(full_config)
def __len__(self):
return self.num_sim
@@ -47,24 +65,34 @@ class RecoveryParamSequence(ParamSequence):
self.num_sim -= 1
self.single_sim = self.num_sim == 1
def __iter__(self) -> Iterator[Tuple[list, dict]]:
for varying_only, full_config in varying_iterator(self.config):
for i in range(self["simulation", "repeat"]):
varying = varying_only + [("num", i)]
print("varying ", varying_only, i)
def __iter__(self) -> Iterator[Tuple[List[Tuple[str, Any]], dict]]:
for variable_list, full_config in self.iterate_without_computing():
sub_folder = os.path.join(
io.get_data_folder(self.id), utils.format_varying_list(varying)
io.get_data_folder(self.id), utils.format_variable_list(variable_list)
)
if not io.propagation_initiated(sub_folder):
yield varying, compute_init_parameters(full_config)
yield variable_list, compute_init_parameters(full_config)
elif not io.propagation_completed(sub_folder, self.config["simulation"]["z_num"]):
yield varying, recover_params(full_config, varying, self.id)
yield variable_list, recover_params(full_config, variable_list, self.id)
else:
continue
def validate(config: dict) -> dict:
"""validates a configuration dictionary and attempts to fill in defaults
Parameters
----------
config : dict
loaded configuration
Returns
-------
dict
updated configuration
"""
_validate_types(config)
return _ensure_consistency(config)
@@ -165,7 +193,7 @@ def _validate_types(config):
for domain, parameters in config.items():
if isinstance(parameters, dict):
for param_name, param_value in parameters.items():
if param_name == "varying":
if param_name == "variable":
for k_vary, v_vary in param_value.items():
if not isinstance(v_vary, list):
raise TypeError(f"Varying parameters should be specified in a list")
@@ -175,7 +203,7 @@ def _validate_types(config):
f"Varying parameters lists should contain at least 1 element"
)
if k_vary not in valid_varying[domain]:
if k_vary not in valid_variable[domain]:
raise TypeError(f"'{k_vary}' is not a valid variable parameter")
[
@@ -189,7 +217,7 @@ def _validate_types(config):
def _contains(sub_conf, param):
return param in sub_conf or param in sub_conf.get("varying", {})
return param in sub_conf or param in sub_conf.get("variable", {})
def _ensure_consistency_fiber(fiber):
@@ -330,8 +358,8 @@ def _ensure_consistency_simulation(simulation):
]:
simulation = defaults.get(simulation, param)
if "raman" in simulation["behaviors"] or any(
["raman" in l for l in simulation.get("varying", {}).get("behaviors", [])]
if "raman" in simulation.get("behaviors", {}) or any(
["raman" in l for l in simulation.get("variable", {}).get("behaviors", [])]
):
simulation = defaults.get(simulation, "raman_type", specified_parameters=["raman"])
return simulation
@@ -359,7 +387,7 @@ def _ensure_consistency(config):
for param_name in sub_dict:
for set_param in config.values():
if isinstance(set_param, dict):
if param_name in set_param and param_name in set_param.get("varying", {}):
if param_name in set_param and param_name in set_param.get("variable", {}):
raise DuplicateParameterError(
f"got multiple values for parameter '{param_name}'"
)
@@ -377,10 +405,9 @@ def _ensure_consistency(config):
return config
def recover_params(params: dict, varying_only: List[Tuple[str, Any]], task_id: int):
print("RECOVERING PARAMETERS")
def recover_params(params: dict, variable_only: List[Tuple[str, Any]], task_id: int):
params = compute_init_parameters(params)
vary_str = utils.format_varying_list(varying_only)
vary_str = utils.format_variable_list(variable_only)
path = os.path.join(io.get_data_folder(task_id), vary_str)
num, last_spectrum = io.load_last_spectrum(path)
params["spec_0"] = last_spectrum
@@ -396,7 +423,7 @@ def compute_init_parameters(config):
Parameters
----------
config : dict
a configuration dictionary containing the pulse, fiber and simulation sections with no varying parameter.
a configuration dictionary containing the pulse, fiber and simulation sections with no variable parameter.
a flattened parameters dictionary may be provided instead
Note : checking the validity of the configuration shall be done before calling this function.

224
src/scgenerator/io.py
View File

@@ -1,4 +1,3 @@
import json
import os
import shutil
from datetime import datetime
@@ -11,29 +10,11 @@ import toml
from send2trash import TrashPermissionError, send2trash
from . import utils
from .const import TMP_FOLDER_KEY_BASE, PARAM_SEPARATOR
from .const import PARAM_SEPARATOR, PREFIX_KEY_BASE, TMP_FOLDER_KEY_BASE, ENVIRON_KEY_BASE
from .errors import IncompleteDataFolderError
from .logger import get_logger
def load_toml(path: str):
"""returns a dictionary parsed from the specified toml file"""
if not path.lower().endswith(".toml"):
path += ".toml"
with open(path, mode="r") as file:
dico = toml.load(file)
return dico
def save_toml(path, dico):
"""saves a dictionary into a toml file"""
if not path.lower().endswith(".toml"):
path += ".toml"
with open(path, mode="w") as file:
toml.dump(dico, file)
return dico
class Paths:
home = os.path.expanduser("~")
_data_files = ["silica.toml", "gas.toml", "hr_t.npz"]
@@ -48,13 +29,15 @@ class Paths:
@classmethod
def get(cls, key):
if key not in cls.paths:
if os.path.exists("paths.json"):
with open("paths.json") as file:
paths_dico = json.load(file)
if os.path.exists("paths.toml"):
with open("paths.toml") as file:
paths_dico = toml.load(file)
for k, v in paths_dico.items():
cls.paths[k] = os.path.abspath(os.path.expanduser(v))
cls.paths[k] = v
if key not in cls.paths:
print(f"{key} was not found in path index, returning current working directory.")
get_logger(__name__).info(
f"{key} was not found in path index, returning current working directory."
)
cls.paths[key] = os.getcwd()
return cls.paths[key]
@@ -80,6 +63,48 @@ class Paths:
return os.path.join(cls.get("plots"), name)
def abspath(rel_path: str):
"""returns the complete path with the correct root. In other words, allows to modify absolute paths
in case the process accessing this function is a sub-process started from another device.
Parameters
----------
rel_path : str
relative path
Returns
-------
str
absolute path
"""
key = utils.formatted_hostname()
prefix = os.getenv(key)
if prefix is None:
p = os.path.abspath(rel_path)
else:
p = os.path.join(prefix, rel_path)
return os.path.normpath(p)
def load_toml(path: str):
"""returns a dictionary parsed from the specified toml file"""
if not path.lower().endswith(".toml"):
path += ".toml"
with open(abspath(path), mode="r") as file:
dico = toml.load(file)
return dico
def save_toml(path, dico):
"""saves a dictionary into a toml file"""
if not path.lower().endswith(".toml"):
path += ".toml"
with open(abspath(path), mode="w") as file:
toml.dump(dico, file)
return dico
def serializable(val):
"""returns True if val is serializable into a Json file"""
types = (np.ndarray, float, int, str, list, tuple)
@@ -132,6 +157,7 @@ def save_parameters(param_dict, file_name="param"):
folder_name, file_name = os.path.split(file_name)
folder_name = "tmp" if folder_name == "" else folder_name
file_name = os.path.splitext(file_name)[0]
folder_name = abspath(folder_name)
if not os.path.exists(folder_name):
os.makedirs(folder_name)
@@ -146,20 +172,22 @@ def save_parameters(param_dict, file_name="param"):
return os.path.join(folder_name, file_name)
def load_previous_parameters(path):
"""loads a parameters json files and converts data to appropriate type
def load_previous_parameters(path: str):
"""loads a parameters toml files and converts data to appropriate type
Parameters
----------
path : path-like
path to the json
path : str
path to the toml
Returns
----------
params : dict
dict
flattened parameters dictionary
"""
params = load_toml(path)
for k, v in params.items():
if isinstance(v, list):
if isinstance(v[0], (float, int)):
if isinstance(v, list) and isinstance(v[0], (float, int)):
params[k] = np.array(v)
return params
@@ -180,101 +208,32 @@ def load_material_dico(name):
return toml.loads(Paths.gets("gas"))[name]
def load_sim_data(folder_name, ind=None, load_param=True):
"""
loads the data already simulated.
defauft shape is (z_targets, n, nt)
def set_environ(config: dict):
"""sets environment variables specified in the config
Parameters
----------
folder_name : (string) folder where the simulation data is stored
ind : list of indices if only certain spectra are desired.
- If left to None, returns every spectrum
- If only 1 int, will cast the (1, n, nt) array into a (n, nt) array
load_param : (bool) return the parameter dictionary as well. returns None
if not available
Returns
----------
spectra : array
squeezed array of complex spectra (n simulation on a nt size grid at each ind)
Raises
----------
FileNotFoundError : folder does not exist or does not contain sufficient
data
config : dict
whole simulation config file
"""
print(f"opening {folder_name}")
# Check if file exists and assert how many z positions there are
if not os.path.exists(folder_name):
raise FileNotFoundError(f"Folder {folder_name} does not exist")
nmax = len(glob(os.path.join(folder_name, "spectra_*.npy")))
if nmax <= 0:
raise FileNotFoundError(f"No appropriate file in specified folder {folder_name}")
if ind is None:
ind = range(nmax)
elif isinstance(ind, int):
ind = [ind]
# Load the spectra
spectra = []
for i in ind:
spectra.append(load_single_spectrum(folder_name, i))
spectra = np.array(spectra)
# Load the parameters dictionary
try:
params = load_previous_parameters(os.path.join(folder_name, "params.toml"))
except FileNotFoundError:
print(f"parameters corresponding to {folder_name} not found")
params = None
print("data successfully loaded")
if load_param:
return spectra.squeeze(), params
else:
return spectra.squeeze()
environ = config.get("environment", {})
for k, v in environ.get("path_prefixes", {}).items():
os.environ[(PREFIX_KEY_BASE + k).upper()] = v
def get_all_environ() -> Dict[str, str]:
"""returns a dictionary of all environment variables set by any instance of scgenerator"""
return dict(filter(lambda el: el[0].startswith(TMP_FOLDER_KEY_BASE), os.environ.items()))
d = dict(filter(lambda el: el[0].startswith(ENVIRON_KEY_BASE), os.environ.items()))
print(d)
return d
def load_single_spectrum(folder, index) -> np.ndarray:
return np.load(os.path.join(folder, f"spectra_{index}.npy"))
def iter_load_sim_data(folder_name, with_params=False) -> Iterable[np.ndarray]:
"""
similar to load_sim_data but works as an iterator
"""
if not os.path.exists(folder_name):
raise FileNotFoundError(f"Folder {folder_name} does not exist")
nmax = len(glob(os.path.join(folder_name, "spectra_*.npy")))
if nmax <= 0:
raise FileNotFoundError(f"No appropriate file in specified folder {folder_name}")
params = {}
if with_params:
try:
params = load_previous_parameters(os.path.join(folder_name, "params.toml"))
except FileNotFoundError:
print(f"parameters corresponding to {folder_name} not found")
params = None
print(f"iterating through {folder_name}")
for i in range(nmax):
if with_params:
yield load_single_spectrum(folder_name, i), params
else:
yield load_single_spectrum(folder_name, i)
return np.load(os.path.join(abspath(folder), f"spectra_{index}.npy"))
def get_data_subfolders(path: str) -> List[str]:
"""returns a list of path/subfolders in the specified directory
"""returns a list of relative path/subfolders in the specified directory
Parameters
----------
@@ -314,7 +273,7 @@ def check_data_integrity(sub_folders: List[str], init_z_num: int):
def propagation_initiated(sub_folder) -> bool:
if os.path.isdir(sub_folder):
if os.path.isdir(abspath(sub_folder)):
return find_last_spectrum_file(sub_folder) > 0
return False
@@ -388,6 +347,7 @@ def merge_same_simulations(path: str):
pt = utils.ProgressTracker(num_operations, logger=logger, prefix="merging data : ")
for base_folder in base_folders:
logger.debug(f"creating new folder {base_folder}")
for j in range(z_num):
spectra = []
for i in range(repeat):
@@ -396,7 +356,7 @@ def merge_same_simulations(path: str):
)
dest_folder = ensure_folder(base_folder, prevent_overwrite=False)
spectra = np.array(spectra).reshape(repeat, len(spectra[0]))
np.save(os.path.join(dest_folder, f"spectra_{j}.npy"), spectra)
np.save(os.path.join(dest_folder, f"spectra_{j}.npy"), spectra.squeeze())
pt.update()
for file_name in ["z.npy", "params.toml"]:
shutil.copy(
@@ -417,7 +377,6 @@ def get_data_folder(task_id: int, name_if_new: str = ""):
tmp = os.getenv(TMP_FOLDER_KEY_BASE + idstr)
if tmp is None:
tmp = ensure_folder("scgenerator_" + name_if_new + idstr)
tmp = os.path.abspath(tmp)
os.environ[TMP_FOLDER_KEY_BASE + idstr] = tmp
return tmp
@@ -453,19 +412,23 @@ def generate_file_path(file_name: str, task_id: int, identifier: str = "") -> st
str
the full path
"""
base_name, ext = os.path.splitext(file_name)
folder = get_data_folder(task_id)
folder = os.path.join(folder, identifier)
folder = ensure_folder(folder, prevent_overwrite=False)
i = 0
base_name = os.path.join(folder, base_name)
new_name = base_name + ext
while os.path.exists(new_name):
print(f"{i=}")
new_name = f"{base_name}_{i}{ext}"
i += 1
# base_name, ext = os.path.splitext(file_name)
# folder = get_data_folder(task_id)
# folder = os.path.join(folder, identifier)
# folder = ensure_folder(folder, prevent_overwrite=False)
# i = 0
# base_name = os.path.join(folder, base_name)
# new_name = base_name + ext
# while os.path.exists(new_name):
# new_name = f"{base_name}_{i}{ext}"
# i += 1
return new_name
path = os.path.join(get_data_folder(task_id), identifier)
path = abspath(path)
os.makedirs(path, exist_ok=True)
path = os.path.join(path, file_name)
return path
def save_data(data: np.ndarray, file_name: str, task_id: int, identifier: str = ""):
@@ -482,14 +445,17 @@ def save_data(data: np.ndarray, file_name: str, task_id: int, identifier: str =
identifier : str, optional
identifier in the main data folder of the task, by default ""
"""
path = generate_file_path(file_name, task_id, identifier)
np.save(path, data)
get_logger(__name__).debug(f"saved data in {path}")
return
def ensure_folder(name, i=0, suffix="", prevent_overwrite=True):
"""creates a folder for simulation data named name and prevents overwrite
by adding a suffix if necessary and returning the name"""
prefix, last_dir = os.path.split(os.path.abspath(name))
prefix, last_dir = os.path.split(abspath(name))
exploded = [prefix]
sub_prefix = prefix
while sub_prefix != os.path.abspath("/"):

34
src/scgenerator/logger.py
View File

@@ -1,5 +1,22 @@
import logging
DEFAULT_LEVEL = logging.INFO
lvl_map = dict(
debug=logging.DEBUG,
info=logging.INFO,
warning=logging.WARNING,
error=logging.ERROR,
fatal=logging.FATAL,
critical=logging.CRITICAL,
)
loggers = []
def _set_debug():
DEFAULT_LEVEL = logging.DEBUG
def get_logger(name=None):
"""returns a logging.Logger instance. This function is there because if scgenerator
@@ -18,9 +35,20 @@ def get_logger(name=None):
"""
name = __name__ if name is None else name
logger = logging.getLogger(name)
if name not in loggers:
loggers.append(logger)
return configure_logger(logger)
# def set_level_all(lvl):
# _default_lvl =
# logging.basicConfig(level=lvl_map[lvl])
# for logger in loggers:
# logger.setLevel(lvl_map[lvl])
# for handler in logger.handlers:
# handler.setLevel(lvl_map[lvl])
def configure_logger(logger, logfile="scgenerator.log"):
"""configures a logging.Logger obj
@@ -39,12 +67,14 @@ def configure_logger(logger, logfile="scgenerator.log"):
if not hasattr(logger, "already_configured"):
if logfile is not None:
file_handler = logging.FileHandler("scgenerator.log", "a+")
file_handler.setFormatter(logging.Formatter("{name}: {message}", style="{"))
file_handler.setFormatter(
logging.Formatter("{levelname}: {name}: {message}", style="{")
)
logger.addHandler(file_handler)
stream_handler = logging.StreamHandler()
logger.addHandler(stream_handler)
logger.setLevel(logging.INFO)
logger.setLevel(DEFAULT_LEVEL)
logger.already_configured = True
return logger

12
src/scgenerator/physics/fiber.py
View File

@@ -272,7 +272,7 @@ def A_eff_hasan(core_radius, capillary_num, capillary_spacing):
def HCPCF_find_with_given_ZDW(
varying,
variable,
target,
search_range,
material_dico,
@@ -286,13 +286,13 @@ def HCPCF_find_with_given_ZDW(
Parameters
----------
varying : str {"pressure", "temperature"}
variable : str {"pressure", "temperature"}
which parameter to vary
target : float
the ZDW target, in m
search_range : array, shape (2,)
(min, max) of the search range
other parameters : see HCPCF_dispersion. Pressure or temperature is used as initial value if it is varying
other parameters : see HCPCF_dispersion. Pressure or temperature is used as initial value if it is variable
Returns
-------
@@ -304,7 +304,7 @@ def HCPCF_find_with_given_ZDW(
#
fixed = [material_dico, model, model_params, ideal]
if varying == "pressure":
if variable == "pressure":
fixed.append(temperature)
x0 = 1e5 if pressure is None else pressure
@@ -321,7 +321,7 @@ def HCPCF_find_with_given_ZDW(
out = current_ZDW - target
return out
elif varying == "temperature":
elif variable == "temperature":
fixed.append(pressure)
x0 = 273.15 if temperature is None else temperature
@@ -339,7 +339,7 @@ def HCPCF_find_with_given_ZDW(
return out
else:
raise AttributeError(f"'varying' arg must be 'pressure' or 'temperature', not {varying}")
raise AttributeError(f"'variable' arg must be 'pressure' or 'temperature', not {variable}")
optimized = optimize.root_scalar(
zdw, x0=x0, args=tuple(fixed), method="brentq", bracket=search_range

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

@@ -395,22 +395,22 @@ class Simulations:
self.logger.info(f"Finished simulations from config {self.name} !")
def _run_available(self):
for varying, params in self.param_seq:
for variable, params in self.param_seq:
io.save_parameters(
params,
io.generate_file_path("params.toml", self.id, utils.format_varying_list(varying)),
io.generate_file_path("params.toml", self.id, utils.format_variable_list(variable)),
)
self.new_sim(varying, params)
self.new_sim(variable, params)
self.finish()
def new_sim(self, varying_list: List[tuple], params: dict):
def new_sim(self, variable_list: List[tuple], params: dict):
"""responsible to launch a new simulation
Parameters
----------
varying_list : list[tuple]
variable_list : list[tuple]
list of tuples (name, value) where name is the name of a
varying parameter and value is its current value
variable parameter and value is its current value
params : dict
a flattened parameter dictionary, as returned by scgenerator.initialize.compute_init_parameters
"""
@@ -434,8 +434,8 @@ class Simulations:
class SequencialSimulations(Simulations, available=True, priority=0):
def new_sim(self, varying_list: List[tuple], params: dict):
v_list_str = utils.format_varying_list(varying_list)
def new_sim(self, variable_list: List[tuple], params: dict):
v_list_str = utils.format_variable_list(variable_list)
self.logger.info(f"launching simulation with {v_list_str}")
self.propagator(
params,
@@ -481,11 +481,11 @@ class RaySimulations(Simulations, available=using_ray, priority=1):
self.jobs = []
self.actors = {}
def new_sim(self, varying_list: List[tuple], params: dict):
def new_sim(self, variable_list: List[tuple], params: dict):
while len(self.jobs) >= self.sim_jobs_total:
self._collect_1_job()
v_list_str = utils.format_varying_list(varying_list)
v_list_str = utils.format_variable_list(variable_list)
new_actor = self.propagator.remote(
params, save_data=True, job_identifier=v_list_str, task_id=self.id
@@ -523,7 +523,7 @@ class RaySimulations(Simulations, available=using_ray, priority=1):
@property
def sim_jobs_total(self):
tot_cpus = sum([node.get("Resources", {}).get("CPU", 0) for node in ray.nodes()])
return min(self.param_seq.num_sim, tot_cpus)
return int(min(self.param_seq.num_sim, tot_cpus))
def new_simulations(
@@ -531,6 +531,7 @@ def new_simulations(
):
config = io.load_toml(config_file)
io.set_environ(config)
param_seq = initialize.ParamSequence(config)
return _new_simulations(param_seq, task_id, data_folder, Method)

103
src/scgenerator/spectra.py Normal file
View File

@@ -0,0 +1,103 @@
import os
from collections.abc import Mapping, Sequence
from glob import glob
from typing import Any, List, Tuple
import numpy as np
from . import io
from .logger import get_logger
class Spectra(Sequence):
def __init__(self, path: str):
self.logger = get_logger(__name__)
self.path = path
if not os.path.isdir(self.path):
raise FileNotFoundError(f"Folder {self.path} does not exist")
self.params = None
try:
self.params = io.load_previous_parameters(os.path.join(self.path, "params.toml"))
except FileNotFoundError:
self.logger.info(f"parameters corresponding to {self.path} not found")
try:
self.z = np.load(os.path.join(path, "z.npy"))
except FileNotFoundError:
if self.params is not None:
self.z = self.params["z_targets"]
else:
raise
self.nmax = len(glob(os.path.join(self.path, "spectra_*.npy")))
if self.nmax <= 0:
raise FileNotFoundError(f"No appropriate file in specified folder {self.path}")
def __iter__(self):
"""
similar to all_spectra but works as an iterator
"""
self.logger.debug(f"iterating through {self.path}")
for i in range(self.nmax):
yield io.load_single_spectrum(self.path, i)
def __len__(self):
return self.nmax
def __getitem__(self, key):
return self.all_spectra(ind=range(self.nmax)[key])
def all_spectra(self, ind=None):
"""
loads the data already simulated.
defauft shape is (z_targets, n, nt)
Parameters
----------
ind : int or list of int
if only certain spectra are desired.
- If left to None, returns every spectrum
- If only 1 int, will cast the (1, n, nt) array into a (n, nt) array
Returns
----------
spectra : array
squeezed array of complex spectra (n simulation on a nt size grid at each ind)
"""
self.logger.debug(f"opening {self.path}")
# Check if file exists and assert how many z positions there are
if ind is None:
ind = range(self.nmax)
elif isinstance(ind, int):
ind = [ind]
# Load the spectra
spectra = []
for i in ind:
spectra.append(io.load_single_spectrum(self.path, i))
spectra = np.array(spectra)
self.logger.debug(f"all spectra from {self.path} successfully loaded")
return spectra.squeeze()
class SpectraCollection(Mapping, Sequence):
def __init__(self, path: str):
self.path = path
self.collection: List[Spectra] = []
if not os.path.isdir(self.path):
raise FileNotFoundError(f"Folder {self.path} does not exist")
self.variable_list
def __getitem__(self, key):
return self.collection[key]
def __len__(self):
pass

123
src/scgenerator/utils.py
View File

@@ -8,13 +8,14 @@ scgenerator module but some function may be used in any python program
import datetime as dt
import itertools
import logging
import socket
from typing import Any, Callable, List, Tuple, Union
import numpy as np
import ray
from copy import deepcopy
from .const import PARAM_SEPARATOR, valid_varying
from .const import PARAM_SEPARATOR, PREFIX_KEY_BASE, valid_variable
from .logger import get_logger
from .math import *
@@ -82,38 +83,22 @@ class ProgressTracker:
return "{}/{}".format(self.current, self.max)
# def ray_safe(func, *args, **kwargs):
# """evaluates functions that return None whether they are Ray workers or normal functions
# Parameters
# ----------
# func : the function or Worker id
# args : arguments to give to the functions
# Returns
# ----------
# nothing
# """
# if hasattr(func, "remote"):
# ray.get(func.remote(*args, **kwargs))
# else:
# func(*args, **kwargs)
def count_variations(config: dict) -> Tuple[int, int]:
"""returns (sim_num, varying_params_num) where sim_num is the total number of simulations required and
varying_params_num is the number of distinct parameters that will vary."""
"""returns (sim_num, variable_params_num) where sim_num is the total number of simulations required and
variable_params_num is the number of distinct parameters that will vary."""
sim_num = 1
varying_params_num = 0
variable_params_num = 0
for section_name in valid_varying:
for array in config.get(section_name, {}).get("varying", {}).values():
for section_name in valid_variable:
for array in config.get(section_name, {}).get("variable", {}).values():
sim_num *= len(array)
varying_params_num += 1
variable_params_num += 1
sim_num *= config["simulation"].get("repeat", 1)
return sim_num, varying_params_num
return sim_num, variable_params_num
def format_varying_list(l: List[tuple]):
def format_variable_list(l: List[tuple]):
joints = 2 * PARAM_SEPARATOR
str_list = []
for p_name, p_value in l:
@@ -123,7 +108,18 @@ def format_varying_list(l: List[tuple]):
return joints[0].join(str_list)
# def varying_list_from_path(s: str) -> List[tuple]:
def format_value(value):
if type(value) == type(False):
return str(value)
elif isinstance(value, (float, int)):
return format(value, ".9g")
elif isinstance(value, (list, tuple, np.ndarray)):
return "-".join([format_value(v) for v in value])
else:
return str(value)
# def variable_list_from_path(s: str) -> List[tuple]:
# s = s.replace("/", "")
# str_list = s.split(PARAM_SEPARATOR)
# out = []
@@ -132,69 +128,59 @@ def format_varying_list(l: List[tuple]):
# return out
def format_value(value):
if type(value) == type(False):
return str(value)
elif isinstance(value, (float, int)):
return format(value, ".5g")
elif isinstance(value, (list, tuple, np.ndarray)):
return "-".join([format_value(v) for v in value])
else:
return str(value)
# def get_value(s: str):
# if s.lower() == "true":
# return True
# if s.lower() == "false":
# return False
# try:
# return int(s)
# except ValueError:
# pass
# try:
# return float(s)
# except ValueError:
# pass
# return s
def get_value(s: str):
if s.lower() == "true":
return True
if s.lower() == "false":
return False
try:
return int(s)
except ValueError:
pass
try:
return float(s)
except ValueError:
pass
return s
def varying_iterator(config):
def variable_iterator(config):
out = deepcopy(config)
varying_dict = {
section_name: out.get(section_name, {}).pop("varying", {}) for section_name in valid_varying
variable_dict = {
section_name: out.get(section_name, {}).pop("variable", {})
for section_name in valid_variable
}
possible_keys = []
possible_ranges = []
for section_name, section in varying_dict.items():
for section_name, section in variable_dict.items():
for key in section:
arr = varying_dict[section_name][key]
arr = variable_dict[section_name][key]
possible_keys.append((section_name, key))
possible_ranges.append(range(len(arr)))
combinations = itertools.product(*possible_ranges)
for combination in combinations:
only_varying = []
only_variable = []
for i, key in enumerate(possible_keys):
parameter_value = varying_dict[key[0]][key[1]][combination[i]]
parameter_value = variable_dict[key[0]][key[1]][combination[i]]
out[key[0]][key[1]] = parameter_value
only_varying.append((key[1], parameter_value))
yield only_varying, out
only_variable.append((key[1], parameter_value))
yield only_variable, out
def parallelize(func, arg_iter, sim_jobs=4, progress_tracker_kwargs=None, const_kwarg={}):
"""given a function and an iterator of arguments, runs the function in parallel
"""given a function and an iterable of arguments, runs the function in parallel
Parameters
----------
func : a function
arg_iter : an iterator that yields a tuple to be unpacked to the function as argument(s)
sim_jobs : number of simultaneous runs
arg_iter : an iterable that yields a tuple to be unpacked to the function as argument(s)
sim_jobs : number of parallel runs
progress_tracker_kwargs : key word arguments to be passed to the ProgressTracker
const_kwarg : keyword arguments to be passed to the function on every run
@@ -235,3 +221,8 @@ def parallelize(func, arg_iter, sim_jobs=4, progress_tracker_kwargs=None, const_
ray.get(pt.update.remote())
return np.array(results)
def formatted_hostname():
s = socket.gethostname().replace(".", "_")
return (PREFIX_KEY_BASE + s).upper()

25
testing/configs/count_variations/120sim_3vary.toml Normal file
View File

@@ -0,0 +1,25 @@
[fiber]
core_radius = 50e-6
length = 50e-2
model = "marcatili"
[gas.variable]
gas_name = ["air", "helium"]
[pulse]
power = 100e3
wavelength = 800e-9
[pulse.variable]
width = [250e-15, 240e-15, 230e-15, 220e-15, 210e-15]
[simulation]
parallel = true
repeat = 4
t_num = 16384
time_window = 37e-12
tolerated_error = 1e-11
z_num = 128
[simulation.variable]
behaviors = [["spm", "raman", "ss"], ["spm", "raman"], ["spm"]]

2
testing/configs/count_variations/1sim_1vary.toml
View File

@@ -10,7 +10,7 @@ gas_name = "air"
power = 100e3
wavelength = 800e-9
[pulse.varying]
[pulse.variable]
width = [250e-15]
[simulation]

2
testing/configs/count_variations/2sim_1vary.toml
View File

@@ -11,7 +11,7 @@ soliton_num = 5
wavelength = 800e-9
width = 250e-15
[pulse.varying]
[pulse.variable]
shape = ["gaussian", "sech"]
[simulation]

2
testing/configs/ensure_consistency/bad1.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
[simulation]

2
testing/configs/ensure_consistency/bad2.toml
View File

@@ -14,7 +14,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
soliton_num = [1, 2, 3, 4]

2
testing/configs/ensure_consistency/bad3.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/bad4.toml
View File

@@ -16,7 +16,7 @@ shape = "gaussian"
wavelength = 1050e-9
width = 120e-15
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/bad5.toml
View File

@@ -17,7 +17,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/bad6.toml
View File

@@ -16,7 +16,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/good1.toml
View File

@@ -14,7 +14,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/good2.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/good3.toml
View File

@@ -13,7 +13,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

4
testing/configs/ensure_consistency/good4.toml
View File

@@ -14,7 +14,7 @@ model = "hasan"
[gas]
gas_name = "helium"
[gas.varying]
[gas.variable]
temperature = [300, 350, 400]
[pulse]
@@ -23,7 +23,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/good5.toml
View File

@@ -14,7 +14,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/ensure_consistency/good6.toml
View File

@@ -16,7 +16,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

29
testing/configs/param_sequence/almost_equal.toml Normal file
View File

@@ -0,0 +1,29 @@
# raman_type should be added
name = "test config"
[fiber]
gamma = 0.018
length = 1
model = "pcf"
pitch = 1.5e-6
pitch_ratio = 0.37
[pulse]
power = 100e3
quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
width = 50e-15
[pulse.variable]
intensity_noise = [0.10000000005e-2, 0.1e-2]
[simulation]
behaviors = ["spm", "raman", "ss"]
parallel = true
repeat = 4
t_num = 16384
time_window = 37e-12
tolerated_error = 1e-11
z_num = 128

29
testing/configs/param_sequence/equal.toml Normal file
View File

@@ -0,0 +1,29 @@
# raman_type should be added
name = "test config"
[fiber]
gamma = 0.018
length = 1
model = "pcf"
pitch = 1.5e-6
pitch_ratio = 0.37
[pulse]
power = 100e3
quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
width = 50e-15
[pulse.variable]
intensity_noise = [0.1e-2, 0.001]
[simulation]
behaviors = ["spm", "raman", "ss"]
parallel = true
repeat = 4
t_num = 16384
time_window = 37e-12
tolerated_error = 1e-11
z_num = 128

27
testing/configs/param_sequence/no_variations.toml Normal file
View File

@@ -0,0 +1,27 @@
# raman_type should be added
name = "test config"
[fiber]
gamma = 0.018
length = 1
model = "pcf"
pitch = 1.5e-6
pitch_ratio = 0.37
[pulse]
intensity_noise = 0.1e-2
power = 100e3
quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
width = 50e-15
[simulation]
behaviors = ["spm", "raman", "ss"]
parallel = true
repeat = 4
t_num = 16384
time_window = 37e-12
tolerated_error = 1e-11
z_num = 128

2
testing/configs/run_simulations/full_anomalous.toml
View File

@@ -19,7 +19,7 @@ length = 0.02
power = 10000
t0 = 2.84e-14
[pulse.varying]
[pulse.variable]
wavelength = [835e-9, 830e-9]
[simulation]

2
testing/configs/validate_types/bad1.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/validate_types/bad2.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/validate_types/bad3.toml
View File

@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = ["gaussian", "sech"]

6
testing/configs/validate_types/bad4.toml
View File

@@ -1,4 +1,4 @@
# parallel should not be varying
# parallel should not be variable
name = "test config"
@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]
@@ -28,5 +28,5 @@ time_window = 37e-12
tolerated_error = 1e-11
z_num = 1
[simulation.varying]
[simulation.variable]
parallel = [true, false]

4
testing/configs/validate_types/bad5.toml
View File

@@ -1,4 +1,4 @@
#varying parameters should be lists
#variable parameters should be lists
name = "test config"
@@ -15,7 +15,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = 0.05e-2
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/validate_types/bad6.toml
View File

@@ -16,7 +16,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
width = [50e-15, 100e-15, 200e-15]
[simulation]

2
testing/configs/validate_types/bad7.toml
View File

@@ -13,7 +13,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = []
width = [50e-15, 100e-15, 200e-15]

2
testing/configs/validate_types/good.toml
View File

@@ -13,7 +13,7 @@ quantum_noise = true
shape = "gaussian"
wavelength = 1050e-9
[pulse.varying]
[pulse.variable]
intensity_noise = [0.05e-2, 0.1e-2]
width = [50e-15, 100e-15, 200e-15]

36
testing/long_tests/test_recovery_param_seq.py Normal file
View File

@@ -0,0 +1,36 @@
import shutil
import unittest
import toml
from scgenerator import initialize, io, logger
from send2trash import send2trash
TMP = "testing/.tmp"
class TestRecoveryParamSequence(unittest.TestCase):
def setUp(self):
shutil.copytree("/Users/benoitsierro/sc_tests/scgenerator_full anomalous55", TMP)
self.conf = toml.load(TMP + "/initial_config.toml")
logger.DEFAULT_LEVEL = logger.logging.FATAL
io.set_data_folder(55, TMP)
def test_remaining_simulations_count(self):
param_seq = initialize.RecoveryParamSequence(self.conf, 55)
self.assertEqual(5, len(param_seq))
def test_only_one_to_complete(self):
param_seq = initialize.RecoveryParamSequence(self.conf, 55)
i = 0
for expected, (vary_list, params) in zip([True, False, False, False, False], param_seq):
i += 1
self.assertEqual(expected, "recovery_last_stored" in params)
self.assertEqual(5, i)
def tearDown(self):
send2trash(TMP)
if __name__ == "__main__":
unittest.main()

36
testing/test_initialize.py
View File

@@ -1,8 +1,10 @@
import unittest
import toml
from copy import deepcopy
import scgenerator.initialize as init
import toml
from scgenerator import utils
from scgenerator.errors import *
from prettyprinter import pprint
def load_conf(name):
@@ -18,6 +20,36 @@ def conf_maker(folder):
return conf
class TestParamSequence(unittest.TestCase):
def iterconf(self, files):
conf = conf_maker("param_sequence")
for path in files:
yield init.ParamSequence(conf(path))
def test_no_repeat_in_sub_folder_names(self):
for param_seq in self.iterconf(["almost_equal", "equal", "no_variations"]):
l = []
s = []
for vary_list, _ in param_seq.iterate_without_computing():
self.assertNotIn(vary_list, l)
self.assertNotIn(utils.format_variable_list(vary_list), s)
l.append(vary_list)
s.append(utils.format_variable_list(vary_list))
def test_init_config_not_affected_by_iteration(self):
for param_seq in self.iterconf(["almost_equal", "equal", "no_variations"]):
config = deepcopy(param_seq.config)
for _ in param_seq.iterate_without_computing():
self.assertEqual(config.items(), param_seq.config.items())
def test_no_variations_yields_only_num_and_id(self):
for param_seq in self.iterconf(["no_variations"]):
for vary_list, _ in param_seq.iterate_without_computing():
self.assertEqual(vary_list[1][0], "num")
self.assertEqual(vary_list[0][0], "id")
self.assertEqual(2, len(vary_list))
class TestInitializeMethods(unittest.TestCase):
def test_validate_types(self):
conf = lambda s: load_conf("validate_types/" + s)

35
testing/test_utils.py
View File

@@ -1,6 +1,8 @@
import unittest
from scgenerator import utils, initialize
import numpy as np
import toml
from scgenerator import initialize, utils
def load_conf(name):
@@ -20,10 +22,33 @@ class TestUtilsMethods(unittest.TestCase):
def test_count_variations(self):
conf = conf_maker("count_variations")
self.assertEqual((1, 0), utils.count_variations(conf("1sim_0vary")))
self.assertEqual((1, 1), utils.count_variations(conf("1sim_1vary")))
self.assertEqual((2, 1), utils.count_variations(conf("2sim_1vary")))
self.assertEqual((2, 0), utils.count_variations(conf("2sim_0vary")))
for sim, vary in [(1, 0), (1, 1), (2, 1), (2, 0), (120, 3)]:
self.assertEqual((sim, vary), utils.count_variations(conf(f"{sim}sim_{vary}vary")))
def test_format_value(self):
values = [
122e-6,
True,
["raman", "ss"],
np.arange(5),
1.123,
1.1230001,
0.002e122,
12.3456e-9,
]
s = [
"0.000122",
"True",
"raman-ss",
"0-1-2-3-4",
"1.123",
"1.1230001",
"2e+119",
"1.23456e-08",
]
for value, target in zip(values, s):
self.assertEqual(target, utils.format_value(value))
if __name__ == "__main__":