Skip to content

cylindra.widgets.batch

CylindraBatchWidget

Methods are available in the namespace ui.batch.

Source code in cylindra/widgets/batch/main.py 
 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
 68
 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
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
203
204
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
238
239
240
241
@magicclass(
    widget_type="split",
    layout="horizontal",
    name="Batch Analysis",
    properties={"min_height": 400},
    symbol=Expr("getattr", [Symbol("ui"), "batch"]),
)
class CylindraBatchWidget(MagicTemplate):
    constructor = field(ProjectSequenceEdit)
    sta = field(BatchSubtomogramAveraging)
    loader_infos = BatchLoaderAccessor()

    def __init__(self):
        self._loaders = LoaderList()
        self._loaders.events.inserted.connect(self.reset_choices)
        self._loaders.events.removed.connect(self.reset_choices)
        self._loaders.events.moved.connect(self.reset_choices)

    def _get_loader_paths(self, *_) -> list[PathInfo]:
        return [prj._get_loader_paths() for prj in self.constructor.projects]

    def _get_expression(self, *_):
        return self.constructor._get_expression()

    @set_design(text=capitalize, location=constructor)
    @thread_worker
    def construct_loader(
        self,
        paths: Annotated[Any, {"bind": _get_loader_paths}],
        predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
        name: str = "Loader",
    ):  # fmt: skip
        """
        Construct a batch loader object from the given paths and predicate.

        Parameters
        ----------
        paths : list of (Path, list[Path]) or list of (Path, list[Path], Path)
            List of tuples of image path, list of molecule paths, and project path. The
            project path is optional.
        predicate : str or polars expression, optional
            Filter predicate of molecules.
        name : str, default "Loader"
            Name of the loader.
        """
        if name == "":
            raise ValueError("Name must be given.")

        yield 0.0, 0.0  # this function yields the progress
        loader = BatchLoader()
        image_paths = dict[int, Path]()
        invert = dict[int, bool]()
        _temp_feat = TempFeatures()
        for img_id, path_info in enumerate(paths):
            path_info = PathInfo(*path_info)
            img = path_info.lazy_imread()
            image_paths[img_id] = Path(path_info.image)
            invert[img_id] = path_info.need_invert
            for molecule_id, mole in enumerate(path_info.iter_molecules(_temp_feat)):
                loader.add_tomogram(img.value, mole, img_id)
                yield img_id / len(paths), molecule_id / len(path_info.molecules)
            yield (img_id + 1) / len(paths), 0.0
        yield 1.0, 1.0

        if predicate is not None:
            if isinstance(predicate, str):
                predicate = eval(predicate, POLARS_NAMESPACE, {})
            loader = loader.filter(predicate)
        new = loader.replace(
            molecules=loader.molecules.drop_features(_temp_feat.to_drop),
            scale=self.constructor.scale.value,
        )

        @thread_worker.callback
        def _on_return():
            self._add_loader(new, name, image_paths, invert)

        return _on_return

    @construct_loader.yielded.connect
    def _on_construct_loader_yielded(self, prog: tuple[float, float]):
        btn = get_button(self.construct_loader, cache=True)
        btn.text = f"Constructing... ({prog[0]:.1%}, {prog[1]:.1%})"

    @construct_loader.finished.connect
    def _on_construct_loader_finished(self):
        btn = get_button(self.construct_loader, cache=True)
        btn.text = "Construct loader"

    @set_design(text=capitalize, location=ProjectSequenceEdit.File)
    def construct_loader_by_list(
        self,
        project_paths: Path.Multiple[FileFilter.PROJECT],
        mole_pattern: str = "*",
        predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
        name: str = "Loader",
    ):
        """
        Construct a batch loader from a list of project paths and a molecule pattern.

        Parameters
        ----------
        project_paths : list of path-like
            All the project paths to be used for construction.
        mole_pattern : str, default "*"
            A glob pattern for molecule file names. For example, "*-ALN1.csv" will only
            collect the molecule file names ends with "-ALN1.csv".
        predicate : str or polars expression, optional
            Filter predicate of molecules.
        name : str, default "Loader"
            Name of the loader.
        """
        self.constructor.add_projects(project_paths, clear=True)
        self.constructor.select_molecules_by_pattern(mole_pattern)
        self.construct_loader(self._get_loader_paths(), predicate=predicate, name=name)
        return None

    @set_design(text=capitalize, location=ProjectSequenceEdit.File)
    def construct_loader_by_pattern(
        self,
        path_pattern: str,
        mole_pattern: str = "*",
        predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
        name: str = "Loader",
    ):
        """
        Construct a batch loader from a pattern of project paths and molecule paths.

        Parameters
        ----------
        path_pattern : str
            A glob pattern for project paths.
        mole_pattern : str, default "*"
            A glob pattern for molecule file names. For example, "*-ALN1.csv" will only
            collect the molecule file names ends with "-ALN1.csv".
        predicate : str or polars expression, optional
            Filter predicate of molecules.
        name : str, default "Loader"
            Name of the loader.
        """
        self.constructor.add_projects_glob(path_pattern, clear=True)
        self.constructor.select_molecules_by_pattern(mole_pattern)
        self.construct_loader(self._get_loader_paths(), predicate=predicate, name=name)
        return None

    def _add_loader(
        self,
        loader: BatchLoader,
        name: str,
        image_paths: dict[int, Path],
        invert: dict[int, bool],
    ):
        self._loaders.append(LoaderInfo(loader, name, image_paths, invert))
        try:
            self.sta["loader_name"].value = self.sta["loader_name"].choices[-1]
        except Exception:
            pass  # Updating the value is not important. Silence just in case.

    @set_design(text=capitalize, location=ProjectSequenceEdit.MacroMenu)
    @do_not_record
    def show_macro(self):
        """Show the macro widget of the batch analyzer."""
        from cylindra import instance

        ui = instance()
        assert ui is not None
        macro_str = self.macro.widget.textedit.value
        ui.OthersMenu.Macro._get_macro_window(macro_str, "Batch")
        return None

    @set_design(text=capitalize, location=ProjectSequenceEdit.MacroMenu)
    @do_not_record
    def show_native_macro(self):
        """Show the native macro widget of the batch analyzer."""
        self.macro.widget.show()
        ACTIVE_WIDGETS.add(self.macro.widget)
        return None

    @set_design(text="Load batch analysis project", location=ProjectSequenceEdit.File)
    @confirm(
        text="Are you sure to clear all loaders?", condition="len(self._loaders) > 0"
    )
    def load_batch_project(self, path: Path.Read[FileFilter.PROJECT]):
        """
        Load a batch project from a JSON file.

        Parameters
        ----------
        path : path-like
            Path to the JSON file.
        """
        self._loaders.clear()
        return CylindraBatchProject.from_file(path)._to_gui(self)

    @set_design(
        text="Save as batch analysis project", location=ProjectSequenceEdit.File
    )
    def save_batch_project(
        self,
        save_path: Path.Save,
        molecules_ext: Literal[".csv", ".parquet"] = ".csv",
    ):
        """
        Save the GUI state to a JSON file.

        Parameters
        ----------
        save_path : path-like
            Path to the JSON file.
        molecules_ext : str, default ".csv"
            Extension of the molecule files.
        """
        return CylindraBatchProject.save_gui(self, Path(save_path), molecules_ext)

