Skip to content

_geometry

deploy_geometry(out_dir, exp, run, ds, det, pixel_size_mm, center, distance, pv_camera_length=None)

Write new geometry files (.geom and .data for CrystFEL and psana).

Should be called with an optimized center and distance.

Parameters:

Name Type Description Default
exp str

Experiment name

required
run int

Run number

required
ds DataSource | MPIDataSource

psana DataSource object.

required
det Detector

psana Detector object.

required
pixel_size_mm float

Detector pixel size in mm.

required
center Tuple[float, float]

Beam center for new geometry.

required
distance float

Detector distance for new geometry.

required
pv_camera_length str | None

PV associated with camera length.

None
Source code in lute/tasks/_geometry.py
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def deploy_geometry(
    out_dir: str,
    exp: str,
    run: int,
    ds: Union[psana.DataSource, psana.MPIDataSource],
    det: psana.Detector,
    pixel_size_mm: float,
    center: Tuple[float, float],
    distance: float,
    pv_camera_length: Optional[str] = None,
) -> None:
    """Write new geometry files (.geom and .data for CrystFEL and psana).

    Should be called with an optimized center and distance.

    Args:
        out_dir (str) Path to output directory.

        exp (str): Experiment name

        run (int): Run number

        ds (psana.DataSource | psana.MPIDataSource): psana DataSource object.

        det (psana.Detector): psana Detector object.

        pixel_size_mm (float): Detector pixel size in mm.

        center (Tuple[float, float]): Beam center for new geometry.

        distance (float): Detector distance for new geometry.

        pv_camera_length (str | None): PV associated with camera length.
    """
    import PSCalib

    # retrieve original geometry
    geom: PSCalib.GeometryAcces.GeometryAccess = det.geometry(run)
    top: PSCalib.GeometryObject.GeometryObject = geom.get_top_geo()
    children: List[PSCalib.GeometryObject.GeometryObject] = top.get_list_of_children()
    child: PSCalib.GeometryObject.GeometryObject = children[0]

    pixel_size_um: float = pixel_size_mm * 1e3

    # determine and deploy shifts in x,y,z
    cy, cx = det.point_indexes(run, pxy_um=(0, 0), fract=True)
    dx = pixel_size_um * (center[0] - cx)  # convert from pixels to microns
    dy = pixel_size_um * (center[1] - cy)  # convert from pixels to microns
    dz = np.mean(-1 * det.coords_z(run)) - 1e3 * distance  # convert from mm to microns
    geom.move_geo(child.oname, 0, dx=-dy, dy=-dx, dz=dz)

    # write optimized geometry files
    psana_file_path: str = f"{out_dir}/r{run:04d}_end.data"
    geom.save_pars_in_file(psana_file_path)

    cfel_file_path: str = f"{out_dir}/r{run:04d}.geom"
    tmp_cfel_file_path: str = f"{cfel_file_path}.tmp"
    generate_geom_file(
        exp,
        run,
        ds,
        det,
        psana_file_path,
        tmp_cfel_file_path,
        distance,
        pv_camera_length,
    )
    modify_crystfel_header(tmp_cfel_file_path, cfel_file_path)
    os.remove(tmp_cfel_file_path)

generate_concentric_sample_pts(peak_radii, center, num_pts=200)

Generate sample points along concentric circles.

Parameters:

Name Type Description Default
peak_radii ndarray[int64]

Radii indices.

required
center List[float]

Center_x, Center_y for the concentric circles.

required
num_pts int

Number of sample points.

200

Returns:

Name Type Description
coords ndarray[float64]

X/Y coordinates of sample points.

Source code in lute/tasks/_geometry.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
def generate_concentric_sample_pts(
    peak_radii: np.ndarray[np.int64],
    center: List[float],
    num_pts: int = 200,
) -> np.ndarray[np.float64]:
    """Generate sample points along concentric circles.

    Args:
        peak_radii (np.ndarray[np.int64]): Radii indices.

        center (List[float]): Center_x, Center_y for the concentric circles.

        num_pts (int): Number of sample points.

    Returns:
        coords (np.ndarray[np.float64]): X/Y coordinates of sample points.
    """
    # X,Y labelling seems backwards
    cx: float = center[0]
    cy: float = center[1]
    # Reshape linear radii (peak indices) for broadcasting
    radii: np.ndarray[np.int64] = np.array([peak_radii]).reshape(-1, 1)
    theta: np.ndarray[np.float64] = np.linspace(0.0, 2 * np.pi, 200)

    coords_x: np.ndarray[np.float64] = radii * np.cos(theta) + cx
    coords_y: np.ndarray[np.float64] = radii * np.sin(theta) + cy

    # Reshape for optimization routines
    coords: np.ndarray[np.float64] = np.zeros((2, num_pts * len(peak_radii)))
    coords[1] = coords_x.reshape(-1)
    coords[0] = coords_y.reshape(-1)

    return coords

