Skip to content

basic_optimalization

Classes¤

BaseOptimalization ¤

BaseOptimalization(
    working_dir="",
    distribution_file="generator.in",
    run_file="photo_track.in",
    log_dir="log",
    log_file_name="log.log",
    console_log=True,
    do_final_run=True,
    n_parallel=5,
)

Bases: OptimalizationCore

Parameters:

Name Type Description Default
working_dir str | Path

Name of the working directory (can be relative path). Defaults to "".

 ''
distribution_file str | Path

Name of the ASTRA distribution file. Defaults to "generator.in".

 'generator.in'
run_file str | Path

Name of the ASTRA run file. Defaults to "photo_track.in".

 'photo_track.in'
log_dir str | Path

Name of the log directory, that will be created inside the working directory (can be relative path). Defaults to "log".

 'log'
log_file_name str

Name of the log file. Defaults to "log.log".

 'log.log'
console_log bool

If False, the log will not be written to the console. Defaults to False.

 True
logger_name str | None

Name of the logger for possible logger sharing. If set to None, new logger will be created. Defaults to None.

 required
do_final_run (bool, True)

If True, one last run with the optimised settings will be performed for the analysis.

 True
n_parallel (int, 5)

Number of parallel ASTRA simulations to be run. Typically limited by the CPU power. Test how many simulations your computer can run at a same time.

 5
Source code in optimalization/basic_optimalization.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, working_dir:str="", distribution_file:str="generator.in", run_file:str="photo_track.in", log_dir:str="log", log_file_name:str="log.log", console_log:bool=True, do_final_run:bool=True, n_parallel:int = 5) -> None:
    """Basic optimization script designed to optimize a set of parameters in a linear matter: one parameter is optimized at a time and the optimal value is used for the following optimization scans. 

    Args:
        working_dir (str | pathlib.Path, optional): Name of the working directory (can be relative path). Defaults to "".
        distribution_file (str | pathlib.Path, optional): Name of the ASTRA distribution file. Defaults to "generator.in".
        run_file (str | pathlib.Path, optional): Name of the ASTRA run file. Defaults to "photo_track.in".
        log_dir (str | pathlib.Path, optional): Name of the log directory, that will be created inside the working directory (can be relative path). Defaults to "log".
        log_file_name (str, optional): Name of the log file. Defaults to "log.log".
        console_log (bool, optional): If False, the log will not be written to the console. Defaults to False.
        logger_name (str | None, optional): Name of the logger for possible logger sharing. If set to None, new logger will be created. Defaults to None.
        do_final_run (bool, True, optional): If True, one last run with the optimised settings will be performed for the analysis.
        n_parallel (int, 5, optional): Number of parallel ASTRA simulations to be run. Typically limited by the CPU power. Test how many simulations your computer can run at a same time.
    """
    super().__init__(working_dir, distribution_file, run_file, log_dir, log_file_name, console_log)

    self.scanning_parameters = {}
    self.optimalization_parameter_weights = {}

    self.do_final_run = do_final_run
    self.final_dir = self.working_dir.joinpath("Final")

    self.n_parallel: int = n_parallel

Functions¤

_set_optimal_parameter_value ¤
_set_optimal_parameter_value(scanning_parameter, optimal)

Set the parameter to given optimal value.

Parameters:

Name Type Description Default
scanning_parameter str

Parameter to be changed to the optimal value.

 required
optimal (str, int)

Optimal value to be set.

 required
Source code in optimalization/basic_optimalization.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def _set_optimal_parameter_value(self, scanning_parameter:str, optimal) -> None:
    """Set the parameter to given optimal value.

    Args:
        scanning_parameter (str): Parameter to be changed to the optimal value.
        optimal (str, int): Optimal value to be set.
    """
    # if only scan, does not change anything
    if self.scanning_parameters[scanning_parameter]["only_scan"]:
        return
    # set to dictionary
    self.scanning_parameters[scanning_parameter]["optimal"] = optimal
    # change in the base dir -> it will be used in following scans
    self.base_dir_process.change_parameter(scanning_parameter, optimal)
    self.logger.info(f"Optimal value of {scanning_parameter} was set to {optimal} (also in the BASE dir) and will be used in following runs.")
add_optimalization_parameter ¤
add_optimalization_parameter(
    optimalization_parameter, weight
)

