cylindra.project

CylindraBatchProject

A project of cylindra batch processing.

Source code in cylindra/project/_batch.py

class CylindraBatchProject(BaseProject):
    """A project of cylindra batch processing."""

    datetime: str
    version: str
    dependency_versions: dict[str, str]
    loaders: list[LoaderInfoModel]
    project_path: Path | None = None

    def resolve_path(self, file_dir: PathLike):
        """Resolve the path of the project."""
        file_dir = Path(file_dir)
        self.loaders = [ldr.resolve_path(file_dir) for ldr in self.loaders]
        return self

    @property
    def macro_path(self) -> Path:
        return self.project_path / "script.py"

    @classmethod
    def from_gui(
        cls,
        gui: "CylindraBatchWidget",
        project_dir: Path,
        mole_ext: str = ".csv",
    ) -> "CylindraBatchProject":
        from datetime import datetime

        _versions = get_versions()

        def as_relative(p: Path):
            assert isinstance(p, Path)
            try:
                out = p.relative_to(project_dir)
            except Exception:
                out = p
            return out

        loaders = list[LoaderInfoModel]()
        for info in gui._loaders:
            name = info.name
            loaders.append(
                LoaderInfoModel(
                    molecule=project_dir / f"Molecules-{name}{mole_ext}",
                    name=name,
                    images=[
                        ImageInfo(
                            id=id,
                            image=as_relative(fp),
                            scale=info.loader.scale,
                            invert=info.invert.get(id, False),
                        )
                        for id, fp in info.image_paths.items()
                    ],
                    scale=info.loader.scale,
                )
            )
        return cls(
            datetime=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
            version=next(iter(_versions.values())),
            dependency_versions=_versions,
            loaders=loaders,
            project_path=project_dir,
        )

    @classmethod
    def save_gui(
        cls: "type[CylindraBatchProject]",
        gui: "CylindraBatchWidget",
        project_dir: Path,
        mole_ext: str = ".csv",
    ) -> None:
        """Save the GUI state to a project directory."""
        self = cls.from_gui(gui, project_dir, mole_ext)

        if not project_dir.exists():
            project_dir.mkdir()

        # save molecules
        for lmodel, info in zip(self.loaders, gui._loaders, strict=True):
            info.loader.molecules.to_file(lmodel.molecule)

        self.project_path.joinpath("script.py").write_text(as_main_function(gui.macro))

        # save objects
        self.to_json(project_dir / "project.json")
        return None

    def _to_gui(self, gui: "CylindraBatchWidget") -> None:
        import impy as ip
        from acryo import BatchLoader, Molecules

        for lmodel in self.loaders:
            loader = BatchLoader(scale=lmodel.scale)
            mole_dict = dict(Molecules.from_file(lmodel.molecule).groupby(Mole.image))
            for imginfo in lmodel.images:
                loader.add_tomogram(
                    image=ip.lazy.imread(imginfo.image, chunks=get_config().dask_chunk)
                    .set_scale(zyx=imginfo.scale)
                    .value,
                    molecules=mole_dict[imginfo.id],
                    image_id=imginfo.id,
                )
            image_paths = {imginfo.id: imginfo.image for imginfo in lmodel.images}
            invert = {imginfo.id: imginfo.invert for imginfo in lmodel.images}
            gui._add_loader(loader, lmodel.name, image_paths, invert)

        txt = self.project_path.joinpath("script.py").read_text()
        macro = mk.parse(txt)
        gui.macro.extend(macro.args)
        gui.reset_choices()
        return None

resolve_path(file_dir)

Resolve the path of the project.

Source code in cylindra/project/_batch.py

def resolve_path(self, file_dir: PathLike):
    """Resolve the path of the project."""
    file_dir = Path(file_dir)
    self.loaders = [ldr.resolve_path(file_dir) for ldr in self.loaders]
    return self

save_gui(gui, project_dir, mole_ext='.csv') classmethod

Save the GUI state to a project directory.

Source code in cylindra/project/_batch.py

@classmethod
def save_gui(
    cls: "type[CylindraBatchProject]",
    gui: "CylindraBatchWidget",
    project_dir: Path,
    mole_ext: str = ".csv",
) -> None:
    """Save the GUI state to a project directory."""
    self = cls.from_gui(gui, project_dir, mole_ext)

    if not project_dir.exists():
        project_dir.mkdir()

    # save molecules
    for lmodel, info in zip(self.loaders, gui._loaders, strict=True):
        info.loader.molecules.to_file(lmodel.molecule)

    self.project_path.joinpath("script.py").write_text(as_main_function(gui.macro))

    # save objects
    self.to_json(project_dir / "project.json")
    return None

CylindraProject

A project of cylindra.

Source code in cylindra/project/_single.py

