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

56.45

/src/robotdataprocess/data_types/Data.py

from __future__ import annotations

from ..conversion_utils import col_to_dec_arr
from decimal import Decimal
from enum import Enum
import matplotlib.pyplot as plt
import numpy as np
from pathlib import Path
from typeguard import typechecked
from typing import List

class CoordinateFrame(Enum):
    """ 
    Enum for different coordinate frames used in robotics. 
    
    Attributes:
        FLU: X forward, Y left, Z up := RHS
        NED: X forward (north), Y right (east), Z down := RHS
        ENU: X right (east), Y forward (north), Z up := RHS
        NONE: No defined coordinate frame.
    """

    FLU = 0
    NED = 1
    ENU = 2
    NONE = 3

class ROSMsgLibType(Enum):
    """ 
    Enum for different ROS message library types.
     
    Attributes:
        ROSBAGS: Use ROS messages from the rosbags library (Pure Python library).
        RCLPY: Use ROS messages from the rclpy library (ROS2 Python client library).
        ROSPY: Use ROS messages from the rospy library (ROS1 Python client library).
    """

    ROSBAGS = 0
    RCLPY = 1
    ROSPY = 2

class Data:
    """
    Generic Data class that provides example methods that should be overwritten by children,
    and can run operations between different data types.
    """

    # Define data attributes for all Data classes
    frame_id: str
    timestamps: np.ndarray[Decimal]

    @typechecked
    def __init__(self, frame_id: str, timestamps: np.ndarray | list):
        
        # Copy initial values into attributes
        self.frame_id = frame_id
        self.timestamps = col_to_dec_arr(timestamps)

        # Check to ensure that all timestamps are sequential
        for i in range(len(self.timestamps) - 1):
            if self.timestamps[i] >= self.timestamps[i+1]:
                raise ValueError(f"Timestamps {self.timestamps[i]} and {self.timestamps[i+1]} do not come in sequential order!")
            
    def len(self):
        """ Returns the number of items in this data class """
        return len(self.timestamps)
    
    @staticmethod
    def get_ros_msg_type(libtype: ROSMsgLibType):
        """ Will return the msgtype for the ROS message for this Data object. """
        raise NotImplementedError("This method needs to be overwritten by the child Data class!")
    
    def get_ros_msg(self, libtype: ROSMsgLibType, i: int):
        """ Will create and return a ROS message object. """
        raise NotImplementedError("This method needs to be overwritten by the child Data class!")
    
    def crop_data(self, start: Decimal, end: Decimal):
        """ Will crop the data so only values within [start, end] inclusive are kept. """
        raise NotImplementedError("This method needs to be overwritten by the child Data class!")

    def hertz_analysis(self) -> None:
        """ Plot historgrams with the sequential data hertz rates and time differences. """

        # Calculate the differences between consecutive timestamps
        if self.len() < 2:
            raise ValueError(f"Not enough data samples to analyze hertz.")
        hertz_diffs = [self.timestamps[i] - self.timestamps[i - 1] for i in range(1, self.len())]
        hertz_values = [1 / diff for diff in hertz_diffs if diff > 0]

        # Output a message if some differences are zero
        num_seq_messages_with_same_timestamps = len([x for x in hertz_diffs if x == 0])
        if num_seq_messages_with_same_timestamps > 0:
            print(f"Warning: Sequential Pairs of timestamps are equivalent {num_seq_messages_with_same_timestamps} times.")

        # Sort each of these Lists
        hertz_diffs.sort()
        hertz_values.sort()

        # To remove potentially noisy values, remove first and last 5
        if len(hertz_diffs) > 5:
            hertz_diffs = hertz_diffs[5:-5]
            hertz_values = hertz_values[5:-5]

        # Create histograms for the hertz values and time differences
        def create_histogram(data: List, title: str, xlabel: str, ylabel: str) -> None:
            """ Create and show a histogram from the provided data. """

            plt.figure(figsize=(10, 6))
            plt.hist(data, bins=100, color='blue', alpha=0.7)
            plt.title(title)
            plt.xlabel(xlabel)
            plt.ylabel(ylabel)
            plt.tight_layout()
            plt.grid(True)
            plt.yscale('log')
            plt.show()

        create_histogram(
            data=hertz_diffs,
            title=f'Time Differences for {self.__class__.__name__}',
            xlabel='Time Difference (seconds)',
            ylabel='Seq. Message Pairs Count (#)'
        )
        create_histogram(
            data=hertz_values,
            title=f'Hertz Analysis for {self.__class__.__name__}',
            xlabel='Hertz (Hz)',
            ylabel='Seq. Message Pairs Count (#)',
        )


