[Python] How to measure objects using python and the pyorbbecsdk library with depth cameras?

I have the FemtoMega depth camera which is supported by the orbbec SDK which also has a Python wrapper: https://github.com/orbbec/pyorbbecsdk. I am trying to create a python program that can measure the length, width, and height of objects (which will always be square/rectangle boxes). The following is what I was able to come up with using the examples in the SDK and help from ChatGPT, but I am not able to focus in on 1 object - I don't know what the camera is actually measuring. Could anyone please help me with this? There is very little information about doing something like this and coming from a JS background, I'm not sure how to proceed.

import time
import cv2
import numpy as np
from pyorbbecsdk import Config, OBSensorType, Pipeline, ColorFrame, FrameSet

ESC_KEY = 27
PRINT_INTERVAL = 1 # seconds
MIN_DEPTH = 20 # 20mm
MAX_DEPTH = 10000 # 10000mm


class TemporalFilter:
def __init__(self, alpha):
self.alpha = alpha
self.previous_frame = None

def process(self, frame):
if self.previous_frame is None:
result = frame
else:
result = cv2.addWeighted(frame, self.alpha, self.previous_frame, 1 - self.alpha, 0)
self.previous_frame = result
return result


def get_stream_dimensions(pipeline, sensor_type=OBSensorType.DEPTH_SENSOR):
"""
Retrieves the dimensions (width, height) of the specified sensor's stream.

aram pipeline: The Pipeline object.
aram sensor_type: The sensor type (default is DEPTH_SENSOR).
:return: Tuple (width, height) if successful, else (None, None).
"""
profile_list = pipeline.get_stream_profile_list(sensor_type)
if profile_list is None:
print(f"No stream profiles found for sensor type: {sensor_type}")
return None, None

stream_profile = profile_list.get_default_video_stream_profile()
if stream_profile is None:
print(f"No default video stream profile found for sensor type: {sensor_type}")
return None, None

# Use getter methods instead of direct attribute access
width = stream_profile.get_width()
height = stream_profile.get_height()
print(f"Stream Dimensions: {width}x{height}")
return width, height