class CylindraProject(BaseProject):
    """A project of cylindra."""

    # this allows extra fields in the json file, for backward compatibility
    model_config = ConfigDict(extra="allow")

    datetime: str
    version: str
    dependency_versions: dict[str, str]
    image: PathLike | None
    image_relative: PathLike | None = None
    scale: float
    invert: bool = False
    multiscales: list[int]
    molecules_info: list[MoleculesInfo] = Field(default_factory=list)
    landscape_info: list[LandscapeInfo] = Field(default_factory=list)
    missing_wedge: MissingWedge = MissingWedge(params={}, kind="none")
    project_path: Path | None = None
    project_description: str = ""

    def resolve_path(self, file_dir: PathLike):
        """Resolve the path of the project."""
        file_dir = Path(file_dir)
        self.image = resolve_path(self.image, file_dir)
        return self

    @classmethod
    def new(
        cls,
        image: PathLike,
        scale: float | None,
        multiscales: list[int],
        missing_wedge: tuple[float, float] | None = None,
        project_path: Path | None = None,
    ):
        """Create a new project."""
        _versions = get_versions()
        if image is None:
            raise ValueError("image must not be None.")
        if scale is None:
            import impy as ip

            img = ip.lazy.imread(image)
            scale = img.scale.x
        return CylindraProject(
            datetime=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
            version=_versions.pop("cylindra", "unknown"),
            dependency_versions=_versions,
            image=image,
            scale=scale,
            multiscales=multiscales,
            missing_wedge=MissingWedge.parse(missing_wedge),
            project_path=project_path,
        )

    def save(
        self,
        project_dir: Path,
        molecules: "dict[str, Molecules]" = {},
    ) -> None:
        """Save this project."""
        from macrokit import parse

        path = Path(self.image).as_posix()
        scale = self.scale
        bin_size = self.multiscales
        tilt_range = self.missing_wedge.as_param()
        with _prep_save_dir(project_dir) as results_dir:
            expr_open = parse(
                f"ui.open_image({path=}, {scale=:.4f}, {bin_size=}, {tilt_range=})",
                squeeze=False,
            )
            expr = as_main_function(expr_open)
            self._script_py_path(results_dir).write_text(expr)
            self_copy = self.model_copy()
            for name, mole in molecules.items():
                save_path = results_dir / name
                if save_path.suffix == "":
                    save_path = save_path.with_suffix(".csv")
                self_copy.molecules_info.append(MoleculesInfo(name=save_path.name))
                mole.to_file(save_path)
            self_copy.to_json(self._project_json_path(results_dir))
        return None

    @classmethod
    def from_gui(
        cls,
        gui: "CylindraMainWidget",
        project_dir: Path,
        mole_ext: str = ".csv",
        save_landscape: bool = False,
    ) -> "CylindraProject":
        """Construct a project from a widget state."""

        _versions = get_versions()
        tomo = gui.tomogram

        # Save path of molecules
        mole_infos = list[MoleculesInfo]()
        for layer in gui.mole_layers:
            mole_infos.append(MoleculesInfo.from_layer(gui, layer, mole_ext))

        # Save paths of landscape
        landscape_infos = list[LandscapeInfo]()
        if save_landscape:
            from cylindra._napari import LandscapeSurface

            for layer in gui.parent_viewer.layers:
                if not isinstance(layer, LandscapeSurface):
                    continue
                landscape_infos.append(LandscapeInfo.from_layer(gui, layer))

        return cls(
            datetime=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
            version=_versions.pop("cylindra", "unknown"),
            dependency_versions=_versions,
            image=tomo.metadata.get("source", None),
            image_relative=_as_relative(tomo.metadata.get("source", None), project_dir),
            scale=tomo.scale,
            invert=tomo.is_inverted,
            multiscales=[x[0] for x in tomo.multiscaled],
            molecules_info=mole_infos,
            landscape_info=landscape_infos,
            missing_wedge=MissingWedge.parse(tomo.tilt),
            project_path=project_dir,
        )

    @classmethod
    def save_gui(
        cls: "type[CylindraProject]",
        gui: "CylindraMainWidget",
        project_dir: Path,
        mole_ext: str = ".csv",
        save_landscape: bool = False,
    ) -> None:
        """
        Serialize the GUI state to a json file.

        Parameters
        ----------
        gui : CylindraMainWidget
            The main widget from which project model will be constructed.
        project_dir : Path
            The path to the project json file.
        """
        self = cls.from_gui(gui, project_dir, mole_ext, save_landscape)

        tomo = gui.tomogram
        localprops = tomo.splines.collect_localprops(allow_none=True)
        globalprops = tomo.splines.collect_globalprops(allow_none=True)

        with _prep_save_dir(project_dir) as results_dir:
            if localprops is not None:
                localprops.write_csv(self._localprops_path(results_dir))
            if globalprops is not None:
                globalprops.write_csv(self._globalprops_path(results_dir))
            for i, spl in enumerate(gui.tomogram.splines):
                spl.to_json(results_dir / f"spline-{i}.json")
            for info in self.molecules_info + self.landscape_info:
                info.save_layer(gui, results_dir)

            _cfg_json = gui.default_config.json_dumps()
            self._default_spline_config_path(results_dir).write_text(_cfg_json)

            # save macro
            expr = as_main_function(
                gui._format_macro(gui.macro[gui._macro_offset :]),
                imports=[plg.import_statement() for plg in gui._plugins_called],
            )
            self._script_py_path(results_dir).write_text(expr)

            self.project_description = gui.GeneralInfo.project_desc.value
            self.to_json(self._project_json_path(results_dir))
        return None

    def _to_gui(
        self,
        gui: "CylindraMainWidget | None" = None,
        filter: "ImageFilter | None" = True,
        read_image: bool = True,
        update_config: bool = True,
    ):
        """Update CylindraMainWidget state based on the project model."""
        from magicclass.utils import thread_worker

        from cylindra.components import SplineConfig

        gui = _get_instance(gui)
        with self.open_project() as project_dir:
            tomogram = self.load_tomogram(project_dir, compute=read_image)
            macro_expr = extract(self._script_py_path(project_dir).read_text()).args
            cfg_path = project_dir / "default_spline_config.json"
            if cfg_path.exists() and update_config:
                default_config = SplineConfig.from_file(cfg_path)
            else:
                default_config = None

            cb = gui._send_tomogram_to_viewer.with_args(
                tomogram, filt=filter, invert=self.invert
            )
            yield cb
            cb.await_call()
            gui._init_macro_state()

            @thread_worker.callback
            def _update_widget():
                if len(tomogram.splines) > 0:
                    gui._update_splines_in_images()
                    with gui.macro.blocked():
                        gui.sample_subtomograms()
                if default_config is not None:
                    gui.default_config = default_config

                gui.macro.extend(macro_expr)

                # load subtomogram analyzer state
                gui.reset_choices()
                gui._need_save = False

            yield _update_widget
            _update_widget.await_call()

            # load molecules and landscapes
            _add_layer = thread_worker.callback(gui.parent_viewer.add_layer)
            with gui._pend_reset_choices():
                for info in self.molecules_info + self.landscape_info:
                    layer = info.to_layer(gui, project_dir)
                    cb = _add_layer.with_args(layer)
                    yield cb
                    cb.await_call(timeout=10)

        @thread_worker.callback
        def out():
            # update project description widget
            gui.GeneralInfo.project_desc.value = self.project_description
            gui.reset_choices()

        return out

    def load_spline(
        self,
        idx: int,
        dir: Path | None = None,
        props: bool = True,
    ) -> "CylSpline":
        """
        Load the spline of the given index.

        >>> spl = project.load_spline(0)  # load the 0-th spline instance

        Parameters
        ----------
        idx : int
            The index of the spline.
        dir : Path, optional
            The project directory. Can be given if the project is already opened.
        props : bool, default True
            Whether to load the properties of the spline.

        Returns
        -------
        CylSpline
            The spline instance.
        """
        from cylindra.components import CylSpline

        if dir is None:
            with self.open_project() as dir:
                return self.load_spline(idx, dir, props)
        spl = CylSpline.from_json(dir / f"spline-{idx}.json")
        if not props:
            return spl
        localprops_path = self._localprops_path(dir)
        globalprops_path = self._globalprops_path(dir)
        if localprops_path.exists():
            _loc = pl.read_csv(localprops_path).filter(pl.col(H.spline_id) == idx)
            _loc = _drop_null_columns(_loc)
        else:
            _loc = pl.DataFrame([])
        if globalprops_path.exists():
            _glob = pl.read_csv(globalprops_path)[idx]
            _glob = _drop_null_columns(_glob)
        else:
            _glob = pl.DataFrame([])

        if H.spl_dist in _loc.columns:
            _loc = _loc.drop(H.spl_dist)
        if H.spl_pos in _loc.columns:
            spl._anchors = _loc[H.spl_pos].to_numpy()
            _loc = _loc.drop(H.spl_pos)
        for c in [H.spline_id, H.pos_id]:
            if c in _loc.columns:
                _loc = _loc.drop(c)
        spl.props.loc = cast_dataframe(_loc)
        spl.props.glob = cast_dataframe(_glob)

        return spl

    def iter_spline_paths(
        self, dir: Path | None = None
    ) -> "Iterable[tuple[int, Path]]":
        """Iterate over the paths of splines and their indices."""
        if dir is None:
            with self.open_project() as dir:
                paths = list(dir.glob("spline-*.json"))
        else:
            paths = list(dir.glob("spline-*.json"))
        # sort by index
        idx_paths = [(int(p.stem.split("-")[1]), p) for p in paths]
        idx_paths.sort(key=lambda x: x[0])
        yield from idx_paths

    def iter_load_splines(
        self,
        dir: Path | None = None,
        drop_columns: bool = True,
    ) -> "Iterable[CylSpline]":
        """Load all splines including its properties iteratively."""
        from cylindra.components import CylSpline

        if dir is None:
            with self.open_project() as dir:
                yield from self.iter_load_splines(dir, drop_columns)
            return

        localprops_path = self._localprops_path(dir)
        globalprops_path = self._globalprops_path(dir)
        if localprops_path.exists():
            _localprops = pl.read_csv(localprops_path)
        else:
            _localprops = None
        if globalprops_path.exists():
            _globalprops = pl.read_csv(globalprops_path)
        else:
            _globalprops = None
        for idx, spl_path in self.iter_spline_paths(dir):
            spl = CylSpline.from_json(spl_path)
            if _localprops is not None:
                _loc = _localprops.filter(pl.col(H.spline_id) == idx)
                _loc = _drop_null_columns(_loc)
                if len(_loc) == 0:
                    _loc = pl.DataFrame([])
            else:
                _loc = pl.DataFrame([])
            if _globalprops is not None:
                _glob = _globalprops.filter(pl.col(H.spline_id) == idx)
                _glob = _drop_null_columns(_glob)
                if len(_glob) == 0:
                    _glob = pl.DataFrame([])
            else:
                _glob = pl.DataFrame([])

            if H.spl_dist in _loc.columns and drop_columns:
                _loc = _loc.drop(H.spl_dist)
            if H.spl_pos in _loc.columns:
                spl._anchors = _loc[H.spl_pos].to_numpy()
                if drop_columns:
                    _loc = _loc.drop(H.spl_pos)
            for c in [H.spline_id, H.pos_id]:
                if c in _loc.columns and drop_columns:
                    _loc = _loc.drop(c)
            spl.props.loc = cast_dataframe(_loc)
            spl.props.glob = cast_dataframe(_glob)
            yield spl

    def iter_load_molecules(
        self, dir: Path | None = None
    ) -> "Iterable[tuple[MoleculesInfo, Molecules]]":
        """Load all molecules iteratively."""
        from acryo import Molecules

        if dir is None:
            with self.open_project() as dir:
                yield from self.iter_load_molecules(dir)
            return

        for info in self.molecules_info:
            path = dir / info.name
            if not path.exists():
                LOGGER.warning(
                    f"Cannot find molecule file: {path.as_posix()}. "
                    "Probably it was moved?"
                )
                continue
            mole = Molecules.from_file(path)
            yield info, mole

    def load_molecules(self, name: str, dir: Path | None = None) -> "Molecules":
        """
        Load the molecule with the given name.

        >>> mole = project.load_molecules("Mole-0")  # load one with name "Mole-0"

        Parameters
        ----------
        name : str
            Name of the molecules layer.
        dir : Path, optional
            The project directory. Can be given if the project is already opened.

        Returns
        -------
        Molecules
            The molecules instance that matches the given name.
        """
        from acryo import Molecules

        if dir is None:
            with self.open_project() as dir:
                return self.load_molecules(name, dir)
        for info in self.molecules_info:
            if info.name == name or info.stem == name:
                path = dir / info.name
                if not path.exists():
                    raise ValueError(
                        f"Cannot find molecule file: {path.as_posix()}. Probably the "
                        "`dir` parameter is not correct."
                    )
                return Molecules.from_file(path)
        raise ValueError(f"Cannot find molecule with name: {name}.")

    def load_tomogram(
        self,
        dir: Path | None = None,
        compute: bool = True,
    ) -> "CylTomogram":
        """
        Load the tomogram object of the project.

        Parameters
        ----------
        dir : Path, optional
            Can be given if the project is already opened.
        compute : bool, optional
            Whether to compute the binned tomograms.
        """
        from cylindra.components import CylTomogram

        if self.image is not None:
            if self.image.exists():
                tomo = CylTomogram.imread(
                    path=self.image,
                    scale=self.scale,
                    tilt=self.missing_wedge.as_param(),
                    binsize=self.multiscales,
                    compute=compute,
                )
            elif _rpath := self._try_resolve_image_relative():
                tomo = CylTomogram.imread(
                    path=_rpath,
                    scale=self.scale,
                    tilt=self.missing_wedge.as_param(),
                    binsize=self.multiscales,
                    compute=compute,
                )
            else:
                LOGGER.warning(
                    f"Cannot find image file: {self.image.as_posix()}. "
                    "Load other components only.",
                )
                tomo = CylTomogram.dummy(
                    scale=self.scale,
                    tilt=self.missing_wedge.as_param(),
                    binsize=self.multiscales,
                    name="<Image not found>",
                )
        else:
            tomo = CylTomogram.dummy(
                scale=self.scale,
                tilt=self.missing_wedge.as_param(),
                binsize=self.multiscales,
                name="<No image>",
            )
        tomo.splines.extend(self.iter_load_splines(dir))
        return tomo

    def _try_resolve_image_relative(self) -> Path | None:
        if self.image_relative is None or self.project_path is None:
            return None
        cur_dir = self.project_path.parent
        for part in Path(self.image_relative).parts:
            if part.startswith(".."):
                cur_dir = cur_dir.parent
            else:
                cur_dir = cur_dir / part
        if cur_dir.exists():
            return cur_dir
        return None

    def make_project_viewer(self):
        """Build a project viewer widget from this project."""
        from cylindra.project._widgets import ProjectViewer

        pviewer = ProjectViewer()
        pviewer._from_project(self)
        return pviewer

    def make_component_viewer(self):
        from cylindra.project._widgets import ComponentsViewer

        """Build a molecules viewer widget from this project."""
        mviewer = ComponentsViewer()
        mviewer._from_project(self)
        return mviewer

    @contextmanager
    def open_project(self) -> Generator[Path, None, None]:
        """Open the project within this context."""
        if self.project_path is None:
            raise ValueError("Project path is not set.")
        ext = self.project_path.suffix
        if ext == "":
            yield self.project_path

        elif ext in (".tar",):
            import tarfile

            with tempfile.TemporaryDirectory() as tmpdir:
                with tarfile.open(self.project_path) as tar:
                    tar.extractall(tmpdir)
                yield Path(tmpdir)

        elif ext in (".zip",):
            import zipfile

            with tempfile.TemporaryDirectory() as tmpdir:
                with zipfile.ZipFile(self.project_path) as _zip:
                    _zip.extractall(tmpdir)
                yield Path(tmpdir)

        else:
            raise ValueError(f"Unsupported extension {ext}.")

        return None

    def rewrite(self, dir: Path):
        """
        Rewrite tar/zip file using given temporary directory.

        This method is only used after some mutable operation on the
        project directory.
        """
        if self.project_path is None:
            raise ValueError("Project path is not set.")

        ext = self.project_path.suffix
        if ext == "":
            return

        elif ext in (".tar",):
            import tarfile

            self.project_path.unlink()

            with tarfile.open(self.project_path, mode="w") as tar:
                for file in Path(dir).glob("*"):
                    tar.add(file, arcname=file.name)

        elif ext in (".zip",):
            import zipfile

            self.project_path.unlink()
            with zipfile.ZipFile(self.project_path, mode="w") as zip:
                for file in Path(dir).glob("*"):
                    zip.write(file, arcname=file.name)

        else:
            raise ValueError(f"Unsupported extension {ext}.")

        return None

    def _localprops_path(self, dir: Path) -> Path:
        """Path to the spline local properties file."""
        return dir / "localprops.csv"

    def _globalprops_path(self, dir: Path) -> Path:
        """Path to the spline global properties file."""
        return dir / "globalprops.csv"

    def _default_spline_config_path(self, dir: Path) -> Path:
        """Path to the default spline config file."""
        return dir / "default_spline_config.json"

    def _script_py_path(self, dir: Path) -> Path:
        """Path to the script.py file."""
        return dir / "script.py"

    def _project_json_path(self, dir: Path) -> Path:
        """Path to the project.json file."""
        return dir / "project.json"