construct_loader(paths, predicate=None, name='Loader')

Construct a batch loader object from the given paths and predicate.

Parameters:

Name Type Description Default
paths list of (Path, list[Path]) or list of (Path, list[Path], Path)

List of tuples of image path, list of molecule paths, and project path. The project path is optional.

 required
predicate str or polars expression

Filter predicate of molecules.

 None
name str

Name of the loader.

 "Loader"
Source code in cylindra/widgets/batch/main.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 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
@set_design(text=capitalize, location=constructor)
@thread_worker
def construct_loader(
    self,
    paths: Annotated[Any, {"bind": _get_loader_paths}],
    predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
    name: str = "Loader",
):  # fmt: skip
    """
    Construct a batch loader object from the given paths and predicate.

    Parameters
    ----------
    paths : list of (Path, list[Path]) or list of (Path, list[Path], Path)
        List of tuples of image path, list of molecule paths, and project path. The
        project path is optional.
    predicate : str or polars expression, optional
        Filter predicate of molecules.
    name : str, default "Loader"
        Name of the loader.
    """
    if name == "":
        raise ValueError("Name must be given.")

    yield 0.0, 0.0  # this function yields the progress
    loader = BatchLoader()
    image_paths = dict[int, Path]()
    invert = dict[int, bool]()
    _temp_feat = TempFeatures()
    for img_id, path_info in enumerate(paths):
        path_info = PathInfo(*path_info)
        img = path_info.lazy_imread()
        image_paths[img_id] = Path(path_info.image)
        invert[img_id] = path_info.need_invert
        for molecule_id, mole in enumerate(path_info.iter_molecules(_temp_feat)):
            loader.add_tomogram(img.value, mole, img_id)
            yield img_id / len(paths), molecule_id / len(path_info.molecules)
        yield (img_id + 1) / len(paths), 0.0
    yield 1.0, 1.0

    if predicate is not None:
        if isinstance(predicate, str):
            predicate = eval(predicate, POLARS_NAMESPACE, {})
        loader = loader.filter(predicate)
    new = loader.replace(
        molecules=loader.molecules.drop_features(_temp_feat.to_drop),
        scale=self.constructor.scale.value,
    )

    @thread_worker.callback
    def _on_return():
        self._add_loader(new, name, image_paths, invert)

    return _on_return

construct_loader_by_list(project_paths, mole_pattern='*', predicate=None, name='Loader')

Construct a batch loader from a list of project paths and a molecule pattern.

Parameters:

Name Type Description Default
project_paths list of path-like

All the project paths to be used for construction.

 required
mole_pattern str

A glob pattern for molecule file names. For example, "*-ALN1.csv" will only collect the molecule file names ends with "-ALN1.csv".

 "*"
predicate str or polars expression

Filter predicate of molecules.

 None
name str

Name of the loader.

 "Loader"
Source code in cylindra/widgets/batch/main.py 
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
@set_design(text=capitalize, location=ProjectSequenceEdit.File)
def construct_loader_by_list(
    self,
    project_paths: Path.Multiple[FileFilter.PROJECT],
    mole_pattern: str = "*",
    predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
    name: str = "Loader",
):
    """
    Construct a batch loader from a list of project paths and a molecule pattern.

    Parameters
    ----------
    project_paths : list of path-like
        All the project paths to be used for construction.
    mole_pattern : str, default "*"
        A glob pattern for molecule file names. For example, "*-ALN1.csv" will only
        collect the molecule file names ends with "-ALN1.csv".
    predicate : str or polars expression, optional
        Filter predicate of molecules.
    name : str, default "Loader"
        Name of the loader.
    """
    self.constructor.add_projects(project_paths, clear=True)
    self.constructor.select_molecules_by_pattern(mole_pattern)
    self.construct_loader(self._get_loader_paths(), predicate=predicate, name=name)
    return None

construct_loader_by_pattern(path_pattern, mole_pattern='*', predicate=None, name='Loader')

Construct a batch loader from a pattern of project paths and molecule paths.

Parameters:

Name Type Description Default
path_pattern str

A glob pattern for project paths.

 required
mole_pattern str

A glob pattern for molecule file names. For example, "*-ALN1.csv" will only collect the molecule file names ends with "-ALN1.csv".

 "*"
predicate str or polars expression

Filter predicate of molecules.

 None
name str

Name of the loader.

 "Loader"
Source code in cylindra/widgets/batch/main.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@set_design(text=capitalize, location=ProjectSequenceEdit.File)
def construct_loader_by_pattern(
    self,
    path_pattern: str,
    mole_pattern: str = "*",
    predicate: Annotated[str | pl.Expr | None, {"bind": _get_expression}] = None,
    name: str = "Loader",
):
    """
    Construct a batch loader from a pattern of project paths and molecule paths.

    Parameters
    ----------
    path_pattern : str
        A glob pattern for project paths.
    mole_pattern : str, default "*"
        A glob pattern for molecule file names. For example, "*-ALN1.csv" will only
        collect the molecule file names ends with "-ALN1.csv".
    predicate : str or polars expression, optional
        Filter predicate of molecules.
    name : str, default "Loader"
        Name of the loader.
    """
    self.constructor.add_projects_glob(path_pattern, clear=True)
    self.constructor.select_molecules_by_pattern(mole_pattern)
    self.construct_loader(self._get_loader_paths(), predicate=predicate, name=name)
    return None

load_batch_project(path)

Load a batch project from a JSON file.

Parameters:

Name Type Description Default
path path - like

Path to the JSON file.

 required
Source code in cylindra/widgets/batch/main.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@set_design(text="Load batch analysis project", location=ProjectSequenceEdit.File)
@confirm(
    text="Are you sure to clear all loaders?", condition="len(self._loaders) > 0"
)
def load_batch_project(self, path: Path.Read[FileFilter.PROJECT]):
    """
    Load a batch project from a JSON file.

    Parameters
    ----------
    path : path-like
        Path to the JSON file.
    """
    self._loaders.clear()
    return CylindraBatchProject.from_file(path)._to_gui(self)

save_batch_project(save_path, molecules_ext='.csv')

Save the GUI state to a JSON file.

Parameters:

Name Type Description Default
save_path path - like

Path to the JSON file.

 required