generate_geom_file(exp, run, ds, det, input_file, output_file, det_dist=None, pv_camera_length=None)

Generate a Crystfel .geom file from either a psana or CrystFEL geometry.

This function also sets the coffset field for each panel based on the estimated detector distance: coffset [m] = 1e-3 * (distance - clen) Supplying det_dist will override the value pulled from the deployed geometry for this run, which is used to compute the coffset parameter.

Parameters:

Name Type Description Default
exp str

Experiment name

required
run int

Run number

required
ds DataSource | MPIDataSource

psana DataSource object.

required
det Detector

psana Detector object.

required
input_file str

Input .geom or .data file

required
output_file str

Output .geom file

required
det_dist Optional[float]

Estimated sample-detector distance in mm

None
pv_camera_length Optional[str]

PV associated with the camera length

None
Source code in lute/tasks/_geometry.py
 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
def generate_geom_file(
    exp: str,
    run: int,
    ds: Union[psana.DataSource, psana.MPIDataSource],
    det: psana.Detector,
    input_file: str,
    output_file: str,
    det_dist: Optional[float] = None,
    pv_camera_length: Optional[str] = None,
) -> None:
    """Generate a Crystfel .geom file from either a psana or CrystFEL geometry.

    This function also sets the coffset field for each panel based
    on the estimated detector distance:
        coffset [m] = 1e-3 * (distance - clen)
    Supplying det_dist will override the value pulled from the deployed
    geometry for this run, which is used to compute the coffset parameter.

    Args:
        exp (str): Experiment name

        run (int): Run number

        ds (psana.DataSource | psana.MPIDataSource): psana DataSource object.

        det (psana.Detector): psana Detector object.

        input_file (str): Input .geom or .data file

        output_file (str): Output .geom file

        det_dist (Optional[float]): Estimated sample-detector distance in mm

        pv_camera_length (Optional[str]): PV associated with the camera length
    """
    from psgeom import camera, sensors

    if input_file.split(".")[-1] == "data":
        geom = camera.CompoundAreaCamera.from_psana_file(input_file)
    elif input_file.split(".")[-1] == "geom":
        if (
            "epix10k2m" in str(det.name).lower()
            or "epix10ka2m" in str(det.name).lower()
        ):
            geom = camera.CompoundAreaCamera.from_crystfel_file(
                input_file, element_type=sensors.Epix10kaSegment
            )
        else:
            geom = camera.CompoundAreaCamera.from_crystfel_file(input_file)
    else:
        raise RuntimeError(
            "Provided a geometry file that did not end in .data or .geom! Cannot guess type."
        )

    if det_dist is None:
        det_dist = -1 * np.mean(det.coords_z(run)) / 1e3
    pv_cl: str
    if pv_camera_length is None:
        if (
            "epix10k2m" in str(det.name).lower()
            or "epix10ka2m" in str(det.name).lower()
        ):
            if "mfx" in str(det.name).lower():
                pv_cl = "MFX:ROB:CONT:POS:Z"
            else:
                raise RuntimeError(f"Cannot guess camera length PV for: {det.name}")
        elif "jungfrau4m" in str(det.name).lower():
            if "cxi" in str(det.name).lower():
                pv_cl = "CXI:DS1:MMS:06.RBV"
            else:
                raise RuntimeError(f"Cannot guess camera length PV for: {det.name}")
        elif "rayonix" in str(det.name).lower():
            if "mfx" in str(det.name).lower():
                pv_cl = "MFX:DET:MMS:04.RBV"
            else:
                raise RuntimeError(f"Cannot guess camera length PV for: {det.name}")
        else:
            raise RuntimeError(f"Cannot guess camera length PV for: {det.name}")
    else:
        pv_cl = pv_camera_length
    coffset: float = (det_dist - ds.env().epicsStore().value(pv_cl)) / 1000.0

    geom.to_crystfel_file(output_file, coffset=coffset)

geometry_optimize_residual(params, powder)

