lpcore.capture.sequenceimagecapture のソースコード

from typing import Any
from lpcore import TaskStatus
from lpcore.capture.capturebase import CaptureBase




class SequenceImageCapture(CaptureBase):
    runtime_parameters_required = {
        "num_images": int,
    }

    def _custom_execute(self, **kwargs: Any) -> tuple[TaskStatus, int]:
        number_of_images = self.parameters["num_images"]

        while self._image_buffer.qsize() < number_of_images:
            self.devices["camera"].trigger()
        
        assert self._image_buffer.qsize() == number_of_images, f"Expected {number_of_images} images, but got {self._image_buffer.qsize()} images."

        return TaskStatus.SUCCESS, number_of_images