From 272542bad51ea3c8141344a5563df2390c66b028 Mon Sep 17 00:00:00 2001 From: obroccolio Date: Fri, 11 Sep 2026 10:08:15 +0800 Subject: [PATCH] feat: add read-only training dataset loader --- README.md | 2 + src/lmpm/training/__init__.py | 1 + src/lmpm/training/dataset.py | 147 +++++++++++++++++++++++++++++++ tests/test_training_dataset.py | 152 +++++++++++++++++++++++++++++++++ 4 files changed, 302 insertions(+) create mode 100644 src/lmpm/training/__init__.py create mode 100644 src/lmpm/training/dataset.py create mode 100644 tests/test_training_dataset.py diff --git a/README.md b/README.md index 1063ed2..5a77257 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ python -m http.server 47319 --directory tech-architecture src/lmpm/ domain/ 材料物性、实验记录、加工参数与质量指标模型 data/ SQLite 初始化与实验记录读写 + training/ 从外部数据平台只读加载训练数据 tests/ 领域校验与数据库往返测试 tech-architecture/ 技术架构报告与本地 Mermaid 资源 ``` @@ -34,6 +35,7 @@ tech-architecture/ 技术架构报告与本地 Mermaid 资源 - `MaterialProperty`：反射率、吸收率、熔点、热导率、密度、粗糙度等物性字段。 - `ExperimentRecord`：加工参数、质量指标与记录时间。 - SQLite 骨架：`materials` 与 `experiments` 两张核心表，支持外键约束与实验记录往返。 +- 训练数据接口：从外部数据平台的 `material_records` 表只读加载记录，并分离元信息、特征和目标。 - 工程配置：`pyproject.toml` 统一依赖、pytest 与 Ruff 配置。 后续将按架构报告逐步补齐 DoE 生成、CSV 交换、设备采集、建模管线和推理服务。 diff --git a/src/lmpm/training/__init__.py b/src/lmpm/training/__init__.py new file mode 100644 index 0000000..ba22430 --- /dev/null +++ b/src/lmpm/training/__init__.py @@ -0,0 +1 @@ +"""Read-only training data interfaces for the external material data platform.""" diff --git a/src/lmpm/training/dataset.py b/src/lmpm/training/dataset.py new file mode 100644 index 0000000..f99e9c4 --- /dev/null +++ b/src/lmpm/training/dataset.py @@ -0,0 +1,147 @@ +"""Read-only access to material records from the external data platform.""" + +import sqlite3 +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +METADATA_FIELDS = ( + "experiment_id", + "test_date", + "material_category", + "material_name", +) +NUMERIC_FEATURE_FIELDS = ( + "apparent_density", + "uv_absorption_355nm", + "material_thickness", + "moisture_content", + "thermal_conductivity", + "initial_decomposition_temp", + "melting_vaporization_temp", + "specific_heat_capacity", + "carbon_residue_rate", + "surface_roughness", + "hardness", + "actual_output_power", + "display_current", + "scanning_speed", + "pulse_frequency", + "pulse_width", + "defocus_amount", + "scan_line_spacing", +) +CATEGORICAL_FEATURE_FIELDS = ( + "filling_method", + "processing_size", +) +BOOLEAN_TARGET_FIELDS = ( + "is_cut_through", + "is_fire_smolder", +) +NUMERIC_TARGET_FIELDS = ( + "carbonized_edge_width", + "etching_depth", + "pattern_clarity_score", + "presentation_balance_score", +) +TARGET_FIELDS = ( + "is_cut_through", + "carbonized_edge_width", + "etching_depth", + "is_fire_smolder", + "pattern_clarity_score", + "presentation_balance_score", +) + + +class TrainingDataError(RuntimeError): + """Raised when the platform schema cannot provide a training dataset.""" + + +class TrainingRecord(BaseModel): + """A single material experiment split into metadata, X, and y.""" + + model_config = ConfigDict(frozen=True) + + metadata: dict[str, str] + features: dict[str, float | str | None] + targets: dict[str, float | bool | None] + + +class TrainingDataset(BaseModel): + """An ordered, experiment-id-addressable view of training records.""" + + records: tuple[TrainingRecord, ...] = Field(default_factory=tuple) + + @property + def size(self) -> int: + return len(self.records) + + @property + def experiment_ids(self) -> list[str]: + return [record.metadata["experiment_id"] for record in self.records] + + @property + def feature_names(self) -> list[str]: + return list(NUMERIC_FEATURE_FIELDS + CATEGORICAL_FEATURE_FIELDS) + + @property + def target_names(self) -> list[str]: + return list(TARGET_FIELDS) + + +def _row_value(row: sqlite3.Row, field: str) -> Any: + try: + return row[field] + except IndexError as error: + raise TrainingDataError( + f"material_records is missing required field: {field}" + ) from error + + +def _numeric_value(row: sqlite3.Row, field: str) -> float | None: + value = _row_value(row, field) + return None if value is None else float(value) + + +def _record_from_row(row: sqlite3.Row) -> TrainingRecord: + metadata = {field: str(_row_value(row, field)) for field in METADATA_FIELDS} + features: dict[str, float | str | None] = { + field: _numeric_value(row, field) for field in NUMERIC_FEATURE_FIELDS + } + features.update( + {field: str(_row_value(row, field)) for field in CATEGORICAL_FEATURE_FIELDS} + ) + targets: dict[str, float | bool | None] = { + field: bool(_row_value(row, field)) for field in BOOLEAN_TARGET_FIELDS + } + targets.update( + {field: _numeric_value(row, field) for field in NUMERIC_TARGET_FIELDS} + ) + return TrainingRecord(metadata=metadata, features=features, targets=targets) + + +def load_sqlite_training_dataset(database_path: Path | str) -> TrainingDataset: + """Load material records without allowing writes to the source database.""" + path = Path(database_path) + if not path.is_file(): + raise FileNotFoundError(f"training database does not exist: {path}") + + uri = f"{path.resolve().as_uri()}?mode=ro" + try: + connection = sqlite3.connect(uri, uri=True) + connection.row_factory = sqlite3.Row + rows = connection.execute( + "SELECT * FROM material_records ORDER BY id" + ).fetchall() + except sqlite3.Error as error: + raise TrainingDataError( + "could not read material_records from the training database" + ) from error + finally: + if "connection" in locals(): + connection.close() + + return TrainingDataset(records=tuple(_record_from_row(row) for row in rows)) diff --git a/tests/test_training_dataset.py b/tests/test_training_dataset.py new file mode 100644 index 0000000..4a4dc62 --- /dev/null +++ b/tests/test_training_dataset.py @@ -0,0 +1,152 @@ +import sqlite3 + +import pytest + +from lmpm.training.dataset import ( + TrainingDataError, + load_sqlite_training_dataset, +) + +MATERIAL_SCHEMA = """ +CREATE TABLE material_records ( + id INTEGER PRIMARY KEY, + entry_batch_id INTEGER NOT NULL, + experiment_id VARCHAR(100) NOT NULL UNIQUE, + test_date DATE NOT NULL, + material_category VARCHAR(100) NOT NULL, + material_name VARCHAR(200) NOT NULL, + apparent_density FLOAT, + uv_absorption_355nm FLOAT, + material_thickness FLOAT NOT NULL, + moisture_content FLOAT, + thermal_conductivity FLOAT, + initial_decomposition_temp FLOAT, + melting_vaporization_temp FLOAT, + specific_heat_capacity FLOAT, + carbon_residue_rate FLOAT, + surface_roughness FLOAT, + hardness FLOAT, + actual_output_power FLOAT NOT NULL, + display_current FLOAT NOT NULL, + scanning_speed FLOAT NOT NULL, + pulse_frequency FLOAT NOT NULL, + pulse_width FLOAT NOT NULL, + defocus_amount FLOAT NOT NULL, + scan_line_spacing FLOAT NOT NULL, + filling_method VARCHAR(100) NOT NULL, + processing_size VARCHAR(100) NOT NULL, + is_cut_through BOOLEAN NOT NULL, + carbonized_edge_width FLOAT NOT NULL, + etching_depth FLOAT NOT NULL, + is_fire_smolder BOOLEAN NOT NULL, + pattern_clarity_score FLOAT NOT NULL, + presentation_balance_score FLOAT NOT NULL, + finished_image_filename VARCHAR(255) NOT NULL, + remarks TEXT NOT NULL, + created_at DATETIME NOT NULL +); +""" + + +def make_record(number: int) -> tuple: + return ( + number, + 1, + f"LAS-2026-{number:04d}", + "2026-09-11", + "polymer", + f"Acrylic {number}", + 1.18 if number == 1 else None, + 0.92, + 3.0, + None, + 0.19, + None, + None, + 1.5, + None, + 1.2, + None, + 8.5 + number, + 2.0, + 80.0, + 20.0, + 5.0, + 0.0, + 0.1, + "双向填充", + "20x20", + number == 1, + 0.2, + 30.0, + False, + 9.0, + 8.0, + f"image-{number}.jpg", + "stable", + "2026-09-11T08:00:00+00:00", + ) + + +def create_database(tmp_path, *, with_table: bool = True) -> object: + database_path = tmp_path / f"platform-{int(with_table)}.db" + connection = sqlite3.connect(database_path) + if with_table: + connection.execute(MATERIAL_SCHEMA) + connection.executemany( + """ + INSERT INTO material_records VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ) + """, + [make_record(1), make_record(2)], + ) + connection.commit() + connection.close() + return database_path + + +def test_sqlite_dataset_maps_features_and_targets(tmp_path): + dataset = load_sqlite_training_dataset(create_database(tmp_path)) + + assert dataset.experiment_ids == ["LAS-2026-0001", "LAS-2026-0002"] + assert dataset.feature_names[:5] == [ + "apparent_density", + "uv_absorption_355nm", + "material_thickness", + "moisture_content", + "thermal_conductivity", + ] + assert dataset.feature_names[-1] == "processing_size" + assert dataset.target_names == [ + "is_cut_through", + "carbonized_edge_width", + "etching_depth", + "is_fire_smolder", + "pattern_clarity_score", + "presentation_balance_score", + ] + assert dataset.records[0].metadata == { + "experiment_id": "LAS-2026-0001", + "test_date": "2026-09-11", + "material_category": "polymer", + "material_name": "Acrylic 1", + } + assert dataset.records[0].features["actual_output_power"] == pytest.approx(9.5) + assert dataset.records[0].targets["is_cut_through"] is True + assert dataset.records[1].targets["is_cut_through"] is False + + +def test_missing_database_raises_expected_error(tmp_path): + missing_path = tmp_path / "missing.db" + + with pytest.raises(FileNotFoundError): + load_sqlite_training_dataset(missing_path) + + +def test_database_without_material_table_raises_expected_error(tmp_path): + database_path = create_database(tmp_path, with_table=False) + + with pytest.raises(TrainingDataError, match="material_records"): + load_sqlite_training_dataset(database_path)