🌐 DDS 5维时空数据空间

CREATE TABLE dds_5d_points ( id TEXT PRIMARY KEY, -- D001, C004, A008, F004 skill_id TEXT NOT NULL, -- 关联原子技能 P001~M073 s_layer TEXT NOT NULL, -- sensor|feature|condition|action|feedback x REAL NOT NULL, -- 空间X坐标 (工厂坐标系) y REAL NOT NULL, -- T时间维度 (语义层高度) z REAL NOT NULL, -- 空间Z坐标 name TEXT NOT NULL, -- 显示名称 desc TEXT, -- 描述 links TEXT, -- JSON数组: 关联的下游节点ID station TEXT, -- 工位编号 I-103 zone TEXT, -- 产线区域 COC/OE/MOD/WH confidence REAL DEFAULT 1.0, -- 条件置信度 threshold TEXT, -- 判定阈值 created DATETIME DEFAULT CURRENT_TIMESTAMP );

#!/usr/bin/env python3 """atomic_skills → 5D spacetime data points batch labeling""" import sqlite3, json, re conn = sqlite3.connect('dds.db') cur = conn.cursor() # 确保表存在 cur.execute('''CREATE TABLE IF NOT EXISTS dds_5d_points ( id TEXT PRIMARY KEY, skill_id TEXT, s_layer TEXT, x REAL, y REAL, z REAL, name TEXT, desc TEXT, links TEXT, station TEXT, zone TEXT, confidence REAL DEFAULT 1.0, threshold TEXT )''') # 语义层高度映射 LAYER_Y = {"sensor":-6, "feature":-2, "condition":2, "action":6, "feedback":10} # 传感器关键词 → 数据点映射 SENSOR_MAP = { "相机": ("RGB-D图像采集", "2D/3D视觉输入"), "显微镜": ("显微放大采集", "高倍率缺陷检测"), "力": ("力/力矩传感", "六维力数据流"), "触觉": ("触觉阵列", "4×4压力分布"), "激光": ("激光轮廓仪", "3D点云扫描"), "温度": ("温度传感", "红外热成像"), "扫码": ("条码/RFID读取", "MES追溯绑定"), } cur.execute("SELECT id,name,category,input_cond,sensors,output_criteria,scene,phase FROM atomic_skills") skills = cur.fetchall() def parse_sensors(sensors_str): """拆分传感器字段提取数据点""" points = [] if not sensors_str: return points for kw, (name, desc) in SENSOR_MAP.items(): if kw in sensors_str: points.append({"name": name, "desc": desc, "layer": "sensor"}) return points def parse_conditions(input_str, output_str): """从输入条件和输出标准提取特征+条件点""" points = [] if input_str: # 按 /、 分割提取独立条件 parts = re.split(r'[/、,，]', input_str) for p in parts[:3]: # 最多3个 p = p.strip() if len(p) > 2: points.append({"name": p + "检测", "desc": "特征提取", "layer": "feature"}) if output_str: thresholds = re.findall(r'[\d.]+%|[\d.]+ms|[\d.]+μm|[\d.]+mm|[\d.]+N|[\d.]+dB', output_str) if thresholds: points.append({ "name": "判定:" + ",".join(thresholds[:3]), "desc": output_str[:60], "layer": "condition", "threshold": ",".join(thresholds[:3]) }) return points count = 0 for s in skills: sid, name, cat, input_c, sensors, output_c, scene, phase = s # Step 1: 传感器 → 感知层 for pt in parse_sensors(sensors): pid = f"D{count:04d}" cur.execute( "INSERT OR REPLACE INTO dds_5d_points VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (pid, sid, "sensor", 0, LAYER_Y["sensor"], 0, pt["name"], pt["desc"], json.dumps([]), None, None, 1.0, None)) count += 1 # Step 2-3: 条件 → 特征层+条件层 for pt in parse_conditions(input_c, output_c): pid = f"D{count:04d}" cur.execute( "INSERT OR REPLACE INTO dds_5d_points VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (pid, sid, pt["layer"], 0, LAYER_Y.get(pt["layer"], 2), 0, pt["name"], pt.get("desc",""), json.dumps([]), None, None, 1.0, pt.get("threshold"))) count += 1 # Step 4: 技能本身 → 动作层 aid = f"A{count:04d}" cur.execute( "INSERT OR REPLACE INTO dds_5d_points VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (aid, sid, "action", 0, LAYER_Y["action"], 0, name, scene or "", json.dumps([]), None, cat, 1.0, None)) count += 1 # Step 5: 成功标准 → 反馈层 fid = f"F{count:04d}" cur.execute( "INSERT OR REPLACE INTO dds_5d_points VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (fid, sid, "feedback", 0, LAYER_Y["feedback"], 0, name + "·验收", output_c[:60] if output_c else "", json.dumps([]), None, cat, 1.0, None)) count += 1 conn.commit() print(f"✅ 批量打标完成: {count} 个5维数据点 (来自 {len(skills)} 条原子技能)") conn.close()

-- 查询"力控插入"相关的所有5维数据点 SELECT s_layer, name, desc, station FROM dds_5d_points WHERE skill_id IN ( SELECT id FROM atomic_skills WHERE name LIKE '%力控%插入%' ) ORDER BY s_layer, id; -- 查询某个工位的全部条件 SELECT id, s_layer, name, desc, threshold FROM dds_5d_points WHERE station = 'I-103' ORDER BY y, id; -- 查询所有P0优先级的动作点 SELECT dp.name, dp.station, a.phase FROM dds_5d_points dp JOIN atomic_skills a ON dp.skill_id = a.id WHERE dp.s_layer = 'action' AND a.phase LIKE 'P0%';