from_gui(gui, project_dir, mole_ext='.csv', save_landscape=False) classmethod

Construct a project from a widget state.

Source code in cylindra/project/_single.py

@classmethod
def from_gui(
    cls,
    gui: "CylindraMainWidget",
    project_dir: Path,
    mole_ext: str = ".csv",
    save_landscape: bool = False,
) -> "CylindraProject":
    """Construct a project from a widget state."""

    _versions = get_versions()
    tomo = gui.tomogram

    # Save path of molecules
    mole_infos = list[MoleculesInfo]()
    for layer in gui.mole_layers:
        mole_infos.append(MoleculesInfo.from_layer(gui, layer, mole_ext))

    # Save paths of landscape
    landscape_infos = list[LandscapeInfo]()
    if save_landscape:
        from cylindra._napari import LandscapeSurface

        for layer in gui.parent_viewer.layers:
            if not isinstance(layer, LandscapeSurface):
                continue
            landscape_infos.append(LandscapeInfo.from_layer(gui, layer))

    return cls(
        datetime=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
        version=_versions.pop("cylindra", "unknown"),
        dependency_versions=_versions,
        image=tomo.metadata.get("source", None),
        image_relative=_as_relative(tomo.metadata.get("source", None), project_dir),
        scale=tomo.scale,
        invert=tomo.is_inverted,
        multiscales=[x[0] for x in tomo.multiscaled],
        molecules_info=mole_infos,
        landscape_info=landscape_infos,
        missing_wedge=MissingWedge.parse(tomo.tilt),
        project_path=project_dir,
    )

iter_load_molecules(dir=None)

Load all molecules iteratively.

Source code in cylindra/project/_single.py

def iter_load_molecules(
    self, dir: Path | None = None
) -> "Iterable[tuple[MoleculesInfo, Molecules]]":
    """Load all molecules iteratively."""
    from acryo import Molecules

    if dir is None:
        with self.open_project() as dir:
            yield from self.iter_load_molecules(dir)
        return

    for info in self.molecules_info:
        path = dir / info.name
        if not path.exists():
            LOGGER.warning(
                f"Cannot find molecule file: {path.as_posix()}. "
                "Probably it was moved?"
            )
            continue
        mole = Molecules.from_file(path)
        yield info, mole

iter_load_splines(dir=None, drop_columns=True)

Load all splines including its properties iteratively.

Source code in cylindra/project/_single.py

def iter_load_splines(
    self,
    dir: Path | None = None,
    drop_columns: bool = True,
) -> "Iterable[CylSpline]":
    """Load all splines including its properties iteratively."""
    from cylindra.components import CylSpline

    if dir is None:
        with self.open_project() as dir:
            yield from self.iter_load_splines(dir, drop_columns)
        return

    localprops_path = self._localprops_path(dir)
    globalprops_path = self._globalprops_path(dir)
    if localprops_path.exists():
        _localprops = pl.read_csv(localprops_path)
    else:
        _localprops = None
    if globalprops_path.exists():
        _globalprops = pl.read_csv(globalprops_path)
    else:
        _globalprops = None
    for idx, spl_path in self.iter_spline_paths(dir):
        spl = CylSpline.from_json(spl_path)
        if _localprops is not None:
            _loc = _localprops.filter(pl.col(H.spline_id) == idx)
            _loc = _drop_null_columns(_loc)
            if len(_loc) == 0:
                _loc = pl.DataFrame([])
        else:
            _loc = pl.DataFrame([])
        if _globalprops is not None:
            _glob = _globalprops.filter(pl.col(H.spline_id) == idx)
            _glob = _drop_null_columns(_glob)
            if len(_glob) == 0:
                _glob = pl.DataFrame([])
        else:
            _glob = pl.DataFrame([])

        if H.spl_dist in _loc.columns and drop_columns:
            _loc = _loc.drop(H.spl_dist)
        if H.spl_pos in _loc.columns:
            spl._anchors = _loc[H.spl_pos].to_numpy()
            if drop_columns:
                _loc = _loc.drop(H.spl_pos)
        for c in [H.spline_id, H.pos_id]:
            if c in _loc.columns and drop_columns:
                _loc = _loc.drop(c)
        spl.props.loc = cast_dataframe(_loc)
        spl.props.glob = cast_dataframe(_glob)
        yield spl

iter_spline_paths(dir=None)

Iterate over the paths of splines and their indices.

Source code in cylindra/project/_single.py

def iter_spline_paths(
    self, dir: Path | None = None
) -> "Iterable[tuple[int, Path]]":
    """Iterate over the paths of splines and their indices."""
    if dir is None:
        with self.open_project() as dir:
            paths = list(dir.glob("spline-*.json"))
    else:
        paths = list(dir.glob("spline-*.json"))
    # sort by index
    idx_paths = [(int(p.stem.split("-")[1]), p) for p in paths]
    idx_paths.sort(key=lambda x: x[0])
    yield from idx_paths

load_molecules(name, dir=None)

Load the molecule with the given name.

mole = project.load_molecules("Mole-0")  # load one with name "Mole-0"

Parameters:

Name	Type	Description	Default
name	str	Name of the molecules layer.	required
dir	Path	The project directory. Can be given if the project is already opened.	None

Returns:

Type	Description
Molecules	The molecules instance that matches the given name.

Source code in cylindra/project/_single.py

def load_molecules(self, name: str, dir: Path | None = None) -> "Molecules":
    """
    Load the molecule with the given name.

    >>> mole = project.load_molecules("Mole-0")  # load one with name "Mole-0"

    Parameters
    ----------
    name : str
        Name of the molecules layer.
    dir : Path, optional
        The project directory. Can be given if the project is already opened.

    Returns
    -------
    Molecules
        The molecules instance that matches the given name.
    """
    from acryo import Molecules

    if dir is None:
        with self.open_project() as dir:
            return self.load_molecules(name, dir)
    for info in self.molecules_info:
        if info.name == name or info.stem == name:
            path = dir / info.name
            if not path.exists():
                raise ValueError(
                    f"Cannot find molecule file: {path.as_posix()}. Probably the "
                    "`dir` parameter is not correct."
                )
            return Molecules.from_file(path)
    raise ValueError(f"Cannot find molecule with name: {name}.")

load_spline(idx, dir=None, props=True)

Load the spline of the given index.

spl = project.load_spline(0)  # load the 0-th spline instance

Parameters:

