chain_scan

Classes

ChainScan

ChainScan(
    working_dir="",
    distribution_file="generator.in",
    run_file="photo_track.in",
    log_dir="log",
    log_file_name="log.log",
    console_log=True,
    logger_name=None,
    n_parallel=5,
)

Bases: ChainCore

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.	True
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

Source code in astra/chain_scan.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=True, logger_name=None, n_parallel=5) -> None:
    """Class for parallel chain scan.

    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.
    """
    super().__init__(working_dir, distribution_file, run_file, log_dir, log_file_name, console_log, logger_name, n_parallel=n_parallel)

    self.scan_parameters = []

Functions

add_scanning_parameter

add_scanning_parameter(
    parameter,
    values=None,
    start=None,
    end=None,
    step=None,
    n_steps=None,
)

Method to add a scanning parameter for the parallel scan.

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

Source code in astra/chain_scan.py

def add_scanning_parameter(self, parameter:str, values=None, start=None, end=None, step=None, n_steps=None) -> None:
    """Method to add a scanning parameter for the parallel scan.

    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.
    """
    # use parent method for initialisation
    parameter_dictionary = self._parameter_values_initialization(parameter, values, start, end, step, n_steps)

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

run

run()

Method to run the chain parallel scan.

Source code in astra/chain_scan.py

def run(self):
    """Method to run the chain parallel scan.
    """
    # initialize values from different layers
    layer_number = 0
    total_scans = np.prod([len(layer["values"]) for layer in self.main_layers])
    self.logger.info("Total number of " + str(total_scans) + " scans will be performed.")
    simulation_preset_parameters = [{} for scan in range(total_scans)]
    while layer_number < len(self.main_layers):
        n_same_sequence = total_scans//np.prod([layer["n_steps"] for layer in self.main_layers[:layer_number+1]])
        n_filled = 0
        main_layer = self.main_layers[layer_number]
        while n_filled < total_scans:
            for i in range(main_layer["n_steps"]):
                for n in range(n_same_sequence):
                    for layer in self.get_same_index_layers(main_layer["layer_idx"]):
                        simulation_preset_parameters[n_filled+n][layer["parameter"]] = layer["values"][i]
                n_filled += n_same_sequence
        layer_number += 1

    # log the presets
    self.write_presets_to_output_file(simulation_preset_parameters)
    # run the scans
    scans_completed = 0
    for preset in simulation_preset_parameters:
        scan_dir = self.setup_scan_dir(preset)
        ps = parallel_scan.ParallelScan(working_dir=scan_dir,
                                n_parallel=self.n_parallel,
                                console_log=False
                                )
        self.logger.info(f"Running scan {scans_completed+1}/{total_scans}.")
        for parameter in self.scan_parameters:
            ps.add_scanning_parameter(parameter["parameter"], values=parameter["values"])

        ps.run_scan()
        self.logger.info("Scan completed!")
        ps.plot()
        self.logger.info("Plots complete!")
        scans_completed += 1

setup_scan_dir

setup_scan_dir(preset_parameters)

Method to setup the scan dir for the prese parameters. Each layer mean new subdirectory structure.

Parameters:

Name	Type	Description	Default
preset_parameters	dict	Dictionary of parameters from layers.	required

Returns:

Type	Description
Path	pathlib.Path: Path to the scan directory.

Source code in astra/chain_scan.py

def setup_scan_dir(self, preset_parameters:dict) -> pathlib.Path:
    """Method to setup the scan dir for the prese parameters. Each layer mean new subdirectory structure.

    Args:
        preset_parameters (dict): Dictionary of parameters from layers.

    Returns:
        pathlib.Path: Path to the scan directory.
    """
    scan_dir = pathlib.Path(self.working_dir)
    log_string = "Setting up directory for: "
    for parameter in preset_parameters.keys(): 
        value = preset_parameters[parameter]
        log_string += (parameter + " = " + str(value) + "; ")   
        # make sure the parameter is set correctly
        self.base_dir_process.change_parameter(parameter, value)
        # add layer to the directory structure
        scan_dir = scan_dir.joinpath(parameter+"_"+str(value))

    # make sure the parent directory exists for safe copying
    scan_dir.parent.mkdir(parents=True, exist_ok=True)
    # copy the scan directory
    shutil.copytree(self.base_dir, scan_dir, dirs_exist_ok=True)
    self.logger.info(log_string)
    self.logger.info("Location: " + str(scan_dir))
    return scan_dir

---

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