Explore 3D vision techniques including depth estimation, stereo vision, point cloud processing, camera calibration, and 3D reconstruction.
3D computer vision extends 2D image understanding to reconstruct and analyze the three-dimensional world. From autonomous vehicles estimating distances to AR applications overlaying virtual objects, 3D vision is fundamental to spatial computing.
This lesson covers the complete stack: from camera calibration and stereo matching to modern deep learning approaches for monocular depth estimation and point cloud processing.
Understanding camera models is essential for 3D reconstruction:
1import numpy as np
2import cv2
3from typing import Tuple, List, Optional
4
5class CameraCalibration:
6 """
7 Camera calibration using checkerboard pattern.
8 Computes intrinsic and extrinsic camera parameters.
9 """
10
11 def __init__(self, checkerboard_size: Tuple[int, int] = (9, 6),
12 square_size: float = 25.0):
13 """
14 Args:
15 checkerboard_size: (columns, rows) of inner corners
16 square_size: Size of each square in mm
17 """
18 self.checkerboard_size = checkerboard_size
19 self.square_size = square_size
20
21 # Prepare object points (3D coordinates on checkerboard)
22 self.objp = np.zeros((checkerboard_size[0] * checkerboard_size[1], 3),
23 np.float32)
24 self.objp[:, :2] = np.mgrid[0:checkerboard_size[0],
25 0:checkerboard_size[1]].T.reshape(-1, 2)
26 self.objp *= square_size
27
28 # Storage for calibration data
29 self.obj_points: List[np.ndarray] = [] # 3D world points
30 self.img_points: List[np.ndarray] = [] # 2D image points
31
32 # Calibration results
33 self.camera_matrix: Optional[np.ndarray] = None
34 self.dist_coeffs: Optional[np.ndarray] = None
35 self.rvecs: List[np.ndarray] = []
36 self.tvecs: List[np.ndarray] = []
37
38 def find_corners(self, image: np.ndarray) -> Tuple[bool, Optional[np.ndarray]]:
39 """
40 Find checkerboard corners in image.
41
42 Returns:
43 success: Whether corners were found
44 corners: (N, 1, 2) array of corner coordinates
45 """
46 gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
47
48 # Find corners
49 ret, corners = cv2.findChessboardCorners(
50 gray, self.checkerboard_size,
51 cv2.CALIB_CB_ADAPTIVE_THRESH + cv2.CALIB_CB_FAST_CHECK +
52 cv2.CALIB_CB_NORMALIZE_IMAGE
53 )
54
55 if ret:
56 # Refine corner positions
57 criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER,
58 30, 0.001)
59 corners = cv2.cornerSubPix(gray, corners, (11, 11), (-1, -1),
60 criteria)
61
62 return ret, corners
63
64 def add_calibration_image(self, image: np.ndarray) -> bool:
65 """Add image to calibration set."""
66 ret, corners = self.find_corners(image)
67
68 if ret:
69 self.obj_points.append(self.objp)
70 self.img_points.append(corners)
71 return True
72
73 return False
74
75 def calibrate(self, image_size: Tuple[int, int]) -> float:
76 """
77 Perform camera calibration.
78
79 Args:
80 image_size: (width, height) of images
81
82 Returns:
83 Reprojection error in pixels
84 """
85 if len(self.obj_points) < 3:
86 raise ValueError("Need at least 3 calibration images")
87
88 ret, mtx, dist, rvecs, tvecs = cv2.calibrateCamera(
89 self.obj_points, self.img_points, image_size, None, None
90 )
91
92 self.camera_matrix = mtx
93 self.dist_coeffs = dist
94 self.rvecs = rvecs
95 self.tvecs = tvecs
96
97 return ret # RMS reprojection error
98
99 def undistort(self, image: np.ndarray) -> np.ndarray:
100 """Remove lens distortion from image."""
101 if self.camera_matrix is None:
102 raise ValueError("Camera not calibrated")
103
104 h, w = image.shape[:2]
105 new_mtx, roi = cv2.getOptimalNewCameraMatrix(
106 self.camera_matrix, self.dist_coeffs, (w, h), 1, (w, h)
107 )
108
109 undistorted = cv2.undistort(
110 image, self.camera_matrix, self.dist_coeffs, None, new_mtx
111 )
112
113 # Crop to valid region
114 x, y, w, h = roi
115 return undistorted[y:y+h, x:x+w]
116
117 def get_camera_parameters(self) -> dict:
118 """Get calibration results."""
119 if self.camera_matrix is None:
120 return {}
121
122 fx = self.camera_matrix[0, 0]
123 fy = self.camera_matrix[1, 1]
124 cx = self.camera_matrix[0, 2]
125 cy = self.camera_matrix[1, 2]
126
127 return {
128 'focal_length': (fx, fy),
129 'principal_point': (cx, cy),
130 'distortion': self.dist_coeffs.flatten(),
131 'camera_matrix': self.camera_matrix
132 }
133
134
135class PinholeCamera:
136 """
137 Pinhole camera model for projection and unprojection.
138 """
139
140 def __init__(self, fx: float, fy: float, cx: float, cy: float):
141 self.fx = fx
142 self.fy = fy
143 self.cx = cx
144 self.cy = cy
145
146 self.K = np.array([
147 [fx, 0, cx],
148 [0, fy, cy],
149 [0, 0, 1]
150 ], dtype=np.float64)
151
152 self.K_inv = np.linalg.inv(self.K)
153
154 def project(self, points_3d: np.ndarray) -> np.ndarray:
155 """
156 Project 3D points to 2D image coordinates.
157
158 Args:
159 points_3d: (N, 3) array of [X, Y, Z] points
160
161 Returns:
162 (N, 2) array of [u, v] pixel coordinates
163 """
164 # Normalize by Z
165 X, Y, Z = points_3d[:, 0], points_3d[:, 1], points_3d[:, 2]
166
167 u = self.fx * X / Z + self.cx
168 v = self.fy * Y / Z + self.cy
169
170 return np.column_stack([u, v])
171
172 def unproject(self, points_2d: np.ndarray, depths: np.ndarray) -> np.ndarray:
173 """
174 Unproject 2D points to 3D given depths.
175
176 Args:
177 points_2d: (N, 2) array of [u, v] pixel coordinates
178 depths: (N,) array of depth values
179
180 Returns:
181 (N, 3) array of [X, Y, Z] points
182 """
183 u, v = points_2d[:, 0], points_2d[:, 1]
184
185 X = (u - self.cx) * depths / self.fx
186 Y = (v - self.cy) * depths / self.fy
187 Z = depths
188
189 return np.column_stack([X, Y, Z])
190
191
192# Example
193calibrator = CameraCalibration(checkerboard_size=(9, 6), square_size=25.0)
194print("Camera Calibration initialized")
195print(f"Checkerboard: {calibrator.checkerboard_size}, square size: {calibrator.square_size}mm")Stereo vision computes depth from two camera views:
1import numpy as np
2import cv2
3from typing import Tuple
4
5class StereoVision:
6 """
7 Stereo vision for depth estimation from two cameras.
8 """
9
10 def __init__(self, baseline: float, focal_length: float):
11 """
12 Args:
13 baseline: Distance between cameras in mm
14 focal_length: Focal length in pixels
15 """
16 self.baseline = baseline
17 self.focal_length = focal_length
18
19 # Stereo matching parameters
20 self.num_disparities = 16 * 8 # Must be divisible by 16
21 self.block_size = 5
22
23 # Create stereo matcher
24 self.stereo = cv2.StereoSGBM_create(
25 minDisparity=0,
26 numDisparities=self.num_disparities,
27 blockSize=self.block_size,
28 P1=8 * 3 * self.block_size ** 2,
29 P2=32 * 3 * self.block_size ** 2,
30 disp12MaxDiff=1,
31 uniquenessRatio=15,
32 speckleWindowSize=100,
33 speckleRange=32,
34 preFilterCap=63,
35 mode=cv2.STEREO_SGBM_MODE_SGBM_3WAY
36 )
37
38 # WLS filter for disparity refinement
39 self.right_matcher = cv2.ximgproc.createRightMatcher(self.stereo)
40 self.wls_filter = cv2.ximgproc.createDisparityWLSFilter(
41 matcher_left=self.stereo
42 )
43 self.wls_filter.setLambda(80000)
44 self.wls_filter.setSigmaColor(1.2)
45
46 def compute_disparity(self, left: np.ndarray,
47 right: np.ndarray) -> np.ndarray:
48 """
49 Compute disparity map from stereo pair.
50
51 Args:
52 left: Left rectified image
53 right: Right rectified image
54
55 Returns:
56 Disparity map (float)
57 """
58 # Convert to grayscale if color
59 if len(left.shape) == 3:
60 left_gray = cv2.cvtColor(left, cv2.COLOR_BGR2GRAY)
61 right_gray = cv2.cvtColor(right, cv2.COLOR_BGR2GRAY)
62 else:
63 left_gray = left
64 right_gray = right
65
66 # Compute disparity
67 disp_left = self.stereo.compute(left_gray, right_gray)
68 disp_right = self.right_matcher.compute(right_gray, left_gray)
69
70 # Apply WLS filter
71 disparity = self.wls_filter.filter(
72 disp_left, left_gray, None, disp_right
73 )
74
75 # Convert to float (SGBM returns fixed-point format)
76 disparity = disparity.astype(np.float32) / 16.0
77
78 return disparity
79
80 def disparity_to_depth(self, disparity: np.ndarray) -> np.ndarray:
81 """
82 Convert disparity to depth using stereo geometry.
83
84 depth = baseline * focal_length / disparity
85 """
86 # Avoid division by zero
87 disparity_safe = np.where(disparity > 0, disparity, 0.1)
88
89 depth = self.baseline * self.focal_length / disparity_safe
90
91 # Invalidate where disparity was invalid
92 depth[disparity <= 0] = 0
93
94 return depth
95
96 def generate_point_cloud(self, disparity: np.ndarray,
97 color_image: np.ndarray,
98 cx: float, cy: float) -> Tuple[np.ndarray, np.ndarray]:
99 """
100 Generate 3D point cloud from disparity map.
101
102 Returns:
103 points: (N, 3) array of 3D coordinates
104 colors: (N, 3) array of RGB colors
105 """
106 h, w = disparity.shape
107
108 # Create coordinate grids
109 u, v = np.meshgrid(np.arange(w), np.arange(h))
110
111 # Convert disparity to 3D coordinates
112 Z = self.baseline * self.focal_length / (disparity + 1e-6)
113 X = (u - cx) * Z / self.focal_length
114 Y = (v - cy) * Z / self.focal_length
115
116 # Stack into point cloud
117 points = np.stack([X, Y, Z], axis=-1).reshape(-1, 3)
118
119 # Get colors
120 if len(color_image.shape) == 3:
121 colors = cv2.cvtColor(color_image, cv2.COLOR_BGR2RGB)
122 else:
123 colors = np.stack([color_image] * 3, axis=-1)
124 colors = colors.reshape(-1, 3) / 255.0
125
126 # Filter invalid points
127 valid = (disparity.flatten() > 0) & (Z.flatten() < 10000)
128 points = points[valid]
129 colors = colors[valid]
130
131 return points, colors
132
133
134class DisparityRefinement:
135 """Advanced disparity map refinement techniques."""
136
137 @staticmethod
138 def fill_holes(disparity: np.ndarray) -> np.ndarray:
139 """Fill invalid disparity regions using interpolation."""
140 # Create mask of invalid pixels
141 invalid = disparity <= 0
142
143 # Use inpainting to fill holes
144 disparity_uint = cv2.normalize(disparity, None, 0, 255,
145 cv2.NORM_MINMAX).astype(np.uint8)
146 filled = cv2.inpaint(disparity_uint, invalid.astype(np.uint8), 5,
147 cv2.INPAINT_TELEA)
148
149 # Scale back
150 scale = disparity.max() / 255.0
151 return filled.astype(np.float32) * scale
152
153 @staticmethod
154 def edge_preserving_filter(disparity: np.ndarray,
155 guide: np.ndarray) -> np.ndarray:
156 """Apply edge-preserving filtering using guide image."""
157 # Joint bilateral filter
158 filtered = cv2.ximgproc.jointBilateralFilter(
159 guide, disparity.astype(np.float32),
160 d=9, sigmaColor=75, sigmaSpace=75
161 )
162 return filtered
163
164
165# Example
166stereo = StereoVision(baseline=120.0, focal_length=500.0)
167print(f"Stereo system: baseline={stereo.baseline}mm, focal={stereo.focal_length}px")
168print(f"Disparity range: 0-{stereo.num_disparities} pixels")Deep learning enables depth estimation from single images:
1import torch
2import torch.nn as nn
3import torch.nn.functional as F
4import numpy as np
5
6class DepthEstimationEncoder(nn.Module):
7 """
8 Encoder network for monocular depth estimation.
9 Uses EfficientNet-style architecture.
10 """
11
12 def __init__(self):
13 super().__init__()
14
15 # Encoder blocks with increasing channels
16 self.enc1 = self._make_block(3, 32, stride=2) # 1/2
17 self.enc2 = self._make_block(32, 64, stride=2) # 1/4
18 self.enc3 = self._make_block(64, 128, stride=2) # 1/8
19 self.enc4 = self._make_block(128, 256, stride=2) # 1/16
20 self.enc5 = self._make_block(256, 512, stride=2) # 1/32
21
22 def _make_block(self, in_ch: int, out_ch: int, stride: int) -> nn.Module:
23 return nn.Sequential(
24 nn.Conv2d(in_ch, out_ch, 3, stride=stride, padding=1),
25 nn.BatchNorm2d(out_ch),
26 nn.ReLU(inplace=True),
27 nn.Conv2d(out_ch, out_ch, 3, padding=1),
28 nn.BatchNorm2d(out_ch),
29 nn.ReLU(inplace=True)
30 )
31
32 def forward(self, x: torch.Tensor) -> list:
33 features = []
34
35 x = self.enc1(x); features.append(x)
36 x = self.enc2(x); features.append(x)
37 x = self.enc3(x); features.append(x)
38 x = self.enc4(x); features.append(x)
39 x = self.enc5(x); features.append(x)
40
41 return features
42
43
44class DepthEstimationDecoder(nn.Module):
45 """
46 Decoder network with skip connections.
47 """
48
49 def __init__(self):
50 super().__init__()
51
52 # Decoder blocks with skip connections
53 self.dec5 = self._make_upblock(512, 256)
54 self.dec4 = self._make_upblock(256 + 256, 128)
55 self.dec3 = self._make_upblock(128 + 128, 64)
56 self.dec2 = self._make_upblock(64 + 64, 32)
57 self.dec1 = self._make_upblock(32 + 32, 16)
58
59 # Final depth prediction
60 self.depth_pred = nn.Sequential(
61 nn.Conv2d(16, 1, 3, padding=1),
62 nn.Sigmoid() # Output in [0, 1], scale to actual depth
63 )
64
65 def _make_upblock(self, in_ch: int, out_ch: int) -> nn.Module:
66 return nn.Sequential(
67 nn.ConvTranspose2d(in_ch, out_ch, 4, stride=2, padding=1),
68 nn.BatchNorm2d(out_ch),
69 nn.ReLU(inplace=True),
70 nn.Conv2d(out_ch, out_ch, 3, padding=1),
71 nn.BatchNorm2d(out_ch),
72 nn.ReLU(inplace=True)
73 )
74
75 def forward(self, features: list) -> torch.Tensor:
76 x = self.dec5(features[4])
77 x = self.dec4(torch.cat([x, features[3]], dim=1))
78 x = self.dec3(torch.cat([x, features[2]], dim=1))
79 x = self.dec2(torch.cat([x, features[1]], dim=1))
80 x = self.dec1(torch.cat([x, features[0]], dim=1))
81
82 depth = self.depth_pred(x)
83 return depth
84
85
86class MonocularDepthNet(nn.Module):
87 """
88 Complete monocular depth estimation network.
89 """
90
91 def __init__(self, max_depth: float = 10.0):
92 super().__init__()
93 self.max_depth = max_depth
94 self.encoder = DepthEstimationEncoder()
95 self.decoder = DepthEstimationDecoder()
96
97 def forward(self, x: torch.Tensor) -> torch.Tensor:
98 """
99 Estimate depth from single RGB image.
100
101 Args:
102 x: RGB image (B, 3, H, W) normalized to [0, 1]
103
104 Returns:
105 Depth map (B, 1, H, W) in meters
106 """
107 features = self.encoder(x)
108 depth_normalized = self.decoder(features)
109
110 # Scale to actual depth
111 depth = depth_normalized * self.max_depth
112
113 return depth
114
115
116class DepthLoss(nn.Module):
117 """
118 Multi-term loss for depth estimation.
119 """
120
121 def __init__(self, alpha: float = 0.5, beta: float = 0.1):
122 super().__init__()
123 self.alpha = alpha
124 self.beta = beta
125
126 def forward(self, pred: torch.Tensor,
127 target: torch.Tensor,
128 mask: torch.Tensor = None) -> dict:
129 """
130 Compute depth estimation loss.
131
132 Args:
133 pred: Predicted depth (B, 1, H, W)
134 target: Ground truth depth
135 mask: Valid depth mask (optional)
136 """
137 if mask is None:
138 mask = torch.ones_like(target)
139
140 # L1 loss
141 l1_loss = F.l1_loss(pred * mask, target * mask, reduction='mean')
142
143 # Gradient loss (edge-aware smoothness)
144 grad_pred_x = torch.abs(pred[:, :, :, :-1] - pred[:, :, :, 1:])
145 grad_pred_y = torch.abs(pred[:, :, :-1, :] - pred[:, :, 1:, :])
146
147 grad_target_x = torch.abs(target[:, :, :, :-1] - target[:, :, :, 1:])
148 grad_target_y = torch.abs(target[:, :, :-1, :] - target[:, :, 1:, :])
149
150 grad_loss = F.l1_loss(grad_pred_x, grad_target_x) + F.l1_loss(grad_pred_y, grad_target_y)
151
152 # Scale-invariant log loss
153 log_pred = torch.log(pred + 1e-6)
154 log_target = torch.log(target + 1e-6)
155 diff = (log_pred - log_target) * mask
156
157 si_loss = torch.mean(diff ** 2) - self.beta * torch.mean(diff) ** 2
158
159 # Total loss
160 total = l1_loss + self.alpha * grad_loss + si_loss
161
162 return {
163 'total': total,
164 'l1': l1_loss,
165 'gradient': grad_loss,
166 'scale_invariant': si_loss
167 }
168
169
170# Example
171depth_model = MonocularDepthNet(max_depth=10.0)
172print(f"Depth model parameters: {sum(p.numel() for p in depth_model.parameters()):,}")
173
174# Test inference
175image = torch.randn(1, 3, 256, 256)
176depth = depth_model(image)
177print(f"Input shape: {image.shape}")
178print(f"Depth output shape: {depth.shape}")Working with 3D point cloud data:
1import numpy as np
2from typing import Tuple, Optional
3import open3d as o3d
4
5class PointCloudProcessor:
6 """
7 Point cloud processing utilities.
8 """
9
10 @staticmethod
11 def create_from_depth(depth: np.ndarray,
12 intrinsics: np.ndarray,
13 color: Optional[np.ndarray] = None) -> np.ndarray:
14 """
15 Create point cloud from depth image.
16
17 Args:
18 depth: (H, W) depth image
19 intrinsics: 3x3 camera intrinsic matrix
20 color: Optional (H, W, 3) color image
21
22 Returns:
23 points: (N, 3) or (N, 6) if color provided
24 """
25 h, w = depth.shape
26 fx, fy = intrinsics[0, 0], intrinsics[1, 1]
27 cx, cy = intrinsics[0, 2], intrinsics[1, 2]
28
29 # Create coordinate grids
30 u, v = np.meshgrid(np.arange(w), np.arange(h))
31
32 # Unproject to 3D
33 Z = depth
34 X = (u - cx) * Z / fx
35 Y = (v - cy) * Z / fy
36
37 # Stack into point cloud
38 points = np.stack([X, Y, Z], axis=-1).reshape(-1, 3)
39
40 # Filter invalid points
41 valid = Z.flatten() > 0
42 points = points[valid]
43
44 if color is not None:
45 colors = color.reshape(-1, 3)[valid] / 255.0
46 points = np.hstack([points, colors])
47
48 return points
49
50 @staticmethod
51 def downsample_voxel(points: np.ndarray,
52 voxel_size: float) -> np.ndarray:
53 """
54 Downsample point cloud using voxel grid.
55 """
56 # Create Open3D point cloud
57 pcd = o3d.geometry.PointCloud()
58 pcd.points = o3d.utility.Vector3dVector(points[:, :3])
59
60 if points.shape[1] > 3:
61 pcd.colors = o3d.utility.Vector3dVector(points[:, 3:6])
62
63 # Voxel downsampling
64 downsampled = pcd.voxel_down_sample(voxel_size)
65
66 # Extract points
67 result = np.asarray(downsampled.points)
68 if downsampled.has_colors():
69 colors = np.asarray(downsampled.colors)
70 result = np.hstack([result, colors])
71
72 return result
73
74 @staticmethod
75 def estimate_normals(points: np.ndarray,
76 radius: float = 0.1,
77 max_nn: int = 30) -> np.ndarray:
78 """
79 Estimate surface normals for point cloud.
80 """
81 pcd = o3d.geometry.PointCloud()
82 pcd.points = o3d.utility.Vector3dVector(points[:, :3])
83
84 pcd.estimate_normals(
85 search_param=o3d.geometry.KDTreeSearchParamHybrid(
86 radius=radius, max_nn=max_nn
87 )
88 )
89
90 return np.asarray(pcd.normals)
91
92 @staticmethod
93 def remove_outliers(points: np.ndarray,
94 nb_neighbors: int = 20,
95 std_ratio: float = 2.0) -> Tuple[np.ndarray, np.ndarray]:
96 """
97 Remove statistical outliers from point cloud.
98
99 Returns:
100 cleaned_points: Points after outlier removal
101 inlier_indices: Indices of retained points
102 """
103 pcd = o3d.geometry.PointCloud()
104 pcd.points = o3d.utility.Vector3dVector(points[:, :3])
105
106 cleaned, indices = pcd.remove_statistical_outlier(
107 nb_neighbors=nb_neighbors, std_ratio=std_ratio
108 )
109
110 inlier_indices = np.asarray(indices)
111 cleaned_points = points[inlier_indices]
112
113 return cleaned_points, inlier_indices
114
115 @staticmethod
116 def segment_plane(points: np.ndarray,
117 distance_threshold: float = 0.02,
118 ransac_n: int = 3,
119 num_iterations: int = 1000
120 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
121 """
122 Segment dominant plane using RANSAC.
123
124 Returns:
125 plane_model: [a, b, c, d] where ax + by + cz + d = 0
126 inliers: Points on the plane
127 outliers: Points not on the plane
128 """
129 pcd = o3d.geometry.PointCloud()
130 pcd.points = o3d.utility.Vector3dVector(points[:, :3])
131
132 plane_model, inlier_indices = pcd.segment_plane(
133 distance_threshold=distance_threshold,
134 ransac_n=ransac_n,
135 num_iterations=num_iterations
136 )
137
138 inlier_indices = np.asarray(inlier_indices)
139 all_indices = np.arange(len(points))
140 outlier_indices = np.setdiff1d(all_indices, inlier_indices)
141
142 return np.array(plane_model), points[inlier_indices], points[outlier_indices]
143
144
145class ICP:
146 """
147 Iterative Closest Point for point cloud registration.
148 """
149
150 @staticmethod
151 def align(source: np.ndarray,
152 target: np.ndarray,
153 threshold: float = 0.05,
154 init_transform: np.ndarray = None,
155 max_iterations: int = 50) -> Tuple[np.ndarray, float]:
156 """
157 Align source point cloud to target using ICP.
158
159 Returns:
160 transformation: 4x4 transformation matrix
161 fitness: Registration fitness score
162 """
163 source_pcd = o3d.geometry.PointCloud()
164 source_pcd.points = o3d.utility.Vector3dVector(source[:, :3])
165
166 target_pcd = o3d.geometry.PointCloud()
167 target_pcd.points = o3d.utility.Vector3dVector(target[:, :3])
168
169 if init_transform is None:
170 init_transform = np.eye(4)
171
172 result = o3d.pipelines.registration.registration_icp(
173 source_pcd, target_pcd, threshold, init_transform,
174 o3d.pipelines.registration.TransformationEstimationPointToPoint(),
175 o3d.pipelines.registration.ICPConvergenceCriteria(
176 max_iteration=max_iterations
177 )
178 )
179
180 return result.transformation, result.fitness
181
182 @staticmethod
183 def transform_points(points: np.ndarray,
184 transformation: np.ndarray) -> np.ndarray:
185 """Apply 4x4 transformation to points."""
186 n = len(points)
187 homogeneous = np.hstack([points[:, :3], np.ones((n, 1))])
188 transformed = (transformation @ homogeneous.T).T
189 return transformed[:, :3]
190
191
192# Example
193print("Point Cloud Processing utilities loaded")
194print("Classes: PointCloudProcessor, ICP")
195
196# Demo with synthetic data
197np.random.seed(42)
198demo_points = np.random.randn(1000, 3) + [5, 0, 10]
199print(f"Demo point cloud: {demo_points.shape[0]} points")
200print(f"Bounds: X=[{demo_points[:,0].min():.2f}, {demo_points[:,0].max():.2f}]")Build an end-to-end pipeline for 3D reconstruction from stereo images.
1import numpy as np
2import cv2
3from typing import Tuple, Dict, Optional
4import matplotlib.pyplot as plt
5from mpl_toolkits.mplot3d import Axes3D
6
7class Reconstruction3DPipeline:
8 """
9 Complete 3D reconstruction pipeline from stereo images.
10 """
11
12 def __init__(self, baseline: float, focal_length: float,
13 cx: float, cy: float):
14 """
15 Args:
16 baseline: Distance between cameras (mm)
17 focal_length: Focal length (pixels)
18 cx, cy: Principal point
19 """
20 self.baseline = baseline
21 self.focal_length = focal_length
22 self.cx = cx
23 self.cy = cy
24
25 # Camera intrinsics
26 self.K = np.array([
27 [focal_length, 0, cx],
28 [0, focal_length, cy],
29 [0, 0, 1]
30 ], dtype=np.float64)
31
32 # Stereo matcher
33 self.stereo = cv2.StereoSGBM_create(
34 minDisparity=0,
35 numDisparities=16 * 8,
36 blockSize=5,
37 P1=8 * 3 * 5 ** 2,
38 P2=32 * 3 * 5 ** 2,
39 disp12MaxDiff=1,
40 uniquenessRatio=10,
41 speckleWindowSize=100,
42 speckleRange=32,
43 preFilterCap=63,
44 mode=cv2.STEREO_SGBM_MODE_SGBM_3WAY
45 )
46
47 def compute_disparity(self, left: np.ndarray,
48 right: np.ndarray) -> np.ndarray:
49 """Compute disparity map from stereo pair."""
50 # Convert to grayscale
51 left_gray = cv2.cvtColor(left, cv2.COLOR_BGR2GRAY)
52 right_gray = cv2.cvtColor(right, cv2.COLOR_BGR2GRAY)
53
54 # Compute disparity
55 disparity = self.stereo.compute(left_gray, right_gray)
56
57 # Convert from fixed-point
58 disparity = disparity.astype(np.float32) / 16.0
59
60 return disparity
61
62 def disparity_to_depth(self, disparity: np.ndarray) -> np.ndarray:
63 """Convert disparity to depth map."""
64 depth = np.zeros_like(disparity)
65 valid = disparity > 0
66
67 depth[valid] = self.baseline * self.focal_length / disparity[valid]
68
69 return depth
70
71 def depth_to_pointcloud(self, depth: np.ndarray,
72 color: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
73 """
74 Convert depth map to 3D point cloud.
75
76 Returns:
77 points: (N, 3) array of 3D coordinates
78 colors: (N, 3) array of RGB colors
79 """
80 h, w = depth.shape
81 u, v = np.meshgrid(np.arange(w), np.arange(h))
82
83 # Unproject
84 Z = depth
85 X = (u - self.cx) * Z / self.focal_length
86 Y = (v - self.cy) * Z / self.focal_length
87
88 # Stack
89 points = np.stack([X, Y, Z], axis=-1).reshape(-1, 3)
90
91 # Get colors
92 colors = cv2.cvtColor(color, cv2.COLOR_BGR2RGB).reshape(-1, 3) / 255.0
93
94 # Filter invalid points
95 valid = (Z.flatten() > 0) & (Z.flatten() < 10000)
96 points = points[valid]
97 colors = colors[valid]
98
99 return points, colors
100
101 def reconstruct(self, left: np.ndarray,
102 right: np.ndarray) -> Dict:
103 """
104 Full reconstruction pipeline.
105
106 Returns:
107 Dictionary with disparity, depth, points, and colors
108 """
109 # Compute disparity
110 disparity = self.compute_disparity(left, right)
111
112 # Convert to depth
113 depth = self.disparity_to_depth(disparity)
114
115 # Generate point cloud
116 points, colors = self.depth_to_pointcloud(depth, left)
117
118 # Statistics
119 valid_points = len(points)
120 depth_range = (depth[depth > 0].min(), depth[depth > 0].max())
121
122 return {
123 'disparity': disparity,
124 'depth': depth,
125 'points': points,
126 'colors': colors,
127 'stats': {
128 'valid_points': valid_points,
129 'depth_range': depth_range
130 }
131 }
132
133 def visualize(self, result: Dict, figsize: Tuple[int, int] = (15, 10)):
134 """Visualize reconstruction results."""
135 fig = plt.figure(figsize=figsize)
136
137 # Disparity
138 ax1 = fig.add_subplot(2, 2, 1)
139 disp_vis = result['disparity'].copy()
140 disp_vis[disp_vis <= 0] = 0
141 ax1.imshow(disp_vis, cmap='jet')
142 ax1.set_title('Disparity Map')
143 ax1.axis('off')
144
145 # Depth
146 ax2 = fig.add_subplot(2, 2, 2)
147 depth_vis = result['depth'].copy()
148 depth_vis[depth_vis <= 0] = np.nan
149 ax2.imshow(depth_vis, cmap='plasma')
150 ax2.set_title('Depth Map')
151 ax2.axis('off')
152
153 # Point cloud (top view)
154 ax3 = fig.add_subplot(2, 2, 3)
155 points = result['points']
156 subsample = np.random.choice(len(points), min(5000, len(points)), replace=False)
157 ax3.scatter(points[subsample, 0], points[subsample, 2],
158 c=result['colors'][subsample], s=0.5)
159 ax3.set_xlabel('X')
160 ax3.set_ylabel('Z (depth)')
161 ax3.set_title('Top View')
162
163 # Point cloud 3D
164 ax4 = fig.add_subplot(2, 2, 4, projection='3d')
165 ax4.scatter(points[subsample, 0], points[subsample, 2],
166 points[subsample, 1], c=result['colors'][subsample], s=0.5)
167 ax4.set_xlabel('X')
168 ax4.set_ylabel('Z')
169 ax4.set_zlabel('Y')
170 ax4.set_title('3D Point Cloud')
171
172 plt.tight_layout()
173 plt.show()
174
175 def save_pointcloud_ply(self, points: np.ndarray, colors: np.ndarray,
176 filename: str):
177 """Save point cloud to PLY file."""
178 with open(filename, 'w') as f:
179 # Header
180 f.write("ply\n")
181 f.write("format ascii 1.0\n")
182 f.write(f"element vertex {len(points)}\n")
183 f.write("property float x\n")
184 f.write("property float y\n")
185 f.write("property float z\n")
186 f.write("property uchar red\n")
187 f.write("property uchar green\n")
188 f.write("property uchar blue\n")
189 f.write("end_header\n")
190
191 # Data
192 for point, color in zip(points, colors):
193 r, g, b = (color * 255).astype(int)
194 f.write(f"{point[0]:.4f} {point[1]:.4f} {point[2]:.4f} {r} {g} {b}\n")
195
196
197def main():
198 # Initialize pipeline
199 pipeline = Reconstruction3DPipeline(
200 baseline=120.0, # 120mm baseline
201 focal_length=500.0, # 500px focal length
202 cx=320, # Principal point x
203 cy=240 # Principal point y
204 )
205
206 print("3D Reconstruction Pipeline initialized")
207 print(f" Baseline: {pipeline.baseline}mm")
208 print(f" Focal length: {pipeline.focal_length}px")
209 print(f" Principal point: ({pipeline.cx}, {pipeline.cy})")
210
211 # Create synthetic stereo pair for demo
212 h, w = 480, 640
213
214 # Left image with some objects
215 left = np.zeros((h, w, 3), dtype=np.uint8)
216 cv2.rectangle(left, (100, 100), (200, 250), (255, 0, 0), -1)
217 cv2.rectangle(left, (350, 150), (500, 350), (0, 255, 0), -1)
218 cv2.circle(left, (550, 200), 50, (0, 0, 255), -1)
219
220 # Right image (shifted to simulate stereo)
221 right = np.zeros((h, w, 3), dtype=np.uint8)
222 cv2.rectangle(right, (90, 100), (190, 250), (255, 0, 0), -1)
223 cv2.rectangle(right, (340, 150), (490, 350), (0, 255, 0), -1)
224 cv2.circle(right, (540, 200), 50, (0, 0, 255), -1)
225
226 # Run reconstruction
227 print("\nRunning reconstruction...")
228 result = pipeline.reconstruct(left, right)
229
230 print(f"\nResults:")
231 print(f" Valid points: {result['stats']['valid_points']:,}")
232 print(f" Depth range: {result['stats']['depth_range'][0]:.1f} - {result['stats']['depth_range'][1]:.1f}mm")
233
234 print("\nPipeline ready! Call pipeline.reconstruct(left, right) with real stereo images.")
235
236if __name__ == "__main__":
237 main()This complete pipeline demonstrates stereo-based 3D reconstruction including disparity computation using SGBM, depth conversion, point cloud generation, and visualization tools.
Next lesson covers Deployment and Optimization techniques including model quantization, ONNX export, and mobile deployment strategies.
Your progress is saved in this browser, on this device only.