qc.camera

CameraTestSuite

CameraTestSuite(
    data_stream: Camera,
    *,
    expected_fps: Optional[int] = None,
    clock_jitter_s: float = 0.0001,
    start_time_s: Optional[float] = None,
    stop_time_s: Optional[float] = None,
    saturation_bounds: tuple[
        Optional[int], Optional[int]
    ] = (5, 250),
)

Bases: Suite

Test suite for validating camera data integrity.

Provides tests for validating video and metadata integrity according to the AIND file format specification for behavior videos.

For more details, see: https://github.com/AllenNeuralDynamics/aind-file-standards/blob/ce0aa517a40064d1ac9764d42c9efe4ae5c61f7b/file_formats/behavior_videos.md

Attributes:

Name	Type	Description
data_stream	Camera	The Camera data stream to test.
expected_fps		Optional expected frames per second for validation.
clock_jitter_s		Maximum allowed time difference between frame timestamps, in seconds.
start_time_s		Optional expected start time for validation, in seconds.
stop_time_s		Optional expected stop time for validation, in seconds.

Examples:

from contraqctor.contract.camera import Camera, CameraParams
from contraqctor.qc.camera import CameraTestSuite
from contraqctor.qc.base import Runner

# Create and load a camera data stream
params = CameraParams(path="recordings/session1/")
camera_stream = Camera("front_camera", reader_params=params).load()

# Create test suite with validation parameters
suite = CameraTestSuite(
    camera_stream,
    expected_fps=30,
    start_time_s=10.0,
    stop_time_s=310.0
)

# Run tests
runner = Runner().add_suite(suite)
results = runner.run_all_with_progress()

Initialize the camera test suite.

Parameters:

Name	Type	Description	Default
data_stream	Camera	The Camera data stream to test.	required
expected_fps	Optional[int]	Optional expected frames per second for validation.	None
clock_jitter_s	float	Maximum allowed time difference between frame timestamps, in seconds.	0.0001
start_time_s	Optional[float]	Optional expected start time for validation, in seconds.	None
stop_time_s	Optional[float]	Optional expected stop time for validation, in seconds.	None
saturation_bounds	tuple[Optional[int], Optional[int]]	Pixel intensity bounds to check for saturation (min, max).	(5, 250)

Source code in src/contraqctor/qc/camera.py

def __init__(
    self,
    data_stream: Camera,
    *,
    expected_fps: t.Optional[int] = None,
    clock_jitter_s: float = 1e-4,
    start_time_s: t.Optional[float] = None,
    stop_time_s: t.Optional[float] = None,
    saturation_bounds: tuple[t.Optional[int], t.Optional[int]] = (5, 250),
):
    """Initialize the camera test suite.

    Args:
        data_stream: The Camera data stream to test.
        expected_fps: Optional expected frames per second for validation.
        clock_jitter_s: Maximum allowed time difference between frame timestamps, in seconds.
        start_time_s: Optional expected start time for validation, in seconds.
        stop_time_s: Optional expected stop time for validation, in seconds.
        saturation_bounds: Pixel intensity bounds to check for saturation (min, max).
    """
    self.data_stream: Camera = data_stream
    self.expected_fps = expected_fps
    self.clock_jitter_s = clock_jitter_s
    self.start_time_s = start_time_s
    self.stop_time_s = stop_time_s
    self.saturation_bounds = saturation_bounds

description property

description: Optional[str]

Get the description of the test suite from its docstring.

Returns:

Type	Description
Optional[str]	Optional[str]: The docstring of the class, or None if not available.

name property

name: str

Get the name of the test suite.

Returns:

Name	Type	Description
str	str	The name of the test suite class.

test_metadata_shape

test_metadata_shape()

Checks if the metadata DataFrame has the expected shape. Including headers.

Source code in src/contraqctor/qc/camera.py

def test_metadata_shape(self):
    """
    Checks if the metadata DataFrame has the expected shape. Including headers.
    """
    if not self.data_stream.has_data:
        return self.fail_test(None, "Data stream does not have loaded data")
    metadata = self.data_stream.data.metadata
    if not isinstance(metadata, pd.DataFrame):
        return self.fail_test(None, "Metadata is not a pandas DataFrame")

    (metadata_cols := list(metadata.columns)).append(metadata.index.name)
    if not all(col in metadata_cols for col in self._expected_columns):
        missing_columns = self._expected_columns - set(metadata_cols)
        return self.fail_test(None, f"Metadata columns do not match expected columns. Missing: {missing_columns}")
    if metadata.empty:
        return self.fail_test(None, "Metadata DataFrame is empty")
    return self.pass_test(None, "Metadata DataFrame has expected shape and columns")