Add parameter which has to be minimized (maximised if weight is set to negative).

Parameters:

Name Type Description Default
optimalization_parameter str

Name of the parameter.

 required
weight float

Weight of the parameter (higher value, higher impact on the optimalization).

 required
Source code in optimalization/basic_optimalization.py 
90
91
92
93
94
95
96
97
98
def add_optimalization_parameter(self, optimalization_parameter:str, weight:float) -> None:
    """Add parameter which has to be minimized (maximised if weight is set to negative).

    Args:
        optimalization_parameter (str): Name of the parameter.
        weight (float): Weight of the parameter (higher value, higher impact on the optimalization).
    """
    self.optimalization_parameter_weights[optimalization_parameter] = weight
    self.logger.info(f"Optimalization parameter {optimalization_parameter} was added with {weight}.")
add_scanning_parameter ¤
add_scanning_parameter(
    scanning_parameter,
    min_value,
    max_value,
    step=None,
    n_steps=None,
    only_scan=False,
)

Method to add a parameter to scan.

Parameters:

Name Type Description Default
scanning_parameter str

Name of the scanning parameter

 required
min_value (str, int)

Starting value of the scan.

 required
max_value (str, int)

Final value of the scan.

 required
step (str, int)

Size of the steps. Defaults to None. If not specified, the number of steps (n_steps) argument is taken.

 None
n_steps (str, int)

Number of steps to be taken. Defaults to None. If not specified, the step size (step) argument is taken.

 None
only_scan bool

If True, only scan is performed and the outcome is not used to optimalization. Defaults to False.

 False

Raises:

Type Description
Exception

