写代码自动识别房间杂乱程度，给出整理顺序，颠覆房间乱到无从下手。
数字文化空间智能整理系统
一、实际应用场景描述
某文创工作室"墨韵雅集"位于北京798艺术区，主营数字艺术展览与手工文创产品。工作室面积80平米，分为展示区、创作区、接待区和储物区。2024年春季，工作室举办"数字山水"主题展期间，由于访客众多、作品频繁更换，空间逐渐陷入混乱：
- 展示区：数字绘画作品、AR装置、VR头显散落各处，线缆缠绕
- 创作区：数位板、iPad Pro、颜料、半成品画布无序堆放
- 接待区：茶具、宣传册、客户资料、充电线混杂
- 储物区：纸箱堆积，过期物料与新到货品混放
店主李艺每天花费1.5小时找东西、理空间，严重影响创作效率和客户体验。传统收纳APP需要手动拍照、分类、规划，操作繁琐且无法理解空间三维关系。
系统运行环境：
- 硬件：普通RGB-D相机（RealSense D435i）+ 树莓派4B/PC
- 软件：Python 3.10, OpenCV, PyTorch, Transformers, LangChain
- 场景：文创工作室、家庭书房、创客空间、艺术教室
- 数据：实时RGB-D图像、空间点云、物品特征库
核心需求：
- 一键扫描，自动识别空间杂乱程度
- 智能规划整理顺序，从"无从下手"到"步步为营"
- 考虑物品关联性、使用频率、美学布局
- 生成可执行的三步整理指令
- 支持数字文化物品的专业分类
- 可视化展示整理前后对比
二、引入痛点
1. 认知过载：面对杂乱空间，人类大脑无法同时处理所有物品的归位逻辑
2. 决策瘫痪：不知道从何开始整理，每个区域都"看起来很乱"
3. 关联断裂：手工制品、数字设备、耗材之间存在隐性关联，人工难以梳理
4. 美学缺失：整理后往往功能性恢复了，但失去了文创空间的创意氛围
5. 重复劳动：每次整理后很快再次混乱，缺乏可持续的空间管理机制
6. 专业盲区：普通收纳系统不懂区分"待售作品"与"创作素材"
7. 空间浪费：不懂利用垂直空间、角落空间进行创意收纳
三、核心逻辑讲解
graph TD
A[RGB-D扫描] --> B[点云重建]
B --> C[物品检测与分割]
C --> D[特征提取与分类]
D --> E[空间关系分析]
E --> F[杂乱度评估]
F --> G[整理优先级排序]
G --> H[三步整理规划]
H --> I[美学优化建议]
I --> J[可执行指令生成]
subgraph "数字文化专业层"
K[作品类型识别] --> L[创作流程分析]
M[数字设备关联] --> N[线缆管理优化]
O[材料生命周期] --> P[存储策略制定]
end
subgraph "空间智能层"
Q[动线分析] --> R[黄金三角优化]
S[视觉焦点] --> T[展示效果评估]
U[可达性] --> V[取用效率计算]
end
D --> K
D --> M
D --> O
E --> Q
E --> S
E --> U
F -->|高杂乱度| W[紧急整理模式]
F -->|中杂乱度| X[标准整理模式]
F -->|低杂乱度| Y[维护模式]
关键技术突破
1. 多模态物品理解：结合视觉、深度、材质信息，识别数字文化特有物品
2. 空间语法分析：基于空间句法理论，分析物品间的功能关联
3. 美学约束优化：在功能性整理基础上，融入构图法则和视觉美学
4. 三步递进策略：将复杂整理任务分解为"清场→分类→归位"的心理可接受步骤
5. 动态优先级算法：考虑物品使用频率、关联度、美学价值的综合排序
四、代码模块化实现
项目结构
digital_space_organizer/
├── config/
│ ├── space_config.yaml
│ ├── item_database.json
│ ├── aesthetic_rules.yaml
│ └── workflow_profiles.json
├── data/
│ ├── scans/
│ │ ├── raw_pointclouds/
│ │ ├── segmented_objects/
│ │ └── processed_features/
│ ├── knowledge_base/
│ │ ├── item_embeddings/
│ │ ├── spatial_graphs/
│ │ └── organization_templates/
│ └── outputs/
│ ├── organization_plans/
│ ├── visualizations/
│ └── reports/
├── src/
│ ├── perception/
│ │ ├── __init__.py
│ │ ├── rgbd_scanner.py
│ │ ├── pointcloud_processor.py
│ │ ├── object_detector.py
│ │ └── feature_extractor.py
│ ├── cognition/
│ │ ├── __init__.py
│ │ ├── item_classifier.py
│ │ ├── spatial_analyzer.py
│ │ ├── clutter_evaluator.py
│ │ └── relationship_mapper.py
│ ├── planning/
│ │ ├── __init__.py
│ │ ├── priority_sorter.py
│ │ ├── three_step_planner.py
│ │ ├── aesthetic_optimizer.py
│ │ └── workflow_generator.py
│ ├── execution/
│ │ ├── __init__.py
│ │ ├── instruction_builder.py
│ │ ├── visualization_engine.py
│ │ └── progress_tracker.py
│ ├── digital_culture/
│ │ ├── __init__.py
│ │ ├── creative_item_recognizer.py
│ │ ├── artwork_classifier.py
│ │ ├── equipment_manager.py
│ │ └── material_lifecycle.py
│ ├── utils/
│ │ ├── __init__.py
│ │ ├── geometry_utils.py
│ │ ├── file_manager.py
│ │ ├── visualizer.py
│ │ └── logger.py
│ └── main.py
├── scripts/
│ ├── calibrate_scanner.py
│ ├── build_knowledge_base.py
│ ├── demo_organization.py
│ └── batch_process.py
├── tests/
│ ├── test_object_detection.py
│ ├── test_clutter_evaluation.py
│ ├── test_three_step_planning.py
│ └── test_aesthetic_optimization.py
├── docs/
│ ├── api_reference.md
│ ├── user_guide.md
│ └── aesthetic_principles.md
├── README.md
├── requirements.txt
└── LICENSE
核心代码实现
1. RGB-D扫描与点云处理 (perception/rgbd_scanner.py)
"""
RGB-D扫描器 - 数字文化空间感知入口
负责获取环境的RGB图像、深度图和重建点云
"""
import numpy as np
import cv2
import open3d as o3d
import logging
from typing import Tuple, Optional, Dict, List
from dataclasses import dataclass, field
from pathlib import Path
import time
from enum import Enum
import threading
from queue import Queue
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ScanQuality(Enum):
"""扫描质量等级"""
LOW = "low" # 640x480, 快速扫描
MEDIUM = "medium" # 1280x720, 平衡
HIGH = "high" # 1920x1080, 精细扫描
@dataclass
class ScanFrame:
"""单帧扫描数据"""
frame_id: str
timestamp: float
rgb_image: np.ndarray
depth_map: np.ndarray
camera_intrinsics: Dict
pose_matrix: np.ndarray
quality: ScanQuality
@dataclass
class SpaceScan:
"""空间扫描结果"""
scan_id: str
created_at: float
frames: List[ScanFrame] = field(default_factory=list)
pointcloud: Optional[o3d.geometry.PointCloud] = None
mesh: Optional[o3d.geometry.TriangleMesh] = None
dimensions: Dict = field(default_factory=dict) # 空间尺寸
obstacles: List[Dict] = field(default_factory=list) # 障碍物信息
class RGBDScanner:
"""
RGB-D扫描器主类
支持多种RGB-D相机，提供高质量空间扫描能力
"""
def __init__(self, config: Dict):
self.config = config
self.camera_type = config.get("camera_type", "realsense_d435i")
self.scan_quality = ScanQuality(config.get("scan_quality", "medium"))
self.scan_resolution = self._get_resolution_from_quality()
self.intrinsics = config.get("camera_intrinsics", {})
# 扫描参数
self.voxel_size = config.get("voxel_size", 0.005) # 点云降采样体素大小
self.depth_scale = config.get("depth_scale", 1000.0) # 深度缩放因子
self.max_depth = config.get("max_depth", 3.0) # 最大有效深度(米)
# 初始化相机
self._initialize_camera()
# 扫描状态
self.is_scanning = False
self.scan_queue = Queue(maxsize=100)
logger.info(f"RGBD Scanner initialized: {self.camera_type}, quality={self.scan_quality.value}")
def _get_resolution_from_quality(self) -> Tuple[int, int]:
"""根据质量等级获取分辨率"""
resolution_map = {
ScanQuality.LOW: (640, 480),
ScanQuality.MEDIUM: (1280, 720),
ScanQuality.HIGH: (1920, 1080)
}
return resolution_map[self.scan_quality]
def _initialize_camera(self):
"""初始化RGB-D相机"""
if self.camera_type == "realsense_d435i":
self._initialize_realsense()
elif self.camera_type == "kinect_v2":
self._initialize_kinect()
else:
# 模拟模式，用于开发和测试
self._initialize_simulation_mode()
def _initialize_realsense(self):
"""初始化Intel RealSense D435i"""
try:
import pyrealsense2 as rs
self.pipeline = rs.pipeline()
config = rs.config()
width, height = self.scan_resolution
config.enable_stream(rs.stream.color, width, height, rs.format.bgr8, 30)
config.enable_stream(rs.stream.depth, width, height, rs.format.z16, 30)
# 获取设备内参
profile = self.pipeline.start(config)
depth_profile = profile.get_stream(rs.stream.depth)
color_profile = profile.get_stream(rs.stream.color)
self.depth_intrinsics = depth_profile.as_video_stream_profile().get_intrinsics()
self.color_intrinsics = color_profile.as_video_stream_profile().get_intrinsics()
# 转换内参格式
self.intrinsics = {
"fx": self.depth_intrinsics.fx,
"fy": self.depth_intrinsics.fy,
"cx": self.depth_intrinsics.ppx,
"cy": self.depth_intrinsics.ppy,
"width": width,
"height": height
}
logger.info("RealSense D435i initialized successfully")
except ImportError:
logger.warning("pyrealsense2 not installed, falling back to simulation mode")
self._initialize_simulation_mode()
except Exception as e:
logger.error(f"Failed to initialize RealSense: {e}")
self._initialize_simulation_mode()
def _initialize_kinect(self):
"""初始化Kinect V2 (Windows only)"""
try:
from pykinect2 import PyKinectV2, PyKinectRuntime
self.kinect = PyKinectRuntime.PyKinectRuntime(
PyKinectV2.FrameSourceTypes_Color | PyKinectV2.FrameSourceTypes_Depth
)
self.kinect_color_intrinsics = {
"fx": 1081.37, "fy": 1081.37,
"cx": 959.5, "cy": 539.5,
"width": 1920, "height": 1080
}
self.intrinsics = self.kinect_color_intrinsics
logger.info("Kinect V2 initialized successfully")
except ImportError:
logger.warning("pykinect2 not installed, falling back to simulation mode")
self._initialize_simulation_mode()
def _initialize_simulation_mode(self):
"""初始化模拟模式（用于开发测试）"""
self.camera_type = "simulation"
self.intrinsics = {
"fx": 525.0, "fy": 525.0,
"cx": 319.5, "cy": 239.5,
"width": self.scan_resolution[0],
"height": self.scan_resolution[1]
}
logger.info("Running in simulation mode")
def capture_frame(self) -> Optional[ScanFrame]:
"""捕获单帧RGB-D数据"""
if self.camera_type == "realsense_d435i":
return self._capture_realsense_frame()
elif self.camera_type == "kinect_v2":
return self._capture_kinect_frame()
else:
return self._capture_simulation_frame()
def _capture_realsense_frame(self) -> Optional[ScanFrame]:
"""从RealSense捕获帧"""
import pyrealsense2 as rs
try:
frames = self.pipeline.wait_for_frames(timeout_ms=5000)
color_frame = frames.get_color_frame()
depth_frame = frames.get_depth_frame()
if not color_frame or not depth_frame:
return None
# 转换为numpy数组
rgb_image = np.asanyarray(color_frame.get_data())
depth_map = np.asanyarray(depth_frame.get_data()).astype(np.float32) / self.depth_scale
# 获取相机位姿（简化，实际需SLAM）
pose_matrix = np.eye(4)
frame = ScanFrame(
frame_id=f"frame_{int(time.time()*1000)}",
timestamp=time.time(),
rgb_image=rgb_image,
depth_map=depth_map,
camera_intrinsics=self.intrinsics,
pose_matrix=pose_matrix,
quality=self.scan_quality
)
return frame
except Exception as e:
logger.error(f"Error capturing RealSense frame: {e}")
return None
def _capture_kinect_frame(self) -> Optional[ScanFrame]:
"""从Kinect捕获帧"""
try:
if self.kinect.has_new_color_frame() and self.kinect.has_new_depth_frame():
rgb_image = self.kinect.get_last_color_frame().reshape(
(1080, 1920, 4)
)[:, :, :3]
depth_map = self.kinect.get_last_depth_frame().reshape(
(424, 512)
).astype(np.float32) / self.depth_scale
# 上采样深度图到RGB分辨率
depth_map = cv2.resize(depth_map, (1920, 1080))
frame = ScanFrame(
frame_id=f"frame_{int(time.time()*1000)}",
timestamp=time.time(),
rgb_image=rgb_image,
depth_map=depth_map,
camera_intrinsics=self.intrinsics,
pose_matrix=np.eye(4),
quality=self.scan_quality
)
return frame
return None
except Exception as e:
logger.error(f"Error capturing Kinect frame: {e}")
return None
def _capture_simulation_frame(self) -> Optional[ScanFrame]:
"""生成模拟扫描帧（用于测试）"""
width, height = self.scan_resolution
# 生成模拟RGB图像（渐变色背景+随机"物品"）
rgb_image = np.zeros((height, width, 3), dtype=np.uint8)
# 添加渐变背景
for y in range(height):
for x in range(width):
rgb_image[y, x] = [
int(200 + 55 * np.sin(x / width * np.pi)),
int(180 + 75 * np.cos(y / height * np.pi)),
int(220 + 35 * np.sin((x + y) / (width + height) * np.pi))
]
# 添加模拟"物品"（彩色矩形块）
np.random.seed(int(time.time()) % 1000)
num_items = np.random.randint(5, 15)
for _ in range(num_items):
x1 = np.random.randint(50, width - 150)
y1 = np.random.randint(50, height - 150)
w = np.random.randint(80, 200)
h = np.random.randint(80, 200)
color = np.random.randint(0, 200, 3).tolist()
cv2.rectangle(rgb_image, (x1, y1), (x1 + w, y1 + h), color, -1)
cv2.rectangle(rgb_image, (x1, y1), (x1 + w, y1 + h), (0, 0, 0), 2)
# 生成模拟深度图
depth_map = np.zeros((height, width), dtype=np.float32)
# 背景深度
depth_map.fill(self.max_depth * 0.8)
# 物品深度（随机距离）
for _ in range(num_items):
x1 = np.random.randint(50, width - 150)
y1 = np.random.randint(50, height - 150)
w = np.random.randint(80, 200)
h = np.random.randint(80, 200)
depth = np.random.uniform(0.5, 2.5)
depth_map[y1:y1+h, x1:x1+w] = depth
# 添加噪声
depth_noise = np.random.normal(0, 0.02, depth_map.shape)
depth_map = np.clip(depth_map + depth_noise, 0, self.max_depth)
frame = ScanFrame(
frame_id=f"sim_frame_{int(time.time()*1000)}",
timestamp=time.time(),
rgb_image=rgb_image,
depth_map=depth_map,
camera_intrinsics=self.intrinsics,
pose_matrix=np.eye(4),
quality=self.scan_quality
)
return frame
def start_continuous_scan(self, callback: Callable[[ScanFrame], None] = None):
"""开始连续扫描"""
self.is_scanning = True
logger.info("Started continuous scanning")
def scan_loop():
while self.is_scanning:
frame = self.capture_frame()
if frame:
if callback:
callback(frame)
else:
self.scan_queue.put(frame)
time.sleep(1.0 / 30) # 30 FPS
self.scan_thread = threading.Thread(target=scan_loop, daemon=True)
self.scan_thread.start()
def stop_continuous_scan(self):
"""停止连续扫描"""
self.is_scanning = False
if hasattr(self, 'scan_thread'):
self.scan_thread.join(timeout=2.0)
logger.info("Stopped continuous scanning")
def reconstruct_pointcloud(self, frames: List[ScanFrame]) -> o3d.geometry.PointCloud:
"""从多帧数据重建点云"""
all_points = []
all_colors = []
for frame in frames:
points, colors = self._frame_to_pointcloud(frame)
if points is not None:
all_points.append(points)
all_colors.append(colors)
if not all_points:
logger.warning("No valid points for reconstruction")
return o3d.geometry.PointCloud()
# 合并所有点
combined_points = np.vstack(all_points)
combined_colors = np.vstack(all_colors)
# 创建Open3D点云
pcd = o3d.geometry.PointCloud()
pcd.points = o3d.utility.Vector3dVector(combined_points)
pcd.colors = o3d.utility.Vector3dVector(combined_colors)
# 降采样
pcd = pcd.voxel_down_sample(voxel_size=self.voxel_size)
# 去除离群点
pcd, _ = pcd.remove_statistical_outlier(nb_neighbors=20, std_ratio=2.0)
# 估计法线
pcd.estimate_normals(
search_param=o3d.geometry.KDTreeSearchParamHybrid(
radius=self.voxel_size * 2, max_nn=30
)
)
logger.info(f"Point cloud reconstructed: {len(pcd.points)} points")
return pcd
def _frame_to_pointcloud(self, frame: ScanFrame) -> Tuple[Optional[np.ndarray], Optional[np.ndarray]]:
"""将单帧数据转换为点云"""
height, width = frame.depth_map.shape
fx, fy = frame.camera_intrinsics["fx"], frame.camera_intrinsics["fy"]
cx, cy = frame.camera_intrinsics["cx"], frame.camera_intrinsics["cy"]
# 创建网格
u, v = np.meshgrid(np.arange(width), np.arange(height))
u = u.flatten()
v = v.flatten()
# 获取深度值
z = frame.depth_map.flatten()
# 过滤无效深度
valid_mask = (z > 0.1) & (z < self.max_depth)
u = u[valid_mask]
v = v[valid_mask]
z = z[valid_mask]
if len(z) == 0:
return None, None
# 反投影到3D坐标
x = (u - cx) * z / fx
y = (v - cy) * z / fy
points = np.stack([x, y, z], axis=1)
# 获取颜色
colors = frame.rgb_image[v, u] / 255.0
return points, colors
def calculate_space_dimensions(self, pcd: o3d.geometry.PointCloud) -> Dict:
"""计算空间尺寸信息"""
if len(pcd.points) == 0:
return {"error": "Empty point cloud"}
points = np.asarray(pcd.points)
# 计算边界框
min_bound = points.min(axis=0)
max_bound = points.max(axis=0)
dimensions = {
"length": max_bound[0] - min_bound[0], # X轴
"width": max_bound[1] - min_bound[1], # Y轴
"height": max_bound[2] - min_bound[2], # Z轴
"volume": (max_bound[0] - min_bound[0]) *
(max_bound[1] - min_bound[1]) *
(max_bound[2] - min_bound[2]),
"floor_area": (max_bound[0] - min_bound[0]) *
(max_bound[1] - min_bound[1]),
"center": (min_bound + max_bound) / 2,
"min_bound": min_bound.tolist(),
"max_bound": max_bound.tolist()
}
return dimensions
def detect_obstacles(self, pcd: o3d.geometry.PointCloud,
floor_height: float = 0.0) -> List[Dict]:
"""检测空间中的障碍物"""
if len(pcd.points) == 0:
return []
points = np.asarray(pcd.points)
# 假设地面高度已知或自动检测
if floor_height == 0.0:
# 简单地面检测：取最低10%点的平均高度
z_values = points[:, 2]
floor_height = np.percentile(z_values, 10)
# 过滤地面附近的点（可能是地面本身）
obstacle_mask = points[:, 2] > floor_height + 0.1
obstacle_points = points[obstacle_mask]
if len(obstacle_points) == 0:
return []
# 聚类分离不同障碍物
from sklearn.cluster import DBSCAN
clustering = DBSCAN(eps=0.3, min_samples=50).fit(obstacle_points)
labels = clustering.labels_
obstacles = []
unique_labels = set(labels)
for label in unique_labels:
if label == -1: # 噪声点
continue
cluster_mask = labels == label
cluster_points = obstacle_points[cluster_mask]
# 计算障碍物边界
min_pt = cluster_points.min(axis=0)
max_pt = cluster_points.max(axis=0)
obstacle = {
"id": f"obstacle_{label}",
"position": ((min_pt + max_pt) / 2).tolist(),
"dimensions": {
"length": max_pt[0] - min_pt[0],
"width": max_pt[1] - min_pt[1],
"height": max_pt[2] - min_pt[2]
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！