Skip to content

plot_properties

Classes¤

PlotProperties ¤

PlotProperties(
    parameter=None,
    zpos="end",
    working_dir="",
    distribution_file="generator.in",
    run_file="photo_track.in",
    log_dir="log",
    log_file_name="plots_log.log",
    console_log=True,
    plots_dir="plots",
    max_value=None,
    min_value=None,
    scanning_parameter_unit=None,
    min_run_number=None,
    max_run_number=None,
    parameter_scaling=1,
    logger_name=None,
)

Bases: PlotCore

Source code in plot_analysis/plot_properties.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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
def __init__(self, parameter=None, zpos="end", working_dir="", 
             distribution_file="generator.in", run_file="photo_track.in", 
             log_dir="log", log_file_name="plots_log.log", console_log=True, 
             plots_dir="plots", max_value=None, min_value=None, 
             scanning_parameter_unit=None, min_run_number=None, max_run_number=None, parameter_scaling=1,
             logger_name=None) -> None:
    super().__init__(working_dir, distribution_file, run_file, log_dir, log_file_name, 
                     console_log, plots_dir, logger_name=logger_name)

    if type(zpos) is int:
        zpos = zpos.lower().strip()
        if zpos != "end":
            print("Zpos argument not valid!")
            exit(1)
    # Check if numerical zpos is valid
    self.zpos = zpos
    self.logger.info(f"Plotting data at z position: {self.zpos}")

    if parameter is None:
        self.scanning_parameter = self.output.columns[1].strip()
    else:
        self.scanning_parameter = parameter.strip() 
    # Set scanning parameter unit
    self.scanning_parameter_unit = scanning_parameter_unit # set parameter unit to be displayed in the plots
    self.parameter_scaling = float(parameter_scaling) # scaling of parameter to possibly transform to other unit

    self.logger.info(f"Plotting dependance on parameter {self.scanning_parameter}")
    self.output = self.output[self.output[self.scanning_parameter].notna()]

    # If set to None, no condition is aplied
    self.min_value = float(min_value) if str(min_value).strip().lower() != "none" else None
    self.logger.info(f"Aplaying value lower limit of {self.min_value}" if self.min_value is not None else "No lower limit for parameter value set.")
    self.max_value = float(max_value) if str(max_value).strip().lower() != "none" else None
    self.logger.info(f"Aplying value upper limit of {self.max_value}" if self.max_value is not None else "No lower limit for parameter value set.")
    self.min_run_number = int(min_run_number) if str(min_run_number).strip().lower() != "none" else None
    self.logger.info(f"Considering only runs with run number larger than {self.min_run_number}" if self.min_run_number is not None else "No lower limit for run number value set.")
    self.max_run_number = int(max_run_number) if str(max_run_number).strip().lower() != "none" else None
    self.logger.info(f"Considering only runs with run number smaller than {self.max_run_number}" if self.max_run_number is not None else "No upper limit for run number value set.")

    # Aply conditions
    self.output = self.output[self.output["Run number"] >= self.min_run_number] if self.min_run_number is not None else self.output
    self.output = self.output[self.output["Run number"] <= self.max_run_number] if self.max_run_number is not None else self.output
    self.output = self.output[self.output[self.scanning_parameter] >= self.min_value] if self.min_value is not None else self.output
    self.output = self.output[self.output[self.scanning_parameter] <= self.max_value] if self.max_value is not None else self.output

    # check for duplicates and keep the first occurance only
    if not self.output["Run number"].is_unique:
        self.logger.warning(f"There are some duplicated run numbers in the output file!")
        self.logger.info(f"Keeping only the first occurances of each run number.")
        self.output = self.output[self.output["Run number"].duplicated()]
    # sort by the run number for possible printing
    self.output = self.output.sort_values("Run number")

    self.logger.info(f"Total number of {self.output.shape[0]} satisfy the conditions.")

    self.data = self.load_data()
    self.save_data()

Functions¤

load_data ¤
load_data()

Method to extract bunch parameters from a performed scan and return them in a format of pandas.DataFrame.

Returns:

Name Type Description
df DataFrame

Dataframe containing bunch parameters obtained in different runs. Columns:

run: run number,

scanning_parameter: value of the scanning parameter

energy: energy

energy_spread: rms energy spread

x_emit: rms normalized emittance in the transverse x plane

y_emit: rms normalized emittance in the transverse y plane

x_rms: rms bunch size in the transverse x plane

y_rms: rms bunch size in the transverse y plane

xBar_rms: rms beam divergence in the x plane

yBar_rms: rms beam divergence in the y plane

Active particle ratio: ratio of active (not lost) particles, if 1, no particles were lost

Beam size (z_rms): rms bunch length
Source code in plot_analysis/plot_properties.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def load_data(self):
    """Method to extract bunch parameters from a performed scan and return them in a format of `pandas.DataFrame`.

    Returns:
        df (pandas.DataFrame): Dataframe containing bunch parameters obtained in different runs. Columns:\n
            `run`: run number, 

            `scanning_parameter`: value of the scanning parameter

            `energy`: energy

            `energy_spread`: rms energy spread

            `x_emit`: rms normalized emittance in the transverse x plane

            `y_emit`: rms normalized emittance in the transverse y plane

            `x_rms`: rms bunch size in the transverse x plane

            `y_rms`: rms bunch size in the transverse y plane

            `xBar_rms`: rms beam divergence in the x plane

            `yBar_rms`: rms beam divergence in the y plane

            `Active particle ratio`: ratio of active (not lost) particles,  if 1, no particles were lost

            `Beam size (z_rms)`: rms bunch length 
    """
    # initialize properties to be plotted
    energy = []
    energy_spread = []
    x_emit = []
    y_emit = []
    x_rms = [] # beam size
    y_rms = [] # beam size
    xBar_rms = [] # beam divergence
    yBar_rms = [] # beam divergence
    active_ratio = [] # ratio of active particles at the end (not lost)
    beam_size = []
    # the following ones are to check what runs and corresponding scanning parameter were included
    included_runs=[]
    scanning_parameter_include=[]

    def get_from(from_df, property_name):
        if self.zpos == "end":
            return float(from_df.tail(1)[property_name])
        else: 
            return from_df.loc[from_df[property_name] == self.zpos]

    self.logger.info("Loading data for plots!")
    # fill properties lists
    for run in self.output["Run number"].to_numpy():
        run = int(run)
        # check if all neccessary file exist
        if not self.output_files_exists(run, Xemit=True, Yemit=True, Zemit=True, TRemit=False):
            self.logger.warn(f"Output files missing for {run}! Skipping.")
            scanning_parameter_include+=[False]
            continue

        included_runs += [run] 
        scanning_parameter_include+=[True]

        energy += [get_from(self.get_Zemit(run), "E")]
        energy_spread += [get_from(self.get_Zemit(run), "E_rms")]
        x_emit += [get_from(self.get_Xemit(run), "epsilon")]
        y_emit += [get_from(self.get_Yemit(run), "epsilon")]
        x_rms += [get_from(self.get_Xemit(run), "x_rms")]
        y_rms += [get_from(self.get_Yemit(run), "y_rms")]
        xBar_rms += [get_from(self.get_Xemit(run), "xBar_rms")]
        yBar_rms += [get_from(self.get_Yemit(run), "yBar_rms")]
        beam_size += [get_from(self.get_Zemit(run), "z_rms")]
        # calculate lost particles
        df = self.get_at_zpos(run, self.zpos)
        total = df.shape[0]
        active = df.loc[df["flag"] > 0].shape[0]
        active_ratio += [active/total]

        self.logger.info(f"Included run number {run}")

    # returnad panda Dataframe
    return pandas.DataFrame({
        "run": included_runs,
        "scanning_parameter": (self.output[self.scanning_parameter].to_numpy()*self.parameter_scaling)[scanning_parameter_include],
        "energy": energy,
        "energy_spread": energy_spread,
        "x_emit": x_emit,
        "y_emit": y_emit,
        "x_rms": x_rms,
        "y_rms": y_rms,
        "xBar_rms": xBar_rms,
        "yBar_rms": yBar_rms,
        "Active particle ratio": active_ratio,
        "Beam size (z_rms)": beam_size,
    })
Last update: October 31, 2023
Created: October 31, 2023