test_check_dropped_frames

test_check_dropped_frames()

Check if there are dropped frames in the metadata DataFrame.

Source code in src/contraqctor/qc/camera.py

def test_check_dropped_frames(self):
    """
    Check if there are dropped frames in the metadata DataFrame.
    """
    metadata = (self.data_stream.data.metadata[list(self._expected_columns - {"ReferenceTime"})]).copy()
    metadata.loc[:, "ReferenceTime"] = metadata.index.values
    diff_metadata = metadata.diff()
    # Convert CameraFrameTime to seconds
    diff_metadata["CameraFrameTime"] = diff_metadata["CameraFrameTime"] * 1e-9

    if not all(diff_metadata["CameraFrameNumber"].dropna() == 1):
        return self.fail_test(
            None, f"Detected {sum(diff_metadata['CameraFrameNumber'].dropna() - 1)} dropped frames metadata."
        )

    inter_clock_diff = diff_metadata["CameraFrameTime"] - diff_metadata["ReferenceTime"]
    if not all(inter_clock_diff.dropna() < self.clock_jitter_s):
        return self.fail_test(
            None,
            f"Detected a difference between CameraFrameTime and ReferenceTime greater than the expected threshold: {self.clock_jitter_s} s.",
        )
    return self.pass_test(None, "No dropped frames detected in metadata.")

test_match_expected_fps

test_match_expected_fps()

Check if the frames per second (FPS) of the video metadata matches the expected FPS.

Source code in src/contraqctor/qc/camera.py

def test_match_expected_fps(self):
    """
    Check if the frames per second (FPS) of the video metadata matches the expected FPS."""
    if self.expected_fps is None:
        return self.skip_test("No expected FPS provided, skipping test.")
    period = np.diff(self.data_stream.data.metadata.index.values)
    if np.std(period) > 1e-4:
        return self.fail_test(None, f"High std in frame period detected: {np.std(period)}")
    if abs(_mean := np.mean(period) - (_expected := (1.0 / self.expected_fps))) > (_expected * 0.01):
        return self.fail_test(None, f"Mean frame period ({_mean}) is different than expected: {_expected}")

    return self.pass_test(None, f"Mean frame period ({_mean}) is within expected range: {_expected}")

test_is_start_bounded

test_is_start_bounded()

Check if the start time of the video is bounded by the provided start time.

Source code in src/contraqctor/qc/camera.py

def test_is_start_bounded(self):
    """
    Check if the start time of the video is bounded by the provided start time."""
    metadata = self.data_stream.data.metadata
    if self.start_time_s is not None:
        if metadata.index[0] < self.start_time_s:
            return self.fail_test(
                None,
                f"Start time is not bounded. First frame time: {metadata.index[0]}, expected start time: {self.start_time_s}",
            )
        else:
            return self.pass_test(
                None,
                f"Start time is bounded. First frame time: {metadata.index[0]}, expected start time: {self.start_time_s}",
            )
    else:
        return self.skip_test("No start time provided, skipping test.")

test_is_stop_bounded

test_is_stop_bounded()

Check if the stop time of the video is bounded by the provided stop time.

Source code in src/contraqctor/qc/camera.py

def test_is_stop_bounded(self):
    """
    Check if the stop time of the video is bounded by the provided stop time."""
    metadata = self.data_stream.data.metadata
    if self.stop_time_s is not None:
        if metadata.index[-1] > self.stop_time_s:
            return self.fail_test(
                None,
                f"Stop time is not bounded. Last frame time: {metadata.index[-1]}, expected stop time: {self.stop_time_s}",
            )
        else:
            return self.pass_test(
                None,
                f"Stop time is bounded. Last frame time: {metadata.index[-1]}, expected stop time: {self.stop_time_s}",
            )
    else:
        return self.skip_test("No stop time provided, skipping test.")

test_video_frame_count

test_video_frame_count()

Check if the number of frames in the video matches the number of rows in the metadata DataFrame.

Source code in src/contraqctor/qc/camera.py

def test_video_frame_count(self):
    """
    Check if the number of frames in the video matches the number of rows in the metadata DataFrame.
    """
    data = self.data_stream.data
    if not data.has_video:
        return self.skip_test("No video data available. Skipping test.")

    if (n_frames := data.video_frame_count) != len(data.metadata):
        return self.fail_test(
            None,
            f"Number of frames in video ({n_frames}) does not match number of rows in metadata ({len(data.metadata)})",
        )
    else:
        return self.pass_test(
            None,
            f"Number of frames in video ({n_frames}) matches number of rows in metadata ({len(data.metadata)})",
        )

