21483849472

Committed 29 Jan 2026 03:20PM UTC coverage: 74.142% (-0.5%) from 74.672%

Build # 21483849472

Build Type

Pull #10

github

Committed by

DanielChaseButterfield

Commit Message

Fix with dependency mismatch with typeguard in conversion_utils, ImageDataOnDisk for .npy files, linear interpolation for odometry data

Pull Request Pull Request #10: (v0.2) Prototype code for ROS2 Publishing, and new ImageDataOnDisk class

Run Details

881 of 1305 new or added lines in 14 files covered. (67.51%)

2 existing lines in 1 file now uncovered.

1749 of 2359 relevant lines covered (74.14%)

1.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.0

/src/robotdataprocess/data_types/ImageData/ImageDataOnDisk.py

from __future__ import annotations

from ...conversion_utils import col_to_dec_arr
from .ImageData import ImageData
import numpy as np
from pathlib import Path
from PIL import Image
from typeguard import typechecked
from typing import Union, List

@typechecked
class ImageDataOnDisk(ImageData):

    class LazyImageArray:
        """A read-only array-like interface that loads PNG images from disk on demand."""
        
        height: int
        width: int
        encoding: ImageData.ImageEncoding
        image_paths: List[Path]

        def __init__(self, height: int, width: int, encoding: ImageData.ImageEncoding, image_paths: List[Path]):
            self.height = height
            self.width = width
            self.encoding = encoding
            self.image_paths = image_paths

        def __getitem__(self, idx) -> np.ndarray:
            # Handle boolean masking (used by your crop_data function)
            if isinstance(idx, np.ndarray) and idx.dtype == bool:
                new_paths = [p for p, keep in zip(self.image_paths, idx) if keep]
                return self.__class__(self.height, self.width, self.encoding, new_paths)

            # Handle slicing (e.g., images[0:10])
            if isinstance(idx, slice):
                return self.__class__(self.height, self.width, self.encoding, self.image_paths[idx])

            # Handle single integer indexing (loading the actual image)
            path: Path = Path(self.image_paths[idx])
            if path.suffix == '.npy':
                return np.load(str(path), 'r')
            elif path.suffix == '.png':
                return np.array(Image.open(str(path)), dtype=self.dtype)
            else: 
                raise NotImplementedError(f"Unsupported file format {path.suffix} in LazyImageArray!")

        def __setitem__(self, idx, value):
            raise RuntimeError("This LazyPNGArray is read-only; writes are forbidden.")
    
        def __len__(self):
            return len(self.image_paths)

        @property
        def shape(self):
            _, channels = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)
            if channels == 1:
                return (len(self), self.height, self.width)
            else:
                return (len(self), self.height, self.width, 3)

        @property
        def dtype(self):
            dtype_val, _ = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)
            return dtype_val
        
    images: ImageDataOnDisk.LazyImageArray # Not initalized here, but put here for visual code highlighting

    def __init__(self, frame_id: str, timestamps: Union[np.ndarray, list], images: ImageDataOnDisk.LazyImageArray):
        super().__init__(frame_id, timestamps, images.height, images.width, images.encoding, images)

    # =========================================================================
    # ============================ Class Methods ============================== 
    # =========================================================================  

    @classmethod
    def from_image_files(cls, image_folder_path: Union[Path, str], frame_id: str) -> ImageDataOnDisk:
        """
        Creates a class structure from a folder with .png files, using the file names
        as the timestamps.

        Args:
            image_folder_path (Path | str): Path to the folder with the images.
            frame_id (str): The frame where this image data was collected.
        Returns:
            ImageDataOnDisk: Instance of this class.
        """

        # Get all png files in the designated folder (sorted)
        all_image_files = [str(p) for p in Path(image_folder_path).glob("*.png")]

        # Extract the timestamps and sort them
        timestamps = col_to_dec_arr([s.split('/')[-1][:-4] for s in all_image_files])
        sorted_indices = np.argsort(timestamps)
        timestamps_sorted = timestamps[sorted_indices]

        # Use sorted_indices to sort all_image_files in the same way
        all_image_files_sorted: List[Path] = [Path(all_image_files[i]) for i in sorted_indices]

        # Make sure the mode is what we expect
        with Image.open(str(all_image_files_sorted[0])) as first_image:
            encoding = ImageData.ImageEncoding.from_pillow_str(first_image.mode)
            if encoding != ImageData.ImageEncoding.RGB8 and encoding != ImageData.ImageEncoding.Mono8:
                raise NotImplementedError(f"Unsupported encoding {encoding} for 'from_image_files' method!")
        
        # Return an ImageDataOnDisk class
        return cls(frame_id, timestamps_sorted, cls.LazyImageArray(first_image.height, first_image.width, encoding, all_image_files_sorted))
    
    @classmethod
    def from_npy_files(cls, npy_folder_path: Union[Path, str], frame_id: str):
        """
        Creates a class structure from .npy files, where each individual image
        is stored in an .npy file with the timestamp as the name

        Args:
            npy_folder_path (Path | str): Path to the folder with the npy images.
            frame_id (str): The frame where this image data was collected.
        Returns:
            ImageData: Instance of this class.
        """

        # Get all npy files in the designated folder (sorted)
        all_image_files = [str(p) for p in Path(npy_folder_path).glob("*.npy")]

        # Extract the timestamps and sort them
        timestamps = col_to_dec_arr([s.split('/')[-1][:-4] for s in all_image_files])
        sorted_indices = np.argsort(timestamps)
        timestamps_sorted = timestamps[sorted_indices]

        # Use sorted_indices to sort all_image_files in the same way
        all_image_files_sorted = [all_image_files[i] for i in sorted_indices]

        # Extract width, height, and channels
        first_image = np.load(all_image_files_sorted[0], 'r')
        assert len(first_image.shape) >= 2
        assert len(first_image.shape) < 4
        height = first_image.shape[0]
        width = first_image.shape[1]
        channels = 1
        if len(first_image.shape) > 2: 
            channels = first_image.shape[2]

        # Extract mode and make sure it matches the supported type for this operation
        encoding = ImageData.ImageEncoding.from_dtype_and_channels(first_image.dtype, channels)
        if encoding != ImageData.ImageEncoding._32FC1:
            raise NotImplementedError(f"Only ImageData.ImageEncoding._32FC1 mode implemented for 'from_npy_files', not {encoding}")

        # Return an ImageDataOnDisk class
        return cls(frame_id, timestamps_sorted, cls.LazyImageArray(height, width, encoding, all_image_files_sorted))