Name	Type	Description	Default
idx	int	The index of the spline.	required
dir	Path	The project directory. Can be given if the project is already opened.	None
props	bool	Whether to load the properties of the spline.	True

Returns:

Type	Description
CylSpline	The spline instance.

Source code in cylindra/project/_single.py

def load_spline(
    self,
    idx: int,
    dir: Path | None = None,
    props: bool = True,
) -> "CylSpline":
    """
    Load the spline of the given index.

    >>> spl = project.load_spline(0)  # load the 0-th spline instance

    Parameters
    ----------
    idx : int
        The index of the spline.
    dir : Path, optional
        The project directory. Can be given if the project is already opened.
    props : bool, default True
        Whether to load the properties of the spline.

    Returns
    -------
    CylSpline
        The spline instance.
    """
    from cylindra.components import CylSpline

    if dir is None:
        with self.open_project() as dir:
            return self.load_spline(idx, dir, props)
    spl = CylSpline.from_json(dir / f"spline-{idx}.json")
    if not props:
        return spl
    localprops_path = self._localprops_path(dir)
    globalprops_path = self._globalprops_path(dir)
    if localprops_path.exists():
        _loc = pl.read_csv(localprops_path).filter(pl.col(H.spline_id) == idx)
        _loc = _drop_null_columns(_loc)
    else:
        _loc = pl.DataFrame([])
    if globalprops_path.exists():
        _glob = pl.read_csv(globalprops_path)[idx]
        _glob = _drop_null_columns(_glob)
    else:
        _glob = pl.DataFrame([])

    if H.spl_dist in _loc.columns:
        _loc = _loc.drop(H.spl_dist)
    if H.spl_pos in _loc.columns:
        spl._anchors = _loc[H.spl_pos].to_numpy()
        _loc = _loc.drop(H.spl_pos)
    for c in [H.spline_id, H.pos_id]:
        if c in _loc.columns:
            _loc = _loc.drop(c)
    spl.props.loc = cast_dataframe(_loc)
    spl.props.glob = cast_dataframe(_glob)

    return spl

load_tomogram(dir=None, compute=True)

Load the tomogram object of the project.

Parameters:

Name	Type	Description	Default
dir	Path	Can be given if the project is already opened.	None
compute	bool	Whether to compute the binned tomograms.	True

Source code in cylindra/project/_single.py

def load_tomogram(
    self,
    dir: Path | None = None,
    compute: bool = True,
) -> "CylTomogram":
    """
    Load the tomogram object of the project.

    Parameters
    ----------
    dir : Path, optional
        Can be given if the project is already opened.
    compute : bool, optional
        Whether to compute the binned tomograms.
    """
    from cylindra.components import CylTomogram

    if self.image is not None:
        if self.image.exists():
            tomo = CylTomogram.imread(
                path=self.image,
                scale=self.scale,
                tilt=self.missing_wedge.as_param(),
                binsize=self.multiscales,
                compute=compute,
            )
        elif _rpath := self._try_resolve_image_relative():
            tomo = CylTomogram.imread(
                path=_rpath,
                scale=self.scale,
                tilt=self.missing_wedge.as_param(),
                binsize=self.multiscales,
                compute=compute,
            )
        else:
            LOGGER.warning(
                f"Cannot find image file: {self.image.as_posix()}. "
                "Load other components only.",
            )
            tomo = CylTomogram.dummy(
                scale=self.scale,
                tilt=self.missing_wedge.as_param(),
                binsize=self.multiscales,
                name="<Image not found>",
            )
    else:
        tomo = CylTomogram.dummy(
            scale=self.scale,
            tilt=self.missing_wedge.as_param(),
            binsize=self.multiscales,
            name="<No image>",
        )
    tomo.splines.extend(self.iter_load_splines(dir))
    return tomo

make_project_viewer()

Build a project viewer widget from this project.

Source code in cylindra/project/_single.py

def make_project_viewer(self):
    """Build a project viewer widget from this project."""
    from cylindra.project._widgets import ProjectViewer

    pviewer = ProjectViewer()
    pviewer._from_project(self)
    return pviewer

new(image, scale, multiscales, missing_wedge=None, project_path=None) classmethod

Create a new project.

Source code in cylindra/project/_single.py

@classmethod
def new(
    cls,
    image: PathLike,
    scale: float | None,
    multiscales: list[int],
    missing_wedge: tuple[float, float] | None = None,
    project_path: Path | None = None,
):
    """Create a new project."""
    _versions = get_versions()
    if image is None:
        raise ValueError("image must not be None.")
    if scale is None:
        import impy as ip

        img = ip.lazy.imread(image)
        scale = img.scale.x
    return CylindraProject(
        datetime=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
        version=_versions.pop("cylindra", "unknown"),
        dependency_versions=_versions,
        image=image,
        scale=scale,
        multiscales=multiscales,
        missing_wedge=MissingWedge.parse(missing_wedge),
        project_path=project_path,
    )

open_project()

Open the project within this context.

Source code in cylindra/project/_single.py

@contextmanager
def open_project(self) -> Generator[Path, None, None]:
    """Open the project within this context."""
    if self.project_path is None:
        raise ValueError("Project path is not set.")
    ext = self.project_path.suffix
    if ext == "":
        yield self.project_path

    elif ext in (".tar",):
        import tarfile

        with tempfile.TemporaryDirectory() as tmpdir:
            with tarfile.open(self.project_path) as tar:
                tar.extractall(tmpdir)
            yield Path(tmpdir)

    elif ext in (".zip",):
        import zipfile

        with tempfile.TemporaryDirectory() as tmpdir:
            with zipfile.ZipFile(self.project_path) as _zip:
                _zip.extractall(tmpdir)
            yield Path(tmpdir)

    else:
        raise ValueError(f"Unsupported extension {ext}.")

    return None

resolve_path(file_dir)

Resolve the path of the project.

Source code in cylindra/project/_single.py

def resolve_path(self, file_dir: PathLike):
    """Resolve the path of the project."""
    file_dir = Path(file_dir)
    self.image = resolve_path(self.image, file_dir)
    return self

rewrite(dir)

Rewrite tar/zip file using given temporary directory.

This method is only used after some mutable operation on the project directory.

Source code in cylindra/project/_single.py

def rewrite(self, dir: Path):
    """
    Rewrite tar/zip file using given temporary directory.

    This method is only used after some mutable operation on the
    project directory.
    """
    if self.project_path is None:
        raise ValueError("Project path is not set.")

    ext = self.project_path.suffix
    if ext == "":
        return

    elif ext in (".tar",):
        import tarfile

        self.project_path.unlink()

        with tarfile.open(self.project_path, mode="w") as tar:
            for file in Path(dir).glob("*"):
                tar.add(file, arcname=file.name)

    elif ext in (".zip",):
        import zipfile

        self.project_path.unlink()
        with zipfile.ZipFile(self.project_path, mode="w") as zip:
            for file in Path(dir).glob("*"):
                zip.write(file, arcname=file.name)

    else:
        raise ValueError(f"Unsupported extension {ext}.")

    return None

save(project_dir, molecules={})

Save this project.

Source code in cylindra/project/_single.py

def save(
    self,
    project_dir: Path,
    molecules: "dict[str, Molecules]" = {},
) -> None:
    """Save this project."""
    from macrokit import parse

    path = Path(self.image).as_posix()
    scale = self.scale
    bin_size = self.multiscales
    tilt_range = self.missing_wedge.as_param()
    with _prep_save_dir(project_dir) as results_dir:
        expr_open = parse(
            f"ui.open_image({path=}, {scale=:.4f}, {bin_size=}, {tilt_range=})",
            squeeze=False,
        )
        expr = as_main_function(expr_open)
        self._script_py_path(results_dir).write_text(expr)
        self_copy = self.model_copy()
        for name, mole in molecules.items():
            save_path = results_dir / name
            if save_path.suffix == "":
                save_path = save_path.with_suffix(".csv")
            self_copy.molecules_info.append(MoleculesInfo(name=save_path.name))
            mole.to_file(save_path)
        self_copy.to_json(self._project_json_path(results_dir))
    return None

save_gui(gui, project_dir, mole_ext='.csv', save_landscape=False) classmethod

Serialize the GUI state to a json file.

Parameters:

Name	Type	Description	Default
gui	CylindraMainWidget	The main widget from which project model will be constructed.	required
project_dir	Path	The path to the project json file.	required

Source code in cylindra/project/_single.py

@classmethod
def save_gui(
    cls: "type[CylindraProject]",
    gui: "CylindraMainWidget",
    project_dir: Path,
    mole_ext: str = ".csv",
    save_landscape: bool = False,
) -> None:
    """
    Serialize the GUI state to a json file.

    Parameters
    ----------
    gui : CylindraMainWidget
        The main widget from which project model will be constructed.
    project_dir : Path
        The path to the project json file.
    """
    self = cls.from_gui(gui, project_dir, mole_ext, save_landscape)

    tomo = gui.tomogram
    localprops = tomo.splines.collect_localprops(allow_none=True)
    globalprops = tomo.splines.collect_globalprops(allow_none=True)

    with _prep_save_dir(project_dir) as results_dir:
        if localprops is not None:
            localprops.write_csv(self._localprops_path(results_dir))
        if globalprops is not None:
            globalprops.write_csv(self._globalprops_path(results_dir))
        for i, spl in enumerate(gui.tomogram.splines):
            spl.to_json(results_dir / f"spline-{i}.json")
        for info in self.molecules_info + self.landscape_info:
            info.save_layer(gui, results_dir)

        _cfg_json = gui.default_config.json_dumps()
        self._default_spline_config_path(results_dir).write_text(_cfg_json)

        # save macro
        expr = as_main_function(
            gui._format_macro(gui.macro[gui._macro_offset :]),
            imports=[plg.import_statement() for plg in gui._plugins_called],
        )
        self._script_py_path(results_dir).write_text(expr)

        self.project_description = gui.GeneralInfo.project_desc.value
        self.to_json(self._project_json_path(results_dir))
    return None