molecules_ext str

Extension of the molecule files.

 ".csv"
Source code in cylindra/widgets/batch/main.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@set_design(
    text="Save as batch analysis project", location=ProjectSequenceEdit.File
)
def save_batch_project(
    self,
    save_path: Path.Save,
    molecules_ext: Literal[".csv", ".parquet"] = ".csv",
):
    """
    Save the GUI state to a JSON file.

    Parameters
    ----------
    save_path : path-like
        Path to the JSON file.
    molecules_ext : str, default ".csv"
        Extension of the molecule files.
    """
    return CylindraBatchProject.save_gui(self, Path(save_path), molecules_ext)

show_macro()

Show the macro widget of the batch analyzer.

Source code in cylindra/widgets/batch/main.py 
187
188
189
190
191
192
193
194
195
196
197
@set_design(text=capitalize, location=ProjectSequenceEdit.MacroMenu)
@do_not_record
def show_macro(self):
    """Show the macro widget of the batch analyzer."""
    from cylindra import instance

    ui = instance()
    assert ui is not None
    macro_str = self.macro.widget.textedit.value
    ui.OthersMenu.Macro._get_macro_window(macro_str, "Batch")
    return None

show_native_macro()

Show the native macro widget of the batch analyzer.

Source code in cylindra/widgets/batch/main.py 
199
200
201
202
203
204
205
@set_design(text=capitalize, location=ProjectSequenceEdit.MacroMenu)
@do_not_record
def show_native_macro(self):
    """Show the native macro widget of the batch analyzer."""
    self.macro.widget.show()
    ACTIVE_WIDGETS.add(self.macro.widget)
    return None

BatchSubtomogramAveraging

Methods are available in the namespace ui.batch.sta.

