chain_core

Classes

ChainCore

ChainCore(
    working_dir="",
    distribution_file="generator.in",
    run_file="photo_track.in",
    log_dir="log",
    log_file_name="log.log",
    console_log=False,
    logger_name=None,
    n_parallel=5,
    chain_output_file="chain_output_file.xlsx",
)

Bases: Core

Parameters:

Name	Type	Description	Default
working_dir	str | Path	Working dir in which to perform the scan. Defaults to "".	''
distribution_file	str | Path	Name of the distribution file. Defaults to "generator.in".	'generator.in'
run_file	str | Path	Name of the run file. Defaults to "photo_track.in".	'photo_track.in'
log_dir	str | Path	Path to the log directory. Defaults to "log".	'log'
log_file_name	str | Path	Name of the log file. Defaults to "log.log".	'log.log'
console_log	bool	Option if to log to console. Defaults to False.	False
logger_name	str | Path	Name of the logger. If set to None, new logger will be created. Defaults to None.	None
n_parallel	int	Number of parallel runs. Defaults to 5.	5
chain_output_file	str | Path	Name of the chain output file. Defaults to "chain_output_file.xlsx".	'chain_output_file.xlsx'

Source code in common/chain_core.py

def __init__(self, working_dir="", distribution_file="generator.in", run_file="photo_track.in", log_dir="log", log_file_name="log.log", console_log=False, logger_name=None, n_parallel = 5, chain_output_file="chain_output_file.xlsx") -> None:
    """Core class for chains scans.

    Args:
        working_dir (str | pathlib.Path, optional): Working dir in which to perform the scan. Defaults to "".
        distribution_file (str | pathlib.Path, optional): Name of the distribution file. Defaults to "generator.in".
        run_file (str | pathlib.Path, optional): Name of the run file. Defaults to "photo_track.in".
        log_dir (str | pathlib.Path, optional): Path to the log directory. Defaults to "log".
        log_file_name (str | pathlib.Path, optional): Name of the log file. Defaults to "log.log".
        console_log (bool, optional): Option if to log to console. Defaults to False.
        logger_name (str | pathlib.Path, optional): Name of the logger. If set to None, new logger will be created. Defaults to None.
        n_parallel (int, optional): Number of parallel runs. Defaults to 5.
        chain_output_file (str | pathlib.Path, optional): Name of the chain output file. Defaults to "chain_output_file.xlsx".
    """
    super().__init__(working_dir, distribution_file, run_file, log_dir, log_file_name, console_log, logger_name)

    self.base_dir = self.working_dir.joinpath("BASE")
    self.check_base_directory()
    self.n_parallel = n_parallel
    self.layers = []
    self.main_layers = []
    self.layer_idxs = []

    self.chain_output_file = self.working_dir.joinpath(chain_output_file)
    self.chain_output_file_lock = self.acquire_fileLock(self.chain_output_file)

Functions

_check_correct_number_of_steps_in_layer

_check_correct_number_of_steps_in_layer(layer_idx)

Method to check if there is the same number of steps in all layers with the same layer index.

Parameters:

Name	Type	Description	Default
layer_idx	int	index of the layers	required

Returns:

Name	Type	Description
bool	bool	True if all layers with the same layer index have the same number of steps, else False

Source code in common/chain_core.py

def _check_correct_number_of_steps_in_layer(self, layer_idx:int) -> bool:
    """Method to check if there is the same number of steps in all layers with the same layer index.

    Args:
        layer_idx (int): index of the layers

    Returns:
        bool: True if all layers with the same layer index have the same number of steps, else False
    """
    if len(self.layer_idxs) == 0:
        return True
    same_index_layers = self.get_same_index_layers(layer_idx)
    if len(same_index_layers) == 0:
        return True

    # this checks if there are multiple number of steps
    if len(list(dict.fromkeys( [layer["n_steps"] for layer in same_index_layers]))) != 1:
        return False
    else:
        return True

add_layer

add_layer(
    parameter,
    values=None,
    start=None,
    end=None,
    step=None,
    n_steps=None,
    layer_idx=None,
)

Adds a layer into the chain scan. Layers are stored in the order they were provided or a layer index can be specified. When layer is not specified, new one will be created automatically.

There are multiple options how to define the range of the parameter to be scanned:

1. Full values list (values)
2. Start and end values (start, end) + step size (step)
3. Start and end values (start, end) + number of steps (n_steps)

This list also corresponds to the order at which the definition of the parameter is checked. Once the parameter is fully defined (one of the points above is fulfilled), the rest of the values may be internally overwritten to match the calculated values from the full definition.

Parameters:

Name	Type	Description	Default
parameter	str	Name of the scanning parameter.	required
values	list	Full list of values to be scanned. If specified, defines fully the parameter. Defaults to None.	None
start	float	Starting value of the scan. Will be taken into account only if values argument was not specified. For full definition of the parameter, the final argument along with step or n_steps must be specified. Defaults to None.	None
end	float	Final value of the scan. Will be taken into account only if values argument was not specified. For full definition of the parameter, the start argument along with step or n_steps must be specified. Defaults to None.	None
step	float | list	Step size to be performed. For constant step size, use float type, for varying step sizes, use list of floats. Will be taken into account only if values argument was not specified. For full definition of the parameter, the start and end arguments must be specified. Defaults to None.	None
n_steps	int	Number of constant steps to be performed. Will be taken into account only if values and steps arguments were not specified. For full definition of the parameter, the start and end arguments must be specified.. Defaults to None.	None
layer_idx	int	Index of the layer: to be used when multiple parameters should be included in the given layer (provided that they have the same number of values to be scanned). Defaults to None -> automatically set.	None