ProjectSequence

Collection of Cylindra projects.

This object is just for project management. BatchLoader, DataFrame and local/global properties can be generated from this object.

Source code in cylindra/project/_sequence.py

class ProjectSequence(MutableSequence[CylindraProject]):
    """
    Collection of Cylindra projects.

    This object is just for project management. BatchLoader, DataFrame and local/global
    properties can be generated from this object.
    """

    def __init__(self, *, check_scale: bool = True):
        self._projects = list[CylindraProject]()
        self._scale_validator = ScaleValidator(check_scale)

    def __repr__(self) -> str:
        if len(self) > 1:
            return (
                f"{type(self).__name__} with {len(self)} projects such as {self[0]!r}"
            )
        return f"{type(self).__name__} (empty)"

    @overload
    def __getitem__(self, key: int) -> CylindraProject:
        ...

    @overload
    def __getitem__(self, key: slice) -> ProjectSequence:
        ...

    def __getitem__(self, key: int):
        out = self._projects[key]
        if isinstance(key, slice):
            out = ProjectSequence(check_scale=self._scale_validator._check)
            out._projects = self._projects[key]
        return out

    def __setitem__(self, key: int, value: CylindraProject) -> None:
        if not isinstance(value, CylindraProject):
            raise TypeError(f"Expected CylindraProject, got {type(value)}.")
        if not isinstance(key, SupportsIndex):
            raise TypeError(f"Expected int, got {type(key)}.")
        self._projects[key] = value

    def __delitem__(self, key: int) -> None:
        del self._projects[key]
        if len(self) == 0:
            self._scale_validator.initialize()

    def __len__(self) -> int:
        return len(self._projects)

    def __iter__(self) -> Iterator[CylindraProject]:
        return iter(self._projects)

    def insert(self, index: int, value: CylindraProject) -> None:
        """Insert a project at the given index."""
        if not isinstance(value, CylindraProject):
            raise TypeError(f"Expected CylindraProject, got {type(value)}.")
        return self._projects.insert(index, value)

    def __add__(self, other: ProjectSequence) -> ProjectSequence:
        """Concatenate two ProjectSequence objects."""
        if not isinstance(other, ProjectSequence):
            raise TypeError(f"Expected ProjectSequence, got {type(other)}.")
        new = ProjectSequence(check_scale=True)
        new._projects = self._projects + other._projects
        if len(self) > 0:
            new._scale_validator.value = self._scale_validator.value
        if len(other) > 0:
            new._scale_validator.value = other._scale_validator.value
        return new

    @classmethod
    def from_paths(
        cls,
        paths: Iterable[str | Path],
        *,
        check_scale: bool = True,
        skip_exc: bool = False,
    ) -> Self:
        """Add all the projects of the given paths."""
        self = cls(check_scale=check_scale)
        if skip_exc:
            for path in paths:
                with suppress(Exception):
                    self.append_file(path)
        else:
            for path in paths:
                self.append_file(path)
        return self

    def append_file(self, path: str | Path) -> Self:
        """Add a project from a file path."""
        prj = CylindraProject.from_file(path)
        self._scale_validator.value = prj.scale
        self._projects.append(prj)
        return self

    def sta_loader(
        self,
        name_filter: Callable[[str], bool] | None = None,
        *,
        curvature: bool = False,
        allow_no_image: bool = False,
    ) -> BatchLoader:
        """
        Construct a STA loader from all the projects.

        Parameters
        ----------
        name_filter : callable, default None
            Function that takes a molecule file name (without extension) and
            returns True if the molecule should be collected. Collect all the
            molecules by default.
        curvature : bool, default False
            If True, the spline curvature will be added to the molecule features.
        allow_no_image : bool, default False
            If True, this method will not raise an error when the image file is not
            found.
        """
        import impy as ip
        from acryo import BatchLoader

        col = BatchLoader(scale=self._scale_validator.value)
        if name_filter is None:

            def name_filter(_):
                return True

        for idx, prj in enumerate(self._projects):
            if prj.image is None or not prj.image.exists():
                if not allow_no_image:
                    raise ValueError(
                        f"Image file not found in project at {prj.project_path}."
                    )
                import numpy as np

                img = np.zeros((0, 0, 0), dtype=np.float32)  # dummy
            else:
                img = ip.lazy.imread(prj.image, chunks=get_config().dask_chunk).value
            with prj.open_project() as dir:
                for info, mole in prj.iter_load_molecules(dir):
                    if not name_filter(info.stem):
                        continue
                    if (
                        curvature
                        and (_spl_i := info.source) is not None
                        and Mole.position in mole.features.columns
                    ):
                        _spl = prj.load_spline(_spl_i, dir=dir, props=False)
                        _u = _spl.y_to_position(mole.features[Mole.position])
                        cv = _spl.curvature(_u)
                        mole.features = mole.features.with_columns(
                            pl.Series(cv, dtype=pl.Float32).alias("spline_curvature")
                        )
                    mole.features = mole.features.with_columns(
                        pl.repeat(info.stem, pl.len()).alias(Mole.id)
                    )
                    col.add_tomogram(img, molecules=mole, image_id=idx)
        return col

    def collect_localprops(
        self,
        allow_none: bool = True,
        id: _IDTYPE = "int",
        spline_details: bool = False,
    ) -> pl.DataFrame:
        """
        Collect all localprops into a single dataframe.

        Parameters
        ----------
        allow_none : bool, default True
            Continue data collection even if property table data file was not
            found in any project. Raise error otherwise.
        id : str, default "int"
            How to describe the source tomogram. If "int", each tomogram will
            be labeled with ascending integers. If "path", each tomogram will
            be labeled with the name of the project directory.
        spline_details : bool, default False
            If True, spline coordinates, its derivatives and the curvature
            will also be collected as well. This will take more memory and time.

        Returns
        -------
        pl.DataFrame
            Dataframe with all the properties.
        """
        dfs_prj = list[pl.DataFrame]()  # localprops of each project
        for idx, prj in enumerate(self._projects):
            with prj.open_project() as dir:
                if not prj._localprops_path(dir).exists():
                    if not allow_none:
                        raise ValueError(
                            f"Localprops not found in project at {prj.project_path}."
                        )
                    continue
                dfs_spl = list[pl.DataFrame]()
                for spl in prj.iter_load_splines(dir, drop_columns=False):
                    _df_spl = spl.props.loc
                    if spline_details:
                        if not spl.has_anchors:
                            raise ValueError(
                                f"Cannot collect spline details because spline {spl!r} "
                                "does not have anchors."
                            )
                        _crds = [spl.map(der=der) for der in [0, 1, 2]]
                        _cv = spl.curvature()
                        _df_spl = _df_spl.with_columns(
                            pl.Series("spline_z", _crds[0][:, 0], dtype=pl.Float32),
                            pl.Series("spline_y", _crds[0][:, 1], dtype=pl.Float32),
                            pl.Series("spline_x", _crds[0][:, 2], dtype=pl.Float32),
                            pl.Series("spline_dz", _crds[1][:, 0], dtype=pl.Float32),
                            pl.Series("spline_dy", _crds[1][:, 1], dtype=pl.Float32),
                            pl.Series("spline_dx", _crds[1][:, 2], dtype=pl.Float32),
                            pl.Series("spline_ddz", _crds[2][:, 0], dtype=pl.Float32),
                            pl.Series("spline_ddy", _crds[2][:, 1], dtype=pl.Float32),
                            pl.Series("spline_ddx", _crds[2][:, 2], dtype=pl.Float32),
                            pl.Series("spline_curvature", _cv, dtype=pl.Float32),
                        )
                    dfs_spl.append(_df_spl)
                _df_prj = pl.concat(dfs_spl, how="diagonal")
                columns = [pl.repeat(idx, pl.len()).cast(pl.UInt16).alias(Mole.image)]
                if H.spline_id in _df_prj.columns:
                    columns.append(pl.col(H.spline_id).cast(pl.UInt16))

            dfs_prj.append(_df_prj.with_columns(columns))
        out = cast_dataframe(pl.concat(dfs_prj, how="diagonal"))
        return self._normalize_id(out, id)

    def collect_globalprops(
        self,
        allow_none: bool = True,
        suffix: str = "",
        id: _IDTYPE = "int",
    ) -> pl.DataFrame:
        """
        Collect all globalprops into a single dataframe.

        Parameters
        ----------
        allow_none : bool, default True
            Continue data collection even if property table data file was not
            found in any project. Raise error otherwise.
        suffix : str, default ""
            Suffix to add to the column names that may be collide with the local
            properties.
        id : str, default "int"
            How to describe the source tomogram. If "int", each tomogram will
            be labeled with ascending integers. If "path", each tomogram will
            be labeled with the name of the project directory.

        Returns
        -------
        pl.DataFrame
            Dataframe with all the properties.
        """
        dataframes = list[pl.DataFrame]()
        for idx, prj in enumerate(self._projects):
            with prj.open_project() as dir:
                path = prj._globalprops_path(dir)
                if path is None:
                    if not allow_none:
                        raise ValueError(
                            f"Globalprops not found in project at {prj.project_path}."
                        )
                    continue
                imagespec = pl.Series(Mole.image, [idx]).cast(pl.UInt16)
                df = pl.read_csv(path).with_columns(imagespec)
            dataframes.append(df)
        out = cast_dataframe(pl.concat(dataframes, how="diagonal"))
        if suffix:
            need_rename = [
                H.spacing, H.twist, H.npf, H.rise, H.skew,
                H.rise_length, H.radius, H.start,
            ]  # fmt: skip
            nmap = {col: col + suffix for col in need_rename if col in out.columns}
            out = out.rename(nmap)
        return self._normalize_id(out, id)

    def collect_joinedprops(
        self,
        allow_none: bool = True,
        id: _IDTYPE = "int",
        spline_details: bool = False,
    ) -> pl.DataFrame:
        """
        Collect all the local and global properties into a single dataframe.

        The global properties are suffixed with "_glob". Note that these columns
        will repeat the same values for each spline. For instance, the "spacing"
        columns will look like following.

        >>> col.collect_joinedprops().select(["spacing", "spacing_glob"])

            shape: (12, 2)
            ┌───────────┬──────────────┐
            │ spacing   ┆ spacing_glob │
            │ ---       ┆ ---          │
            │ f32       ┆ f32          │
            ╞═══════════╪══════════════╡
            │ 4.093385  ┆ 4.1024575    │
            │ 4.0987015 ┆ 4.1024575    │
            │ 4.1013646 ┆ 4.1024575    │
            │ …         ┆ …            │
            │ 4.074887  ┆ 4.089436     │
            │ 4.0987015 ┆ 4.089436     │
            └───────────┴──────────────┘

        Parameters
        ----------
        allow_none : bool, default True
            Forwarded to `collect_localprops` and `collect_globalprops`.
        id : str, default "int"
            How to describe the source tomogram. If "int", each tomogram will
            be labeled with ascending integers. If "path", each tomogram will
            be labeled with the name of the project directory.
        spline_details : bool, default False
            Forwarded to `collect_localprops`.

        Returns
        -------
        pl.DataFrame
            Dataframe with all the properties.
        """
        props = self.collect_props(
            allow_none=allow_none, spline_details=spline_details, suffix="_glob"
        )
        on = [H.spline_id, Mole.image]
        out = props.loc.join(props.glob, on=on, suffix="_glob")
        return self._normalize_id(out, id)

    def collect_props(
        self,
        allow_none: bool = True,
        spline_details: bool = False,
        suffix="",
    ) -> CollectedProps:
        """
        Collect all the local and global properties.

        Parameters
        ----------
        allow_none : bool, default True
            Forwarded to `collect_localprops` and `collect_globalprops`.
        spline_details : bool, default False
            Forwarded to `collect_localprops`.

        Returns
        -------
        CollectedProps
            Tuple of the collected local and global properties.
        """
        loc = self.collect_localprops(
            allow_none=allow_none, id="int", spline_details=spline_details
        )
        glb = self.collect_globalprops(allow_none=allow_none, id="int", suffix=suffix)
        if spline_details:
            lengths = list[float]()
            for _, spl in self.iter_splines():
                lengths.append(spl.length())
            col = pl.Series("spline_length", lengths, dtype=pl.Float32)
            glb = glb.with_columns(col)
        return CollectedProps(loc, glb)

    def collect_molecules(
        self,
        name_filter: Callable[[str], bool] | None = None,
        *,
        curvature: bool = False,
    ) -> Molecules:
        """
        Collect all the molecules in this project sequence.

        Parameters
        ----------
        name_filter : callable, optional
            Function that takes a molecule file name (without extension) and
            returns True if the molecule should be collected. Collect all the
            molecules by default.
        curvature : bool, default False
            If True, the spline curvature will be added to the molecule features.
        """
        loader = self.sta_loader(name_filter, curvature=curvature, allow_no_image=True)
        return loader.molecules

    def iter_splines(self) -> Iterable[tuple[SplineKey, CylSpline]]:
        """Iterate over all the splines in all the projects."""
        for i_prj, prj in enumerate(self._projects):
            with prj.open_project() as dir:
                for i_spl, spl in enumerate(prj.iter_load_splines(dir)):
                    yield SplineKey(i_prj, i_spl), spl

    def iter_molecules(
        self,
        name_filter: Callable[[str], bool] | None = None,
    ) -> Iterable[tuple[MoleculesKey, Molecules]]:
        """
        Iterate over all the molecules in all the projects.

        Parameters
        ----------
        name_filter : callable, optional
            Function that takes a molecule file name (without extension) and
            returns True if the molecule should be collected. Collect all the
            molecules by default.
        """
        for sl, (mole, _) in self.iter_molecules_with_splines(
            name_filter, skip_no_spline=False
        ):
            yield sl, mole

    def iter_molecules_with_splines(
        self,
        name_filter: Callable[[str], bool] | None = None,
        *,
        skip_no_spline: bool = True,
    ) -> Iterator[MoleculesItem]:
        """
        Iterate over all the molecules and its source spline.

        Parameters
        ----------
        name_filter : callable, optional
            Function that takes a molecule file name (without extension) and
            returns True if the molecule should be collected. Collect all the
            molecules by default.
        skip_no_spline : bool, default True
            If True, molecules without a source spline will be skipped.
        """
        if name_filter is None:

            def name_filter(_):
                return True

        for i_prj, prj in enumerate(self._projects):
            with prj.open_project() as dir_:
                for info, mole in prj.iter_load_molecules():
                    if not name_filter(info.name):
                        continue
                    if (src := info.source) is None and skip_no_spline:
                        continue
                    spl = prj.load_spline(src, dir=dir_)
                    yield MoleculesItem(MoleculesKey(i_prj, info.stem), (mole, spl))

    def collect_spline_coords(self, ders: int | Iterable[int] = 0) -> pl.DataFrame:
        """
        Collect spline coordinates or its derivative(s) as a dataframe.

        Coordinates will be labeled as "z", "y", "x". The 1st derivative will be
        labeled as "dz", "dy", "dx", and so on.

        Parameters
        ----------
        ders : int or iterable of int, default 0
            Derivative order(s) to collect. If multiple values are given, all the
            derivatives will be concatenated in a single dataframe.
        """
        dfs = list[pl.DataFrame]()
        if not hasattr(ders, "__iter__"):
            ders = [ders]
        for (i, j), spl in self.iter_splines():
            nanc = spl.anchors.size
            df = pl.DataFrame(
                [
                    pl.repeat(i, nanc, eager=True, dtype=pl.UInt16).alias(Mole.image),
                    pl.repeat(j, nanc, eager=True, dtype=pl.UInt16).alias(H.spline_id),
                ]
            )
            for der in ders:
                d = "d" * der
                coords = spl.map(der=der)
                df = df.with_columns(
                    pl.Series(coords[:, 0], dtype=pl.Float32).alias(f"{d}z"),
                    pl.Series(coords[:, 1], dtype=pl.Float32).alias(f"{d}y"),
                    pl.Series(coords[:, 2], dtype=pl.Float32).alias(f"{d}x"),
                )
            dfs.append(df)
        return pl.concat(dfs, how="vertical")

    def _normalize_id(self, out: pl.DataFrame, id: _IDTYPE) -> pl.DataFrame:
        match id:
            case "int":
                pass
            case "path":
                _map = dict[int, str]()
                _appeared = set[str]()
                for i, prj in enumerate(self._projects):
                    path = prj.project_path
                    if path is None:
                        raise ValueError(
                            f"The {i}-th project {prj!r} does not have a path."
                        )
                    path = Path(path)
                    if path.suffix == ".json":
                        label = path.parent.name
                    elif path.suffix == "":
                        label = path.name
                    else:
                        label = path.stem
                    label = _make_unique_label(label, _appeared)
                    _map[i] = label
                    _appeared.add(label)
                _image_col = pl.col(Mole.image).replace_strict(
                    _map, return_dtype=pl.Enum(list(_map.values()))
                )
                out = out.with_columns(_image_col)
            case _:
                raise ValueError(f"Invalid id type {id!r}.")
        return out

