diff --git a/binarycpython/utils/functions.py b/binarycpython/utils/functions.py
index 8e10644b21dc24638665b248416e7d412ee54081..f0a857d285f2e30a947962e05ce6159eeec27268 100644
--- a/binarycpython/utils/functions.py
+++ b/binarycpython/utils/functions.py
@@ -71,16 +71,18 @@ def remove_file(file: str, verbosity: int = 0) -> None:
except FileNotFoundError as inst:
print("Error while deleting file {}: {}".format(file, inst))
else:
- verbose_print("File/directory {} doesn't exist. Can't remove it.", verbosity, 1)
+ verbose_print("File/directory {} doesn't exist. Can't remove it.".format(file), verbosity, 1)
-def temp_dir() -> str:
+def temp_dir(*args: str) -> str:
"""
- Function to return the path the custom logging library shared object
- and script will be written to.
+ Function to create directory within the TMP directory of the filesystem
Makes use of os.makedirs exist_ok which requires python 3.2+
+ Args:
+ function arguments: str input where each next input will be a child of the previous full_path. e.g. temp_dir('tests', 'grid') will become '/tmp/binary_c_python/tests/grid'
+
Returns:
the path of a subdirectory called binary_c_python in the TMP of the filesystem
"""
@@ -88,12 +90,16 @@ def temp_dir() -> str:
tmp_dir = tempfile.gettempdir()
path = os.path.join(tmp_dir, "binary_c_python")
+ # loop over the other paths if there are any:
+ if args:
+ for extra_dir in args:
+ path = os.path.join(path, extra_dir)
+
#
os.makedirs(path, exist_ok=True)
return path
-
def create_hdf5(data_dir: str, name: str) -> None:
"""
Function to create an hdf5 file from the contents of a directory:
@@ -1128,12 +1134,17 @@ def merge_dicts(dict_1: dict, dict_2: dict) -> dict:
# See whether the types are actually the same
if not type(dict_1[key]) is type(dict_2[key]):
- print(
- "Error key: {} value: {} and key: {} value: {} are not of the same type and cannot be merged".format(
- key, dict_1[key], key, dict_2[key]
+ # Exceptions:
+ if (type(dict_1[key]) in [int, float]) and (type(dict_2[key]) in [int, float]):
+ new_dict[key] = dict_1[key] + dict_2[key]
+
+ else:
+ print(
+ "Error key: {} value: {} type: {} and key: {} value: {} type: {} are not of the same type and cannot be merged".format(
+ key, dict_1[key], type(dict_1[key]), key, dict_2[key], type(dict_2[key])
+ )
)
- )
- raise ValueError
+ raise ValueError
# Here we check for the cases that we want to explicitly catch. Ints will be added,
# floats will be added, lists will be appended (though that might change) and dicts will be
@@ -1170,6 +1181,34 @@ def merge_dicts(dict_1: dict, dict_2: dict) -> dict:
#
return new_dict
+def extract_ensemble_json_from_string(binary_c_output: str) -> dict:
+ """
+ Function to extract the ensemble_json information from a raw binary_c output string
+
+ Args:
+ binary_c_output: raw binary_c output string
+
+ Returns:
+ json dictionary with the parsed ENSEMBLE_JSON data
+ """
+
+ json = None
+
+ try:
+ ensemble_jsons_strings = [
+ line for line in binary_c_output.splitlines() if line.startswith("ENSEMBLE_JSON")
+ ]
+
+ json = handle_ensemble_string_to_json(
+ ensemble_jsons_strings[0][len("ENSEMBLE_JSON ") :]
+ )
+
+ if len(ensemble_jsons_strings)>1:
+ verbose_print("Warning: There is more than one line starting with ENSEMBLE_JSON. Taking the first, but you should check this out.",1, 0)
+ except IndexError:
+ verbose_print("Error: Couldn't extract the ensemble information from the output string", 0, 0)
+
+ return json
class binarycDecoder(json.JSONDecoder):
"""
diff --git a/binarycpython/utils/grid.py b/binarycpython/utils/grid.py
index 90eb785e97fa2538096bc4621ed50889dfa3ecf3..c6c7635bc9598251e6b6f16cb2cf23f731ed7993 100644
--- a/binarycpython/utils/grid.py
+++ b/binarycpython/utils/grid.py
@@ -57,6 +57,7 @@ from binarycpython.utils.functions import (
binarycDecoder,
merge_dicts,
BinaryCEncoder,
+ extract_ensemble_json_from_string
)
# from binarycpython.utils.hpc_functions import (
# get_condor_version,
@@ -612,11 +613,11 @@ class Population:
pass to binary_c.
"""
- for thread_nr in self.grid_options["amt_cores"]:
+ for thread_id in range(self.grid_options["amt_cores"]):
persistent_data_memaddr = (
- _binary_c_bindings.binary_c_return_persistent_data_memaddr()
+ _binary_c_bindings.return_persistent_data_memaddr()
)
- self.persistent_data_memory_dict[thread_nr] = persistent_data_memaddr
+ self.persistent_data_memory_dict[thread_id] = persistent_data_memaddr
verbose_print(
"Created the following dict with persistent memaddresses: {}".format(
self.persistent_data_memory_dict
@@ -631,29 +632,32 @@ class Population:
pass to binary_c.
TODO: fix the function
+ TODO: remove this function probably.
+ TODO: add option to delay combining all the ensemble outputs.
"""
combined_ensemble_json = {}
- for key in self.persistent_data_memory_dict:
- persistent_data_memaddr = self.persistent_data_memory_dict[key]
+ for thread_id in self.persistent_data_memory_dict:
+ persistent_data_memaddr = self.persistent_data_memory_dict[thread_id]
verbose_print(
"Freeing {} (thread {})and merging output to combined dict".format(
- persistent_data_memaddr, key
+ persistent_data_memaddr, thread_id
),
self.grid_options["verbosity"],
1,
)
# Get the output and decode it correctly to get the numbers correct
- ensemble_json_output = _binary_c_bindings.binary_c_free_persistent_data_memaddr_and_return_json_output(
+ ensemble_raw_output = _binary_c_bindings.free_persistent_data_memaddr_and_return_json_output(
persistent_data_memaddr
)
- parsed_json = json.loads(
- ensemble_json_output.splitlines()[0][len("ENSEMBLE_JSON ") :],
- cls=binarycDecoder,
- )
+
+ print("ensemble_raw_output: ", ensemble_raw_output)
+
+ # Readout the raw string and get the ensemble_json str:
+ parsed_json = extract_ensemble_json_from_string(ensemble_raw_output)
# Combine the output with the main output
combined_ensemble_json = merge_dicts(combined_ensemble_json, parsed_json)
@@ -878,10 +882,15 @@ class Population:
for output_dict in result:
combined_output_dict = merge_dicts(combined_output_dict, output_dict)
+ # handle the ensemble if necessary
+ if self.bse_options.get("ensemble", None):
+ self._free_persistent_data_memory_and_combine_results_and_output()
+
print(combined_output_dict)
# Put the values back as object properties
self.grid_options["results"] = combined_output_dict["results"]
+ self.grid_options["ensemble_results"] = combined_output_dict["ensemble_results"]
self.grid_options["_failed_count"] = combined_output_dict["_failed_count"]
self.grid_options["_failed_prob"] = combined_output_dict["_failed_prob"]
self.grid_options["_failed_systems_error_codes"] = list(
@@ -901,6 +910,11 @@ class Population:
binary_cmdline_string = self._return_argline(full_system_dict)
+ persistent_data_memaddr = -1
+ if self.bse_options.get("ensemble", 0) == 1:
+ persistent_data_memaddr = self.persistent_data_memory_dict[self.process_ID]
+ print("thread {}: persistent_data_memaddr: ".format(self.process_ID), persistent_data_memaddr)
+
# Get results binary_c
out = _binary_c_bindings.run_system(
argstring=binary_cmdline_string,
@@ -909,11 +923,18 @@ class Population:
],
store_memaddr=self.grid_options["_store_memaddr"],
population=1, # since this system is part of a population, we set this flag to prevent the store from being freed
+ persistent_data_memaddr=persistent_data_memaddr
)
# Check for errors
_ = self._check_binary_c_error(out, full_system_dict)
+ # print("custom output:")
+ # ensemble_raw_output = _binary_c_bindings.free_persistent_data_memaddr_and_return_json_output(
+ # persistent_data_memaddr
+ # )
+ # print("ensemble_raw_output custom: ", ensemble_raw_output)
+
# Have some user-defined function do stuff with the data.
if self.grid_options["parse_function"]:
self.grid_options["parse_function"](self, out)
@@ -932,6 +953,13 @@ class Population:
ID # Store the ID as a object property again, lets see if that works.
)
+ if self.bse_options.get('ensemble', 0) == 1:
+ # set persistent data memaddr if necessary.
+ persistent_data_memaddr = (
+ _binary_c_bindings.return_persistent_data_memaddr()
+ )
+ self.persistent_data_memory_dict = {self.process_ID: persistent_data_memaddr}
+
# apparently we have to re-load this for every process, otherwise NameErrors arise (seems like a bug but I'm not sure)
self._load_grid_function()
@@ -977,7 +1005,6 @@ class Population:
# Evolve the system
self._evolve_system_mp(full_system_dict)
- # TODO: fix the 'repeat' and 'weight' tracking here
# Keep track of systems:
probability_of_systems_run += full_system_dict["probability"]
number_of_systems_run += 1
@@ -988,9 +1015,30 @@ class Population:
# Has to be here because this one is used for the (localcounter+ID) % (self..)
localcounter += 1
+ # Handle ensemble output: is ensemble==1, then either directly write that data to a file, or combine everything into 1 file.
+ ensemble_json = {} # Make sure it exists already
+ if self.bse_options.get('ensemble', 0) == 1:
+ ensemble_raw_output = _binary_c_bindings.free_persistent_data_memaddr_and_return_json_output(
+ self.persistent_data_memory_dict[self.process_ID]
+ )
+
+ #
+ if self.grid_options['combine_ensemble_with_thread_joining'] == True:
+ ensemble_json = extract_ensemble_json_from_string(ensemble_raw_output) # Load this into a dict so that we can combine it later
+
+ else:
+ # If we do not allow this, automatically we will export this to the data_dir, in some formatted way
+ output_file = os.path.join(self.custom_options['data_dir'], "ensemble_output_{}_{}.json".format(self.grid_options['_population_id'], self.process_ID))
+ print("Thread {}: Chosen to output the ensemble results directly to file: {}".format(self.process_ID, output_file))
+
+ # Write to file
+ with open(output_file, "w") as f:
+ f.write(ensemble_raw_output)
+
# Return a set of results and errors
output_dict = {
"results": self.grid_options["results"],
+ "ensemble_results": ensemble_json,
"_failed_count": self.grid_options["_failed_count"],
"_failed_prob": self.grid_options["_failed_prob"],
"_failed_systems_error_codes": self.grid_options[
@@ -1077,15 +1125,15 @@ class Population:
if not self.grid_options["parse_function"]:
print("Warning: No parse function set. Make sure you intended to do this.")
- #######################
- ### Custom logging code:
+ # #######################
+ # ### Custom logging code:
self._set_custom_logging()
- ### Load store
+ ### Load store: Make sure this actually works.
self.grid_options["_store_memaddr"] = _binary_c_bindings.return_store_memaddr()
- ### ensemble:
- ## check the settings:
+ ### ensemble: make some checks for this
+ ## check the settings and set all the warnings.
if self.bse_options.get("ensemble", None):
if not self.bse_options["ensemble_defer"] == 1:
verbose_print(
@@ -1095,13 +1143,24 @@ class Population:
)
raise ValueError
- if not self.grid_options["ensemble_output_name"]:
+ if not self.custom_options["ensemble_output_name"]:
verbose_print(
"Error: if you want to run an ensemble in a population, please set set 'ensemble_output_name'. It will be combined with 'data_dir' to write the output of the ensembles to", self.grid_options['verbosity'], 0)
raise ValueError
- ## Load persistent_data_memaddr if necessary:
- self._load_persistent_data_memory_dict()
+ if (not any([key.startswith("ensemble_filter_") for key in self.bse_options])):
+ verbose_print(
+ "Warning: Running the ensemble without any filter requires alot of available RAM", self.grid_options['verbosity'], 0)
+
+ if self.bse_options.get("ensemble_filters_off", None):
+ if self.bse_options["ensemble_filters_off"] == 0:
+ verbose_print(
+ "Warning: Running the ensemble without any filter requires alot of available RAM", self.grid_options['verbosity'], 0)
+
+ if self.grid_options['combine_ensemble_with_thread_joining'] == False:
+ if not self.custom_options.get('data_dir', None):
+ verbose_print(
+ "Error: chosen to write the ensemble output directly to files but data_dir isnt set", self.grid_options['verbosity'], 0)
# Check which type of population generation
if self.grid_options["evolution_type"] == "grid":
@@ -1195,10 +1254,6 @@ class Population:
- unload dry grid function/module
"""
- # Output the ensemble if necessary:
- if self.bse_options.get("ensemble", None):
- self._free_persistent_data_memory_and_combine_results_and_output()
-
# Reset values
self.grid_options["_count"] = 0
self.grid_options["_probtot"] = 0
diff --git a/binarycpython/utils/grid_options_defaults.py b/binarycpython/utils/grid_options_defaults.py
index d295e12119d9babff8816989647b2816ee268f43..f462c5cebc3f3be5f5abde68886afda37b8f3149 100644
--- a/binarycpython/utils/grid_options_defaults.py
+++ b/binarycpython/utils/grid_options_defaults.py
@@ -27,6 +27,7 @@ grid_options_defaults_dict = {
"parse_function": None, # FUnction to parse the output with.
"tmp_dir": temp_dir(), # Setting the temp dir of the program
"_main_pid": -1, # Placeholder for the main process id of the run.
+ "combine_ensemble_with_thread_joining": True, # Flag on whether to combine everything and return it to the user or if false: write it to data_dir/ensemble_output_{popuation_id}_{thread_id}.json
# "output_dir":
"_commandline_input": "",
##########################
@@ -85,6 +86,7 @@ grid_options_defaults_dict = {
"weight": 1.0, # weighting for the probability
"repeat": 1, # number of times to repeat each system (probability is adjusted to be 1/repeat)
"results": {}, # dict to store the results. Every process fills this on its own and then it will be joined later
+ "ensemble_results": {}, # Dict to store the ensemble results
"_start_time_evolution": 0, # Start time of the grid
"_end_time_evolution": 0, # end time of the grid
"_errors_found": False, # Flag whether there are any errors from binary_c
@@ -463,6 +465,9 @@ grid_options_descriptions = {
"weight": "Weight factor for each system. The calculated probability is mulitplied by this. If the user wants each system to be repeated several times, then this variable should not be changed, rather change the _repeat variable instead, as that handles the reduction in probability per system. This is useful for systems that have a process with some random element in it.", # TODO: add more info here, regarding the evolution splitting.
"repeat": "Factor of how many times a system should be repeated. Consider the evolution splitting binary_c argument for supernovae kick repeating.", # TODO: make sure this is used.
"evolution_type": "Variable containing the type of evolution used of the grid. Multiprocessing or linear processing",
+ "combine_ensemble_with_thread_joining": "BOolean flag on whether to combine everything and return it to the user or if false: write it to data_dir/ensemble_output_{popuation_id}_{thread_id}.json",
+ "ensemble_results": "Dictinary that stores the ensemble results if combine_ensemble_with_thread_joining==True",
+
}
#################################