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

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




class TimedImageCapture(CaptureBase):
    runtime_parameters_required = {"capture_duration_ms": float}

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

        time_start = time.time()
        while (time.time() - time_start) * 1e3 < capture_duration_ms:
            self.devices["camera"].trigger()

        assert (
            time.time() - time_start
        ) * 1e3 >= capture_duration_ms, f"Expected {capture_duration_ms} ms, but got {(time.time() - time_start) * 1e3} ms."

        number_of_images = self._image_buffer.qsize()
        return TaskStatus.SUCCESS, number_of_images