__add__(other)

Concatenate two ProjectSequence objects.

Source code in cylindra/project/_sequence.py

def __add__(self, other: ProjectSequence) -> ProjectSequence:
    """Concatenate two ProjectSequence objects."""
    if not isinstance(other, ProjectSequence):
        raise TypeError(f"Expected ProjectSequence, got {type(other)}.")
    new = ProjectSequence(check_scale=True)
    new._projects = self._projects + other._projects
    if len(self) > 0:
        new._scale_validator.value = self._scale_validator.value
    if len(other) > 0:
        new._scale_validator.value = other._scale_validator.value
    return new

append_file(path)

Add a project from a file path.

Source code in cylindra/project/_sequence.py

def append_file(self, path: str | Path) -> Self:
    """Add a project from a file path."""
    prj = CylindraProject.from_file(path)
    self._scale_validator.value = prj.scale
    self._projects.append(prj)
    return self

collect_globalprops(allow_none=True, suffix='', id='int')

Collect all globalprops into a single dataframe.

Parameters:

Name	Type	Description	Default
allow_none	bool	Continue data collection even if property table data file was not found in any project. Raise error otherwise.	True
suffix	str	Suffix to add to the column names that may be collide with the local properties.	""
id	str	How to describe the source tomogram. If "int", each tomogram will be labeled with ascending integers. If "path", each tomogram will be labeled with the name of the project directory.	"int"

Returns:

Type	Description
DataFrame	Dataframe with all the properties.

Source code in cylindra/project/_sequence.py