Source code in common/chain_core.py

def add_layer(self, parameter:str, values=None, start=None, end=None, step=None, n_steps=None, layer_idx = None) -> None:
    """Adds a layer into the chain scan. Layers are stored in the order they were provided or a layer index can be specified. 
    When layer is not specified, new one will be created automatically.

    There are multiple options how to define the range of the parameter to be scanned:

    1. Full values list (`values`)
    2. Start and end values (`start`, `end`) + step size (`step`)
    3. Start and end values (`start`, `end`) + number of steps (`n_steps`)

    This list also corresponds to the order at which the definition of the parameter is checked. Once the parameter is fully defined (one of the points above is fulfilled), the rest of the values may be internally overwritten to match the calculated values from the full definition. 

    Args:
        parameter (str): Name of the scanning parameter.
        values (list, optional): Full list of values to be scanned. If specified, defines fully the parameter. Defaults to None.
        start (float, optional): Starting value of the scan. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `final` argument along with `step` or `n_steps` must be specified. Defaults to None.
        end (float, optional): Final value of the scan. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `start` argument along with `step` or `n_steps` must be specified. Defaults to None.
        step (float | list, optional): Step size to be performed. For constant step size, use `float` type, for varying step sizes, use list of floats. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `start` and `end` arguments must be specified. Defaults to None.
        n_steps (int, optional): Number of constant steps to be performed. Will be taken into account only if `values` and `steps` arguments were not specified. For full definition of the parameter, the `start` and `end` arguments must be specified.. Defaults to None.
        layer_idx (int, optional): Index of the layer: to be used when multiple parameters should be included in the given layer (provided that they have the same number of values to be scanned). Defaults to None -> automatically set.
    """
    # use parent method for initialisation
    parameter_dictionary = self._parameter_values_initialization(parameter, values, start, end, step, n_steps)
    # make sure the layer index is correctly set
    if layer_idx is None:
        # if not set by the user, set it automatically
        layer_idx = max(self.layer_idxs)+1 if len(self.layer_idxs) != 0 else 0
    parameter_dictionary["layer_idx"] = layer_idx

    self.logger.info(f"Adding layer: {parameter_dictionary['parameter']}, start: {parameter_dictionary['start']}, end: {parameter_dictionary['end']} (number of steps: {parameter_dictionary['n_steps']}), layer index: {layer_idx}.")
    # add it to the set of layers
    self.layers += [parameter_dictionary]

    # check if there is correct number of steps for the given layer
    if not self._check_correct_number_of_steps_in_layer(layer_idx):
        raise ValueError(f"Trying to add parameter {parameter} into layer number {layer_idx} with incorrect number of steps ({parameter_dictionary['n_steps']})!")

    # if this is the first layer with the layer_idx, save it as the main layer
    if layer_idx not in self.layer_idxs:
        self.main_layers += [parameter_dictionary]
    self.layer_idxs.append(layer_idx)

get_same_index_layers

get_same_index_layers(layer_idx)

Method to get all the layers in the layer_idx.

Parameters:

Name	Type	Description	Default
layer_idx	int	index of the layer	required

Returns:

Name	Type	Description
list	list	list of layers (layer is represented by a dictionary)

Source code in common/chain_core.py

def get_same_index_layers(self, layer_idx:int) -> list:
    """Method to get all the layers in the layer_idx.

    Args:
        layer_idx (int): index of the layer

    Returns:
        list: list of layers (layer is represented by a dictionary)
    """
    return [layer for layer in self.layers if layer["layer_idx"] == layer_idx]

write_presets_to_output_file

write_presets_to_output_file(presets)

Method to write the preset values for each parallel scan to an output file for later use.

Parameters:

Name	Type	Description	Default
presets	list	List of dictionaries that define the values in each layer.	required

Source code in common/chain_core.py

def write_presets_to_output_file(self, presets:list)->None:
    """Method to write the preset values for each parallel scan to an output file for later use.

    Args:
        presets (list): List of dictionaries that define the values in each layer.
    """
    if len(presets) == 0:
        return
    # get all defined parameters
    parameters = {}
    for param in presets[0].keys():
        parameters[param] = []

    # fill in the values  
    for preset in presets:
        for key in parameters.keys():
            parameters[key].append(preset[key])

    # create a panda DataFrame for simpler manipulation
    df = pd.DataFrame(parameters)

    # write safely with filelock
    with self.chain_output_file_lock:
        # load possible previous output file and merge
        if self.chain_output_file.is_file():
            dff = pd.read_excel(self.chain_output_file, index_col=0)
            df = pd.concat([df, dff])
            df = df.drop_duplicates(subset=df.columns, keep="last")
        df.to_excel(self.chain_output_file)

---

Last update: October 31, 2023
Created: October 31, 2023