timelapse

plantimager.controller.scanner.timelapse

Handles the coordination and the scheduling of multiple scans through the TimeLapse class

TimeLapse

TimeLapse(cnc, db_url, cameras, path, timelapse_name, config, power_manager, parent=None)

Bases: QObject

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

def __init__(self, cnc: AbstractCNC, db_url: str, cameras: list[PiCameraComm], path: Path,
             timelapse_name: str, config: dict[str, Any], power_manager: PowerManager, parent=None):
    super().__init__(parent)
    self.cnc = cnc
    self.db_url = db_url
    self.cameras = cameras
    self.path = path
    self.id = timelapse_name
    self.config = config

    # timelapse settings
    self._setup_timelapse_settings()

    # Dynamically import and instantiate the path class
    path_module = importlib.import_module("plantimager.controller.scanner.path")
    path_cfg = config["ScanPath"]
    self.scan_path = getattr(path_module, path_cfg["class_name"])(**path_cfg["kwargs"])
    self.pathInfoChanged.emit(self.path_info)

    # Update progress tracking based on path length
    self._max_progress = len(self.scan_path)
    self.progressChanged.emit(self.current_idx, self._max_progress)

    # Store metadata for the scan
    self.dataset_metadata = config["Metadata"]["object"]  # Biological metadata
    self.hw_metadata = config["Metadata"]["hardware"]  # Hardware metadata

    # Configure cameras
    for camera in self.cameras:
        if camera.name in config:
            res = config[camera.name]["res_x"], config[camera.name]["res_y"]
            camera.resolution = res

    self._state = TimeLapseState.STANDBY
    self.power_manager = power_manager

    self._next_scan_timer = QTimer(self, singleShot=True)
    self._next_scan_timer.timeout.connect(self._trigger_next_scan)
    self._powerup_timer = QTimer(self, singleShot=True)
    self._powerup_timer.timeout.connect(power_manager.prepare_for_scan)
    self._setup_next_scan_timer()

scan

scan(index)

Executes a scheduled scan based on the provided index.

This method updates the current scanning index and triggers the progressChanged signal. It determines the interval to the next scheduled time and either waits for the appropriate time, proceeds with the scan, or skips it if the grace period has been violated.

Parameters:

Name	Type	Description	Default
index	int	The index of the scan to execute.	required

Raises:

Type	Description
ValueError	If the provided index is invalid.

Notes

• The method computes the interval between the current time and the next scheduled time. If the interval is greater than the grace_period, the method pauses execution until the scheduled time.
• If the time has already passed and exceeded the grace period, the scan is skipped.
• Sleeps the thread until the scheduled time arrives if the scan occurs too early.

See Also

Scan.scan : Executes the actual scan process.

Examples:

Suppose you have a scheduler with predefined schedule_times and grace_period, you can invoke the following method by supplying a valid scan index:

>>> scheduler.scan(2)

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

@Slot(int)
def scan(self, index: int):
    """
    Executes a scheduled scan based on the provided index.

    This method updates the current scanning index and triggers the `progressChanged` signal.
    It determines the interval to the next scheduled time and either waits for the appropriate
    time, proceeds with the scan, or skips it if the grace period has been violated.

    Parameters
    ----------
    index : int
        The index of the scan to execute.

    Raises
    ------
    ValueError
        If the provided `index` is invalid.

    Notes
    -----
    - The method computes the interval between the current time and the next scheduled time.
      If the interval is greater than the `grace_period`, the method pauses execution until
      the scheduled time.
    - If the time has already passed and exceeded the grace period, the scan is skipped.
    - Sleeps the thread until the scheduled time arrives if the scan occurs too early.

    See Also
    --------
    Scan.scan : Executes the actual scan process.

    Examples
    --------
    Suppose you have a scheduler with predefined `schedule_times` and `grace_period`, you can
    invoke the following method by supplying a valid scan index:

    >>> scheduler.scan(2)
    """
    self.current_idx = index
    self.progressChanged.emit(self.current_idx, self.max_progress)

    next_time = self.schedule_times[self.next_idx]
    current_time = datetime.datetime.now()
    interval = next_time - current_time
    if interval.total_seconds() > self.grace_period:
        logger.warning("Woke up too early, sleeping until time.")
        time.sleep(interval.total_seconds())
    elif interval.total_seconds() <= -self.grace_period:
        # date and grace period have passed
        logger.warning(
            f"Scan {self.next_idx} planned at {next_time.isoformat()} timed out at {current_time.isoformat()}. Skipping."
        )
        return
    scan = Scan(
        self.power_manager.cnc, self.db_url, self.cameras, self.path, f"{self.id}_{current_time.isoformat()}", self.config, parent=self
    )
    scan.scan()
    self.scans.append(scan)

