feat: add read-only training dataset loader

2026-09-11 10:08:15 +08:00
parent 7bdd705cef
commit 272542bad5
4 changed files with 302 additions and 0 deletions
@@ -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 交换、设备采集、建模管线和推理服务。
@@ -0,0 +1 @@
+"""Read-only training data interfaces for the external material data platform."""
@@ -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))
@@ -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)
				`@@ -0,0 +1 @@`
				`"""Read-only training data interfaces for the external material data platform."""`