Integration test for detect+track video

.github/workflows/test_and_deploy.yml

@@ -37,6 +37,13 @@ jobs:
           - os: macos-latest  # M1 macOS
             python-version: "3.10"
     steps:
+      - name: Cache test data
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.crabs-exploration-test-data/*
+          key: cached-test-data
+          enableCrossOsArchive: true
       - uses: neuroinformatics-unit/actions/test@v2
         with:
           python-version: ${{ matrix.python-version }}

conftest.py

@@ -1,5 +1,6 @@
 """Pytest configuration file."""
 
 pytest_plugins = [
+    "tests.fixtures.integration",
     "tests.fixtures.frame_extraction",
 ]

crabs/tracker/track_video.py

@@ -372,7 +372,7 @@ def tracking_parse_args(args):
             "Optionally, it can include a video file named "
             "<video-name>_tracks.mp4, and all frames from the video "
             "under a <video-name>_frames subdirectory. "
-            "Default: ./tracking_output_<timestamp>. "
+            "Default: tracking_output_<timestamp>. "
         ),
     )
     parser.add_argument(

pyproject.toml

@@ -46,6 +46,8 @@ dev = [
     "ruff",
     "setuptools_scm",
     "check-manifest",
+    "pooch",
+    "tqdm",
     # "codespell",
     # "pandas-stubs",
     # "types-attrs",

tests/fixtures/integration.py

@@ -0,0 +1,49 @@
+"""Pytest fixtures for integration tests."""
+
+from datetime import datetime
+from pathlib import Path
+
+import pooch
+import pytest
+
+GIN_TEST_DATA_REPO = "https://gin.g-node.org/SainsburyWellcomeCentre/crabs-exploration-test-data"
+
+
+@pytest.fixture(scope="session")
+def pooch_registry() -> dict:
+    """Pooch registry for the test data.
+
+    This fixture is common for all the test session. The
+    file registry is downloaded fresh for every test session.
+
+    Returns
+    -------
+    dict
+        URL and hash of the GIN repository with the test data
+
+    """
+    # Initialise pooch registry
+    registry = pooch.create(
+        Path.home() / ".crabs-exploration-test-data",
+        base_url=f"{GIN_TEST_DATA_REPO}/raw/master/test_data",
+    )
+
+    # Download the registry file from GIN
+    # Force to download it fresh every time by using a timestamped filename
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    file_registry = pooch.retrieve(
+        url=f"{GIN_TEST_DATA_REPO}/raw/master/files-registry.txt",
+        known_hash=None,
+        fname=f"files-registry_{timestamp}.txt",
+        path=Path.home() / ".crabs-exploration-test-data",
+    )
+
+    # Load registry file onto pooch registry
+    registry.load_registry(
+        file_registry,
+    )
+
+    # Delete registry file
+    Path(file_registry).unlink()
+
+    return registry

tests/test_integration/test_annotations.py

@@ -400,7 +400,7 @@ def test_exclude_pattern(via_json_1: str, via_json_2: str, tmp_path: Path):
     # combine input json files, excluding those that end with _2.json
     json_out_fullpath = combine_multiple_via_jsons(
         [via_json_1, via_json_2],
-        exclude_pattern="\w+_2.json$",
+        exclude_pattern=r"\w+_2.json$",
         json_out_dir=str(tmp_path),
     )
 

tests/test_integration/test_frame_extraction.py

@@ -56,7 +56,7 @@ def assert_output_files(list_input_videos: list, cli_dict: dict) -> None:
 
     # check filename format of images: <video_name>_frame_{frame_idx:08d}
     list_regex_patterns = [
-        Path(input_video_str).stem + "_frame_[\d]{8}$"
+        Path(input_video_str).stem + r"_frame_[\d]{8}$"
         for input_video_str in list_input_videos
     ]
     for f in list_imgs:

tests/test_integration/test_inference.py

@@ -0,0 +1,151 @@
+import re
+import subprocess
+from pathlib import Path
+
+import cv2
+import pooch
+import pytest
+
+from crabs.tracker.utils.io import open_video
+
+
+@pytest.fixture()
+def input_data_paths(pooch_registry: pooch.Pooch):
+    """Input data for a detector+tracking run.
+
+    The data is fetched from the pooch registry.
+
+    Returns
+    -------
+    dict
+        Dictionary with the paths to the input video, annotations,
+        tracking configuration and trained model.
+
+    """
+    input_data_paths = {}
+    video_root_name = "04.09.2023-04-Right_RE_test_3_frames"
+    input_data_paths["video_root_name"] = video_root_name
+
+    # get trained model from pooch registry
+    # download and unzip ml-runs
+    list_files_ml_runs = pooch_registry.fetch(
+        "ml-runs.zip",
+        processor=pooch.Unzip(extract_dir=""),
+        progressbar=True,
+    )
+    # get path to the last checkpoint
+    input_data_paths["ckpt"] = next(
+        x for x in list_files_ml_runs if x.endswith("last.ckpt")
+    )
+
+    # get input video, annotations and config from registry
+    map_key_to_filepath = {
+        "video": f"{video_root_name}/{video_root_name}.mp4",
+        "annotations": f"{video_root_name}/{video_root_name}_ground_truth.csv",
+        "tracking_config": f"{video_root_name}/tracking_config.yaml",
+    }
+    for key, filepath in map_key_to_filepath.items():
+        input_data_paths[key] = pooch_registry.fetch(filepath)
+
+    return input_data_paths
+
+
+@pytest.mark.parametrize(
+    "flags_to_append",
+    [
+        [],
+        ["--save_video"],
+        ["--save_frames"],
+        ["--save_video", "--save_frames"],
+    ],
+)
+def test_detect_and_track_video(
+    input_data_paths: dict,
+    tmp_path: Path,
+    flags_to_append: list,
+):
+    """Test the detect-and-track-video entry point when groundtruth is passed.
+
+    The test checks:
+    - the exit code of the detect-and-track-video command
+    - the existence of csv file with predictions
+    - the existence of csv file with tracking metrics
+    - the existence of video file if requested
+    - the existence of exported frames if requested
+
+    """
+    # Define main detect-and-track-video command
+    main_command = [
+        "detect-and-track-video",
+        f"--trained_model_path={input_data_paths['ckpt']}",
+        f"--video_path={input_data_paths['video']}",
+        f"--config_file={input_data_paths['tracking_config']}",
+        f"--annotations_file={input_data_paths['annotations']}",
+        "--accelerator=cpu",
+    ]
+    # append required flags
+    main_command.extend(flags_to_append)
+
+    # run command
+    completed_process = subprocess.run(
+        main_command,
+        check=True,
+        cwd=tmp_path,
+        # set cwd to Pytest's temporary directory
+        # so the output is saved there
+    )
+
+    # check the command runs successfully
+    assert completed_process.returncode == 0
+
+    # check the tracking output directory is created and has expected name
+    output_dir_name_expected = "tracking_output"
+    expected_pattern = re.compile(
+        rf"{output_dir_name_expected}_\d{{8}}_\d{{6}}$"
+    )
+    list_cwd_subdirs = [x for x in tmp_path.iterdir() if x.is_dir()]
+    tracking_output_dir = list_cwd_subdirs[0]
+    assert len(list_cwd_subdirs) == 1
+    assert expected_pattern.match(tracking_output_dir.stem)
+
+    # check csv with predictions exists
+    predictions_csv = (
+        tmp_path
+        / tracking_output_dir
+        / f"{input_data_paths['video_root_name']}_tracks.csv"
+    )
+    assert (predictions_csv).exists()
+
+    # check csv with tracking metrics exists
+    tracking_metrics_csv = (
+        tmp_path / tracking_output_dir / "tracking_metrics_output.csv"
+    )
+    assert (tracking_metrics_csv).exists()
+
+    # if the video is requested: check it exists
+    if "--save_video" in flags_to_append:
+        assert (
+            tmp_path
+            / tracking_output_dir
+            / f"{input_data_paths['video_root_name']}_tracks.mp4"
+        ).exists()
+
+    # if the frames are requested: check they exist
+    if "--save_frames" in flags_to_append:
+        input_video_object = open_video(input_data_paths["video"])
+        total_n_frames = int(input_video_object.get(cv2.CAP_PROP_FRAME_COUNT))
+
+        # check frames subdirectory exists
+        frames_subdir = (
+            tmp_path
+            / tracking_output_dir
+            / f"{input_data_paths['video_root_name']}_frames"
+        )
+        assert frames_subdir.exists()
+
+        # check files are named as expected
+        expected_pattern = re.compile(r"frame_\d{8}.png")
+        list_files = [x for x in frames_subdir.iterdir() if x.is_file()]
+
+        assert len(list_files) == total_n_frames
+        assert all(expected_pattern.match(x.name) for x in list_files)