Source code in cylindra/widgets/batch/sta.py 
 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
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
203
204
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
238
239
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
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
@magicclass(name="Batch Subtomogram Analysis")
@_shared_doc.update_cls
class BatchSubtomogramAveraging(MagicTemplate):
    def _get_parent(self):
        from .main import CylindraBatchWidget

        return self.find_ancestor(CylindraBatchWidget, cache=True)

    def _get_loader_names(self, _=None) -> list[str]:
        try:
            parent = self._get_parent()
        except Exception:
            return []
        return parent.loader_infos.names()

    # Menus
    BatchSubtomogramAnalysis = field(
        BatchSubtomogramAnalysis, name="Subtomogram Analysis"
    )
    BatchRefinement = field(BatchRefinement, name="Refinement")
    BatchLoaderMenu = field(BatchLoaderMenu, name="Loader")

    @magicclass(layout="horizontal", properties={"margins": (0, 0, 0, 0)})
    class Header(MagicTemplate):
        loader_name = abstractapi()
        show_loader_info = abstractapi()
        remove_loader = abstractapi()

    loader_name = vfield(str, location=Header, record=False).with_choices(
        choices=_get_loader_names
    )

    def _get_current_loader_name(self, _=None) -> str:
        return self.loader_name

    @set_design(text="??", max_width=36, location=Header)
    @do_not_record
    def show_loader_info(self):
        """Show information about this loader"""
        info = self._get_parent().loader_infos[self.loader_name]
        loader = info.loader
        img_info = "\n" + "\n".join(
            f"{img_id}: {img_path}" for img_id, img_path in info.image_paths.items()
        )

        info_text = (
            f"name: {info.name}\nmolecule: n={loader.count()}\nimages:{img_info}"
        )
        view = DataFrameView(value=loader.molecules.to_dataframe())
        txt = ConsoleTextEdit(value=info_text)
        txt.read_only = True
        cnt = Container(widgets=[txt, view], layout="horizontal", labels=False)
        cnt.native.setParent(self.native, cnt.native.windowFlags())
        cnt.show()

    @set_design(text="✕", max_width=36, location=Header)
    def remove_loader(
        self, loader_name: Annotated[str, {"bind": _get_current_loader_name}]
    ):
        """Remove this loader"""
        del self._get_parent().loader_infos[loader_name]

    params = field(StaParameters)

    def _get_selected_loader_choice(self, *_) -> list[str]:
        try:
            loader = self.get_loader(self.loader_name)
            return loader.molecules.features.columns
        except Exception:
            return []

    def _get_template_path(self, _=None):
        return self.params.template_path.value

    def _get_mask_params(self, _=None):
        return self.params._get_mask_params()

    @set_design(text="Split loader", location=BatchLoaderMenu)
    def split_loader(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        by: Annotated[str, {"choices": _get_selected_loader_choice}],
        delete_old: bool = False,
    ):
        """
        Split the selected loader by the values of the given column.

        Parameters
        ----------
        loader_name : str
            Name of the input loader
        by : str
            Column name to split the loader
        delete_old : bool, default False
            If true, the original loader will be deleted.
        """
        parent = self._get_parent()
        loaders = parent._loaders
        batch_info = loaders[loader_name]
        batch_loader = batch_info.loader
        n_unique = batch_loader.molecules.features[by].n_unique()
        if n_unique > 48:
            raise ValueError(
                f"Too many groups ({n_unique}). Did you choose a float column?"
            )
        for _key, loader in batch_loader.groupby(by):
            existing_id = set(loader.features[Mole.image])
            image_paths = {
                k: v for k, v in batch_info.image_paths.items() if v in existing_id
            }
            invert = {k: v for k, v in batch_info.invert.items() if v in existing_id}
            parent._add_loader(loader, f"{loader_name}_{_key}", image_paths, invert)

        if delete_old:
            idx = -1
            for i, info in enumerate(loaders):
                if info.loader is batch_loader:
                    idx = i
                    break
            else:
                idx = -1
            if idx < 0:
                raise RuntimeError("Loader not found.")
            del loaders[idx]

    @set_design(text="Filter loader", location=BatchLoaderMenu)
    def filter_loader(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        expression: PolarsExprStr,
    ):
        """
        Filter the selected loader and add the filtered one to the list.

        Parameters
        ----------
        loader_name : str
            Name of the input loader
        expression : str
            polars expression that will be used to filter the loader. For example,
            `col("score") > 0.7` will filter out all low-score molecules.
        """
        loaderlist = self._get_parent()._loaders
        info = loaderlist[loader_name]
        loader = info.loader
        new = loader.filter(norm_expr(expression))
        existing_id = set(new.features[Mole.image])
        loaderlist.add_loader(
            new,
            name=f"{info.name}-Filt",
            image_paths={k: v for k, v in info.image_paths.items() if v in existing_id},
            invert={k: v for k, v in info.invert.items() if v in existing_id},
        )
        return None

    @nogui
    def get_loader(self, name: str) -> BatchLoader:
        """Return the acryo.BatchLoader object with the given name"""
        if not isinstance(name, str):
            raise TypeError(f"Name must be a string, got {type(name).__name__}")
        return self._get_parent().loader_infos[name].loader

    @set_design(text="Average all molecules", location=BatchSubtomogramAnalysis)
    @dask_thread_worker.with_progress(desc="Averaging all molecules in projects")
    def average_all(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        size: _SubVolumeSize = None,
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
        bin_size: _BINSIZE = 1,
    ):
        t0 = timer()
        loader = self._get_parent().loader_infos[loader_name].loader
        shape = self._get_shape_in_px(size, loader)
        img = ip.asarray(
            loader.replace(output_shape=shape, order=interpolation)
            .binning(bin_size, compute=False)
            .average(),
            axes="zyx",
        ).set_scale(zyx=loader.scale * bin_size, unit="nm")
        t0.toc()
        return self._show_rec.with_args(img, f"[AVG]{loader_name}")

    @set_design(text="Average group-wise", location=BatchSubtomogramAnalysis)
    @dask_thread_worker.with_progress(desc="Grouped subtomogram averaging")
    def average_groups(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        size: _SubVolumeSize = None,
        by: PolarsExprStr = "col('pf-id')",
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
        bin_size: _BINSIZE = 1,
    ):
        """
        Groupwise subtomogram averaging using molecules grouped by the given expression.

        This method first group molecules by its features, and then average each group.
        This method is useful for such as get average of each protofilament and
        segmented subtomogram averaging.

        Parameters
        ----------
        {loader_name}{size}
        by : str or polars expression
            Expression to group molecules.
        {interpolation}{bin_size}
        """
        t0 = timer()
        loader = self._get_parent().loader_infos[loader_name].loader
        shape = self._get_shape_in_px(size, loader)
        img = ip.asarray(
            loader.replace(output_shape=shape, order=interpolation)
            .binning(bin_size, compute=False)
            .groupby(norm_expr(by))
            .average()
            .value_stack(axis=0),
            axes="pzyx",
        ).set_scale(zyx=loader.scale * bin_size, unit="nm")
        t0.toc()
        return self._show_rec.with_args(img, f"[AVG]{loader_name}", store=False)

    @set_design(text="Split and average molecules", location=BatchSubtomogramAnalysis)
    @dask_thread_worker.with_progress(desc="Split-and-average")
    def split_and_average(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        n_pairs: Annotated[int, {"min": 1, "label": "number of image pairs"}] = 1,
        size: _SubVolumeSize = None,
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
        bin_size: _BINSIZE = 1,
    ):
        """
        Split molecules into two groups and average separately.

        Parameters
        ----------
        {loader_name}{size}
        n_pairs : int, default is 1
            How many pairs of average will be calculated.
        {size}{interpolation}{bin_size}
        """
        t0 = timer()
        loader = self._get_parent().loader_infos[loader_name].loader
        shape = self._get_shape_in_px(size, loader)

        axes = "ipzyx" if n_pairs > 1 else "pzyx"
        img = ip.asarray(
            loader.replace(output_shape=shape, order=interpolation)
            .binning(bin_size, compute=False)
            .average_split(n_pairs),
            axes=axes,
        ).set_scale(zyx=loader.scale * bin_size, unit="nm")
        t0.toc()
        return self._show_rec.with_args(img, f"[Split]{loader_name}", store=False)

    @set_design(text="Align all molecules", location=BatchRefinement)
    @dask_thread_worker.with_progress(desc="Aligning all molecules")
    def align_all(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        template_path: Annotated[str | Path, {"bind": _get_template_path}],
        mask_params: Annotated[Any, {"bind": _get_mask_params}],
        max_shifts: _MaxShifts = (1.0, 1.0, 1.0),
        rotations: _Rotations = ((0.0, 0.0), (0.0, 0.0), (0.0, 0.0)),
        cutoff: _CutoffFreq = 0.5,
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 3,
        method: Annotated[str, {"choices": METHOD_CHOICES}] = "zncc",
        bin_size: _BINSIZE = 1,
    ):  # fmt: skip
        """
        Align all the molecules in the selected loader.

        Parameters
        ----------
        {loader_name}{template_path}{mask_params}{max_shifts}{rotations}{cutoff}
        {interpolation}{method}{bin_size}
        """
        t0 = timer()
        loaderlist = self._get_parent()._loaders
        info = loaderlist[loader_name]
        loader = info.loader
        template, mask = loader.normalize_input(
            template=self.params._norm_template_param(template_path),
            mask=self.params._get_mask(params=mask_params),
        )
        aligned = (
            loader.replace(output_shape=template.shape, order=interpolation)
            .binning(bin_size, compute=False)
            .align(
                template=template,
                mask=mask,
                max_shifts=max_shifts,
                rotations=rotations,
                cutoff=cutoff,
                alignment_model=_get_alignment(method),
            )
        )
        loaderlist.add_loader(
            aligned,
            name=_coerce_aligned_name(info.name, loaderlist),
            image_paths=info.image_paths,
            invert=info.invert,
        )
        t0.toc()
        return None

    @set_design(text="Calculate FSC", location=BatchSubtomogramAnalysis)
    @dask_thread_worker.with_progress(desc="Calculating FSC")
    def calculate_fsc(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        template_path: Annotated[str | Path | None, {"bind": _get_template_path}] = None,
        mask_params: Annotated[Any, {"bind": _get_mask_params}] = None,
        size: _SubVolumeSize = None,
        seed: Annotated[Optional[int], {"text": "Do not use random seed."}] = 0,
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
        n_pairs: Annotated[int, {"min": 1, "label": "number of image pairs"}] = 1,
        show_average: bool = True,
        dfreq: FSCFreq = None,
    ):  # fmt: skip
        """
        Calculate Fourier Shell Correlation using the selected loader.

        Parameters
        ----------
        {loader_name}{template_path}{mask_params}{size}
        seed : int, optional
            Random seed used for subtomogram sampling.
        {interpolation}
        n_pairs : int, default 1
            How many sets of image pairs will be generated to average FSC.
        show_average : bool, default True
            If true, subtomogram averaging will be shown after FSC calculation.
        dfreq : float, default 0.02
            Precision of frequency to calculate FSC. "0.02" means that FSC will be
            calculated at frequency 0.01, 0.03, 0.05, ..., 0.45.
        """
        t0 = timer()
        loader = (
            self._get_parent()
            .loader_infos[loader_name]
            .loader.replace(order=interpolation)
        )

        template, mask = loader.normalize_input(
            template=self.params._norm_template_param(template_path, allow_none=True),
            mask=self.params._get_mask(params=mask_params),
        )

        fsc, avg = loader.reshape(
            template=template if size is None else None,
            mask=mask,
            shape=None if size is None else self._get_shape_in_px(size, loader),
        ).fsc_with_average(mask=mask, seed=seed, n_set=n_pairs, dfreq=dfreq)

        if show_average:
            img_avg = ip.asarray(avg, axes="zyx").set_scale(zyx=loader.scale)
        else:
            img_avg = None

        result = FscResult.from_dataframe(fsc, loader.scale)
        criteria = [0.5, 0.143]
        t0.toc()

        @thread_worker.callback
        def _calculate_fsc_on_return():
            _Logger.print_html(f"<b>Fourier Shell Correlation of {loader_name!r}</b>")
            with _Logger.set_plt():
                result.plot(criteria)
            for _c in criteria:
                _r = result.get_resolution(_c)
                _Logger.print_html(f"Resolution at FSC={_c:.3f} ... <b>{_r:.3f} nm</b>")

            if img_avg is not None:
                _rec_layer = self._show_rec(img_avg, name=f"[AVG]{loader_name}")
                _rec_layer.metadata["fsc"] = result

        return _calculate_fsc_on_return

    @set_design(text="PCA/K-means classification", location=BatchSubtomogramAnalysis)
    @dask_thread_worker.with_progress(descs=_classify_pca_fmt)
    def classify_pca(
        self,
        loader_name: Annotated[str, {"bind": _get_current_loader_name}],
        template_path: Annotated[str | Path | None, {"bind": _get_template_path}] = None,
        mask_params: Annotated[Any, {"bind": _get_mask_params}] = None,
        size: Annotated[Optional[nm], {"text": "Use mask shape", "options": {"value": 12.0, "max": 100.0}, "label": "size (nm)"}] = None,
        cutoff: _CutoffFreq = 0.5,
        interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 3,
        bin_size: _BINSIZE = 1,
        n_components: Annotated[int, {"min": 2, "max": 20}] = 2,
        n_clusters: Annotated[int, {"min": 2, "max": 100}] = 2,
        seed: Annotated[Optional[int], {"text": "Do not use random seed."}] = 0,
    ):  # fmt: skip
        """
        Classify molecules in the loader using PCA and K-means clustering.

        Parameters
        ----------
        {loader_name}{template_path}{mask_params}{size}{cutoff}{interpolation}{bin_size}
        n_components : int, default 2
            The number of PCA dimensions.
        n_clusters : int, default 2
            The number of clusters.
        seed : int, default
            Random seed.
        """
        from cylindra.widgets.subwidgets import PcaViewer

        t0 = timer()
        loader = self._get_parent().loader_infos[loader_name].loader
        template, mask = loader.normalize_input(
            template=self.params._norm_template_param(template_path, allow_none=True),
            mask=self.params._get_mask(params=mask_params),
        )
        shape = None
        if mask is None:
            shape = self._get_shape_in_px(size, loader)
        out, pca = (
            loader.reshape(
                template=template if mask is None and shape is None else None,
                mask=mask,
                shape=shape,
            )
            .replace(order=interpolation)
            .binning(binsize=bin_size, compute=False)
            .classify(
                mask=mask,
                seed=seed,
                cutoff=cutoff,
                n_components=n_components,
                n_clusters=n_clusters,
                label_name="cluster",
            )
        )

        avgs = ip.asarray(
            out.groupby("cluster").average().value_stack(axis=0),
            axes=["cluster", "z", "y", "x"],
        ).set_scale(zyx=loader.scale, unit="nm")

        t0.toc()

        @thread_worker.callback
        def _on_return():
            loader.molecules.features = out.molecules.features
            pca_viewer = PcaViewer(pca)
            pca_viewer.native.setParent(self.native, pca_viewer.native.windowFlags())
            pca_viewer.show()
            self._show_rec(avgs, name=f"[PCA]{loader_name}", store=False)

            ACTIVE_WIDGETS.add(pca_viewer)

        return _on_return

    @magictoolbar
    class STATools(MagicTemplate):
        show_template = abstractapi()
        show_template_original = abstractapi()
        show_mask = abstractapi()

    @set_design(icon="ic:baseline-view-in-ar", location=STATools)
    @do_not_record
    def show_template(self):
        """Load and show template image in the scale of the tomogram."""
        template = self._get_template_image()
        self._show_rec(template, name="Template image", store=False)

    @set_design(icon="material-symbols:view-in-ar", location=STATools)
    @do_not_record
    def show_template_original(self):
        """Load and show template image in the original scale."""
        _input = self.params._get_template_input(allow_multiple=True)
        if _input is None:
            raise ValueError("No template path provided.")
        elif isinstance(_input, Path):
            self._show_rec(ip.imread(_input), name="Template image", store=False)
        else:
            for i, fp in enumerate(_input):
                img = ip.imread(fp)
                self._show_rec(img, name=f"Template image [{i}]", store=False)

    @set_design(icon="fluent:shape-organic-20-filled", location=STATools)
    @do_not_record
    def show_mask(self):
        """Load and show mask image in the scale of the tomogram."""
        loader = self.get_loader(self.loader_name)
        _, mask = loader.normalize_input(
            self.params._norm_template_param(
                self.params._get_template_input(allow_multiple=False),
                allow_none=True,
            ),
            self.params._get_mask(),
        )
        if mask is None:
            raise ValueError("No mask to show.")
        mask = ip.asarray(mask, axes="zyx").set_scale(zyx=loader.scale, unit="nm")
        self._show_rec(mask, name="Mask image", store=False)

    @thread_worker.callback
    def _show_rec(self, img: ip.ImgArray, name: str, store: bool = True):
        return self.params._show_reconstruction(img, name, store)

    def _get_shape_in_px(
        self, default: "nm | None", loader: BatchLoader
    ) -> tuple[int, ...]:
        if default is None:
            tmp = self._get_template_image()
            return tmp.sizesof("zyx")
        else:
            return (roundint(default / loader.scale),) * 3

    def _get_template_image(self) -> ip.ImgArray:
        scale = self.get_loader(self.loader_name).scale

        template = self.params._norm_template_param(
            self.params._get_template_input(allow_multiple=True),
            allow_none=False,
            allow_multiple=True,
        ).provide(scale)
        if isinstance(template, list):
            template = ip.asarray(np.stack(template, axis=0), axes="zyx")
        else:
            template = ip.asarray(template, axes="zyx")
        return template.set_scale(zyx=scale, unit="nm")

    @setup_function_gui(split_loader)
    def _(self, gui):
        gui[0].changed.connect(gui[1].reset_choices)