def collect_globalprops(
    self,
    allow_none: bool = True,
    suffix: str = "",
    id: _IDTYPE = "int",
) -> pl.DataFrame:
    """
    Collect all globalprops into a single dataframe.

    Parameters
    ----------
    allow_none : bool, default True
        Continue data collection even if property table data file was not
        found in any project. Raise error otherwise.
    suffix : str, default ""
        Suffix to add to the column names that may be collide with the local
        properties.
    id : str, default "int"
        How to describe the source tomogram. If "int", each tomogram will
        be labeled with ascending integers. If "path", each tomogram will
        be labeled with the name of the project directory.

    Returns
    -------
    pl.DataFrame
        Dataframe with all the properties.
    """
    dataframes = list[pl.DataFrame]()
    for idx, prj in enumerate(self._projects):
        with prj.open_project() as dir:
            path = prj._globalprops_path(dir)
            if path is None:
                if not allow_none:
                    raise ValueError(
                        f"Globalprops not found in project at {prj.project_path}."
                    )
                continue
            imagespec = pl.Series(Mole.image, [idx]).cast(pl.UInt16)
            df = pl.read_csv(path).with_columns(imagespec)
        dataframes.append(df)
    out = cast_dataframe(pl.concat(dataframes, how="diagonal"))
    if suffix:
        need_rename = [
            H.spacing, H.twist, H.npf, H.rise, H.skew,
            H.rise_length, H.radius, H.start,
        ]  # fmt: skip
        nmap = {col: col + suffix for col in need_rename if col in out.columns}
        out = out.rename(nmap)
    return self._normalize_id(out, id)

collect_joinedprops(allow_none=True, id='int', spline_details=False)

Collect all the local and global properties into a single dataframe.

The global properties are suffixed with "_glob". Note that these columns will repeat the same values for each spline. For instance, the "spacing" columns will look like following.

col.collect_joinedprops().select(["spacing", "spacing_glob"])

shape: (12, 2)
┌───────────┬──────────────┐
│ spacing   ┆ spacing_glob │
│ ---       ┆ ---          │
│ f32       ┆ f32          │
╞═══════════╪══════════════╡
│ 4.093385  ┆ 4.1024575    │
│ 4.0987015 ┆ 4.1024575    │
│ 4.1013646 ┆ 4.1024575    │
│ …         ┆ …            │
│ 4.074887  ┆ 4.089436     │
│ 4.0987015 ┆ 4.089436     │
└───────────┴──────────────┘

Parameters:

Name	Type	Description	Default
allow_none	bool	Forwarded to collect_localprops and collect_globalprops.	True
id	str	How to describe the source tomogram. If "int", each tomogram will be labeled with ascending integers. If "path", each tomogram will be labeled with the name of the project directory.	"int"
spline_details	bool	Forwarded to collect_localprops.	False

Returns:

Type	Description
DataFrame	Dataframe with all the properties.

Source code in cylindra/project/_sequence.py

def collect_joinedprops(
    self,
    allow_none: bool = True,
    id: _IDTYPE = "int",
    spline_details: bool = False,
) -> pl.DataFrame:
    """
    Collect all the local and global properties into a single dataframe.

    The global properties are suffixed with "_glob". Note that these columns
    will repeat the same values for each spline. For instance, the "spacing"
    columns will look like following.

    >>> col.collect_joinedprops().select(["spacing", "spacing_glob"])

        shape: (12, 2)
        ┌───────────┬──────────────┐
        │ spacing   ┆ spacing_glob │
        │ ---       ┆ ---          │
        │ f32       ┆ f32          │
        ╞═══════════╪══════════════╡
        │ 4.093385  ┆ 4.1024575    │
        │ 4.0987015 ┆ 4.1024575    │
        │ 4.1013646 ┆ 4.1024575    │
        │ …         ┆ …            │
        │ 4.074887  ┆ 4.089436     │
        │ 4.0987015 ┆ 4.089436     │
        └───────────┴──────────────┘

    Parameters
    ----------
    allow_none : bool, default True
        Forwarded to `collect_localprops` and `collect_globalprops`.
    id : str, default "int"
        How to describe the source tomogram. If "int", each tomogram will
        be labeled with ascending integers. If "path", each tomogram will
        be labeled with the name of the project directory.
    spline_details : bool, default False
        Forwarded to `collect_localprops`.

    Returns
    -------
    pl.DataFrame
        Dataframe with all the properties.
    """
    props = self.collect_props(
        allow_none=allow_none, spline_details=spline_details, suffix="_glob"
    )
    on = [H.spline_id, Mole.image]
    out = props.loc.join(props.glob, on=on, suffix="_glob")
    return self._normalize_id(out, id)

collect_localprops(allow_none=True, id='int', spline_details=False)

Collect all localprops into a single dataframe.

Parameters:

Name	Type	Description	Default
allow_none	bool	Continue data collection even if property table data file was not found in any project. Raise error otherwise.	True
id	str	How to describe the source tomogram. If "int", each tomogram will be labeled with ascending integers. If "path", each tomogram will be labeled with the name of the project directory.	"int"
spline_details	bool	If True, spline coordinates, its derivatives and the curvature will also be collected as well. This will take more memory and time.	False

Returns:

Type	Description
DataFrame	Dataframe with all the properties.

Source code in cylindra/project/_sequence.py

def collect_localprops(
    self,
    allow_none: bool = True,
    id: _IDTYPE = "int",
    spline_details: bool = False,
) -> pl.DataFrame:
    """
    Collect all localprops into a single dataframe.

    Parameters
    ----------
    allow_none : bool, default True
        Continue data collection even if property table data file was not
        found in any project. Raise error otherwise.
    id : str, default "int"
        How to describe the source tomogram. If "int", each tomogram will
        be labeled with ascending integers. If "path", each tomogram will
        be labeled with the name of the project directory.
    spline_details : bool, default False
        If True, spline coordinates, its derivatives and the curvature
        will also be collected as well. This will take more memory and time.

    Returns
    -------
    pl.DataFrame
        Dataframe with all the properties.
    """
    dfs_prj = list[pl.DataFrame]()  # localprops of each project
    for idx, prj in enumerate(self._projects):
        with prj.open_project() as dir:
            if not prj._localprops_path(dir).exists():
                if not allow_none:
                    raise ValueError(
                        f"Localprops not found in project at {prj.project_path}."
                    )
                continue
            dfs_spl = list[pl.DataFrame]()
            for spl in prj.iter_load_splines(dir, drop_columns=False):
                _df_spl = spl.props.loc
                if spline_details:
                    if not spl.has_anchors:
                        raise ValueError(
                            f"Cannot collect spline details because spline {spl!r} "
                            "does not have anchors."
                        )
                    _crds = [spl.map(der=der) for der in [0, 1, 2]]
                    _cv = spl.curvature()
                    _df_spl = _df_spl.with_columns(
                        pl.Series("spline_z", _crds[0][:, 0], dtype=pl.Float32),
                        pl.Series("spline_y", _crds[0][:, 1], dtype=pl.Float32),
                        pl.Series("spline_x", _crds[0][:, 2], dtype=pl.Float32),
                        pl.Series("spline_dz", _crds[1][:, 0], dtype=pl.Float32),
                        pl.Series("spline_dy", _crds[1][:, 1], dtype=pl.Float32),
                        pl.Series("spline_dx", _crds[1][:, 2], dtype=pl.Float32),
                        pl.Series("spline_ddz", _crds[2][:, 0], dtype=pl.Float32),
                        pl.Series("spline_ddy", _crds[2][:, 1], dtype=pl.Float32),
                        pl.Series("spline_ddx", _crds[2][:, 2], dtype=pl.Float32),
                        pl.Series("spline_curvature", _cv, dtype=pl.Float32),
                    )
                dfs_spl.append(_df_spl)
            _df_prj = pl.concat(dfs_spl, how="diagonal")
            columns = [pl.repeat(idx, pl.len()).cast(pl.UInt16).alias(Mole.image)]
            if H.spline_id in _df_prj.columns:
                columns.append(pl.col(H.spline_id).cast(pl.UInt16))

        dfs_prj.append(_df_prj.with_columns(columns))
    out = cast_dataframe(pl.concat(dfs_prj, how="diagonal"))
    return self._normalize_id(out, id)

collect_molecules(name_filter=None, *, curvature=False)

Collect all the molecules in this project sequence.

Parameters:

Name	Type	Description	Default
name_filter	callable	Function that takes a molecule file name (without extension) and returns True if the molecule should be collected. Collect all the molecules by default.	None
curvature	bool	If True, the spline curvature will be added to the molecule features.	False

Source code in cylindra/project/_sequence.py

def collect_molecules(
    self,
    name_filter: Callable[[str], bool] | None = None,
    *,
    curvature: bool = False,
) -> Molecules:
    """
    Collect all the molecules in this project sequence.

    Parameters
    ----------
    name_filter : callable, optional
        Function that takes a molecule file name (without extension) and
        returns True if the molecule should be collected. Collect all the
        molecules by default.
    curvature : bool, default False
        If True, the spline curvature will be added to the molecule features.
    """
    loader = self.sta_loader(name_filter, curvature=curvature, allow_no_image=True)
    return loader.molecules

collect_props(allow_none=True, spline_details=False, suffix='')

Collect all the local and global properties.

Parameters:

Name	Type	Description	Default
allow_none	bool	Forwarded to collect_localprops and collect_globalprops.	True
spline_details	bool	Forwarded to collect_localprops.	False

Returns:

Type	Description
CollectedProps	Tuple of the collected local and global properties.

Source code in cylindra/project/_sequence.py