def print_sample_depth_values(depth_data, width, height):
"""
Prints sample depth values from different regions of the frame.

aram depth_data: Numpy array of depth data.
aram width: Width of the depth frame.
aram height: Height of the depth frame.
"""
sample_points = [
(0, 0), # Top-left corner
(height // 2, width // 2), # Center
(height - 1, width - 1), # Bottom-right corner
(height // 4, width // 4), # Quarter position
(3 * height // 4, 3 * width // 4) # Three-quarters position
]

print("Sample Depth Values:")
for y, x in sample_points:
distance = depth_data[y, x]
print(f" Point ({x}, {y}): {distance} mm")


def preprocess_depth_data(depth_data, lower_bound=MIN_DEPTH, upper_bound=MAX_DEPTH):
"""
Applies preprocessing steps to the depth data, including thresholding.

aram depth_data: Numpy array of raw depth data.
aram lower_bound: Minimum depth to consider (in mm).
aram upper_bound: Maximum depth to consider (in mm).
:return: Preprocessed binary image for contour detection.
"""
# Apply Gaussian Blur to reduce noise
blurred = cv2.GaussianBlur(depth_data, (5, 5), 0)

# Apply binary thresholding based on depth bounds
thresh = cv2.inRange(blurred, lower_bound, upper_bound)

return thresh


def preprocess_color_data(color_frame):
"""
Converts color frame to HSV and creates a mask to filter out the pallet.

aram color_frame: ColorFrame object.
:return: Binary mask where the pallet is masked out.
"""
color_data = np.frombuffer(color_frame.get_data(), dtype=np.uint8)
color_data = color_data.reshape((color_frame.get_height(), color_frame.get_width(), 3))
hsv = cv2.cvtColor(color_data, cv2.COLOR_BGR2HSV)

# Define the color range for the pallet (e.g., brown)
lower_brown = np.array([10, 100, 20])
upper_brown = np.array([20, 255, 200])

# Create mask to exclude the pallet
mask = cv2.inRange(hsv, lower_brown, upper_brown)
mask_inv = cv2.bitwise_not(mask)

return mask_inv


def detect_and_calculate_dimensions(depth_data, depth_scale, min_contour_area=500):
"""
Detects the largest object in the depth data and calculates its dimensions.

aram depth_data: Numpy array of depth data.
aram depth_scale: Scale factor to convert depth units to millimeters.
aram min_contour_area: Minimum area to consider a contour valid.
:return: Tuple containing dimensions dictionary and largest contour or None.
"""
# Preprocess depth data
thresh = preprocess_depth_data(depth_data)

# Find contours to detect objects
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

if contours:
# Filter out small contours based on area
valid_contours = [cnt for cnt in contours if cv2.contourArea(cnt) > min_contour_area]
if not valid_contours:
print("No valid contours detected.")
return None

# Assume the largest valid contour corresponds to the object
largest_contour = max(valid_contours, key=cv2.contourArea)
x, y, w, h = cv2.boundingRect(largest_contour)

# Calculate center distance based on the bounding box center
center_y = y + h // 2
center_x = x + w // 2
center_distance = depth_data[center_y, center_x]

# Calculate object dimensions in millimeters
object_width = w * depth_scale
object_height = h * depth_scale

dimensions = {
"center_distance_mm": center_distance,
"width_mm": object_width,
"height_mm": object_height,
"bounding_box": (x, y, w, h)
}

return dimensions, largest_contour
else:
print("No contours found.")
return None


def main():
config = Config()
pipeline = Pipeline()
temporal_filter = TemporalFilter(alpha=0.5)
try:
# Retrieve and enable the depth stream profile
width, height = get_stream_dimensions(pipeline, OBSensorType.DEPTH_SENSOR)
if width is None or height is None:
print("Failed to get stream dimensions.")
return

profile_list = pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
depth_profile = profile_list.get_default_video_stream_profile()
print("Depth profile: ", depth_profile)

# Use the retrieved stream profile to enable the stream
config.enable_stream(depth_profile)
except Exception as e:
print(f"Error during stream configuration: {e}")
return

try:
pipeline.start(config)
last_print_time = time.time()
while True:
try:
frames = pipeline.wait_for_frames(1000)
if frames is None:
continue
depth_frame = frames.get_depth_frame()
if depth_frame is None:
continue

# Convert depth frame data to numpy array
depth_data = np.frombuffer(depth_frame.get_data(), dtype=np.uint16)
depth_data = depth_data.reshape((height, width))

# Apply scaling and filtering
depth_scale = depth_frame.get_depth_scale()
depth_data = depth_data.astype(np.float32) * depth_scale
depth_data = np.where((depth_data > MIN_DEPTH) & (depth_data < MAX_DEPTH), depth_data, 0)
depth_data = depth_data.astype(np.uint16)

# Apply temporal filtering
depth_data = temporal_filter.process(depth_data)

# Detect object and calculate dimensions
result = detect_and_calculate_dimensions(depth_data, depth_scale)
if result:
dimensions, largest_contour = result
center_distance = dimensions["center_distance_mm"]
object_width = dimensions["width_mm"]
object_height = dimensions["height_mm"]

# Print dimensions at defined intervals
current_time = time.time()
if current_time - last_print_time >= PRINT_INTERVAL:
print(f"Center distance: {center_distance} mm")
print(f"Object Width: {object_width:.2f} mm")
print(f"Object Height: {object_height:.2f} mm")
last_print_time = current_time

# Draw bounding box and center point on the depth image
depth_image = cv2.normalize(depth_data, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
depth_image = cv2.applyColorMap(depth_image, cv2.COLORMAP_JET)
x, y, w, h = dimensions["bounding_box"]
cv2.rectangle(depth_image, (x, y), (x + w, y + h), (0, 255, 0), 2)
center_y = y + h // 2
center_x = x + w // 2
cv2.circle(depth_image, (center_x, center_y), 5, (255, 0, 0), -1)
else:
center_distance = 0
print("No objects detected.")

# Display the depth image with annotations
cv2.imshow("Depth Viewer", depth_image)
key = cv2.waitKey(1)
if key == ord('q') or key == ESC_KEY:
print("Exiting the program.")
break
except KeyboardInterrupt:
print("Interrupted by user.")
break
except Exception as e:
print(f"An error occurred during streaming: {e}")
finally:
pipeline.stop()
cv2.destroyAllWindows()
print("Pipeline stopped and windows closed.")


if __name__ == "__main__":
main()

Continue reading...