align_all(loader_name, template_path, mask_params, max_shifts=(1.0, 1.0, 1.0), rotations=((0.0, 0.0), (0.0, 0.0), (0.0, 0.0)), cutoff=0.5, interpolation=3, method='zncc', bin_size=1)

Align all the molecules in the selected loader.

Parameters:

Name Type Description Default
loader_name str

Name of the batch subtomogram loader to be used.

 required
template_path Path or str

Path to template image.

 required
mask_params str or (float, float)

Mask image path or dilation/Gaussian blur parameters. If a path is given, image must in the same shape as the template.

 required
max_shifts int or tuple of int

Maximum shift between subtomograms and template in nm. ZYX order.

 (1.0, 1.0, 1.0)
rotations ((float, float), (float, float), (float, float))

Rotation in external degree around each axis.

 ((0.0, 0.0), (0.0, 0.0), (0.0, 0.0))
cutoff float

Cutoff frequency of low-pass filter applied in each subtomogram.

 0.5
interpolation int

Interpolation order.

 3
method str

Correlation metrics for alignment.

 'zncc'
bin_size int

Bin size of multiscale image to be used. Set to >1 to boost performance.

 1
Source code in cylindra/widgets/batch/sta.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
@set_design(text="Align all molecules", location=BatchRefinement)
@dask_thread_worker.with_progress(desc="Aligning all molecules")
def align_all(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    template_path: Annotated[str | Path, {"bind": _get_template_path}],
    mask_params: Annotated[Any, {"bind": _get_mask_params}],
    max_shifts: _MaxShifts = (1.0, 1.0, 1.0),
    rotations: _Rotations = ((0.0, 0.0), (0.0, 0.0), (0.0, 0.0)),
    cutoff: _CutoffFreq = 0.5,
    interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 3,
    method: Annotated[str, {"choices": METHOD_CHOICES}] = "zncc",
    bin_size: _BINSIZE = 1,
):  # fmt: skip
    """
    Align all the molecules in the selected loader.

    Parameters
    ----------
    {loader_name}{template_path}{mask_params}{max_shifts}{rotations}{cutoff}
    {interpolation}{method}{bin_size}
    """
    t0 = timer()
    loaderlist = self._get_parent()._loaders
    info = loaderlist[loader_name]
    loader = info.loader
    template, mask = loader.normalize_input(
        template=self.params._norm_template_param(template_path),
        mask=self.params._get_mask(params=mask_params),
    )
    aligned = (
        loader.replace(output_shape=template.shape, order=interpolation)
        .binning(bin_size, compute=False)
        .align(
            template=template,
            mask=mask,
            max_shifts=max_shifts,
            rotations=rotations,
            cutoff=cutoff,
            alignment_model=_get_alignment(method),
        )
    )
    loaderlist.add_loader(
        aligned,
        name=_coerce_aligned_name(info.name, loaderlist),
        image_paths=info.image_paths,
        invert=info.invert,
    )
    t0.toc()
    return None