test_histogram_and_create_asset

test_histogram_and_create_asset()

Checks the histogram of the video and ensures color is well distributed. It also saves an asset with a single frame of the video and color histogram.

Source code in src/contraqctor/qc/camera.py

def test_histogram_and_create_asset(self):
    """Checks the histogram of the video and ensures color is well distributed.
    It also saves an asset with a single frame of the video and color histogram."""

    data = self.data_stream.data
    if not data.has_video:
        return self.skip_test("No video data available. Skipping test.")

    with data.as_video_capture() as video:
        video.set(cv2.CAP_PROP_POS_FRAMES, video.get(cv2.CAP_PROP_FRAME_COUNT) // 2)
        ret, frame = video.read()

        if not ret:
            return self.fail_test(None, "Failed to read a frame from the video")
        max_d = 2 ** (frame.dtype.itemsize * 8)

        fig, ax = plt.subplots(2, frame.shape[2], figsize=(15, 7))

        for channel in range(frame.shape[2]):
            hist = cv2.calcHist([frame], [channel], None, [max_d], [0, max_d])
            hist /= hist.sum()
            ax[0, channel].imshow(frame[:, :, channel], cmap="gray")
            ax[0, channel].axis("off")
            ax[1, channel].plot(hist, color="k", label=f"Channel-{channel}")
            ax[1, channel].set_xlim([0, max_d])
            ax[1, channel].set_xlabel("Pixel Value")
            ax[1, channel].set_ylabel("Normalized Frequency")
            ax[1, channel].set_title(f"Histogram channel-{channel}")
        fig.subplots_adjust(top=0.9)  # Leave space for suptitle
        fig.suptitle("Pixel value histogram")
        fig.tight_layout()

    return self.pass_test(
        None, "Histogram and asset created successfully.", context=ContextExportableObj.as_context(fig)
    )

test_create_pixel_saturation_visualizer

test_create_pixel_saturation_visualizer()

Creates a visualization highlighting saturated and underexposed pixels in the video frame.

Source code in src/contraqctor/qc/camera.py

def test_create_pixel_saturation_visualizer(self):
    """Creates a visualization highlighting saturated and underexposed pixels in the video frame."""
    data = self.data_stream.data
    if not data.has_video:
        return self.skip_test("No video data available. Skipping test.")

    with data.as_video_capture() as video:
        video.set(cv2.CAP_PROP_POS_FRAMES, video.get(cv2.CAP_PROP_FRAME_COUNT) // 2)
        ret, frame = video.read()

        if not ret:
            return self.fail_test(None, "Failed to read a frame from the video")

        lower_bound, upper_bound = self.saturation_bounds

        fig, ax = plt.subplots(1, frame.shape[2], figsize=(15, 5))

        for channel in range(frame.shape[2]):
            channel_data = frame[:, :, channel]

            channel_saturated = np.zeros(frame.shape[:2], dtype=bool)
            channel_underexposed = np.zeros(frame.shape[:2], dtype=bool)

            if upper_bound is not None:
                channel_saturated = channel_data >= upper_bound
            if lower_bound is not None:
                channel_underexposed = channel_data <= lower_bound

            # Create RGB image: grayscale with saturated pixels in red and underexposed in blue
            colored_frame = np.stack([channel_data, channel_data, channel_data], axis=-1)
            colored_frame[channel_saturated] = [255, 0, 0]  # Red for saturated
            colored_frame[channel_underexposed] = [0, 0, 255]  # Blue for underexposed

            ax[channel].imshow(colored_frame)
            ax[channel].axis("off")
            ax[channel].set_title(f"Channel-{channel}")

        fig.subplots_adjust(top=0.9)  # Leave space for suptitle
        fig.suptitle("Pixel Saturation Visualization (bounds: {})".format(self.saturation_bounds))
        fig.tight_layout()

        return self.pass_test(
            None, "Histogram and asset created successfully.", context=ContextExportableObj.as_context(fig)
        )

get_tests

get_tests() -> Generator[ITest, None, None]

Find all methods starting with 'test'.

Yields:

Name	Type	Description
ITest	ITest	Test methods found in the suite.

Source code in src/contraqctor/qc/base.py

def get_tests(self) -> t.Generator[ITest, None, None]:
    """Find all methods starting with 'test'.

    Yields:
        ITest: Test methods found in the suite.
    """
    for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
        if name.startswith("test"):
            yield method

pass_test

pass_test() -> Result

pass_test(result: Any) -> Result

pass_test(result: Any, message: str) -> Result

pass_test(result: Any, *, context: Any) -> Result

pass_test(
    result: Any, message: str, *, context: Any
) -> Result

pass_test(
    result: Any = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a passing test result.

Parameters:

Name	Type	Description	Default
result	Any	The value to include in the test result.	None
message	Optional[str]	Optional message describing why the test passed.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with PASSED status.

Source code in src/contraqctor/qc/base.py

def pass_test(
    self, result: t.Any = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a passing test result.

    Args:
        result: The value to include in the test result.
        message: Optional message describing why the test passed.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with PASSED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.PASSED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

warn_test

warn_test() -> Result

warn_test(result: Any) -> Result

warn_test(result: Any, message: str) -> Result

warn_test(result: Any, *, context: Any) -> Result

warn_test(
    result: Any, message: str, *, context: Any
) -> Result

warn_test(
    result: Any = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a warning test result.

Creates a result with WARNING status, or FAILED if warnings are elevated.

Parameters:

Name	Type	Description	Default
result	Any	The value to include in the test result.	None
message	Optional[str]	Optional message describing the warning.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with WARNING or FAILED status.

Source code in src/contraqctor/qc/base.py

def warn_test(
    self, result: t.Any = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a warning test result.

    Creates a result with WARNING status, or FAILED if warnings are elevated.

    Args:
        result: The value to include in the test result.
        message: Optional message describing the warning.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with WARNING or FAILED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.WARNING if not _elevate_warning.get() else Status.FAILED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

fail_test

fail_test() -> Result

fail_test(result: Any) -> Result

fail_test(result: Any, message: str) -> Result

fail_test(
    result: Any, message: str, *, context: Any
) -> Result

fail_test(
    result: Optional[Any] = None,
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a failing test result.

Parameters:

Name	Type	Description	Default
result	Optional[Any]	The value to include in the test result.	None
message	Optional[str]	Optional message describing why the test failed.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with FAILED status.

Source code in src/contraqctor/qc/base.py

def fail_test(
    self, result: t.Optional[t.Any] = None, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None
) -> Result:
    """Create a failing test result.

    Args:
        result: The value to include in the test result.
        message: Optional message describing why the test failed.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with FAILED status.
    """
    calling_func_name, description = self._get_caller_info()

    return Result(
        status=Status.FAILED,
        result=result,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

skip_test

skip_test() -> Result

skip_test(message: str) -> Result

skip_test(message: str, *, context: Any) -> Result

skip_test(
    message: Optional[str] = None,
    *,
    context: Optional[Any] = None,
) -> Result

Create a skipped test result.

Creates a result with SKIPPED status, or FAILED if skips are elevated.

Parameters:

Name	Type	Description	Default
message	Optional[str]	Optional message explaining why the test was skipped.	None
context	Optional[Any]	Optional contextual data for the test result.	None

Returns:

Name	Type	Description
Result	Result	A Result object with SKIPPED or FAILED status.

Source code in src/contraqctor/qc/base.py

def skip_test(self, message: t.Optional[str] = None, *, context: t.Optional[t.Any] = None) -> Result:
    """Create a skipped test result.

    Creates a result with SKIPPED status, or FAILED if skips are elevated.

    Args:
        message: Optional message explaining why the test was skipped.
        context: Optional contextual data for the test result.

    Returns:
        Result: A Result object with SKIPPED or FAILED status.
    """
    calling_func_name, description = self._get_caller_info()
    return Result(
        status=Status.SKIPPED if not _elevate_skippable.get() else Status.FAILED,
        result=None,
        test_name=calling_func_name,
        suite_name=self.name,
        message=message,
        context=context,
        description=description,
    )

setup_suite

setup_suite() -> None

Run once before any test method in the suite.

Mimics :meth:unittest.TestCase.setUpClass. Override this to perform expensive or failure-prone preparation — e.g. loading a data stream's .data — a single time for the whole suite, rather than in __init__ (which is not protected by exception handling) or in :meth:setup (which re-runs before every test).

The :class:Runner invokes this inside an exception-handling block. If it raises, every test in the suite is reported as an error instead of being run, and :meth:teardown_suite is not called. Suite construction and test discovery are unaffected.

Source code in src/contraqctor/qc/base.py

def setup_suite(self) -> None:
    """Run once before any test method in the suite.

    Mimics :meth:`unittest.TestCase.setUpClass`. Override this to perform
    expensive or failure-prone preparation — e.g. loading a data stream's
    ``.data`` — a single time for the whole suite, rather than in ``__init__``
    (which is not protected by exception handling) or in :meth:`setup` (which
    re-runs before every test).

    The :class:`Runner` invokes this inside an exception-handling block. If it
    raises, every test in the suite is reported as an error instead of being
    run, and :meth:`teardown_suite` is *not* called. Suite construction and
    test discovery are unaffected.
    """
    pass

teardown_suite

teardown_suite() -> None

Run once after all test methods in the suite have run.

Mimics :meth:unittest.TestCase.tearDownClass. Only invoked if :meth:setup_suite completed successfully.

Source code in src/contraqctor/qc/base.py

def teardown_suite(self) -> None:
    """Run once after all test methods in the suite have run.

    Mimics :meth:`unittest.TestCase.tearDownClass`. Only invoked if
    :meth:`setup_suite` completed successfully.
    """
    pass

setup

setup() -> None

Run before each test method.

Mimics :meth:unittest.TestCase.setUp. This method can be overridden by subclasses to implement setup logic that runs before each test. For work that should happen only once for the whole suite, override :meth:setup_suite instead.

Source code in src/contraqctor/qc/base.py

def setup(self) -> None:
    """Run before each test method.

    Mimics :meth:`unittest.TestCase.setUp`. This method can be overridden by
    subclasses to implement setup logic that runs before each test. For work
    that should happen only once for the whole suite, override
    :meth:`setup_suite` instead.
    """
    pass

teardown

teardown() -> None

Run after each test method.

Mimics :meth:unittest.TestCase.tearDown. This method can be overridden by subclasses to implement teardown logic that runs after each test. For work that should happen only once for the whole suite, override :meth:teardown_suite instead.

Source code in src/contraqctor/qc/base.py

def teardown(self) -> None:
    """Run after each test method.

    Mimics :meth:`unittest.TestCase.tearDown`. This method can be overridden
    by subclasses to implement teardown logic that runs after each test. For
    work that should happen only once for the whole suite, override
    :meth:`teardown_suite` instead.
    """
    pass

run_test

run_test(
    test_method: ITest,
) -> Generator[Result, None, None]

Run a single test method and yield its results.

Handles setup, test execution, result processing, and teardown.

Parameters:

Name	Type	Description	Default
test_method	ITest	The test method to run.	required

Yields:

Name	Type	Description
Result	Result	Result objects produced by the test method.

Source code in src/contraqctor/qc/base.py

def run_test(self, test_method: ITest) -> t.Generator[Result, None, None]:
    """Run a single test method and yield its results.

    Handles setup, test execution, result processing, and teardown.

    Args:
        test_method: The test method to run.

    Yields:
        Result: Result objects produced by the test method.
    """
    test_name = test_method.__name__
    suite_name = self.name
    test_description = getattr(test_method, "__doc__", None)

    try:
        self.setup()
        result = test_method()
        if inspect.isgenerator(result):
            for sub_result in result:
                yield self._process_test_result(sub_result, test_method, test_name, test_description)
        else:
            yield self._process_test_result(result, test_method, test_name, test_description)
    except Exception as e:
        tb = traceback.format_exc()
        yield Result(
            status=Status.ERROR,
            result=None,
            test_name=test_name,
            suite_name=suite_name,
            description=test_description,
            message=f"Error during test execution: {str(e)}",
            exception=e,
            traceback=tb,
            test_reference=test_method,
            suite_reference=self,
        )
    finally:
        self.teardown()

run_all

run_all() -> Generator[Result, None, None]

Run all test methods in the suite.

Runs :meth:setup_suite once, then all test methods in sequence, then :meth:teardown_suite. If :meth:setup_suite raises, every test is yielded as an error result and no tests are run.

Yields:

Name	Type	Description
Result	Result	Result objects produced by all test methods.

Source code in src/contraqctor/qc/base.py

def run_all(self) -> t.Generator[Result, None, None]:
    """Run all test methods in the suite.

    Runs :meth:`setup_suite` once, then all test methods in sequence, then
    :meth:`teardown_suite`. If :meth:`setup_suite` raises, every test is
    yielded as an error result and no tests are run.

    Yields:
        Result: Result objects produced by all test methods.
    """
    tests = list(self.get_tests())
    setup_failure = self._try_setup_suite()
    if setup_failure is not None:
        exception, tb = setup_failure
        for test in tests:
            yield self._suite_setup_error_result(test, exception, tb)
        return

    try:
        for test in tests:
            yield from self.run_test(test)
    finally:
        self.teardown_suite()