Raised if neither number of steps (n_step) nor step size (step) is given.
Source code in optimalization/basic_optimalization.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def add_scanning_parameter(self, scanning_parameter:str, min_value:str, max_value:str, step=None, n_steps=None, only_scan=False) -> None:
    """Method to add a parameter to scan.

    Args:
        scanning_parameter (str): Name of the scanning parameter
        min_value (str, int): Starting value of the scan.
        max_value (str, int): Final value of the scan.
        step (str, int, optional): Size of the steps. Defaults to None. If not specified, the number of steps (n_steps) argument is taken.
        n_steps (str, int, optional): Number of steps to be taken. Defaults to None. If not specified, the step size (step) argument is taken.
        only_scan (bool, optional): If True, only scan is performed and the outcome is not used to optimalization. Defaults to False.

    Raises:
        Exception: Raised if neither number of steps (n_step) nor step size (step) is given.
    """
    # initialize step size and number of steps
    if step is None and n_steps is None:
        raise Exception(f"Neither step nor number of steps of {scanning_parameter} scanning parameter set. Unable to perform the scan.")
    if step is not None:
        n_steps = int((max_value - min_value)//step) + 1
    else:
        step = (Decimal(str(max_value)) - Decimal(str(min_value)))/Decimal(str(n_steps -1))

    # add parameter setting to the directory and saved to later used
    self.scanning_parameters[scanning_parameter] = {
        "scanning_parameter" : str(scanning_parameter).strip(),
        "min_value" : float(min_value),
        "max_value" : float(max_value),
        "step" : float(step),
        "n_steps": int(n_steps),
        "optimal": None,
        "only_scan": only_scan
    }
    self.logger.info(f"Parameter {scanning_parameter} will be scanned from {min_value} to {max_value} with step {step} (total steps {n_steps}).")
    if only_scan:
        self.logger.info(f"Parameter {scanning_parameter} will be only scanned, not optimalized.")
find_optimal_value ¤
find_optimal_value(scanning_parameter, scanning_dir)

Method to find and optimal value of the scanning_parameter.

Parameters:

Name Type Description Default
scanning_parameter str

Name of the parameter.

 required
scanning_dir Path

Location of the performed scan.

 required

Returns:

Name Type Description
float float

Optimal value of the scanning_parameter.
Source code in optimalization/basic_optimalization.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def find_optimal_value(self, scanning_parameter:str, scanning_dir: pathlib.Path) -> float:
    """Method to find and optimal value of the scanning_parameter.

    Args:
        scanning_parameter (str): Name of the parameter.
        scanning_dir (pathlib.Path): Location of the performed scan.

    Returns:
        float: Optimal value of the scanning_parameter.
    """
    # load the data
    loader = plot_properties.PlotProperties(scanning_parameter, 
                                            working_dir=scanning_dir, console_log=False,
                                            )
    df = loader.load_data()
    # create pandas.Series of weighted values of the parameters to be minimized
    df["weighted"] = 0
    for param in self.optimalization_parameter_weights.keys():
        df["weighted"] += df[param]*self.optimalization_parameter_weights[param]
    # apply ban for losing particles
    df["weighted"] += (1-df["Active particle ratio"])*1e10
    return df["scanning_parameter"][df["weighted"].idxmin()]
optimize ¤
optimize()

Method to run the optimalization.

Source code in optimalization/basic_optimalization.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def optimize(self) -> None:
    """Method to run the optimalization.
    """
    # runs scan for each saved parameter
    for scanning_parameter in self.scanning_parameters.keys():
        self.logger.info(f"Running scan for parameter {scanning_parameter}.")
        self.run_scan_of_parameter(scanning_parameter)

    self.logger.info("Optimization complete.")
    # log optimal values of optimized parameters
    for param in self.scanning_parameters:
        if not self.scanning_parameters[scanning_parameter]["only_scan"]:
            self.logger.info(f"Optimal value of parameter {param}: {self.scanning_parameters[param]['optimal']}")
    self.logger.info(f"Result of scan to be found in {self.working_dir}")

    #run the final run with optimized parameters
    if self.do_final_run:
        self.logger.info(f"Running final run with optimised parametres.")
        self.run_final()
        self.save_final_run_characteristics()
run_final ¤
run_final()

Method to run the final scan.

Source code in optimalization/basic_optimalization.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def run_final(self) -> None:
    """Method to run the final scan.
    """
    # prepare directory
    shutil.copytree(self.base_dir, self.final_dir)

    # run ASTRA
    runner = Runner({"RUN": "1"}, working_dir=self.final_dir, distribution_file=self.distribution_file,
                    run_file=self.run_file, run_number=1, renaming_input_files=False)
    runner.run()

    # plot phase space
    plotter = plot_phase_space.PlotPhaseSpace(1, working_dir=self.final_dir)
    plotter.plot()

    self.logger.info(f"Final run with optimized parameters was finished in {self.final_dir}.")
run_scan_of_parameter ¤
run_scan_of_parameter(scanning_parameter)

Method to run the scan of parameter. Returns the optimal value.

Parameters:

Name Type Description Default
scanning_parameter str

Name of the parameter to be scanned and optimized

 required

Returns:

Name Type Description
float float

optimal value of the scanning_parameter
Source code in optimalization/basic_optimalization.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def run_scan_of_parameter(self, scanning_parameter:str) -> float:
    """Method to run the scan of parameter. Returns the optimal value.

    Args:
        scanning_parameter (str): Name of the parameter to be scanned and optimized

    Returns:
        float: optimal value of the scanning_parameter
    """
    scanning_dir = self.make_new_scan_dir(scanning_parameter)

    # initialize paralel scan
    scan = parallel_scan.ParallelScan(
        scanning_parameter, 
        self.scanning_parameters[scanning_parameter]["min_value"],
        self.scanning_parameters[scanning_parameter]["max_value"],
        self.scanning_parameters[scanning_parameter]["step"],
        working_dir=scanning_dir,
        n_parallel=self.n_parallel,
        console_log=False
        )
    scan.run_scan()
    scan.plot()

    # get optimal value and set it 
    optimal_value = self.find_optimal_value(scanning_parameter, scanning_dir)
    self.logger.info(f"Optimal value of {scanning_parameter} was found at {optimal_value} (step size: {self.scanning_parameters[scanning_parameter]['step']}).")
    self._set_optimal_parameter_value(scanning_parameter, optimal_value)
    return optimal_value
save_final_run_characteristics ¤
save_final_run_characteristics()

Method to save the final run charateristics.

Source code in optimalization/basic_optimalization.py 
191
192
193
194
195
def save_final_run_characteristics(self) -> None:
    """Method to save the final run charateristics.
    """
    characteristics = run_characteristics.RunCharacteristics(working_dir=self.final_dir, logger_name=self.logger.name)
    characteristics.save()
Last update: October 31, 2023
Created: October 31, 2023