average_groups(loader_name, size=None, by="col('pf-id')", interpolation=1, bin_size=1)

Groupwise subtomogram averaging using molecules grouped by the given expression.

This method first group molecules by its features, and then average each group. This method is useful for such as get average of each protofilament and segmented subtomogram averaging.

Parameters:

Name Type Description Default
loader_name str

Name of the batch subtomogram loader to be used.

 required
size nm

Size of the template in nm. Use the size of template image by default.

 None
by str or polars expression

Expression to group molecules.

 "col('pf-id')"
interpolation int

Interpolation order.

 1
bin_size int

Bin size of multiscale image to be used. Set to >1 to boost performance.

 1
Source code in cylindra/widgets/batch/sta.py 
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
@set_design(text="Average group-wise", location=BatchSubtomogramAnalysis)
@dask_thread_worker.with_progress(desc="Grouped subtomogram averaging")
def average_groups(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    size: _SubVolumeSize = None,
    by: PolarsExprStr = "col('pf-id')",
    interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
    bin_size: _BINSIZE = 1,
):
    """
    Groupwise subtomogram averaging using molecules grouped by the given expression.

    This method first group molecules by its features, and then average each group.
    This method is useful for such as get average of each protofilament and
    segmented subtomogram averaging.

    Parameters
    ----------
    {loader_name}{size}
    by : str or polars expression
        Expression to group molecules.
    {interpolation}{bin_size}
    """
    t0 = timer()
    loader = self._get_parent().loader_infos[loader_name].loader
    shape = self._get_shape_in_px(size, loader)
    img = ip.asarray(
        loader.replace(output_shape=shape, order=interpolation)
        .binning(bin_size, compute=False)
        .groupby(norm_expr(by))
        .average()
        .value_stack(axis=0),
        axes="pzyx",
    ).set_scale(zyx=loader.scale * bin_size, unit="nm")
    t0.toc()
    return self._show_rec.with_args(img, f"[AVG]{loader_name}", store=False)

calculate_fsc(loader_name, template_path=None, mask_params=None, size=None, seed=0, interpolation=1, n_pairs=1, show_average=True, dfreq=None)

Calculate Fourier Shell Correlation using the selected loader.

Parameters:

Name Type Description Default
loader_name str

Name of the batch subtomogram loader to be used.

 required
template_path Path or str

Path to template image.

 None
mask_params str or (float, float)

Mask image path or dilation/Gaussian blur parameters. If a path is given, image must in the same shape as the template.

 None
size nm

Size of the template in nm. Use the size of template image by default.

 None
seed int

Random seed used for subtomogram sampling.

 0
interpolation int

Interpolation order.

 1
n_pairs int

How many sets of image pairs will be generated to average FSC.

 1
show_average bool

If true, subtomogram averaging will be shown after FSC calculation.

 True
dfreq float

Precision of frequency to calculate FSC. "0.02" means that FSC will be calculated at frequency 0.01, 0.03, 0.05, ..., 0.45.

 0.02
