11.08., 9:00 - 11:00: Due to updates GitLab will be unavailable for some minutes between 09:00 and 11:00.

Commit 974b435a authored by Daniel Lehmberg's avatar Daniel Lehmberg

Squashed 'Tools/SUQController/' changes from f75af9422..2bcc6f3ee

2bcc6f3ee Merge branch 'omnet_add_on_final' into 'master'
ca96ef284 Merge branch 'master' into omnet_add_on_final
34d78229e minor changes
42f00a252 minor changes (pyDOE not global dependency)
980041184 minor changes
2ef30133c corrected vadere server name
5666a686b add vadere server id
db95e7a4e moved file to repo Forschung
1781dbeb3 social dist calibration
99a919b35 refactored
83d944bd7 moved util function into rover class CoupledDictVariation
66bac6af3 read data for succesfull sim runs
3ff758a1a refactored
45d5b59d0 use vadere default seed instead of seed "42"
17dd13cfe get enviromenment_manager info
933181b19 create subset of simulations
108747469 set seeds with omnet ini file
b1c5bee19 only copy folder structure and xml files
ef848d982 use nlevels
08df62298 added some explanations
c3b4d6bc3 Merge branch 'onnet_add_on3' into omnet_add_on_final
639822bc8 Auto stash before merge of "omnet_add_on_final" and "origin/omnet_add_on_final"
c044e1bc3 Apply suggestion to suqc/parameter/sampling.py
733769944 further refactoring
51123ef2d applied some suggestions
6eecb3e8c do not copy cache results
6d70e73a4 Apply suggestion to suqc/environment.py
7629d24b6 Merge branch 'omnet_add_on_final0' into omnet_add_on_final
c2468db5c update
dd8afe40c Update README.md
b6c68b8f2 Update README.md
c852a6d74 Update README.md
1fae57c3d Update README.md
d4d1944c4 Update README.md
4605c5942 Apply suggestion to requirements.txt
ce488ff72 remove output after each simulation run in rover
c4d3c688f implemented AbstractScenarioCreation
8c7f67c1d abstract 2
f14585306 VaderQoI s parent class is Qoi
a7b990113 use scenario file defined in config
a6b02378f moved classed into sampling.py
41776cf94 set runscript to class attribute
4b54ba0fb give qoi to external python script
5039ddf2d added GenerealOutputFile
6a1c75540 use general QoI
34d88a1fc removed waiting
1176810db format code
664225e14 allow multiple runs
101905a4b wrong value
fceb0dbbc final setup of parameter studies
d46f84f2c removed figures
95d0bede4 removed seed
83fd95dca added shadowing study
94300edbd fixed bugs and refactored
b488aae21 extended sampling methods
85086f7ab before introducing new sampling feature
c68f3abb7 repition test
0516249ab some parameterstudies to compare with LTE
872dc90ce use different numbr of repititions
98b8d22e8 refactored for rover sim
015c13f10 adapted postprocessing for Python 3.8
f6da2f347 fixed bug
af30aa71b prepared for FP
55cd34485 added postprocessing
12012a827 added postprocessing
6e9565929 added utility for vector parameters
419bd6d74 added dependent parameters
ff94bcd59 read qoi
606324287 changed scenario
5951e9187 no parralel runs
e066af5e9 produce error in deep_look_up
8f9822a69 extended framework with abstract consolewrapper
97c6bc77d moved methods in abstract class
5f50ddecc sorted scripts
f588d5ed1 also parse inifiles
4b34227b4 create rover structure in output folder
577133bac minor: add TODO
d5114d6a7 Merge branch 'delete_me' into test
c4d7b7cbf added classes and tried to access data
216dc10b6 Merge branch 'CM' into 'master'
9fa54dc33 added forgotten if query for reading data
824d6f9cb adapt read from existing output
791e3e13b apply black and isort
9ff248e3d Merge branch 'master' into omnet_add_on (apply black isort)
ea9e16349 apply black and isort
64d7b4afc Merge remote-tracking branch 'origin/omnet_add_on' into omnet_add_on
1623c7f3e created sampling with units
c4f62c728 add ConfigParser for OMNeT++ ini files.
b9b446001 Merge remote-tracking branch 'origin/master'
f62bf74b3 scenario runs for FolderExistScenarios

