From 691a812fcf6b58345893c748eadcc51fdf65f73e Mon Sep 17 00:00:00 2001 From: yunmei Date: Tue, 4 Aug 2026 18:14:37 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#29=20TDen?= =?UTF-8?q?gine=20=E8=B6=85=E7=BA=A7=E8=A1=A8=20schema=20=E4=BE=9D?= =?UTF-8?q?=E6=8D=AE=E7=82=B9=E4=BD=8D=E5=AD=97=E5=85=B8=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E7=94=9F=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/data-bus/README.md | 8 ++- core/data-bus/__init__.py | 4 ++ core/data-bus/_sanity_check.py | 12 +++++ core/data-bus/tdengine_schema.py | 43 +++++++++++++++ core/data-bus/tests/test_tdengine_schema.py | 59 +++++++++++++++++++++ 5 files changed, 125 insertions(+), 1 deletion(-) diff --git a/core/data-bus/README.md b/core/data-bus/README.md index bf85a52..fb10717 100644 --- a/core/data-bus/README.md +++ b/core/data-bus/README.md @@ -10,7 +10,7 @@ topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模 | 文件 | 职责 | |------|------| | `templating.py` | 模板命名推导：Kafka topic + 分区策略、TDengine 超级表/子表、PostgreSQL schema/表、MinIO 桶/对象键 | -| `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表）+ 批量 INSERT SQL | +| `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表，**依据点位字典**）+ 批量 INSERT SQL | | `postgres_schema.py` | 关系库 schema（模板 / 模型 / 用户 / 权限）+ 角色授权语句 | | `batch_writer.py` | 批量写入缓冲：批量聚合（默认 5000 条/0.1s，对齐 PRD 5.2「5k/100ms」基线）、幂等去重、失败重试 —— **数据不丢不重** | @@ -18,9 +18,15 @@ topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模 ```python from data_bus.batch_writer import BatchWriter, TdengineSink +from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows from data_bus.templating import TemplateNaming naming = TemplateNaming(template="ti-cl4") # 换行业只改模板名 + +# 依据点位字典自动生成 TDengine 完整 schema（超级表 + 每测点子表） +point_dict_rows = [{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "unit": "℃", "dataType": "float"}] +schema_ddls = generate_schema(naming, specs_from_point_dict_rows(point_dict_rows), retention_days=90) + sink = TdengineSink(naming, executor=run_sql) # executor 注入 TDengine 连接适配器 writer = BatchWriter(sink, batch_size=5000, flush_interval=0.1) diff --git a/core/data-bus/__init__.py b/core/data-bus/__init__.py index c79856e..588e060 100644 --- a/core/data-bus/__init__.py +++ b/core/data-bus/__init__.py @@ -28,8 +28,10 @@ from .postgres_schema import generate_grant_ddl, generate_schema_ddl from .tdengine_schema import ( PointSpec, build_batch_insert, + generate_schema, generate_subtable_ddls, generate_supertable_ddl, + specs_from_point_dict_rows, ) from .templating import TemplateNaming, sanitize, sanitize_sql @@ -40,6 +42,8 @@ __all__ = [ "PointSpec", "generate_supertable_ddl", "generate_subtable_ddls", + "generate_schema", + "specs_from_point_dict_rows", "build_batch_insert", "generate_schema_ddl", "generate_grant_ddl", diff --git a/core/data-bus/_sanity_check.py b/core/data-bus/_sanity_check.py index dd516c4..4a73dad 100644 --- a/core/data-bus/_sanity_check.py +++ b/core/data-bus/_sanity_check.py @@ -17,6 +17,7 @@ spec.loader.exec_module(pkg) from data_bus import BatchWriter, MemorySink, TdengineSink, TemplateNaming, __version__ # noqa: E402 from data_bus.postgres_schema import generate_schema_ddl # noqa: E402 +from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows # noqa: E402 ddl = generate_schema_ddl(TemplateNaming(template="ti-cl4")) assert "DEFAULT '{}'::jsonb" in ddl, "jsonb default escaped wrongly" @@ -24,3 +25,14 @@ assert "REFERENCES tpl_ti_cl4.templates(template)" in ddl print("version:", __version__) print("ddl ok; lines:", len(ddl.splitlines())) print("exports ok:", [c.__name__ for c in (BatchWriter, MemorySink, TdengineSink)]) + +# 子任务 #29：依据点位字典自动生成 TDengine 完整 schema（超级表 + 子表） +_td_rows = [ + {"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "unit": "℃", "dataType": "float"}, + {"device_id": "CLF-02", "point_id": "CLF-02.RUN", "unit": "%", "dataType": "bool"}, +] +_td_ddls = generate_schema(TemplateNaming(template="ti-cl4"), specs_from_point_dict_rows(_td_rows), retention_days=90) +assert _td_ddls[0].startswith("CREATE STABLE IF NOT EXISTS ti_cl4_points") +assert len(_td_ddls) == 3 # 1 超级表 + 2 子表 +assert all("USING ti_cl4_points" in d for d in _td_ddls[1:]) +print("tdengine schema ok; statements:", len(_td_ddls)) diff --git a/core/data-bus/tdengine_schema.py b/core/data-bus/tdengine_schema.py index 354803d..be0c315 100644 --- a/core/data-bus/tdengine_schema.py +++ b/core/data-bus/tdengine_schema.py @@ -120,6 +120,49 @@ def generate_subtable_ddls( return out +# --------------------------------------------------------------------------- +# 依据点位字典自动生成完整 schema（子任务 #29 交付入口） +# --------------------------------------------------------------------------- + +def specs_from_point_dict_rows(rows: List[dict]) -> List[PointSpec]: + """把点位字典 CSV 行（edge-gateway 点位字典表头）转换为 PointSpec 列表。 + + 与 `core/edge-gateway/point_dict/loader.py` 的 CSV 表头对齐： + ``device_id / point_id / unit / dataType``（兼容小写 ``data_type``）。 + 其余列（name / sampleRate / qualityCode / opcNode / protocol）与 + schema 生成无关，宽松忽略；非法行按 ``from_dict`` 宽松兜底， + 不做静默丢弃（保持与点位数一致，便于外层校验定位）。 + + Args: + rows: 点位字典行（每行一个测点，dict 键为 CSV 表头）。 + + Returns: + 与输入顺序一致的 PointSpec 列表。 + """ + return [PointSpec.from_dict(raw) for raw in rows] + + +def generate_schema( + naming: TemplateNaming, + points: List[PointSpec], + retention_days: Optional[int] = None, +) -> List[str]: + """依据点位字典一次性生成完整 TDengine schema（超级表 + 每测点子表）。 + + Args: + naming: 模板命名器（决定超级表/子表名）。 + points: 点位字典（测点维度集合，可由 specs_from_point_dict_rows 得到）。 + retention_days: 数据保留天数（KEEP，可选）。 + + Returns: + DDL 语句列表：第 0 条为超级表 CREATE STABLE，其余为子表 + CREATE TABLE ... USING ... TAGS(...)，可直接按序执行。 + """ + return [generate_supertable_ddl(naming, retention_days=retention_days)] + generate_subtable_ddls( + naming, points + ) + + def build_batch_insert( naming: TemplateNaming, rows: List[dict], diff --git a/core/data-bus/tests/test_tdengine_schema.py b/core/data-bus/tests/test_tdengine_schema.py index 3ebd65e..9a08634 100644 --- a/core/data-bus/tests/test_tdengine_schema.py +++ b/core/data-bus/tests/test_tdengine_schema.py @@ -10,8 +10,10 @@ import _bootstrap # noqa: F401 from data_bus.tdengine_schema import ( PointSpec, build_batch_insert, + generate_schema, generate_subtable_ddls, generate_supertable_ddl, + specs_from_point_dict_rows, ) from data_bus.templating import TemplateNaming @@ -22,6 +24,19 @@ def sample_point(device_id="CLF-01", point_id="CLF-01.TEMP", unit="℃", data_ty return PointSpec(device_id=device_id, point_id=point_id, unit=unit, data_type=data_type) +# 与 edge-gateway 点位字典 CSV 表头对齐的样例行 +SAMPLE_POINT_DICT_ROWS = [ + {"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "name": "炉温", "unit": "℃", + "dataType": "float", "sampleRate": "1000", "qualityCode": "true", "opcNode": "ns=2;s=CLF.Temp"}, + {"device_id": "CLF-01", "point_id": "CLF-01.PRES", "name": "炉压", "unit": "kPa", + "dataType": "float", "sampleRate": "1000", "qualityCode": "true"}, + {"device_id": "CLF-02", "point_id": "CLF-02.RUN", "name": "运行状态", "unit": "%", + "dataType": "bool", "sampleRate": "1000"}, + {"device_id": "S7-01", "point_id": "S7-01.PUMP_A", "name": "泵A频率", "unit": "Hz", + "dataType": "float", "sampleRate": "1000", "protocol": "s7"}, +] + + class SupertableDDLTest(unittest.TestCase): def test_ddl_shape(self): ddl = generate_supertable_ddl(NAMING) @@ -72,5 +87,49 @@ class BatchInsertTest(unittest.TestCase): self.assertEqual(build_batch_insert(NAMING, []), []) +class PointDictSchemaTest(unittest.TestCase): + """依据点位字典自动生成完整 schema（子任务 #29 交付口径）。""" + + def test_specs_from_point_dict_rows(self): + specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS) + self.assertEqual(len(specs), 4) + self.assertEqual(specs[0].device_id, "CLF-01") + self.assertEqual(specs[0].point_id, "CLF-01.TEMP") + self.assertEqual(specs[0].unit, "℃") + self.assertEqual(specs[0].data_type, "float") + # 兼容小写 data_type 键 + specs2 = specs_from_point_dict_rows( + [{"device_id": "CLF-01", "point_id": "P1", "unit": "℃", "data_type": "int"}] + ) + self.assertEqual(specs2[0].data_type, "int") + # 缺失列宽松兜底 + specs3 = specs_from_point_dict_rows([{"device_id": "D1", "point_id": "P2"}]) + self.assertEqual(specs3[0].unit, "") + self.assertEqual(specs3[0].data_type, "float") + + def test_generate_schema_full(self): + specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS) + ddls = generate_schema(NAMING, specs) + # 第 0 条是超级表，其余是子表；4 个测点 → 1 + 4 条 + self.assertEqual(len(ddls), 5) + self.assertTrue(ddls[0].startswith("CREATE STABLE IF NOT EXISTS ti_cl4_points")) + self.assertTrue(all(d.startswith("CREATE TABLE IF NOT EXISTS ti_cl4_pt_") for d in ddls[1:])) + # 子表均 USING 同一超级表，且含点位字典标签 + self.assertIn("USING ti_cl4_points", ddls[1]) + self.assertIn("'CLF-01'", ddls[1]) + # bool 点位标签保留 data_type + self.assertTrue(any("'bool'" in d for d in ddls)) + + def test_generate_schema_retention(self): + specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS) + ddls = generate_schema(NAMING, specs, retention_days=365) + self.assertIn("KEEP(365)", ddls[0]) + + def test_generate_schema_empty_points(self): + ddls = generate_schema(NAMING, []) + self.assertEqual(len(ddls), 1) # 只有超级表，无子表 + self.assertIn("CREATE STABLE", ddls[0]) + + if __name__ == "__main__": unittest.main()