parse_duration

parse_duration(duration_string, /, duration_regexp=DURATION_REGEXP)

Parses a duration string and converts it into a datetime.timedelta object.

This function takes a duration string in a specific format and uses a regular expression pattern to extract the components (e.g., days, hours, minutes, seconds). The parsed components are then converted into a datetime.timedelta object for further manipulation.

Parameters:

Name	Type	Description	Default
duration_string	str	A string representing the duration in the format Xd-Xh-Xm-Xs (e.g., 2d-3h-1m-0s), where X represents integers for days, hours, minutes, and seconds. Each component may be omitted if not needed (e.g., 3h-20m).	required
duration_regexp	Pattern	A compiled regular expression pattern used to match and extract components from duration_string. Defaults to DURATION_REGEXP. Group Names must be keyword arguments of the datetime.timedelta objects constructor. (e.g., days, hours, minutes, seconds)	DURATION_REGEXP

Returns:

Type	Description
timedelta	A datetime.timedelta object representing the parsed time duration.

Raises:

Type	Description
RuntimeError	If duration_string does not match the expected format defined by duration_regexp.

Notes

• The DURATION_REGEXP constant, if used as the default duration_regexp, must be pre-defined in the module. It should include named capturing groups for days (d), hours (h), minutes (m), and seconds (s).

Examples:

>>> import re
>>> import datetime
>>> DURATION_REGEXP = re.compile(r'(?:(?P<days>\d+)d)?\W?(?:(?P<hours>\d+)h)?\W?(?:(?P<minutes>\d+)m)?\W?(?P<seconds>\d+)s?')
>>> parse_duration("2d-3h-1m-0s", duration_regexp=DURATION_REGEXP)
datetime.timedelta(days=2, seconds=10980)

>>> parse_duration("3h-20m", duration_regexp=DURATION_REGEXP)
datetime.timedelta(seconds=12000)

>>> parse_duration("invalid-string", duration_regexp=DURATION_REGEXP)
Traceback (most recent call last):
...
RuntimeError: Failed to parse duration_string: invalid-string. Did not follow the format 2d-3h-1m-0s

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

def parse_duration(duration_string, /, duration_regexp: re.Pattern = DURATION_REGEXP):
    """
    Parses a duration string and converts it into a `datetime.timedelta` object.

    This function takes a duration string in a specific format and uses a regular expression
    pattern to extract the components (e.g., days, hours, minutes, seconds). The parsed
    components are then converted into a `datetime.timedelta` object for further manipulation.

    Parameters
    ----------
    duration_string : str
        A string representing the duration in the format `Xd-Xh-Xm-Xs` (e.g., `2d-3h-1m-0s`),
        where `X` represents integers for days, hours, minutes, and seconds. Each component
        may be omitted if not needed (e.g., `3h-20m`).
    duration_regexp : re.Pattern, optional
        A compiled regular expression pattern used to match and extract components from
        `duration_string`. Defaults to `DURATION_REGEXP`.
        Group Names must be keyword arguments of the datetime.timedelta objects constructor.
        (e.g., days, hours, minutes, seconds)

    Returns
    -------
    datetime.timedelta
        A `datetime.timedelta` object representing the parsed time duration.

    Raises
    ------
    RuntimeError
        If `duration_string` does not match the expected format defined by `duration_regexp`.

    Notes
    -----
    - The `DURATION_REGEXP` constant, if used as the default `duration_regexp`, must be
      pre-defined in the module. It should include named capturing groups for days (`d`),
      hours (`h`), minutes (`m`), and seconds (`s`).

    Examples
    --------
    >>> import re
    >>> import datetime
    >>> DURATION_REGEXP = re.compile(r'(?:(?P<days>\d+)d)?\W?(?:(?P<hours>\d+)h)?\W?(?:(?P<minutes>\d+)m)?\W?(?P<seconds>\d+)s?')
    >>> parse_duration("2d-3h-1m-0s", duration_regexp=DURATION_REGEXP)
    datetime.timedelta(days=2, seconds=10980)

    >>> parse_duration("3h-20m", duration_regexp=DURATION_REGEXP)
    datetime.timedelta(seconds=12000)

    >>> parse_duration("invalid-string", duration_regexp=DURATION_REGEXP)
    Traceback (most recent call last):
    ...
    RuntimeError: Failed to parse duration_string: invalid-string. Did not follow the format 2d-3h-1m-0s
    """
    match = duration_regexp.match(duration_string)
    if match:
        return datetime.timedelta(**{
            k: int(v) if v else 0 for k, v in match.groupdict().items()
        })
    else:
        raise RuntimeError(f"Failed to parse duration_string: {duration_string}. Did not follow the format 2d-3h-1m-0s")