Source code in cylindra/widgets/batch/sta.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
@set_design(text="Calculate FSC", location=BatchSubtomogramAnalysis)
@dask_thread_worker.with_progress(desc="Calculating FSC")
def calculate_fsc(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    template_path: Annotated[str | Path | None, {"bind": _get_template_path}] = None,
    mask_params: Annotated[Any, {"bind": _get_mask_params}] = None,
    size: _SubVolumeSize = None,
    seed: Annotated[Optional[int], {"text": "Do not use random seed."}] = 0,
    interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
    n_pairs: Annotated[int, {"min": 1, "label": "number of image pairs"}] = 1,
    show_average: bool = True,
    dfreq: FSCFreq = None,
):  # fmt: skip
    """
    Calculate Fourier Shell Correlation using the selected loader.

    Parameters
    ----------
    {loader_name}{template_path}{mask_params}{size}
    seed : int, optional
        Random seed used for subtomogram sampling.
    {interpolation}
    n_pairs : int, default 1
        How many sets of image pairs will be generated to average FSC.
    show_average : bool, default True
        If true, subtomogram averaging will be shown after FSC calculation.
    dfreq : float, default 0.02
        Precision of frequency to calculate FSC. "0.02" means that FSC will be
        calculated at frequency 0.01, 0.03, 0.05, ..., 0.45.
    """
    t0 = timer()
    loader = (
        self._get_parent()
        .loader_infos[loader_name]
        .loader.replace(order=interpolation)
    )

    template, mask = loader.normalize_input(
        template=self.params._norm_template_param(template_path, allow_none=True),
        mask=self.params._get_mask(params=mask_params),
    )

    fsc, avg = loader.reshape(
        template=template if size is None else None,
        mask=mask,
        shape=None if size is None else self._get_shape_in_px(size, loader),
    ).fsc_with_average(mask=mask, seed=seed, n_set=n_pairs, dfreq=dfreq)

    if show_average:
        img_avg = ip.asarray(avg, axes="zyx").set_scale(zyx=loader.scale)
    else:
        img_avg = None

    result = FscResult.from_dataframe(fsc, loader.scale)
    criteria = [0.5, 0.143]
    t0.toc()

    @thread_worker.callback
    def _calculate_fsc_on_return():
        _Logger.print_html(f"<b>Fourier Shell Correlation of {loader_name!r}</b>")
        with _Logger.set_plt():
            result.plot(criteria)
        for _c in criteria:
            _r = result.get_resolution(_c)
            _Logger.print_html(f"Resolution at FSC={_c:.3f} ... <b>{_r:.3f} nm</b>")

        if img_avg is not None:
            _rec_layer = self._show_rec(img_avg, name=f"[AVG]{loader_name}")
            _rec_layer.metadata["fsc"] = result

    return _calculate_fsc_on_return

classify_pca(loader_name, template_path=None, mask_params=None, size=None, cutoff=0.5, interpolation=3, bin_size=1, n_components=2, n_clusters=2, seed=0)

Classify molecules in the loader using PCA and K-means clustering.

Parameters:

Name Type Description Default
loader_name str

Name of the batch subtomogram loader to be used.

 required
template_path Path or str

Path to template image.

 None
mask_params str or (float, float)

Mask image path or dilation/Gaussian blur parameters. If a path is given, image must in the same shape as the template.

 None
size nm

Size of the template in nm. Use the size of template image by default.

 None
cutoff float

Cutoff frequency of low-pass filter applied in each subtomogram.

 0.5
interpolation int

Interpolation order.

 3
bin_size int

Bin size of multiscale image to be used. Set to >1 to boost performance.

 1
n_components int

The number of PCA dimensions.

 2
n_clusters int

The number of clusters.

 2
seed (int, default)

Random seed.

 0
Source code in cylindra/widgets/batch/sta.py 
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
@set_design(text="PCA/K-means classification", location=BatchSubtomogramAnalysis)
@dask_thread_worker.with_progress(descs=_classify_pca_fmt)
def classify_pca(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    template_path: Annotated[str | Path | None, {"bind": _get_template_path}] = None,
    mask_params: Annotated[Any, {"bind": _get_mask_params}] = None,
    size: Annotated[Optional[nm], {"text": "Use mask shape", "options": {"value": 12.0, "max": 100.0}, "label": "size (nm)"}] = None,
    cutoff: _CutoffFreq = 0.5,
    interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 3,
    bin_size: _BINSIZE = 1,
    n_components: Annotated[int, {"min": 2, "max": 20}] = 2,
    n_clusters: Annotated[int, {"min": 2, "max": 100}] = 2,
    seed: Annotated[Optional[int], {"text": "Do not use random seed."}] = 0,
):  # fmt: skip
    """
    Classify molecules in the loader using PCA and K-means clustering.

    Parameters
    ----------
    {loader_name}{template_path}{mask_params}{size}{cutoff}{interpolation}{bin_size}
    n_components : int, default 2
        The number of PCA dimensions.
    n_clusters : int, default 2
        The number of clusters.
    seed : int, default
        Random seed.
    """
    from cylindra.widgets.subwidgets import PcaViewer

    t0 = timer()
    loader = self._get_parent().loader_infos[loader_name].loader
    template, mask = loader.normalize_input(
        template=self.params._norm_template_param(template_path, allow_none=True),
        mask=self.params._get_mask(params=mask_params),
    )
    shape = None
    if mask is None:
        shape = self._get_shape_in_px(size, loader)
    out, pca = (
        loader.reshape(
            template=template if mask is None and shape is None else None,
            mask=mask,
            shape=shape,
        )
        .replace(order=interpolation)
        .binning(binsize=bin_size, compute=False)
        .classify(
            mask=mask,
            seed=seed,
            cutoff=cutoff,
            n_components=n_components,
            n_clusters=n_clusters,
            label_name="cluster",
        )
    )

    avgs = ip.asarray(
        out.groupby("cluster").average().value_stack(axis=0),
        axes=["cluster", "z", "y", "x"],
    ).set_scale(zyx=loader.scale, unit="nm")

    t0.toc()

    @thread_worker.callback
    def _on_return():
        loader.molecules.features = out.molecules.features
        pca_viewer = PcaViewer(pca)
        pca_viewer.native.setParent(self.native, pca_viewer.native.windowFlags())
        pca_viewer.show()
        self._show_rec(avgs, name=f"[PCA]{loader_name}", store=False)

        ACTIVE_WIDGETS.add(pca_viewer)

    return _on_return

filter_loader(loader_name, expression)

Filter the selected loader and add the filtered one to the list.

Parameters:

Name Type Description Default
loader_name str

Name of the input loader

 required
expression str

polars expression that will be used to filter the loader. For example, col("score") > 0.7 will filter out all low-score molecules.

 required
Source code in cylindra/widgets/batch/sta.py 
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
@set_design(text="Filter loader", location=BatchLoaderMenu)
def filter_loader(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    expression: PolarsExprStr,
):
    """
    Filter the selected loader and add the filtered one to the list.

    Parameters
    ----------
    loader_name : str
        Name of the input loader
    expression : str
        polars expression that will be used to filter the loader. For example,
        `col("score") > 0.7` will filter out all low-score molecules.
    """
    loaderlist = self._get_parent()._loaders
    info = loaderlist[loader_name]
    loader = info.loader
    new = loader.filter(norm_expr(expression))
    existing_id = set(new.features[Mole.image])
    loaderlist.add_loader(
        new,
        name=f"{info.name}-Filt",
        image_paths={k: v for k, v in info.image_paths.items() if v in existing_id},
        invert={k: v for k, v in info.invert.items() if v in existing_id},
    )
    return None

get_loader(name)

Return the acryo.BatchLoader object with the given name

Source code in cylindra/widgets/batch/sta.py 
236
237
238
239
240
241
@nogui
def get_loader(self, name: str) -> BatchLoader:
    """Return the acryo.BatchLoader object with the given name"""
    if not isinstance(name, str):
        raise TypeError(f"Name must be a string, got {type(name).__name__}")
    return self._get_parent().loader_infos[name].loader