1	from __future__ import annotations	2✔
2
3	from ...conversion_utils import col_to_dec_arr	2✔
4	from .ImageData import ImageData	2✔
5	import numpy as np	2✔
6	from pathlib import Path	2✔
7	from PIL import Image	2✔
8	from typeguard import typechecked	2✔
9	from typing import Union, List	2✔
10
11	@typechecked	2✔
12	class ImageDataOnDisk(ImageData):	2✔
13
14	class LazyImageArray:	2✔
15	"""A read-only array-like interface that loads PNG images from disk on demand."""
16
17	height: int	2✔
18	width: int	2✔
19	encoding: ImageData.ImageEncoding	2✔
20	image_paths: List[Path]	2✔
21
22	def __init__(self, height: int, width: int, encoding: ImageData.ImageEncoding, image_paths: List[Path]):	2✔
23	self.height = height	2✔
24	self.width = width	2✔
25	self.encoding = encoding	2✔
26	self.image_paths = image_paths	2✔
27
28	def __getitem__(self, idx) -> np.ndarray:	2✔
29	# Handle boolean masking (used by your crop_data function)
30	if isinstance(idx, np.ndarray) and idx.dtype == bool:	2✔
NEW 31	new_paths = [p for p, keep in zip(self.image_paths, idx) if keep]	×
NEW 32	return self.__class__(self.height, self.width, self.encoding, new_paths)	×
33
34	# Handle slicing (e.g., images[0:10])
35	if isinstance(idx, slice):	2✔
NEW 36	return self.__class__(self.height, self.width, self.encoding, self.image_paths[idx])	×
37
38	# Handle single integer indexing (loading the actual image)
39	path: Path = Path(self.image_paths[idx])	2✔
40	if path.suffix == '.npy':	2✔
NEW 41	return np.load(str(path), 'r')	×
42	elif path.suffix == '.png':	2✔
43	return np.array(Image.open(str(path)), dtype=self.dtype)	2✔
44	else:
NEW 45	raise NotImplementedError(f"Unsupported file format {path.suffix} in LazyImageArray!")	×
46
47	def __setitem__(self, idx, value):	2✔
NEW 48	raise RuntimeError("This LazyPNGArray is read-only; writes are forbidden.")	×
49
50	def __len__(self):	2✔
51	return len(self.image_paths)	2✔
52
53	@property	2✔
54	def shape(self):	2✔
NEW 55	_, channels = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)	×
NEW 56	if channels == 1:	×
NEW 57	return (len(self), self.height, self.width)	×
58	else:
NEW 59	return (len(self), self.height, self.width, 3)	×
60
61	@property	2✔
62	def dtype(self):	2✔
63	dtype_val, _ = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)	2✔
64	return dtype_val	2✔
65
66	images: ImageDataOnDisk.LazyImageArray # Not initalized here, but put here for visual code highlighting	2✔
67
68	def __init__(self, frame_id: str, timestamps: Union[np.ndarray, list], images: ImageDataOnDisk.LazyImageArray):	2✔
69	super().__init__(frame_id, timestamps, images.height, images.width, images.encoding, images)	2✔
70
71	# =========================================================================
72	# ============================ Class Methods ==============================
73	# =========================================================================
74
75	@classmethod	2✔
76	def from_image_files(cls, image_folder_path: Union[Path, str], frame_id: str) -> ImageDataOnDisk:	2✔
77	"""
78	Creates a class structure from a folder with .png files, using the file names
79	as the timestamps.
80
81	Args:
82	image_folder_path (Path \| str): Path to the folder with the images.
83	frame_id (str): The frame where this image data was collected.
84	Returns:
85	ImageDataOnDisk: Instance of this class.
86	"""
87
88	# Get all png files in the designated folder (sorted)
89	all_image_files = [str(p) for p in Path(image_folder_path).glob("*.png")]	2✔
90
91	# Extract the timestamps and sort them
92	timestamps = col_to_dec_arr([s.split('/')[-1][:-4] for s in all_image_files])	2✔
93	sorted_indices = np.argsort(timestamps)	2✔
94	timestamps_sorted = timestamps[sorted_indices]	2✔
95
96	# Use sorted_indices to sort all_image_files in the same way
97	all_image_files_sorted: List[Path] = [Path(all_image_files[i]) for i in sorted_indices]	2✔
98
99	# Make sure the mode is what we expect
100	with Image.open(str(all_image_files_sorted[0])) as first_image:	2✔
101	encoding = ImageData.ImageEncoding.from_pillow_str(first_image.mode)	2✔
102	if encoding != ImageData.ImageEncoding.RGB8 and encoding != ImageData.ImageEncoding.Mono8:	2✔
NEW 103	raise NotImplementedError(f"Unsupported encoding {encoding} for 'from_image_files' method!")	×
104
105	# Return an ImageDataOnDisk class
106	return cls(frame_id, timestamps_sorted, cls.LazyImageArray(first_image.height, first_image.width, encoding, all_image_files_sorted))	2✔
107
108	@classmethod	2✔
109	def from_npy_files(cls, npy_folder_path: Union[Path, str], frame_id: str):	2✔
110	"""
111	Creates a class structure from .npy files, where each individual image
112	is stored in an .npy file with the timestamp as the name
113
114	Args:
115	npy_folder_path (Path \| str): Path to the folder with the npy images.
116	frame_id (str): The frame where this image data was collected.
117	Returns:
118	ImageData: Instance of this class.
119	"""
120
121	# Get all npy files in the designated folder (sorted)
NEW 122	all_image_files = [str(p) for p in Path(npy_folder_path).glob("*.npy")]	×
123
124	# Extract the timestamps and sort them
NEW 125	timestamps = col_to_dec_arr([s.split('/')[-1][:-4] for s in all_image_files])	×
NEW 126	sorted_indices = np.argsort(timestamps)	×
NEW 127	timestamps_sorted = timestamps[sorted_indices]	×
128
129	# Use sorted_indices to sort all_image_files in the same way
NEW 130	all_image_files_sorted = [all_image_files[i] for i in sorted_indices]	×
131
132	# Extract width, height, and channels
NEW 133	first_image = np.load(all_image_files_sorted[0], 'r')	×
NEW 134	assert len(first_image.shape) >= 2	×
NEW 135	assert len(first_image.shape) < 4	×
NEW 136	height = first_image.shape[0]	×
NEW 137	width = first_image.shape[1]	×
NEW 138	channels = 1	×
NEW 139	if len(first_image.shape) > 2:	×
NEW 140	channels = first_image.shape[2]	×
141
142	# Extract mode and make sure it matches the supported type for this operation
NEW 143	encoding = ImageData.ImageEncoding.from_dtype_and_channels(first_image.dtype, channels)	×
NEW 144	if encoding != ImageData.ImageEncoding._32FC1:	×
NEW 145	raise NotImplementedError(f"Only ImageData.ImageEncoding._32FC1 mode implemented for 'from_npy_files', not {encoding}")	×
146
147	# Return an ImageDataOnDisk class
NEW 148	return cls(frame_id, timestamps_sorted, cls.LazyImageArray(height, width, encoding, all_image_files_sorted))	×

lunarlab-gatech / robotdataprocess / 21483849472

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous