timelapse_store

plantimager.controller.scanner.timelapse_store

ScanRecord dataclass

ScanRecord(scan_id, started_at=None, finished_at=None, status='pending', error=None)

Only the data needed to resume/re‑run a timelapse.

from_scan classmethod

from_scan(scan_obj)

Create a record from a live Scan instance.

Source code in plantimager/controller/scanner/timelapse_store.py

@classmethod
def from_scan(cls, scan_obj: Any) -> "ScanRecord":
    """Create a record from a live `Scan` instance."""
    # The real Scan class already stores timestamps as floats;
    # we convert them to ISO strings for readability.
    return cls(
        scan_id=scan_obj.scan_id,
        started_at=_dt_to_iso(
            datetime.fromtimestamp(scan_obj._start_time) if getattr(scan_obj, "_start_time", None) else None
        ),
        finished_at=_dt_to_iso(
            datetime.fromtimestamp(scan_obj._stop_time) if getattr(scan_obj, "_stop_time", None) else None
        ),
        status=getattr(scan_obj, "status", "pending"),
        error=getattr(scan_obj, "error", None),
    )

TimelapseStore dataclass

TimelapseStore(_file_name=getenv('PI3_TIMELAPSE_FILE_NAME', TIMELAPSE_STORAGE_JSON), version=1, timelapse_id='', mode='', state='', schedule_times=list(), next_idx=0, current_idx=0, warmup_sec=0, standby_threshold_sec=0, grace_period=0, start_at=None, scans=list(), extra=dict())

Encapsulates atomic persistence of a TimeLapse object.

from_timelapse staticmethod

from_timelapse(tl_obj)

Create a store instance from a live TimeLapse object. Only the fields that are required for a restart are persisted.

Source code in plantimager/controller/scanner/timelapse_store.py

@staticmethod
def from_timelapse(tl_obj: Any) -> "TimelapseStore":
    """
    Create a store instance from a *live* TimeLapse object.
    Only the fields that are required for a restart are persisted.
    """
    # Serialize schedule_times as ISO strings
    schedule_iso = [_dt_to_iso(dt) for dt in tl_obj.schedule_times]

    # Build thin ScanRecord objects – we avoid persisting the whole Scan
    # to keep the JSON small and independent of heavy objects.
    scan_records = [asdict(ScanRecord.from_scan(s)) for s in getattr(tl_obj, "scans", [])]

    return TimelapseStore(
        timelapse_id=tl_obj.id,
        mode=_enum_to_str(tl_obj.mode),
        state=_enum_to_str(tl_obj._state),
        schedule_times=schedule_iso,
        next_idx=tl_obj.next_idx,
        current_idx=tl_obj.current_idx,
        warmup_sec=tl_obj.warmup_sec,
        standby_threshold_sec=tl_obj.standby_threshold_sec,
        grace_period=tl_obj.grace_period,
        start_at=_dt_to_iso(tl_obj.start_at),
        scans=scan_records,
        extra={},
    )

new_store_from_last classmethod

new_store_from_last()

Load a TimelapseStore instance from a stored JSON file in the storage directory.

This method attempts to recreate a TimelapseStore object from a persisted state stored in a JSON file. If the JSON file cannot be found or is invalid, the method logs the issue and returns None. In case of a version mismatch between the stored data and the current implementation, the method attempts a best-effort loading process.

Returns:

Type	Description
Optional[TimelapseStore]	A TimelapseStore instance reconstructed from the JSON file if successful; otherwise, None.

Notes

• The JSON file' is located at ~/.local/share/plant-imager3-app/$PI3_TIMELAPSE_FILE_NAME. If the environment variable is not set, it defaults to "timelapse_storage.json".
• A basic version check is performed to ensure backward compatibility. If a version mismatch is detected, a warning is logged, and the loading process continues on a best-effort basis.
• Any exception encountered during file reading or JSON parsing will be caught, and a corresponding error message will be logged.

Source code in plantimager/controller/scanner/timelapse_store.py

