20934864165

Committed 12 Jan 2026 09:01PM UTC coverage: 70.168% (-4.5%) from 74.672%

Build # 20934864165

Build Type

Pull #10

github

Committed by

DanielChaseButterfield

Commit Message

Swap ROS1 while loop for publishing with rospy.Timer for real-time image publishing

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

Run Details

524 of 780 new or added lines in 12 files covered. (67.18%)

3 existing lines in 2 files now uncovered.

1423 of 2028 relevant lines covered (70.17%)

1.4 hits per line

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

47.37

/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 = self.image_paths[idx]
            return np.array(Image.open(str(path)), dtype=self.dtype)

        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))

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✔
NEW 23	self.height = height	×
NEW 24	self.width = width	×
NEW 25	self.encoding = encoding	×
NEW 26	self.image_paths = image_paths	×
27
28	def __getitem__(self, idx) -> np.ndarray:	2✔
29	# Handle boolean masking (used by your crop_data function)
NEW 30	if isinstance(idx, np.ndarray) and idx.dtype == bool:	×
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])
NEW 35	if isinstance(idx, slice):	×
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)
NEW 39	path: Path = self.image_paths[idx]	×
NEW 40	return np.array(Image.open(str(path)), dtype=self.dtype)	×
41
42	def __setitem__(self, idx, value):	2✔
NEW 43	raise RuntimeError("This LazyPNGArray is read-only; writes are forbidden.")	×
44
45	def __len__(self):	2✔
NEW 46	return len(self.image_paths)	×
47
48	@property	2✔
49	def shape(self):	2✔
NEW 50	_, channels = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)	×
NEW 51	if channels == 1:	×
NEW 52	return (len(self), self.height, self.width)	×
53	else:
NEW 54	return (len(self), self.height, self.width, 3)	×
55
56	@property	2✔
57	def dtype(self):	2✔
NEW 58	dtype_val, _ = ImageData.ImageEncoding.to_dtype_and_channels(self.encoding)	×
NEW 59	return dtype_val	×
60
61	images: ImageDataOnDisk.LazyImageArray # Not initalized here, but put here for visual code highlighting	2✔
62
63	def __init__(self, frame_id: str, timestamps: Union[np.ndarray, list], images: ImageDataOnDisk.LazyImageArray):	2✔
NEW 64	super().__init__(frame_id, timestamps, images.height, images.width, images.encoding, images)	×
65
66	# =========================================================================
67	# ============================ Class Methods ==============================
68	# =========================================================================
69
70	@classmethod	2✔
71	def from_image_files(cls, image_folder_path: Union[Path, str], frame_id: str) -> ImageDataOnDisk:	2✔
72	"""
73	Creates a class structure from a folder with .png files, using the file names
74	as the timestamps.
75
76	Args:
77	image_folder_path (Path \| str): Path to the folder with the images.
78	frame_id (str): The frame where this image data was collected.
79	Returns:
80	ImageDataOnDisk: Instance of this class.
81	"""
82
83	# Get all png files in the designated folder (sorted)
NEW 84	all_image_files = [str(p) for p in Path(image_folder_path).glob("*.png")]	×
85
86	# Extract the timestamps and sort them
NEW 87	timestamps = col_to_dec_arr([s.split('/')[-1][:-4] for s in all_image_files])	×
NEW 88	sorted_indices = np.argsort(timestamps)	×
NEW 89	timestamps_sorted = timestamps[sorted_indices]	×
90
91	# Use sorted_indices to sort all_image_files in the same way
NEW 92	all_image_files_sorted: List[Path] = [Path(all_image_files[i]) for i in sorted_indices]	×
93
94	# Make sure the mode is what we expect
NEW 95	with Image.open(str(all_image_files_sorted[0])) as first_image:	×
NEW 96	encoding = ImageData.ImageEncoding.from_pillow_str(first_image.mode)	×
NEW 97	if encoding != ImageData.ImageEncoding.RGB8 and encoding != ImageData.ImageEncoding.Mono8:	×
NEW 98	raise NotImplementedError(f"Unsupported encoding {encoding} for 'from_image_files' method!")	×
99
100	# Return an ImageDataOnDisk class
NEW 101	return cls(frame_id, timestamps_sorted, cls.LazyImageArray(first_image.height, first_image.width, encoding, all_image_files_sorted))	×

lunarlab-gatech / robotdataprocess / 20934864165

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