remove_loader(loader_name)

Remove this loader

Source code in cylindra/widgets/batch/sta.py 
136
137
138
139
140
141
@set_design(text="✕", max_width=36, location=Header)
def remove_loader(
    self, loader_name: Annotated[str, {"bind": _get_current_loader_name}]
):
    """Remove this loader"""
    del self._get_parent().loader_infos[loader_name]

show_loader_info()

Show information about this loader

Source code in cylindra/widgets/batch/sta.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@set_design(text="??", max_width=36, location=Header)
@do_not_record
def show_loader_info(self):
    """Show information about this loader"""
    info = self._get_parent().loader_infos[self.loader_name]
    loader = info.loader
    img_info = "\n" + "\n".join(
        f"{img_id}: {img_path}" for img_id, img_path in info.image_paths.items()
    )

    info_text = (
        f"name: {info.name}\nmolecule: n={loader.count()}\nimages:{img_info}"
    )
    view = DataFrameView(value=loader.molecules.to_dataframe())
    txt = ConsoleTextEdit(value=info_text)
    txt.read_only = True
    cnt = Container(widgets=[txt, view], layout="horizontal", labels=False)
    cnt.native.setParent(self.native, cnt.native.windowFlags())
    cnt.show()

show_mask()

Load and show mask image in the scale of the tomogram.

Source code in cylindra/widgets/batch/sta.py 
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
@set_design(icon="fluent:shape-organic-20-filled", location=STATools)
@do_not_record
def show_mask(self):
    """Load and show mask image in the scale of the tomogram."""
    loader = self.get_loader(self.loader_name)
    _, mask = loader.normalize_input(
        self.params._norm_template_param(
            self.params._get_template_input(allow_multiple=False),
            allow_none=True,
        ),
        self.params._get_mask(),
    )
    if mask is None:
        raise ValueError("No mask to show.")
    mask = ip.asarray(mask, axes="zyx").set_scale(zyx=loader.scale, unit="nm")
    self._show_rec(mask, name="Mask image", store=False)

show_template()

Load and show template image in the scale of the tomogram.

Source code in cylindra/widgets/batch/sta.py 
542
543
544
545
546
547
@set_design(icon="ic:baseline-view-in-ar", location=STATools)
@do_not_record
def show_template(self):
    """Load and show template image in the scale of the tomogram."""
    template = self._get_template_image()
    self._show_rec(template, name="Template image", store=False)

show_template_original()

Load and show template image in the original scale.

Source code in cylindra/widgets/batch/sta.py 
549
550
551
552
553
554
555
556
557
558
559
560
561
@set_design(icon="material-symbols:view-in-ar", location=STATools)
@do_not_record
def show_template_original(self):
    """Load and show template image in the original scale."""
    _input = self.params._get_template_input(allow_multiple=True)
    if _input is None:
        raise ValueError("No template path provided.")
    elif isinstance(_input, Path):
        self._show_rec(ip.imread(_input), name="Template image", store=False)
    else:
        for i, fp in enumerate(_input):
            img = ip.imread(fp)
            self._show_rec(img, name=f"Template image [{i}]", store=False)

split_and_average(loader_name, n_pairs=1, size=None, interpolation=1, bin_size=1)

Split molecules into two groups and average separately.

Parameters:

Name Type Description Default
loader_name str

Name of the batch subtomogram loader to be used.

 required
size nm

Size of the template in nm. Use the size of template image by default.

 None
n_pairs int

How many pairs of average will be calculated.

 is 1
size nm

Size of the template in nm. Use the size of template image by default.

 None
interpolation int

Interpolation order.

 1
bin_size int

Bin size of multiscale image to be used. Set to >1 to boost performance.

 1
Source code in cylindra/widgets/batch/sta.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@set_design(text="Split and average molecules", location=BatchSubtomogramAnalysis)
@dask_thread_worker.with_progress(desc="Split-and-average")
def split_and_average(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    n_pairs: Annotated[int, {"min": 1, "label": "number of image pairs"}] = 1,
    size: _SubVolumeSize = None,
    interpolation: Annotated[int, {"choices": INTERPOLATION_CHOICES}] = 1,
    bin_size: _BINSIZE = 1,
):
    """
    Split molecules into two groups and average separately.

    Parameters
    ----------
    {loader_name}{size}
    n_pairs : int, default is 1
        How many pairs of average will be calculated.
    {size}{interpolation}{bin_size}
    """
    t0 = timer()
    loader = self._get_parent().loader_infos[loader_name].loader
    shape = self._get_shape_in_px(size, loader)

    axes = "ipzyx" if n_pairs > 1 else "pzyx"
    img = ip.asarray(
        loader.replace(output_shape=shape, order=interpolation)
        .binning(bin_size, compute=False)
        .average_split(n_pairs),
        axes=axes,
    ).set_scale(zyx=loader.scale * bin_size, unit="nm")
    t0.toc()
    return self._show_rec.with_args(img, f"[Split]{loader_name}", store=False)

split_loader(loader_name, by, delete_old=False)

Split the selected loader by the values of the given column.

Parameters:

Name Type Description Default
loader_name str

Name of the input loader

 required
by str

Column name to split the loader

 required
delete_old bool

If true, the original loader will be deleted.

 False
Source code in cylindra/widgets/batch/sta.py 
158
159
160
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
203
204
@set_design(text="Split loader", location=BatchLoaderMenu)
def split_loader(
    self,
    loader_name: Annotated[str, {"bind": _get_current_loader_name}],
    by: Annotated[str, {"choices": _get_selected_loader_choice}],
    delete_old: bool = False,
):
    """
    Split the selected loader by the values of the given column.

    Parameters
    ----------
    loader_name : str
        Name of the input loader
    by : str
        Column name to split the loader
    delete_old : bool, default False
        If true, the original loader will be deleted.
    """
    parent = self._get_parent()
    loaders = parent._loaders
    batch_info = loaders[loader_name]
    batch_loader = batch_info.loader
    n_unique = batch_loader.molecules.features[by].n_unique()
    if n_unique > 48:
        raise ValueError(
            f"Too many groups ({n_unique}). Did you choose a float column?"
        )
    for _key, loader in batch_loader.groupby(by):
        existing_id = set(loader.features[Mole.image])
        image_paths = {
            k: v for k, v in batch_info.image_paths.items() if v in existing_id
        }
        invert = {k: v for k, v in batch_info.invert.items() if v in existing_id}
        parent._add_loader(loader, f"{loader_name}_{_key}", image_paths, invert)

    if delete_old:
        idx = -1
        for i, info in enumerate(loaders):
            if info.loader is batch_loader:
                idx = i
                break
        else:
            idx = -1
        if idx < 0:
            raise RuntimeError("Loader not found.")
        del loaders[idx]