Target function for OptimizeAgBhGeometryExhaustive geometry fitting.

Parameters:

Name Type Description Default
params Parameters

Parameters. [center_x, center_y, peaks...] Center values are floats. Peaks are integers.

required
powder ndarray[float64]

Powder image.

required

Returns:

Name Type Description
pixel_values ndarray[float64]

Residuals for fitting.

Source code in lute/tasks/_geometry.py
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
73
def geometry_optimize_residual(
    params: lmfit.Parameters, powder: np.ndarray[np.float64]
) -> np.ndarray[np.float64]:
    """Target function for OptimizeAgBhGeometryExhaustive geometry fitting.

    Args:
        params (lmfit.Parameters): Parameters. [center_x, center_y, peaks...]
            Center values are floats. Peaks are integers.

        powder (np.ndarray[np.float64]): Powder image.

    Returns:
        pixel_values (np.ndarray[np.float64]): Residuals for fitting.
    """
    # Unpack the parameters
    params_l: List[float] = [val.value for _, val in params.items()]
    center_guess: List[float] = params_l[:2]
    # Indices are in radii units since they are for a radial profile
    indices: List[float] = params_l[2:]
    coords: np.ndarray[np.float64] = generate_concentric_sample_pts(
        indices, center_guess
    )

    # Use residual for fitting - difference between intensity in ring
    # and powder max
    pixel_values: np.ndarray[np.float64] = map_coordinates(powder, coords)
    pixel_values -= np.max(powder)
    return pixel_values

modify_crystfel_coffset_res(input_file, output_file, coffset, res)

Overwrite coffset and res entries in a CrystFEL .geom file.

This is a hack to fix Rayonix geom files generated using the wrong pixel size.

Parameters:

Name Type Description Default
input_file str

Input .geom file

required
output_file str

Output modified .geom file

required
coffset float

coffset (camera offset) value in meters.

required
res float

Pixel resolution in um

required
Source code in lute/tasks/_geometry.py
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def modify_crystfel_coffset_res(
    input_file: str, output_file: str, coffset: float, res: float
) -> None:
    """Overwrite coffset and res entries in a CrystFEL .geom file.

    This is a hack to fix Rayonix geom files generated using the wrong pixel size.

    Args:
        input_file (str): Input .geom file

        output_file (str): Output modified .geom file

        coffset (float): coffset (camera offset) value in meters.

        res (float): Pixel resolution in um
    """
    outfile = open(output_file, "w")

    with open(input_file, "r") as infile:
        for line in infile.readlines():

            if "coffset" in line:
                start = line.split("=")[0].strip(" ")
                outfile.write(f"{start} = {coffset}\n")

            elif "res = " in line:
                start = line.split("=")[0].strip(" ")
                outfile.write(f"{start} = {res}\n")

            else:
                outfile.write(line)

    outfile.close()

modify_crystfel_header(input_file, output_file)

Modify the header of a psgeom-generated Crystfel (.geom) file.

This function performs the following modifications
  1. Uncomment lines indicating the mask file and LCLS parameters.
  2. Add entries indicating the location of peaks in the cxi files.

Parameters:

Name Type Description Default
input_file str

Input .geom file generated by psgeom

required
output_file str

Output modified .geom file

required
Source code in lute/tasks/_geometry.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def modify_crystfel_header(input_file, output_file):
    """Modify the header of a psgeom-generated Crystfel (.geom) file.

    This function performs the following modifications:
        1. Uncomment lines indicating the mask file and LCLS parameters.
        2. Add entries indicating the location of peaks in the cxi files.

    Args:
        input_file (str): Input .geom file generated by psgeom

        output_file (str): Output modified .geom file
    """
    outfile = open(output_file, "w")

    with open(input_file, "r") as infile:
        for line in infile.readlines():

            # uncomment by removing semicolon and space
            if line[0] == ";":
                if line.split()[1] in [
                    "clen",
                    "photon_energy",
                    "adu_per_eV",
                    "mask",
                    "mask_good",
                    "mask_bad",
                ]:
                    outfile.write(line[2:])
                else:
                    outfile.write(line)

            # add these header lines for latest crystfel
            elif "data = /entry_1/data_1/data" in line:
                outfile.write(line)
                outfile.write("\n")
                outfile.write("peak_list = /entry_1/result_1\n")
                outfile.write("peak_list_type = cxi\n")

            else:
                outfile.write(line)

    outfile.close()