@classmethod
def new_store_from_last(cls) -> Optional["TimelapseStore"]:
    """
    Load a `TimelapseStore` instance from a stored JSON file in the storage directory.

    This method attempts to recreate a `TimelapseStore` object from a persisted state
    stored in a JSON file. If the JSON file cannot be found or is invalid, the method
    logs the issue and returns `None`. In case of a version mismatch between the stored
    data and the current implementation, the method attempts a best-effort loading process.

    Returns
    -------
    Optional[TimelapseStore]
        A `TimelapseStore` instance reconstructed from the JSON file if successful;
        otherwise, `None`.

    Notes
    -----
    - The JSON file' is located at ``~/.local/share/plant-imager3-app/$PI3_TIMELAPSE_FILE_NAME``.
    If the environment variable is not set, it defaults to `"timelapse_storage.json"`.
    - A basic version check is performed to ensure backward compatibility. If a version
      mismatch is detected, a warning is logged, and the loading process continues on a
      best-effort basis.
    - Any exception encountered during file reading or JSON parsing will be caught, and
      a corresponding error message will be logged.
    """
    store_path = get_storage_dir() / os.getenv(
        "PI3_TIMELAPSE_FILE_NAME", TIMELAPSE_STORAGE_JSON
    )
    if not store_path.is_file():
        logger.info("No persisted timelapse file found – starting fresh.")
        return None

    try:
        with store_path.open("r", encoding="utf-8") as f:
            raw = json.load(f)

        # Basic version check – allow future‑compatible extensions
        if raw.get("version", 1) != cls().version:
            logger.warning("Timelapse file version mismatch; attempting best‑effort load.")

        # Build the object
        obj = cls(
            timelapse_id=raw.get("timelapse_id", ""),
            mode=raw.get("mode", ""),
            state=raw.get("state", ""),
            schedule_times=raw.get("schedule_times", []),
            next_idx=raw.get("next_idx", 0),
            current_idx=raw.get("current_idx", 0),
            warmup_sec=raw.get("warmup_sec", 0),
            standby_threshold_sec=raw.get("standby_threshold_sec", 0),
            grace_period=raw.get("grace_period", 0),
            start_at=raw.get("start_at"),
            scans=raw.get("scans", []),
            extra=raw.get("extra", {}),
            version=raw.get("version", 1),
        )
        logger.debug(f"Timelapse state loaded from {store_path}")
        return obj
    except Exception as exc:
        logger.error(f"Failed to read timelapse state: {exc}")
        return None

save

save()

Write the current state to disk atomically.

Source code in plantimager/controller/scanner/timelapse_store.py

def save(self) -> None:
    """Write the current state to disk atomically."""
    try:
        get_storage_dir().mkdir(parents=True, exist_ok=True)

        # Write to a temporary file in the same directory (rename is atomic on POSIX)
        fd, tmp_path = tempfile.mkstemp(dir=get_storage_dir(), prefix=self._file_name, suffix=".tmp")
        with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:
            json.dump(self._as_serialisable_dict(), tmp_file, indent=2, sort_keys=True)
            tmp_file.flush()
            os.fsync(tmp_file.fileno())

        # Replace the old file
        tmp_path_obj = pathlib.Path(tmp_path)
        tmp_path_obj.replace(self._full_path)
        logger.debug(f"Timelapse state saved to {self._full_path}")
    except Exception as exc:
        logger.error(f"Failed to save timelapse state: {exc}")
        raise

to_timelapse_kwargs

to_timelapse_kwargs()

Convert the stored JSON into a dict that can be fed to a new TimeLapse instance for a quick “resume” path.

Source code in plantimager/controller/scanner/timelapse_store.py

def to_timelapse_kwargs(self) -> Dict[str, Any]:
    """
    Convert the stored JSON into a dict that can be fed to a new
    `TimeLapse` instance for a quick “resume” path.
    """
    # Convert ISO strings back to datetime objects where required.
    schedule = [_iso_to_dt(s) for s in self.schedule_times]
    start = _iso_to_dt(self.start_at)

    return {
        "timelapse_id": self.timelapse_id,
        "mode": self.mode,
        "state": self.state,
        "schedule_times": schedule,
        "next_idx": self.next_idx,
        "current_idx": self.current_idx,
        "warmup_sec": self.warmup_sec,
        "standby_threshold_sec": self.standby_threshold_sec,
        "grace_period": self.grace_period,
        "start_at": start,
        "scans": [ScanRecord(**s) for s in self.scans],
        "extra": self.extra,
    }

get_storage_dir

get_storage_dir()

Return the path to the storage directory at ~/.local/share/plant-imager3-app

Source code in plantimager/controller/scanner/timelapse_store.py

def get_storage_dir() -> pathlib.Path:
    """Return the path to the storage directory at ``~/.local/share/plant-imager3-app``"""
    return pathlib.Path(os.getenv("XDG_DATA_HOME", pathlib.Path.home() / ".local" / "share")) / "plant-imager3-app"