import os
import copy
import json
import sys
import datetime
import time
import random
import inspect
import numpy as np
import multiprocessing as mp
from pathos.multiprocessing import ProcessingPool as Pool
import binary_c_python_api
import binarycpython
from binarycpython.utils.grid_options_defaults import grid_options_defaults_dict
from binarycpython.utils.custom_logging_functions import (
autogen_C_logging_code,
binary_c_log_code,
create_and_load_logging_function,
temp_dir,
)
from binarycpython.utils.functions import (
get_defaults,
parse_binary_c_version_info,
output_lines,
remove_file,
filter_arg_dict,
)
# Todo-list
# TODO: add functionality to 'on-init' set arguments
# TODO: add functionality to return the initial_abundance_hash
# TODO: add functionality to return the isotope_hash
# TODO: add functionality to return the isotope_list
# TODO: add functionality to return the nuclear_mass_hash
# TODO: add functionality to return the nuclear_mass_list
# TODO: add functionality to return the source_list
# TODO: add functionality to return the ensemble_list
# TODO: change the grid_options dict structure so that there is room for descriptions
# TODO: consider spreading the functions over more files.
# Make this function also an API call. Doest seem to get written to a buffer that is stored into a python object. rather its just written to stdout
class Population(object):
def __init__(self):
"""
Initialisation function of the population class
"""
self.defaults = get_defaults()
# Different sections of options
self.bse_options = (
{}
) # bse_options is just empty. Setting stuff will check against the defaults to see if the input is correct.
self.grid_options = grid_options_defaults_dict.copy()
self.custom_options = {}
# Argline dict
self.argline_dict = {}
# Set main process id
self.grid_options["main_pid"] = os.getpid()
###################################################
# Argument functions
###################################################
# General flow of generating the arguments for the binary_c call:
# - user provides parameter and value via set (or manually but that is risky)
# - The parameter names of these input get compared to the parameter names in the self.defaults; with this, we know that its a valid
# parameter to give to binary_c.
# - For a single system, the bse_options will be written as a arg line
# - For a population the bse_options will get copied to a temp_bse_options dict and updated with all the parameters generated by the grid
# I will NOT create the argument line by fully writing ALL the defaults and overriding user input, that seems not necessary
# because by using the get_defaults() function we already know for sure which parameter names are valid for the binary_c version
# And because binary_c uses internal defaults, its not necessary to explicitly pass them.
# I do however suggest everyone to export the binary_c defaults to a file, so that you know exactly which values were the defaults.
# TODO: maybe make a set_bse option.
def set_bse_option(self, key, arg):
self.bse_options[key] = arg
def set(self, **kwargs):
"""
Function to set the values of the population. This is the preferred method to set values of functions, as it
provides checks on the input.
the bse_options will get populated with all the those that have a key that is present in the self.defaults
the grid_options will get updated with all the those that have a key that is present in the self.grid_options
If neither of above is met; the key and the value get stored in a custom_options dict.
"""
for key in kwargs.keys():
# Filter out keys for the bse_options
if key in self.defaults.keys():
if self.grid_options["verbose"] > 0:
print("adding: {}={} to BSE_options".format(key, kwargs[key]))
self.bse_options[key] = kwargs[key]
# Filter out keys for the grid_options
elif key in self.grid_options.keys():
if self.grid_options["verbose"] > 0:
print("adding: {}={} to grid_options".format(key, kwargs[key]))
self.grid_options[key] = kwargs[key]
# The of the keys go into a custom_options dict
else:
print(
"!! Key doesnt match previously known parameter: adding: {}={} to custom_options".format(
key, kwargs[key]
)
)
self.custom_options[key] = kwargs[key]
def parse_cmdline(self):
"""
Function to handle settings values via the command line:
TODO: remove the need for --cmdline
"""
import argparse
parser = argparse.ArgumentParser()
parser.add_argument(
"--cmdline",
help='Setting values via the commandline. Input like --cmdline "metallicity=0.02"',
)
args = parser.parse_args()
# How its set up now is that as input you need to give --cmdline "metallicity=0.002"
# Its checked if this exists and handled accordingly.
if args.cmdline:
if self.grid_options["verbose"] > 0:
print("Found cmdline args. Parsing them now")
# Grab the input and split them up, while accepting only non-empty entries
cmdline_args = args.cmdline
split_args = [
cmdline_arg
for cmdline_arg in cmdline_args.split(" ")
if not cmdline_arg == ""
]
# Make dict and fill it
cmdline_dict = {}
for cmdline_arg in split_args:
split = cmdline_arg.split("=")
parameter = split[0]
value = split[1]
# Add to dict
cmdline_dict[parameter] = value
# unpack the dictionary into the setting function that handles where the values are set
self.set(**cmdline_dict)
def return_argline(self, parameter_dict=None):
"""
Function to create the string for the arg line from a parameter dict
"""
if not parameter_dict:
parameter_dict = self.bse_options
argline = "binary_c "
for param_name in sorted(parameter_dict):
argline += "{} {} ".format(param_name, parameter_dict[param_name])
argline = argline.strip()
return argline
def generate_population_arglines_file(self, output_file):
"""
Function to generate a file that contains all the argument lines that would be given to binary_c if the population had been run
"""
pass
def add_grid_variable(
self,
name,
longname,
valuerange,
resolution,
spacingfunc,
probdist,
dphasevol,
parameter_name,
precode=None,
condition=None,
):
"""spec
Function to add grid variables to the grid_options.
TODO: Fix this complex function.
The execution of the grid generation will be through a nested forloop,
and will rely heavily on the eval() functionality of python. Which, in terms of safety is very bad, but in terms of flexibility is very good.
name:
name of parameter
example: name = 'lnm1'
longname:
Long name of parameter
example: longname = 'Primary mass'
range:
Range of values to take
example: range = [log($mmin),log($mmax)]
resolution:
Resolution of the sampled range (amount of samples)
example: resolution = $resolution->{m1}
spacingfunction:
Function determining how the range is sampled
example: spacingfunction = "const(log($mmin),log($mmax),$resolution->{m1})"
precode:
# TODO: think of good description.
example: precode = '$m1=exp($lnm1);'
probdist:
FUnction determining the probability that gets asigned to the sampled parameter
example: probdist = 'Kroupa2001($m1)*$m1'
dphasevol:
part of the parameter space that the total probability is calculated with
example: dphasevol = '$dlnm1'
condition:
condition that has to be met in order for the grid generation to continue
example: condition = '$self->{_grid_options}{binary}==1'
"""
# Add grid_variable
grid_variable = {
"name": name,
"longname": longname,
"valuerange": valuerange,
"resolution": resolution,
"spacingfunc": spacingfunc,
"precode": precode,
"probdist": probdist,
"dphasevol": dphasevol,
"parameter_name": parameter_name,
"condition": condition,
"grid_variable_number": len(self.grid_options["grid_variables"]),
}
# Load it into the grid_options
self.grid_options["grid_variables"][grid_variable["name"]] = grid_variable
if self.grid_options["verbose"] > 0:
print("Added grid variable: {}".format(json.dumps(grid_variable, indent=4)))
###################################################
# Return functions
###################################################
def return_population_settings(self):
"""
Function that returns all the options that have been set.
Can be combined with json to make a nice file.
"""
options = {
"bse_options": self.bse_options,
"grid_options": self.grid_options,
"custom_options": self.custom_options,
}
return options
def return_binary_c_version_info(self, parsed=False):
"""
Function that returns the version information of binary_c
"""
version_info = binary_c_python_api.return_version_info().strip()
if parsed:
version_info = parse_binary_c_version_info(version_info)
return version_info
def return_binary_c_defaults(self):
"""
Function that returns the defaults of the binary_c version that is used.
"""
return self.defaults
def return_all_info(
self,
include_population_settings=True,
include_binary_c_defaults=True,
include_binary_c_version_info=True,
include_binary_c_help_all=True,
):
"""
Function that returns all the information about the population and binary_c
"""
from binarycpython.utils.functions import get_help_all
#
all_info = {}
#
if include_population_settings:
population_settings = self.return_population_settings()
all_info["population_settings"] = population_settings
#
if include_binary_c_defaults:
binary_c_defaults = self.return_binary_c_defaults()
all_info["binary_c_defaults"] = binary_c_defaults
if include_binary_c_version_info:
binary_c_version_info = self.return_binary_c_version_info(parsed=True)
all_info["binary_c_version_info"] = binary_c_version_info
if include_binary_c_help_all:
binary_c_help_all_info = get_help_all(print_help=False, return_dict=True)
all_info["binary_c_help_all"] = binary_c_help_all_info
return all_info
def export_all_info(
self,
use_datadir=True,
outfile=None,
include_population_settings=True,
include_binary_c_defaults=True,
include_binary_c_version_info=True,
include_binary_c_help_all=True,
):
"""
Function that exports the all_info to a json file
TODO: if any of the values in the dicts here is of a not-serializable form, then we need to change that to a string or something
so, use a recursive function that goes over the all_info dict and finds those that fit
TODO: Fix to write things to the directory. which options do which etc
"""
all_info = self.return_all_info(
include_population_settings=include_population_settings,
include_binary_c_defaults=include_binary_c_defaults,
include_binary_c_version_info=include_binary_c_version_info,
include_binary_c_help_all=include_binary_c_help_all,
)
# Copy dict
all_info_cleaned = copy.deepcopy(all_info)
# Clean the all_info_dict: (i.e. transform the function objects to strings)
if all_info_cleaned.get("population_settings", None):
if all_info_cleaned["population_settings"]["grid_options"][
"parse_function"
]:
all_info_cleaned["population_settings"]["grid_options"][
"parse_function"
] = str(
all_info_cleaned["population_settings"]["grid_options"][
"parse_function"
]
)
if use_datadir:
base_name = os.path.splitext(self.custom_options["base_filename"])[0]
settings_name = base_name + "_settings.json"
# Check directory, make if necessary
os.makedirs(self.custom_options["data_dir"], exist_ok=True)
settings_fullname = os.path.join(
self.custom_options["data_dir"], settings_name
)
if self.grid_options["verbose"] > 0:
print("Writing settings to {}".format(settings_fullname))
# if not outfile.endswith('json'):
with open(settings_fullname, "w") as f:
f.write(json.dumps(all_info_cleaned, indent=4))
else:
if self.grid_options["verbose"] > 0:
print("Writing settings to {}".format(outfile))
# if not outfile.endswith('json'):
with open(outfile, "w") as f:
f.write(json.dumps(all_info_cleaned, indent=4))
def set_custom_logging(self):
"""
Function/routine to set all the custom logging so that the function memory pointer is known to the grid.
"""
# C_logging_code gets priority of C_autogen_code
if self.grid_options["verbose"] > 0:
print("Creating and loading custom logging functionality")
if self.grid_options["C_logging_code"]:
# Generate entire shared lib code around logging lines
custom_logging_code = binary_c_log_code(
self.grid_options["C_logging_code"],
verbose=self.grid_options["verbose"],
)
# Load memory adress
(
self.grid_options["custom_logging_func_memaddr"],
self.grid_options["custom_logging_shared_library_file"],
) = create_and_load_logging_function(
custom_logging_code, verbose=self.grid_options["verbose"]
)
elif self.grid_options["C_auto_logging"]:
# Generate real logging code
logging_line = autogen_C_logging_code(
self.grid_options["C_auto_logging"],
verbose=self.grid_options["verbose"],
)
# Generate entire shared lib code around logging lines
custom_logging_code = binary_c_log_code(
logging_line, verbose=self.grid_options["verbose"]
)
# Load memory adress
(
self.grid_options["custom_logging_func_memaddr"],
self.grid_options["custom_logging_shared_library_file"],
) = create_and_load_logging_function(
custom_logging_code, verbose=self.grid_options["verbose"]
)
###################################################
# Evolution functions
###################################################
def setup(self):
"""
Function to set up the necessary stuff for the population evolution:
# TODO: Make other kinds of populations possible. i.e, read out type of grid, and set up accordingly
# TODO: make this function more general. Have it explicitly set the system_generator function
"""
#######################
### Custom logging code:
self.set_custom_logging()
### Load store
self.grid_options["store_memaddr"] = binary_c_python_api.return_store("")
#######################
# Dry run and getting starcount
self.grid_options["probtot"] = 0
self.generate_grid_code(dry_run=True)
self.load_grid_function()
self.dry_run()
print(
"Total starcount for this run will be: {}".format(
self.grid_options["total_starcount"]
)
)
#######################
# Reset values and prepare the grid function
self.grid_options[
"probtot"
] = 0 # To make sure that the values are reset. TODO: fix this in a cleaner way
self.grid_options[
"start_time_evolution"
] = time.time() # Setting start time of grid
#
self.generate_grid_code(dry_run=False)
#
self.load_grid_function()
def cleanup(self):
"""
Function that handles all the cleaning up after the grid has been generated and/or run
- reset values to 0
- remove grid file
- unload grid function/module
- remove dry grid file
- unload dry grid function/module
"""
# Reset values
self.grid_options["count"] = 0
self.grid_options["probtot"] = 0
self.grid_options["system_generator"] = None
# Remove files
# Unload functions
def evolve_system_mp(self, binary_cmdline_string):
"""
Function that the multiprocessing evolution method calls to evolve a system
"""
out = binary_c_python_api.run_system(
argstring=binary_cmdline_string,
custom_logging_func_memaddr=self.grid_options[
"custom_logging_func_memaddr"
],
store_memaddr=self.grid_options["store_memaddr"],
population=1,
)
if self.grid_options["parse_function"]:
self.grid_options["parse_function"](self, out)
def yield_system_mp(self):
"""
Function that the multiprocessing evolution method calls to yield systems
"""
for i, system in enumerate(self.grid_options["system_generator"](self)):
full_system_dict = self.bse_options.copy()
full_system_dict.update(system)
binary_cmdline_string = self.return_argline(full_system_dict)
self.print_info(
i + 1, self.grid_options["total_starcount"], full_system_dict
)
yield binary_cmdline_string
print("generator done")
def evolve_single(self, clean_up_custom_logging_files=True):
"""
Function to run a single system
The output of the run gets returned, unless a parse function is given to this function.
"""
### Custom logging code:
self.set_custom_logging()
# Get argument line
argline = self.return_argline(self.bse_options)
print("Running {}".format(argline))
# Run system
out = binary_c_python_api.run_system(
argstring=argline,
custom_logging_func_memaddr=self.grid_options[
"custom_logging_func_memaddr"
],
store_memaddr=self.grid_options["store_memaddr"],
population=0,
)
# TODO: add call to function that cleans up the temp customlogging dir, and unloads the loaded libraries.
# TODO: make a switch to turn this off
if clean_up_custom_logging_files:
self.clean_up_custom_logging(evol_type="single")
# Parse
if self.grid_options["parse_function"]:
return self.grid_options["parse_function"](self, out)
else:
return out
def evolve_population_mp(self):
"""
Function to evolve the population with multiprocessing approach. Using pathos to be able to include class-owned functions.
"""
import multiprocessing as mp
from pathos.pools import _ProcessPool as Pool
from pathos.helpers import mp as pathos_multiprocess
# TODO: make further use of a queue to handle jobs or at least get information on the process ids etc
# https://stackoverflow.com/questions/10190981/get-a-unique-id-for-worker-in-python-multiprocessing-pool
# https://stackoverflow.com/questions/8640367/python-manager-dict-in-multiprocessing/9536888 for muting values through dicts
# https://python-forum.io/Thread-Dynamic-updating-of-a-nested-dictionary-in-multiprocessing-pool
# https://stackoverflow.com/questions/28740955/working-with-pathos-multiprocessing-tool-in-python-and
manager = pathos_multiprocess.Manager()
self.grid_options["result_dict"] = manager.dict()
# Create pool
p = Pool(processes=self.grid_options["amt_cores"])
# Execute
# TODO: calculate the chunksize value based on: total starcount and cores used.
r = list(
p.imap_unordered(
self.evolve_system_mp, self.yield_system_mp(), chunksize=20
)
)
# Handle clean termination of the whole multiprocessing (making sure there are no zombie processes (https://en.wikipedia.org/wiki/Zombie_process))
p.close()
p.join()
def evolve_population_lin(self):
"""
Function to evolve the population linearly (i.e. 1 core, no multiprocessing)
"""
for i, system in enumerate(self.grid_options["system_generator"](self)):
full_system_dict = self.bse_options.copy()
full_system_dict.update(system)
binary_cmdline_string = self.return_argline(full_system_dict)
out = binary_c_python_api.run_system(
argstring=binary_cmdline_string,
custom_logging_func_memaddr=self.grid_options[
"custom_logging_func_memaddr"
],
store_memaddr=self.grid_options["store_memaddr"],
population=1,
)
self.print_info(
i + 1, self.grid_options["total_starcount"], full_system_dict
)
def evolve_population(self):
"""
Function to evolve populations. This is the main function. Handles the setting up, evolving and cleaning up of a population of stars.
"""
##
# Prepare code/initialise grid.
# set custom logging, set up store_memaddr, build grid code. dry run grid code.
self.setup()
##
# Evolve systems: via grid_options one can choose to do this linearly, or multiprocessing method.
if (
self.grid_options["evolution_type"]
in self.grid_options["evolution_type_options"]
):
if self.grid_options["evolution_type"] == "mp":
self.evolve_population_mp()
elif self.grid_options["evolution_type"] == "linear":
self.evolve_population_lin()
else:
print(
"Warning. you chose a wrong option for the grid evolution types. Please choose from the following: {}.".format(
self.grid_options["evolution_type_options"]
)
)
##
# Clean up code: remove files, unset values.
self.cleanup()
###################################################
# Function to test evolution algorithms
###################################################
def test_evolve_single(self):
"""
Function to test the evolution of a system. Calls the api binding directly.
"""
if self.grid_options["verbose"] > 0:
print("running a single system as a test")
m1 = 15.0 # Msun
m2 = 14.0 # Msun
separation = 0 # 0 = ignored, use period
orbital_period = 4530.0 # days
eccentricity = 0.0
metallicity = 0.02
max_evolution_time = 15000
argstring = "binary_c M_1 {0:g} M_2 {1:g} separation {2:g} orbital_period {3:g} eccentricity {4:g} metallicity {5:g} max_evolution_time {6:g} ".format(
m1,
m2,
separation,
orbital_period,
eccentricity,
metallicity,
max_evolution_time,
)
output = binary_c_python_api.run_system(argstring)
print("\n\nBinary_c output:")
print(output)
###################################################
# Gridcode functions
#
# Function below are used to run populations with
# a variable grid
###################################################
def generate_grid_code(self, dry_run=False):
"""
Function that generates the code from which the population will be made.
dry_run: when True, it will return the starcount at the end so that we know what the total amount of systems is.
The phasevol values are handled by generating a second array
# DONE: make a generator for this.
# TODO: Add correct logging everywhere
# TODO: add part to handle separation if orbital_period is added. Idea. use default values for orbital parameters and possibly overwrite those or something.
# TODO: add centering center left right for the spacing
# TODO: add sensible description to this function.
# TODO: Check whether all the probability and phasevol values are correct.
Results in a generated file that contains a system_generator function.
"""
if self.grid_options["verbose"] > 0:
print("Generating grid code")
# Some local values
code_string = ""
depth = 0
indent = " "
total_grid_variables = len(self.grid_options["grid_variables"])
# Import packages
code_string += "import math\n"
code_string += "import numpy as np\n"
code_string += "from binarycpython.utils.distribution_functions import *\n"
code_string += "from binarycpython.utils.spacing_functions import *\n"
code_string += "from binarycpython.utils.useful_funcs import *\n"
code_string += "\n\n"
# Make the function
code_string += "def grid_code(self):\n"
# Increase depth
depth += 1
# Write some info in the function
code_string += (
indent * depth
+ "# Grid code generated on {}\n".format(
datetime.datetime.now().isoformat()
)
+ indent * depth
+ "# This function generates the systems that will be evolved with binary_c\n\n"
)
# Set some values in the generated code:
code_string += indent * depth + "# Setting initial values\n"
code_string += indent * depth + "total_starcount = 0\n"
code_string += indent * depth + "starcounts = [0 for i in range({})]\n".format(
total_grid_variables
)
code_string += indent * depth + "probabilities = {}\n"
code_string += (
indent * depth
+ "probabilities_list = [0 for i in range({})]\n".format(
total_grid_variables
)
)
code_string += (
indent * depth
+ "probabilities_sum = [0 for i in range({})]\n".format(
total_grid_variables
)
)
code_string += indent * depth + "parameter_dict = {}\n"
code_string += indent * depth + "phasevol = 1\n"
code_string += indent * depth + "\n"
code_string += indent * depth + "# setting probability lists\n"
# Prepare the probability
for el in sorted(
self.grid_options["grid_variables"].items(),
key=lambda x: x[1]["grid_variable_number"],
):
# Make probabilities dict
grid_variable = el[1]
code_string += indent * depth + 'probabilities["{}"] = 0\n'.format(
grid_variable["parameter_name"]
)
#################################################################################
# Start of code generation
#################################################################################
code_string += indent * depth + "\n"
# Generate code
print("Generating grid code")
for loopnr, el in enumerate(
sorted(
self.grid_options["grid_variables"].items(),
key=lambda x: x[1]["grid_variable_number"],
)
):
print("Constructing/adding: {}".format(el[0]))
grid_variable = el[1]
#################################################################################
# Check condition and generate forloop
# If the grid variable has a condition, write the check and the action
if grid_variable["condition"]:
# Add comment
code_string += (
indent * depth
+ "# Condition for {}".format(grid_variable["parameter_name"])
+ "\n"
)
# Add condition check
code_string += (
indent * depth
+ "if not {}:".format(grid_variable["condition"])
+ "\n"
)
# Add condition failed action:
code_string += (
indent * (depth + 1)
+ 'print("Condition for {} not met!")'.format(
grid_variable["parameter_name"]
)
+ "\n"
)
code_string += indent * (depth + 1) + "raise ValueError" + "\n"
# Add some whiteline
code_string += indent * (depth + 1) + "\n"
#########################
# Setting up the forloop
# Add comment for forloop
code_string += (
indent * depth
+ "# for loop for {}".format(grid_variable["parameter_name"])
+ "\n"
)
code_string += (
indent * depth
+ "sampled_values_{} = {}".format(
grid_variable["name"], grid_variable["spacingfunc"]
)
+ "\n"
)
# TODO: Make clear that the phasevol only works good if you sample linearly in that thing.
code_string += (
indent * depth
+ "phasevol_{} = sampled_values_{}[1]-sampled_values_{}[0]".format(
grid_variable["name"], grid_variable["name"], grid_variable["name"]
)
+ "\n"
)
# # Some print statement
# code_string += (
# indent * depth
# + "print('phasevol_{}:', phasevol_{})".format(grid_variable["name"], grid_variable["name"])
# + "\n"
# )
# Adding for loop structure
code_string += (
indent * depth
+ "for {} in sampled_values_{}:".format(
grid_variable["name"], grid_variable["name"]
)
+ "\n"
)
#########################
# Setting up pre-code and value in some cases
# Add pre-code
if grid_variable["precode"]:
code_string += (
indent * (depth + 1)
+ "{}".format(
grid_variable["precode"].replace("\n", "\n" + indent * (depth))
)
+ "\n"
)
# Set phasevol
code_string += indent * (depth + 1) + "phasevol *= phasevol_{}\n".format(
grid_variable["name"],
)
#######################
# Probabilities
# Calculate probability
code_string += indent * (depth + 1) + "\n"
code_string += indent * (depth + 1) + "# Setting probabilities\n"
code_string += (
indent * (depth + 1)
+ "d{} = phasevol_{} * {}".format(
grid_variable["name"],
grid_variable["name"],
grid_variable["probdist"],
)
+ "\n"
)
# Saving probability sum
code_string += (
indent * (depth + 1)
+ "probabilities_sum[{}] += d{}".format(
grid_variable["grid_variable_number"], grid_variable["name"]
)
+ "\n"
)
if grid_variable["grid_variable_number"] == 0:
code_string += (
indent * (depth + 1)
+ "probabilities_list[0] = d{}".format(grid_variable["name"])
+ "\n"
)
else:
code_string += (
indent * (depth + 1)
+ "probabilities_list[{}] = probabilities_list[{}] * d{}".format(
grid_variable["grid_variable_number"],
grid_variable["grid_variable_number"] - 1,
grid_variable["name"],
)
+ "\n"
)
#######################
# Increment starcount for this parameter
code_string += "\n"
code_string += indent * (
depth + 1
) + "# Increment starcount for {}\n".format(grid_variable["parameter_name"])
code_string += (
indent * (depth + 1)
+ "starcounts[{}] += 1".format(grid_variable["grid_variable_number"],)
+ "\n"
)
# Add value to dict
code_string += (
indent * (depth + 1)
+ 'parameter_dict["{}"] = {}'.format(
grid_variable["parameter_name"], grid_variable["parameter_name"]
)
+ "\n"
)
# Add some space
code_string += "\n"
# The final parts of the code, where things are returned, are within the deepest loop,
# but in some cases code from a higher loop needs to go under it again
# SO I think its better to put an ifstatement here that checks whether this is the last loop.
if loopnr == len(self.grid_options["grid_variables"]) - 1:
#################################################################################
# Here are the calls to the queuing or other solution. this part is for every system
# Add comment
code_string += indent * (depth + 1) + "#" * 40 + "\n"
code_string += (
indent * (depth + 1)
+ "# Code below will get evaluated for every generated system\n"
)
# Calculate value
code_string += (
indent * (depth + 1)
+ 'probability = self.grid_options["weight"] * probabilities_list[{}]'.format(
grid_variable["grid_variable_number"]
)
+ "\n"
)
code_string += (
indent * (depth + 1)
+ 'repeat_probability = probability / self.grid_options["repeat"]'
+ "\n"
)
code_string += indent * (depth + 1) + "total_starcount += 1\n"
# set probability and phasevol values
code_string += (
indent * (depth + 1)
+ 'parameter_dict["{}"] = {}'.format("probability", "probability")
+ "\n"
)
code_string += (
indent * (depth + 1)
+ 'parameter_dict["{}"] = {}'.format("phasevol", "phasevol")
+ "\n"
)
# Some prints. will be removed
# code_string += indent * (depth + 1) + "print(probabilities)\n"
# code_string += (
# indent * (depth + 1) + 'print("total_starcount: ", total_starcount)\n'
# )
# code_string += indent * (depth + 1) + "print(probability)\n"
# Increment total probability
code_string += (
indent * (depth + 1) + "self.increment_probtot(probability)\n"
)
if not dry_run:
# Handling of what is returned, or what is not.
# TODO: think of whether this is a good method
code_string += indent * (depth + 1) + "yield(parameter_dict)\n"
# The below solution might be a good one to add things to specific queues
# $self->queue_evolution_code_run($self->{_flexigrid}->{thread_q},
# $system);
# If its a dry run, dont do anything with it
else:
code_string += indent * (depth + 1) + "pass\n"
code_string += indent * (depth + 1) + "#" * 40 + "\n"
# increment depth
depth += 1
depth -= 1
code_string += "\n"
# Write parts to write below the part that yield the results. this has to go in a reverse order:
# Here comes the stuff that is put after the deepest nested part that calls returns stuff.
for loopnr, el in enumerate(
sorted(
self.grid_options["grid_variables"].items(),
key=lambda x: x[1]["grid_variable_number"],
reverse=True,
)
):
grid_variable = el[1]
code_string += indent * (depth + 1) + "#" * 40 + "\n"
code_string += (
indent * (depth + 1)
+ "# Code below is for finalising the handling of this iteration of the parameter\n"
)
# Set phasevol
# TODO: fix. this isnt supposed to be the value that we give it here. discuss
code_string += indent * (depth + 1) + "phasevol /= phasevol_{}\n".format(
grid_variable["name"]
)
code_string += indent * (depth + 1) + "\n"
depth -= 1
################
# Finalising print statements
#
# code_string += indent * (depth + 1) + "\n"
code_string += indent * (depth + 1) + "#" * 40 + "\n"
code_string += (
indent * (depth + 1)
+ "print('Grid has handled {} stars'.format(total_starcount))\n"
)
code_string += (
indent * (depth + 1)
+ "print('with a total probability of {}'.format(self.grid_options['probtot']))\n"
)
if dry_run:
code_string += indent * (depth + 1) + "return total_starcount\n"
#################################################################################
# Stop of code generation. Here the code is saved and written
# Save the gridcode to the grid_options
if self.grid_options["verbose"] > 0:
print("Saving grid code to grid_options")
self.grid_options["code_string"] = code_string
# Write to file
gridcode_filename = os.path.join(
self.grid_options["tmp_dir"], "example_grid.py"
)
self.grid_options["gridcode_filename"] = gridcode_filename
if self.grid_options["verbose"] > 0:
print("Writing grid code to {}".format(gridcode_filename))
with open(gridcode_filename, "w") as f:
f.write(code_string)
def load_grid_function(self):
"""
Test function to run grid stuff. mostly to test the import
"""
# Code to load the
import importlib.util
if self.grid_options["verbose"] > 0:
print(
"Loading grid code function from {}".format(
self.grid_options["gridcode_filename"]
)
)
spec = importlib.util.spec_from_file_location(
"binary_c_python_grid",
os.path.join(self.grid_options["gridcode_filename"]),
)
grid_file = importlib.util.module_from_spec(spec)
spec.loader.exec_module(grid_file)
generator = grid_file.grid_code
self.grid_options["system_generator"] = generator
if self.grid_options["verbose"] > 0:
print("Grid code loaded")
def dry_run(self):
"""
Function to dry run the grid and know how many stars it will run
Requires the grid to be built as a dry run grid
"""
system_generator = self.grid_options["system_generator"]
total_starcount = system_generator(self)
self.grid_options["total_starcount"] = total_starcount
def print_info(self, run_number, total_systems, full_system_dict):
"""
Function to print info about the current system and the progress of the grid.
# color info tricks from https://ozzmaker.com/add-colour-to-text-in-python/
https://stackoverflow.com/questions/287871/how-to-print-colored-text-in-terminal-in-python
"""
# Define frequency
if self.grid_options["verbose"] == 1:
print_freq = 1
else:
print_freq = 10
# Calculate amount of time left
# calculate amount of time passed
time_passed = time.time() - self.grid_options["start_time_evolution"]
if run_number % print_freq == 0:
binary_cmdline_string = self.return_argline(full_system_dict)
info_string = "{color_part_1} {text_part_1}{end_part_1}{color_part_2} {text_part_2}{end_part_2}".format(
color_part_1="\033[1;32;41m",
text_part_1="{}/{}".format(run_number, total_systems),
end_part_1="\033[0m",
color_part_2="\033[1;32;42m",
text_part_2="{}".format(binary_cmdline_string),
end_part_2="\033[0m",
)
print(info_string)
###################################################
# Montecarlo functions
#
# Functions below are used to run populations with
# Monte carlo
###################################################
###################################################
# Population from file functions
#
# Functions below are used to run populations from
# a file containing binary_c calls
###################################################
###################################################
# Unordered functions
#
# Functions that arent ordered yet
###################################################
def write_binary_c_calls_to_file(
self, output_dir=None, output_filename=None, include_defaults=False
):
"""
Function that loops over the gridcode and writes the generated parameters to a file. In the form of a commandline call
Only useful when you have a variable grid as system_generator. MC wouldnt be that useful
Also, make sure that in this export there are the basic parameters like m1,m2,sep, orb-per, ecc, probability etc.
On default this will write to the datadir, if it exists
# warning; dont use yet. not fully tested.
"""
if self.grid_options["system_generator"]:
# Check if there is an output dir configured
if self.custom_options.get("data_dir", None):
binary_c_calls_output_dir = self.custom_options["data_dir"]
# otherwise check if theres one passed to the function
else:
if not output_dir:
print(
"Error. No data_dir configured and you gave no output_dir. Aborting"
)
raise ValueError
else:
binary_c_calls_output_dir = output_dir
# check if theres a filename passed to the function
if output_filename:
binary_c_calls_filename = output_filename
# otherwise use default value
else:
binary_c_calls_filename = "binary_c_calls.txt"
binary_c_calls_full_filename = os.path.join(
binary_c_calls_output_dir, binary_c_calls_filename
)
print("Writing binary_c calls to {}".format(binary_c_calls_full_filename))
# Write to file
with open(binary_c_calls_full_filename, "w") as f:
# Get defaults and clean them, then overwrite them with the set values.
if include_defaults:
# TODO: make sure that the defaults here are cleaned up properly
cleaned_up_defaults = self.cleaned_up_defaults()
full_system_dict = cleaned_up_defaults.copy()
full_system_dict.update(self.bse_options.copy())
else:
full_system_dict = self.bse_options.copy()
for system in self.grid_options["system_generator"]:
# update values with current system values
full_system_dict.update(system)
binary_cmdline_string = self.return_argline(full_system_dict)
f.write(binary_cmdline_string + "\n")
else:
print("Error. No grid function found!")
raise KeyError
def cleanup_defaults(self):
"""
Function to clean up the default values:
from a dictionary, removes the entries that have the following values:
- "NULL"
- ""
- "Function"
Uses the function from utils.functions
"""
binary_c_defaults = self.return_binary_c_defaults().copy()
cleaned_dict = filter_arg_dict(binary_c_defaults)
return cleaned_dict
def clean_up_custom_logging(self, evol_type):
"""
Function to clean up the custom logging.
Has two types:
'single':
- removes the compiled shared library (which name is stored in grid_options['custom_logging_shared_library_file'])
- TODO: unloads/frees the memory allocated to that shared library (which is stored in grid_options['custom_logging_func_memaddr'])
- sets both to None
'multiple':
- TODO: make this and design this
"""
if evol_type == "single":
if self.grid_options["verbose"] > 0:
print("Cleaning up the custom logging stuff. type: single")
# TODO: Unset custom logging code
# TODO: Unset function memory adress
# print(self.grid_options["custom_logging_func_memaddr"])
# remove shared library files
if self.grid_options["custom_logging_shared_library_file"]:
remove_file(
self.grid_options["custom_logging_shared_library_file"],
self.grid_options["verbose"],
)
if evol_type == "population":
if self.grid_options["verbose"] > 0:
print("Cleaning up the custom logging stuffs. type: population")
# TODO: make sure that these also work. not fully sure if necessary tho. whether its a single file, or a dict of files/memaddresses
def increment_probtot(self, prob):
"""
Function to add to the total probability
"""
self.grid_options["probtot"] += prob
def increment_count(self):
"""
Function to add to the total amount of stars
"""
self.grid_options["count"] += 1
# def join_result_dicts(self):
# """
# Function to join the result dictionaries
# """
################################################################################################