git-subtree-dir: Tools/SUQController
git-subtree-split: 2bcc6f3eeb520df2c455a112016b86fda95cc607
parent 7a9937ed
......@@ -45,3 +45,27 @@ not installed successfully.
### Introduction
See [SRC_PATH]/tutorial
#### Using SUQC and Vadere
Here a few hints for your .scenario file for Vadere:
1. ScenarioChecker
Before running your scenario automatically on suqc, activate the ``ScenarioChecker`` (Project > Activate ScenarioChecker) and run it in the ``VadereGui``.
The ScenarioChecker will point out potential problems with your scenario file.
2. Default parameters
Make sure to set ``realTimeSimTimeRatio`` to 0.0. (Values > 0.0 slow down the simulation for the visualisation)
Another point that may cost a lot of computation time is the ``optimizationType``, consider using ``DISCRETE`` (discrete optimization) instead of ``NELDER_MEAD``. Please note, that ``varyStepDirection`` should always be activated with discrete optimization.
Remove ``attributesCar`` from the .scenario file if you are not using any vehicles to avoid confusion of attributes.
3. Visual check
Visually check the results of your simulation, maybe check upper and lower parameter bounds.
4. Clean topography
Remove elements in your topography that are not used. Sometimes through the interaction with the mouse, tiny obstacles or targets are created unintentionally.
Check the elements in your topography, you can take a look at the ``ElementTree`` in the Topography creator tab. Remove all elements that are unused, especially focusing on targets.
5. Data processors
Remove all data processors and output files that you don't use. In particular, remove the overlap processors, they are intended for testing purposes.
6. Reproducibility
Make sure that your runs are reproducible - work with a ``fixedSeed`` by activating ``useFixedSeed`` or save all the ``simulationSeed``s that have been used.
(Another way is to provide a ``fixedSeed`` for each runs with suqc, in this case make sure that ``useFixedSeed`` is true.)
setuptools>=39.1.0
fabric==2.4.0
pandas>=0.24.0,<1.0.0
pandas>=1.0.3
numpy>=1.17.0
scikit-learn>=0.21.0
......@@ -2,15 +2,15 @@
import os
from setuptools import setup, find_packages
from setuptools import find_packages, setup
from suqc.configuration import SuqcConfig
from suqc import __version__
from suqc.configuration import SuqcConfig
# To generate a new requirements.txt file run in console (install vis 'pip3 install pipreqs'):
# pipreqs --use-local --force /home/daniel/REPOS/suq-controller
with open('requirements.txt', "r") as f:
with open("requirements.txt", "r") as f:
requirements = f.read().splitlines()
# Writes a file that gives information about the version such that "suqc.__version__" provides the current version,
......@@ -27,7 +27,7 @@ setup(
url="www.vadere.org",
packages=find_packages(),
install_requires=requirements,
data_files=[('suqc', ["suqc/PACKAGE.txt"])]
data_files=[("suqc", ["suqc/PACKAGE.txt"])],
)
os.remove(SuqcConfig.path_package_indicator_file())
......
......@@ -5,4 +5,4 @@ from suqc.parameter.postchanges import PostScenarioChangesBase
from suqc.qoi import *
from suqc.request import *
__version__ = "2.0"
__version__ = "2.1"
#!/usr/bin/env python3
import os
import json
import os
import os.path as p
import pathlib
# configuration of the suq-controller
DEFAULT_SUQC_CONFIG = {"default_vadere_src_path": "TODO", # TODO Feature: #25
"server": {
"host": "",
"user": "",
"port": -1}}
DEFAULT_SUQC_CONFIG = {
"default_vadere_src_path": "TODO", # TODO Feature: #25
"server": {"host": "", "user": "", "port": -1},
}
def check_setup(_paths_class):
if not os.path.exists(_paths_class.path_cfg_folder()) and _paths_class.is_package_paths():
if (
not os.path.exists(_paths_class.path_cfg_folder())
and _paths_class.is_package_paths()
):
print(f"INFO: Setting up configuration folder {_paths_class.path_cfg_folder()}")
# the next two checks will fail automatically too, because the folder is empty
os.mkdir(_paths_class.path_cfg_folder())
if not os.path.exists(_paths_class.path_suq_config_file()):
print(f"INFO: Setting up default configuration file located at {_paths_class.path_suq_config_file()}")
print(
f"INFO: Setting up default configuration file located at "
f"{_paths_class.path_suq_config_file()}"
)
_paths_class.store_config(DEFAULT_SUQC_CONFIG)
if not os.path.exists(_paths_class.path_container_folder()):
print(f"INFO: Setting up the default container path (which will store output of simulation runs). "
f"Location {_paths_class.path_container_folder()}")
print(
f"INFO: Setting up the default container path "
f"(which will store output of simulation runs). "
f"Location {_paths_class.path_container_folder()}"
)
os.mkdir(_paths_class.path_container_folder())
return _paths_class
......@@ -128,4 +135,4 @@ class SuqcConfig(object):
if __name__ == "__main__":
print(SuqcConfig)
\ No newline at end of file
print(SuqcConfig)
#!/usr/bin/env python3
import json
import platform
import glob
import json
import os
import shutil
import subprocess
import time
import os
from shutil import rmtree
from shutil import copytree, ignore_patterns, rmtree
from typing import *
from suqc.configuration import SuqcConfig
from suqc.utils.general import user_query_yes_no, get_current_suqc_state, str_timestamp
import pandas as pd
from suqc.configuration import SuqcConfig
from suqc.opp.config_parser import OppConfigFileBase, OppConfigType, OppParser
from suqc.utils.general import (
get_current_suqc_state,
str_timestamp,
user_query_yes_no,
include_patterns,
removeEmptyFolders,
)
# configuration of the suq-controller
DEFAULT_SUQ_CONFIG = {"default_vadere_src_path": "TODO",
"server": {
"host": "",
"user": "",
"port": -1
}}
DEFAULT_SUQ_CONFIG = {
"default_vadere_src_path": "TODO",
"server": {"host": "", "user": "", "port": -1},
}
class AbstractConsoleWrapper(object):
@classmethod
def infer_model(cls, model) -> "AbstractConsoleWrapper":
if isinstance(model, str):
if model == "Coupled":
return CoupledConsoleWrapper(model)
else:
return VadereConsoleWrapper.infer_model(model)
elif isinstance(model, VadereConsoleWrapper) or isinstance(
model, CoupledConsoleWrapper
):
return model
else:
raise ValueError(
f"Model must be of type string or VadereConsoleWrapper or CoupledConsoleWrapper. Got type {type(model)}."
)
class CoupledConsoleWrapper(AbstractConsoleWrapper):
def __init__(self, model):
self.simulator = model
def run_simulation(
self, dirname, start_file, required_files: Union[str, List[str]]
):
terminal_command = ["python3", start_file, "--qoi"]
terminal_command.extend(required_files)
terminal_command.extend(["--run-name", os.path.basename(dirname)])
terminal_command.extend(["--create-vadere-container"])
class VadereConsoleWrapper(object):
time_started = time.time()
t = time.strftime("%H:%M:%S", time.localtime(time_started))
print(f"{t}\t Call {os.path.basename(dirname)}/{start_file} ")
return_code = subprocess.check_call(
terminal_command,
env=os.environ,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
cwd=dirname,
timeout=10800 # stop simulation after 3h
)
process_duration = time.time() - time_started
output_subprocess = None
return return_code, process_duration, output_subprocess
# Current log level choices, requires to manually add, if there are changes
ALLOWED_LOGLVL = ["OFF", "FATAL", "TOPOGRAPHY_ERROR", "TOPOGRAPHY_WARN", "INFO", "DEBUG", "ALL"]
def __init__(self, model_path: str, loglvl="INFO", jvm_flags: Optional[List] =
None, timeout_sec=None):
class VadereConsoleWrapper(AbstractConsoleWrapper):
# Current log level choices, requires to manually add, if there are changes in Vadere
ALLOWED_LOGLVL = [
"OFF",
"FATAL",
"TOPOGRAPHY_ERROR",
"TOPOGRAPHY_WARN",
"INFO",
"DEBUG",
"ALL",
]
def __init__(
self,
model_path: str,
loglvl="INFO",
jvm_flags: Optional[List] = None,
timeout_sec=None,
):
self.jar_path = os.path.abspath(model_path)
if not os.path.exists(self.jar_path):
raise FileNotFoundError(
f"Vadere console .jar file {self.jar_path} does not exist.")
f"Vadere console .jar file {self.jar_path} does not exist."
)
loglvl = loglvl.upper()
if loglvl not in self.ALLOWED_LOGLVL:
raise ValueError(f"set loglvl={loglvl} not contained "
f"in allowed: {self.ALLOWED_LOGLVL}")
raise ValueError(
f"set loglvl={loglvl} not contained "
f"in allowed: {self.ALLOWED_LOGLVL}"
)
if jvm_flags is not None and not isinstance(jvm_flags, list):
raise TypeError(
f"jvm_flags are required to be a list. Got: {type(jvm_flags)}")
f"jvm_flags are required to be a list. Got: {type(jvm_flags)}"
)
if timeout_sec is None:
pass # do nothing, no timeout
pass # do nothing, no timeout
elif not isinstance(timeout_sec, int) or timeout_sec <= 0:
raise TypeError("vadere_run_timeout_sec must be of type int and positive "
"value")
raise TypeError(
"vadere_run_timeout_sec must be of type int and positive " "value"
)
self.loglvl = loglvl
# Additional Java Virtual Machine options / flags
self.jvm_flags = jvm_flags if jvm_flags is not None else []
self.timeout_sec = timeout_sec
@classmethod
def infer_model(cls, model):
if isinstance(model, str):
if os.path.exists(model):
return VadereConsoleWrapper.from_model_path(os.path.abspath(model))
else:
return VadereConsoleWrapper.from_default_models(model)
elif isinstance(model, VadereConsoleWrapper):
return model
else:
raise ValueError(f"Failed to infer Vadere model. \n {model}")
def run_simulation(self, scenario_fp, output_path):
start = time.time()
......@@ -69,9 +156,9 @@ class VadereConsoleWrapper(object):
output_subprocess = dict()
try:
subprocess.check_output(subprocess_cmd,
timeout=self.timeout_sec,
stderr=subprocess.PIPE)
subprocess.check_output(
subprocess_cmd, timeout=self.timeout_sec, stderr=subprocess.PIPE
)
process_duration = time.time() - start
# if return_code != 0 a subprocess.CalledProcessError is raised
......@@ -100,28 +187,13 @@ class VadereConsoleWrapper(object):
def from_model_path(cls, model_path):
return cls(model_path)
@classmethod
def infer_model(cls, model):
if isinstance(model, str):
if os.path.exists(model):
return VadereConsoleWrapper.from_model_path(os.path.abspath(model))
else:
return VadereConsoleWrapper.from_default_models(model)
elif isinstance(model, VadereConsoleWrapper):
return model
else:
raise ValueError(f"Failed to infer Vadere model. \n {model}")
class EnvironmentManager(object):
PREFIX_BASIS_SCENARIO = "BASIS_"
VADERE_SCENARIO_FILE_TYPE = ".scenario"
vadere_output_folder = "vadere_output"
class AbstractEnvironmentManager(object):
def __init__(self, base_path, env_name: str):
self.base_path, self.env_name = self.handle_path_and_env_input(base_path, env_name)
self.base_path, self.env_name = self.handle_path_and_env_input(
base_path, env_name
)
self.env_name = env_name
self.env_path = self.output_folder_path(self.base_path, self.env_name)
......@@ -135,48 +207,54 @@ class EnvironmentManager(object):
print(f"INFO: Set environment path to {self.env_path}")
if not os.path.exists(self.env_path):
raise FileNotFoundError(f"Environment {self.env_path} does not exist. Use function "
f"'EnvironmentManager.create_new_environment'")
self._scenario_basis = None
raise FileNotFoundError(
f"Environment {self.env_path} does not exist. Use function "
f"'EnvironmentManager.create_new_environment'"
)
self._vadere_scenario_basis = None
@property
def basis_scenario(self):
if self._scenario_basis is None:
path_basis_scenario = self.path_basis_scenario
def vadere_basis_scenario(self):
if self._vadere_scenario_basis is None:
path_basis_scenario = self.vadere_path_basis_scenario
with open(path_basis_scenario, "r") as f:
basis_file = json.load(f)
self._scenario_basis = basis_file
self._vadere_scenario_basis = basis_file
return self._scenario_basis
return self._vadere_scenario_basis
@property
def path_basis_scenario(self):
sc_files = glob.glob(os.path.join(self.env_path, f"*{self.VADERE_SCENARIO_FILE_TYPE}"))
def vadere_path_basis_scenario(self):
sc_files = glob.glob(
os.path.join(self.env_path, f"*{self.VADERE_SCENARIO_FILE_TYPE}")
)
if len(sc_files) != 1:
raise RuntimeError(f"None or too many '{self.VADERE_SCENARIO_FILE_TYPE}' files "
"found in environment.")
raise RuntimeError(
f"None or too many '{self.VADERE_SCENARIO_FILE_TYPE}' files "
"found in environment."
)
return sc_files[0]
@classmethod
def from_full_path(cls, env_path):
assert os.path.isdir(env_path)
base_path = os.path.dirname(env_path)
if env_path.endswith(os.pathsep):
env_path = env_path.rstrip(os.path.sep)
env_name = os.path.basename(env_path)
cls(base_path=base_path, env_name=env_name)
def create_variation_env_from_info_file(cls, path_info_file):
raise NotImplemented
@classmethod
def create_new_environment(cls, base_path=None, env_name=None, handle_existing="ask_user_replace"):
def create_new_environment(
cls, base_path=None, env_name=None, handle_existing="ask_user_replace"
):
base_path, env_name = cls.handle_path_and_env_input(base_path, env_name)
# TODO: Refactor, make handle_existing an Enum
assert handle_existing in ["ask_user_replace", "force_replace", "write_in_if_exist_else_create", "write_in"]
assert handle_existing in [
"ask_user_replace",
"force_replace",
"write_in_if_exist_else_create",
"write_in",
]
# set to True if env already exists, and it shouldn't be overwritten
about_creating_env = False
......@@ -191,7 +269,9 @@ class EnvironmentManager(object):
if env_exists:
cls.remove_environment(base_path, env_name, force=True)
elif handle_existing == "write_in":
assert env_exists, f"base_path={base_path} env_name={env_name} does not exist"
assert (
env_exists
), f"base_path={base_path} env_name={env_name} does not exist"
env_man = cls(base_path=base_path, env_name=env_name)
elif handle_existing == "write_in_if_exist_else_create":
if env_exists:
......@@ -207,13 +287,137 @@ class EnvironmentManager(object):
return env_man
@classmethod
def remove_environment(cls, base_path, name, force=False):
target_path = cls.output_folder_path(base_path, name)
if force or user_query_yes_no(
question=f"Are you sure you want to remove the current environment? Path: \n "
f"{target_path}"
):
try:
rmtree(target_path)
except FileNotFoundError:
print(f"INFO: Tried to remove environment {name}, but did not exist.")
return True
return False
@classmethod
def create_variation_env(cls, basis_scenario: Union[str, dict], base_path=None, env_name=None,
handle_existing="ask_user_replace"):
def from_full_path(cls, env_path):
assert os.path.isdir(env_path)
base_path = os.path.dirname(env_path)
if env_path.endswith(os.pathsep):
env_path = env_path.rstrip(os.path.sep)
env_name = os.path.basename(env_path)
cls(base_path=base_path, env_name=env_name)
@staticmethod
def handle_path_and_env_input(base_path, env_name):
if env_name is None:
env_name = "_".join(["output", str_timestamp()])
if base_path is None:
base_path = SuqcConfig.path_container_folder()
return base_path, env_name
@staticmethod
def output_folder_path(base_path, env_name):
base_path, env_name = VadereEnvironmentManager.handle_path_and_env_input(
base_path, env_name
)
assert os.path.isdir(base_path)
output_folder_path = os.path.join(base_path, env_name)
return output_folder_path
def scenario_variation_path(self, par_id, run_id):
return os.path.join(
self.get_env_outputfolder_path(),
self._scenario_variation_filename(par_id, run_id),
)
def save_scenario_variation(self, par_id, run_id, content):
scenario_path = self.scenario_variation_path(par_id, run_id)
assert not os.path.exists(
scenario_path
), f"File {scenario_path} already exists!"
with open(scenario_path, "w") as outfile:
json.dump(content, outfile, indent=4)
return scenario_path
def get_temp_folder(self):
raise NotImplemented
def get_env_outputfolder_path(self):
raise NotImplemented
def get_variation_output_folder(self, parameter_id, run_id):
scenario_filename = self._scenario_variation_filename(
parameter_id=parameter_id, run_id=run_id
)
scenario_filename = scenario_filename.replace(
self.VADERE_SCENARIO_FILE_TYPE, ""
)
return os.path.join(
self.get_env_outputfolder_path(), "".join([scenario_filename, "_output"])
)
def _scenario_variation_filename(self, parameter_id, run_id):
digits_parameter_id = str(parameter_id).zfill(self.nr_digits_variation)
digits_run_id = str(run_id).zfill(self.nr_digits_variation)
numbered_scenario_name = "_".join([digits_parameter_id, digits_run_id])
return "".join([numbered_scenario_name, self.VADERE_SCENARIO_FILE_TYPE])
def get_env_info(self):
return self.env_info_df
@classmethod
def set_env_info(cls, basis_scenario, base_path, env_name, ini_scenario):
info = {
"basis_scenario": basis_scenario,
"ini_path": ini_scenario,
"base_path": base_path,
"env_name": env_name,
}
info = pd.DataFrame(data=info, index=[0])
cls.env_info_df = info
class VadereEnvironmentManager(AbstractEnvironmentManager):
PREFIX_BASIS_SCENARIO = "BASIS_"
VADERE_SCENARIO_FILE_TYPE = ".scenario"
simulation_runs_output_folder = "vadere_output"
def __init__(self, base_path, env_name: str):
super().__init__(base_path, env_name)
@classmethod
def create_variation_env(
cls,
basis_scenario: Union[str, dict],
base_path=None,
env_name=None,
handle_existing="ask_user_replace",
):
cls.set_env_info(
basis_scenario=basis_scenario,
base_path=base_path,
env_name=env_name,
ini_scenario="",
)
# Check if environment already exists
env_man = cls.create_new_environment(base_path=base_path, env_name=env_name, handle_existing=handle_existing)
env_man = cls.create_new_environment(
base_path=base_path, env_name=env_name, handle_existing=handle_existing
)
path_output_folder = env_man.env_path
# Add basis scenario used for the variation (i.e. sampling)
......@@ -221,15 +425,18 @@ class EnvironmentManager(object):
if not os.path.isfile(basis_scenario):
raise FileExistsError("Filepath to .scenario does not exist")
elif basis_scenario.split(".")[-1] != cls.VADERE_SCENARIO_FILE_TYPE[1:]:
raise ValueError("basis_scenario has to be a Vadere '*"
f"{cls.VADERE_SCENARIO_FILE_TYPE}' file")
raise ValueError(
"basis_scenario has to be a Vadere '*"
f"{cls.VADERE_SCENARIO_FILE_TYPE}' file"
)
with open(basis_scenario, "r") as file:
basis_scenario = file.read()
# add prefix to scenario file:
basis_fp = os.path.join(path_output_folder,
f"{cls.PREFIX_BASIS_SCENARIO}{env_name}.scenario")
basis_fp = os.path.join(
path_output_folder, f"{cls.PREFIX_BASIS_SCENARIO}{env_name}.scenario"
)
# FILL IN THE STANDARD FILES IN THE NEW SCENARIO:
with open(basis_fp, "w") as file:
......@@ -245,53 +452,218 @@ class EnvironmentManager(object):
if not SuqcConfig.is_package_paths():
cfg["suqc_state"] = get_current_suqc_state()
with open(os.path.join(path_output_folder, "suqc_commit_hash.json"), 'w') as outfile:
s = "\n".join(["commit hash at creation", cfg["suqc_state"]["git_hash"]])
with open(
os.path.join(path_output_folder, "suqc_commit_hash.json"), "w"
) as outfile:
s = "\n".join(
["commit hash at creation", cfg["suqc_state"]["git_hash"]]
)
outfile.write(s)
# Create the folder where all output is stored
os.mkdir(os.path.join(path_output_folder, EnvironmentManager.vadere_output_folder))
os.mkdir(
os.path.join(
path_output_folder,
VadereEnvironmentManager.simulation_runs_output_folder,
)
)
return cls(base_path, env_name)
def get_env_outputfolder_path(self):
rel_path = os.path.join(
self.env_path, VadereEnvironmentManager.simulation_runs_output_folder
)
return os.path.abspath(rel_path)
class CoupledEnvironmentManager(AbstractEnvironmentManager):
PREFIX_BASIS_SCENARIO = ""
VADERE_SCENARIO_FILE_TYPE = ".scenario"
simulation_runs_output_folder = "simulation_runs"
simulation_runs_single_folder_name = "Sample_"
run_file = "run_script.py"
temp_folder_rover = "temp"
def __init__(self, base_path, env_name: str):
super().__init__(base_path, env_name)
self._omnet_ini_basis = None
@property
def omnet_basis_ini(self):
if self._omnet_ini_basis is None:
path_basis_ini = self.omnet_path_ini
ini_file = OppConfigFileBase.from_path(
ini_path=path_basis_ini,
config="final",
cfg_type=OppConfigType.EXT_DEL_LOCAL,
)
self._omnet_ini_basis = ini_file
return self._omnet_ini_basis
@property
def omnet_path_ini(self):
sc_files = glob.glob(os.path.join(self.env_path, "*ini"))
if len(sc_files) != 1:
raise RuntimeError(f"None or too many 'ini' files " "found in environment.")
return sc_files[0]
@classmethod
def remove_environment(cls, base_path, name, force=False):
target_path = cls.output_folder_path(base_path, name)
def create_variation_env_from_info_file(cls, path_info_file):
if force or user_query_yes_no(question=f"Are you sure you want to remove the current environment? Path: \n "
f"{target_path}"):
try:
rmtree(target_path)
except FileNotFoundError:
print(f"INFO: Tried to remove environment {name}, but did not exist.")
return True
return False
d = pd.read_pickle(path_info_file)
@staticmethod
def handle_path_and_env_input(base_path, env_name):
if env_name is None:
env_name = "_".join(["output", str_timestamp()])
env = cls.create_variation_env(
basis_scenario=d["basis_scenario"].values[0],
ini_scenario=d["ini_path"].values[0],