"""CSV 読込（変換行列・学習データ）． SPEC_02「データファイル形式」に対応する読込と形式検証をまとめる． 旧実装（`org/src` の `Readfile()` / `CsvRead()`）と同じ実効データになるよう， 読み込む行数・列数と dtype（float32）を踏襲する． """ from __future__ import annotations import csv import logging from dataclasses import dataclass from pathlib import Path import numpy as np from numpy.typing import NDArray logger = logging.getLogger(__name__) # --- 形式に関する定数（SPEC_02「データファイル形式」）--- #: 変換行列 CSV の行数（多項式 17 項．SPEC_02 (1)） TRANSFORM_COEFFICIENT_COUNT = 17 #: 変換行列 CSV の列数（a[i], b[i], c[i] の 3 列） TRANSFORM_COLUMN_COUNT = 3 #: data.csv の学習サンプル数（旧実装 EXP_DATA_ROW = 84） SAMPLE_COUNT = 84 #: SVM 入力の次元（L*a*b*．旧実装 EXP_INPUT_DIM = 3） LAB_DIMENSION = 3 #: 舌色の分類数（紫・暗赤・紅・淡紅・淡白） COLOR_COUNT = 5 #: 医師の人数（旧実装 EXP_OUTPUT_DIM 190 = 38 名 × 5 色） DOCTOR_COUNT = 38 #: 医師評価列の数（旧実装 EXP_OUTPUT_DIM = 190） EVALUATION_COLUMN_COUNT = DOCTOR_COUNT * COLOR_COUNT #: data.csv から読み込む列数（旧実装は 3 + 190 = 193 列だけを読み，以降の列は無視する） TRAINING_COLUMN_COUNT = LAB_DIMENSION + EVALUATION_COLUMN_COUNT #: data.csv のヘッダ行数 HEADER_ROW_COUNT = 1 # 読込に使う文字コードの候補（同梱データは UTF-8．旧実装由来のファイルは Shift-JIS） _ENCODINGS = ("utf-8", "cp932") class CsvIoError(Exception): """CSV 読込に関するエラーの基底（`core` の専用例外）．""" class DataFileNotFoundError(CsvIoError): """CSV ファイルが存在しない，または読み取れない．""" class DataFormatError(CsvIoError): """CSV の形式が仕様と異なる（行数・列数不一致，数値でないセル等）．""" @dataclass(frozen=True) class TrainingData: """`data.csv` の学習データ（SPEC_02 (3) の SVM 入力）． Attributes: lab: 形状 (84, 3)・dtype float32．各行が 1 サンプルの L*a*b*． evaluations: 形状 (84, 190)・dtype int32．医師 38 名 × 5 色の 0/1 評価． 列 `doctor_index * 5 + color_index` が医師ごとの one-hot に対応する． """ lab: NDArray[np.float32] evaluations: NDArray[np.int32] @property def sample_count(self) -> int: """学習サンプル数（= 84）．""" return int(self.lab.shape[0]) @property def output_dimension(self) -> int: """SVM の出力次元（= 医師数 × 5 色 = 190）．""" return int(self.evaluations.shape[1]) @property def doctor_count(self) -> int: """医師の人数（= 38）．""" return self.output_dimension // COLOR_COUNT def color_columns(self, color_index: int) -> NDArray[np.int32]: """指定した色（0〜4）に対応する全医師分の評価列を返す（形状 (84, 医師数)）．""" if not 0 <= color_index < COLOR_COUNT: raise ValueError(f"color_index は 0〜{COLOR_COUNT - 1} の範囲: {color_index}") return self.evaluations[:, color_index::COLOR_COUNT] def load_transform_matrix(path: Path | str) -> NDArray[np.float64]: """変換行列 CSV（17 行 × 3 列・ヘッダなし）を読み込む． Args: path: 変換行列 CSV のパス（ユーザーが診断時に選択するファイル）． Returns: 形状 (17, 3)・dtype float64 の配列．列 0 が a[i]（X 用）， 列 1 が b[i]（Y 用），列 2 が c[i]（Z 用）の係数（SPEC_02 (1)）． Raises: DataFileNotFoundError: ファイルが存在しない・読み取れない場合． DataFormatError: 行数・列数が仕様と異なる，または数値でないセルがある場合． """ csv_path = Path(path) rows = _read_rows(csv_path) if len(rows) != TRANSFORM_COEFFICIENT_COUNT: raise DataFormatError( f"変換行列 CSV は {TRANSFORM_COEFFICIENT_COUNT} 行必要ですが " f"{len(rows)} 行です: {csv_path}" ) matrix = np.empty((TRANSFORM_COEFFICIENT_COUNT, TRANSFORM_COLUMN_COUNT), dtype=np.float64) for row_index, row in enumerate(rows): if len(row) != TRANSFORM_COLUMN_COUNT: raise DataFormatError( f"変換行列 CSV の {row_index + 1} 行目は " f"{TRANSFORM_COLUMN_COUNT} 列必要ですが {len(row)} 列です: {csv_path}" ) for column_index, cell in enumerate(row): matrix[row_index, column_index] = _parse_float( cell, csv_path, row_index + 1, column_index + 1 ) logger.info("変換行列を読み込みました: %s", csv_path) return matrix def load_training_data(path: Path | str) -> TrainingData: """学習データ `data.csv`（ヘッダ 1 行 + 84 サンプル）を読み込む． 列 1〜3 が L*a*b*，列 4〜193 が医師 38 名 × 5 色の 0/1 評価である． 旧実装（`CsvRead(..., 84, 3 + 190, 2, 1)`）と同じく先頭 193 列のみを読み， 194 列目以降（全サンプルで全 0 の未使用ブロック）は無視する． Args: path: `data.csv` のパス（同梱データ．`core` はパスを固定しない）． Returns: Lab と医師評価を保持する `TrainingData`． Raises: DataFileNotFoundError: ファイルが存在しない・読み取れない場合． DataFormatError: 行数・列数が仕様と異なる，数値でないセル， または評価列に 0/1 以外の値がある場合． """ csv_path = Path(path) rows = _read_rows(csv_path) expected_row_count = HEADER_ROW_COUNT + SAMPLE_COUNT if len(rows) != expected_row_count: raise DataFormatError( f"data.csv はヘッダ 1 行 + {SAMPLE_COUNT} サンプルの " f"{expected_row_count} 行必要ですが {len(rows)} 行です: {csv_path}" ) lab = np.empty((SAMPLE_COUNT, LAB_DIMENSION), dtype=np.float32) evaluations = np.empty((SAMPLE_COUNT, EVALUATION_COLUMN_COUNT), dtype=np.int32) for sample_index, row in enumerate(rows[HEADER_ROW_COUNT:]): line_number = sample_index + HEADER_ROW_COUNT + 1 if len(row) < TRAINING_COLUMN_COUNT: raise DataFormatError( f"data.csv の {line_number} 行目は {TRAINING_COLUMN_COUNT} 列以上必要ですが " f"{len(row)} 列です: {csv_path}" ) for lab_index, cell in enumerate(row[:LAB_DIMENSION]): lab[sample_index, lab_index] = _parse_float(cell, csv_path, line_number, lab_index + 1) for offset, cell in enumerate(row[LAB_DIMENSION:TRAINING_COLUMN_COUNT]): evaluations[sample_index, offset] = _parse_binary( cell, csv_path, line_number, LAB_DIMENSION + offset + 1 ) logger.info( "学習データを読み込みました: %s（%d サンプル × %d 列）", csv_path, SAMPLE_COUNT, TRAINING_COLUMN_COUNT, ) return TrainingData(lab=lab, evaluations=evaluations) def _read_rows(csv_path: Path) -> list[list[str]]: """CSV を読み，空行を除いた行のリスト（各行はセルのリスト）を返す．""" text = _read_text(csv_path) rows = list(csv.reader(text.splitlines())) return [row for row in rows if any(cell.strip() for cell in row)] def _read_text(csv_path: Path) -> str: """CSV をテキストとして読む（UTF-8 優先，失敗時は Shift-JIS を試す）．""" try: data = csv_path.read_bytes() except OSError as error: raise DataFileNotFoundError(f"CSV ファイルを読み込めません: {csv_path}") from error for encoding in _ENCODINGS: try: return data.decode(encoding) except UnicodeDecodeError: continue raise DataFormatError(f"CSV の文字コードを判別できません（UTF-8・Shift-JIS 以外）: {csv_path}") def _cell_position(csv_path: Path, line_number: int, column_number: int) -> str: """エラーメッセージ用にセル位置を表す文字列を組み立てる．""" return f"{csv_path} の {line_number} 行 {column_number} 列" def _parse_float(cell: str, csv_path: Path, line_number: int, column_number: int) -> float: """セルを実数として解釈する．数値でなければ `DataFormatError`．""" try: return float(cell.strip()) except ValueError as error: raise DataFormatError( f"{_cell_position(csv_path, line_number, column_number)}が数値ではありません: {cell!r}" ) from error def _parse_binary(cell: str, csv_path: Path, line_number: int, column_number: int) -> int: """医師評価のセル（0 または 1）を解釈する．それ以外は `DataFormatError`．""" value = _parse_float(cell, csv_path, line_number, column_number) if value not in (0.0, 1.0): raise DataFormatError( f"{_cell_position(csv_path, line_number, column_number)}は " f"0 か 1 である必要があります: {cell!r}" ) return int(value)