def collect_props(
    self,
    allow_none: bool = True,
    spline_details: bool = False,
    suffix="",
) -> CollectedProps:
    """
    Collect all the local and global properties.

    Parameters
    ----------
    allow_none : bool, default True
        Forwarded to `collect_localprops` and `collect_globalprops`.
    spline_details : bool, default False
        Forwarded to `collect_localprops`.

    Returns
    -------
    CollectedProps
        Tuple of the collected local and global properties.
    """
    loc = self.collect_localprops(
        allow_none=allow_none, id="int", spline_details=spline_details
    )
    glb = self.collect_globalprops(allow_none=allow_none, id="int", suffix=suffix)
    if spline_details:
        lengths = list[float]()
        for _, spl in self.iter_splines():
            lengths.append(spl.length())
        col = pl.Series("spline_length", lengths, dtype=pl.Float32)
        glb = glb.with_columns(col)
    return CollectedProps(loc, glb)

collect_spline_coords(ders=0)

Collect spline coordinates or its derivative(s) as a dataframe.

Coordinates will be labeled as "z", "y", "x". The 1st derivative will be labeled as "dz", "dy", "dx", and so on.

Parameters:

Name	Type	Description	Default
ders	int or iterable of int	Derivative order(s) to collect. If multiple values are given, all the derivatives will be concatenated in a single dataframe.	0

Source code in cylindra/project/_sequence.py

def collect_spline_coords(self, ders: int | Iterable[int] = 0) -> pl.DataFrame:
    """
    Collect spline coordinates or its derivative(s) as a dataframe.

    Coordinates will be labeled as "z", "y", "x". The 1st derivative will be
    labeled as "dz", "dy", "dx", and so on.

    Parameters
    ----------
    ders : int or iterable of int, default 0
        Derivative order(s) to collect. If multiple values are given, all the
        derivatives will be concatenated in a single dataframe.
    """
    dfs = list[pl.DataFrame]()
    if not hasattr(ders, "__iter__"):
        ders = [ders]
    for (i, j), spl in self.iter_splines():
        nanc = spl.anchors.size
        df = pl.DataFrame(
            [
                pl.repeat(i, nanc, eager=True, dtype=pl.UInt16).alias(Mole.image),
                pl.repeat(j, nanc, eager=True, dtype=pl.UInt16).alias(H.spline_id),
            ]
        )
        for der in ders:
            d = "d" * der
            coords = spl.map(der=der)
            df = df.with_columns(
                pl.Series(coords[:, 0], dtype=pl.Float32).alias(f"{d}z"),
                pl.Series(coords[:, 1], dtype=pl.Float32).alias(f"{d}y"),
                pl.Series(coords[:, 2], dtype=pl.Float32).alias(f"{d}x"),
            )
        dfs.append(df)
    return pl.concat(dfs, how="vertical")

from_paths(paths, *, check_scale=True, skip_exc=False) classmethod

Add all the projects of the given paths.

Source code in cylindra/project/_sequence.py

@classmethod
def from_paths(
    cls,
    paths: Iterable[str | Path],
    *,
    check_scale: bool = True,
    skip_exc: bool = False,
) -> Self:
    """Add all the projects of the given paths."""
    self = cls(check_scale=check_scale)
    if skip_exc:
        for path in paths:
            with suppress(Exception):
                self.append_file(path)
    else:
        for path in paths:
            self.append_file(path)
    return self

insert(index, value)

Insert a project at the given index.

Source code in cylindra/project/_sequence.py

def insert(self, index: int, value: CylindraProject) -> None:
    """Insert a project at the given index."""
    if not isinstance(value, CylindraProject):
        raise TypeError(f"Expected CylindraProject, got {type(value)}.")
    return self._projects.insert(index, value)

iter_molecules(name_filter=None)

Iterate over all the molecules in all the projects.

Parameters:

Name	Type	Description	Default
name_filter	callable	Function that takes a molecule file name (without extension) and returns True if the molecule should be collected. Collect all the molecules by default.	None

Source code in cylindra/project/_sequence.py

def iter_molecules(
    self,
    name_filter: Callable[[str], bool] | None = None,
) -> Iterable[tuple[MoleculesKey, Molecules]]:
    """
    Iterate over all the molecules in all the projects.

    Parameters
    ----------
    name_filter : callable, optional
        Function that takes a molecule file name (without extension) and
        returns True if the molecule should be collected. Collect all the
        molecules by default.
    """
    for sl, (mole, _) in self.iter_molecules_with_splines(
        name_filter, skip_no_spline=False
    ):
        yield sl, mole

iter_molecules_with_splines(name_filter=None, *, skip_no_spline=True)

Iterate over all the molecules and its source spline.

Parameters:

Name	Type	Description	Default
name_filter	callable	Function that takes a molecule file name (without extension) and returns True if the molecule should be collected. Collect all the molecules by default.	None
skip_no_spline	bool	If True, molecules without a source spline will be skipped.	True

Source code in cylindra/project/_sequence.py

def iter_molecules_with_splines(
    self,
    name_filter: Callable[[str], bool] | None = None,
    *,
    skip_no_spline: bool = True,
) -> Iterator[MoleculesItem]:
    """
    Iterate over all the molecules and its source spline.

    Parameters
    ----------
    name_filter : callable, optional
        Function that takes a molecule file name (without extension) and
        returns True if the molecule should be collected. Collect all the
        molecules by default.
    skip_no_spline : bool, default True
        If True, molecules without a source spline will be skipped.
    """
    if name_filter is None:

        def name_filter(_):
            return True

    for i_prj, prj in enumerate(self._projects):
        with prj.open_project() as dir_:
            for info, mole in prj.iter_load_molecules():
                if not name_filter(info.name):
                    continue
                if (src := info.source) is None and skip_no_spline:
                    continue
                spl = prj.load_spline(src, dir=dir_)
                yield MoleculesItem(MoleculesKey(i_prj, info.stem), (mole, spl))

iter_splines()

Iterate over all the splines in all the projects.

Source code in cylindra/project/_sequence.py

def iter_splines(self) -> Iterable[tuple[SplineKey, CylSpline]]:
    """Iterate over all the splines in all the projects."""
    for i_prj, prj in enumerate(self._projects):
        with prj.open_project() as dir:
            for i_spl, spl in enumerate(prj.iter_load_splines(dir)):
                yield SplineKey(i_prj, i_spl), spl

sta_loader(name_filter=None, *, curvature=False, allow_no_image=False)

Construct a STA loader from all the projects.

Parameters:

Name	Type	Description	Default
name_filter	callable	Function that takes a molecule file name (without extension) and returns True if the molecule should be collected. Collect all the molecules by default.	None
curvature	bool	If True, the spline curvature will be added to the molecule features.	False
allow_no_image	bool	If True, this method will not raise an error when the image file is not found.	False

Source code in cylindra/project/_sequence.py

def sta_loader(
    self,
    name_filter: Callable[[str], bool] | None = None,
    *,
    curvature: bool = False,
    allow_no_image: bool = False,
) -> BatchLoader:
    """
    Construct a STA loader from all the projects.

    Parameters
    ----------
    name_filter : callable, default None
        Function that takes a molecule file name (without extension) and
        returns True if the molecule should be collected. Collect all the
        molecules by default.
    curvature : bool, default False
        If True, the spline curvature will be added to the molecule features.
    allow_no_image : bool, default False
        If True, this method will not raise an error when the image file is not
        found.
    """
    import impy as ip
    from acryo import BatchLoader

    col = BatchLoader(scale=self._scale_validator.value)
    if name_filter is None:

        def name_filter(_):
            return True

    for idx, prj in enumerate(self._projects):
        if prj.image is None or not prj.image.exists():
            if not allow_no_image:
                raise ValueError(
                    f"Image file not found in project at {prj.project_path}."
                )
            import numpy as np

            img = np.zeros((0, 0, 0), dtype=np.float32)  # dummy
        else:
            img = ip.lazy.imread(prj.image, chunks=get_config().dask_chunk).value
        with prj.open_project() as dir:
            for info, mole in prj.iter_load_molecules(dir):
                if not name_filter(info.stem):
                    continue
                if (
                    curvature
                    and (_spl_i := info.source) is not None
                    and Mole.position in mole.features.columns
                ):
                    _spl = prj.load_spline(_spl_i, dir=dir, props=False)
                    _u = _spl.y_to_position(mole.features[Mole.position])
                    cv = _spl.curvature(_u)
                    mole.features = mole.features.with_columns(
                        pl.Series(cv, dtype=pl.Float32).alias("spline_curvature")
                    )
                mole.features = mole.features.with_columns(
                    pl.repeat(info.stem, pl.len()).alias(Mole.id)
                )
                col.add_tomogram(img, molecules=mole, image_id=idx)
    return col

extract(text)

Extract the content of main function.

Source code in cylindra/project/_utils.py

def extract(text: str) -> Expr:
    """Extract the content of main function."""

    macro_expr = parse(text, squeeze=False)
    if macro_expr.args[0].head is Head.import_:
        for line in macro_expr.args:
            if line.head is Head.function and str(line.args[0].args[0]) == "main":
                macro_expr = line.args[1]
                break

    return macro_expr

get_project_file(path)

Return the path to the project file.

Source code in cylindra/project/_utils.py

def get_project_file(path: str | Path):
    """Return the path to the project file."""
    path = Path(path)
    if path.is_dir():
        path = path / "project.json"
        if not path.exists():
            raise FileNotFoundError(
                f"Directory {path} seems not a cylindra project directory. A "
                "project directory should contain a 'project.json' file."
            )
    return path