1	from __future__ import annotations	2✔
2
3	from ..conversion_utils import col_to_dec_arr	2✔
4	from decimal import Decimal	2✔
5	from enum import Enum	2✔
6	import matplotlib.pyplot as plt	2✔
7	import numpy as np	2✔
8	from pathlib import Path	2✔
9	from typeguard import typechecked	2✔
10	from typing import List	2✔
11
12	class CoordinateFrame(Enum):	2✔
13	"""
14	Enum for different coordinate frames used in robotics.
15
16	Attributes:
17	FLU: X forward, Y left, Z up := RHS
18	NED: X forward (north), Y right (east), Z down := RHS
19	ENU: X right (east), Y forward (north), Z up := RHS
20	NONE: No defined coordinate frame.
21	"""
22
23	FLU = 0	2✔
24	NED = 1	2✔
25	ENU = 2	2✔
26	NONE = 3	2✔
27
28	class ROSMsgLibType(Enum):	2✔
29	"""
30	Enum for different ROS message library types.
31
32	Attributes:
33	ROSBAGS: Use ROS messages from the rosbags library (Pure Python library).
34	RCLPY: Use ROS messages from the rclpy library (ROS2 Python client library).
35	ROSPY: Use ROS messages from the rospy library (ROS1 Python client library).
36	"""
37
38	ROSBAGS = 0	2✔
39	RCLPY = 1	2✔
40	ROSPY = 2	2✔
41
42	class Data:	2✔
43	"""
44	Generic Data class that provides example methods that should be overwritten by children,
45	and can run operations between different data types.
46	"""
47
48	# Define data attributes for all Data classes
49	frame_id: str	2✔
50	timestamps: np.ndarray[Decimal]	2✔
51
52	@typechecked	2✔
53	def __init__(self, frame_id: str, timestamps: np.ndarray \| list):	2✔
54
55	# Copy initial values into attributes
56	self.frame_id = frame_id	2✔
57	self.timestamps = col_to_dec_arr(timestamps)	2✔
58
59	# Check to ensure that all timestamps are sequential
60	for i in range(len(self.timestamps) - 1):	2✔
61	if self.timestamps[i] >= self.timestamps[i+1]:	2✔
62	raise ValueError(f"Timestamps {self.timestamps[i]} and {self.timestamps[i+1]} do not come in sequential order!")	2✔
63
64	def len(self):	2✔
65	""" Returns the number of items in this data class """
66	return len(self.timestamps)	2✔
67
68	@staticmethod	2✔
69	def get_ros_msg_type(libtype: ROSMsgLibType):	2✔
70	""" Will return the msgtype for the ROS message for this Data object. """
UNCOV 71	raise NotImplementedError("This method needs to be overwritten by the child Data class!")	×
72
73	def get_ros_msg(self, libtype: ROSMsgLibType, i: int):	2✔
74	""" Will create and return a ROS message object. """
UNCOV 75	raise NotImplementedError("This method needs to be overwritten by the child Data class!")	×
76
77	def crop_data(self, start: Decimal, end: Decimal):	2✔
78	""" Will crop the data so only values within [start, end] inclusive are kept. """
79	raise NotImplementedError("This method needs to be overwritten by the child Data class!")	×
80
81	def hertz_analysis(self) -> None:	2✔
82	""" Plot historgrams with the sequential data hertz rates and time differences. """
83
84	# Calculate the differences between consecutive timestamps
NEW 85	if self.len() < 2:	×
NEW 86	raise ValueError(f"Not enough data samples to analyze hertz.")	×
NEW 87	hertz_diffs = [self.timestamps[i] - self.timestamps[i - 1] for i in range(1, self.len())]	×
NEW 88	hertz_values = [1 / diff for diff in hertz_diffs if diff > 0]	×
89
90	# Output a message if some differences are zero
NEW 91	num_seq_messages_with_same_timestamps = len([x for x in hertz_diffs if x == 0])	×
NEW 92	if num_seq_messages_with_same_timestamps > 0:	×
NEW 93	print(f"Warning: Sequential Pairs of timestamps are equivalent {num_seq_messages_with_same_timestamps} times.")	×
94
95	# Sort each of these Lists
NEW 96	hertz_diffs.sort()	×
NEW 97	hertz_values.sort()	×
98
99	# To remove potentially noisy values, remove first and last 5
NEW 100	if len(hertz_diffs) > 5:	×
NEW 101	hertz_diffs = hertz_diffs[5:-5]	×
NEW 102	hertz_values = hertz_values[5:-5]	×
103
104	# Create histograms for the hertz values and time differences
NEW 105	def create_histogram(data: List, title: str, xlabel: str, ylabel: str) -> None:	×
106	""" Create and show a histogram from the provided data. """
107
NEW 108	plt.figure(figsize=(10, 6))	×
NEW 109	plt.hist(data, bins=100, color='blue', alpha=0.7)	×
NEW 110	plt.title(title)	×
NEW 111	plt.xlabel(xlabel)	×
NEW 112	plt.ylabel(ylabel)	×
NEW 113	plt.tight_layout()	×
NEW 114	plt.grid(True)	×
NEW 115	plt.yscale('log')	×
NEW 116	plt.show()	×
117
NEW 118	create_histogram(	×
119	data=hertz_diffs,
120	title=f'Time Differences for {self.__class__.__name__}',
121	xlabel='Time Difference (seconds)',
122	ylabel='Seq. Message Pairs Count (#)'
123	)
NEW 124	create_histogram(	×
125	data=hertz_values,
126	title=f'Hertz Analysis for {self.__class__.__name__}',
127	xlabel='Hertz (Hz)',
128	ylabel='Seq. Message Pairs Count (#)',
129	)
130

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