"""`tias-dataset` コマンド: manifest の生成とデータセットの集計（SPEC_01「CLI」）.
サブコマンド:
- `build-manifest --config PATH [--overwrite]`: `images/` を走査して manifest を書き出す
- `survey --config PATH [--out-root runs]`: データセットを集計し，`runs/<実験ID>/` に集計値だけを
書き出す（`dataset_summary.json`・`config.yaml`・`log.txt`・`figures/*.png`）．最後に実験 ID を
標準出力へ出す（`tias-remote pull <実験ID>` に使うため）
実データを読むためサーバで実行する．ログ・出力に患者 ID・ファイル名・画像単位の値を出さない．
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import re
import sys
import time
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime
from pathlib import Path
from typing import Any
import yaml
from pydantic import ValidationError
from tias_judge_model.data.manifest import build_manifest_records, read_manifest, write_manifest
from tias_judge_model.data.survey import (
LAB_CHANNELS,
collect_image_stats,
lab_values_by_group,
summarize_dataset,
)
from tias_judge_model.data.survey_figures import (
LAB_BOXPLOT_FILENAME,
LAB_MEANS_AB_FILENAME,
plot_group_means_ab,
plot_lab_by_group,
)
from tias_judge_model.schema import GIT_COMMIT_ENV_VAR, ExperimentConfig, load_config
logger = logging.getLogger(__name__)
SURVEY_RUN_NAME = "survey"
NO_GIT = "nogit"
SUMMARY_FILENAME = "dataset_summary.json"
CONFIG_FILENAME = "config.yaml"
LOG_FILENAME = "log.txt"
FIGURES_DIRNAME = "figures"
# 実験 ID に埋め込むため，ファイル名・RUN_ID として安全な文字だけを許す（git の短縮ハッシュと
# `-dirty` を想定）．
_GIT_COMMIT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
_LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
_EPILOG = """\
使用例（サーバ上で実行する．手元では configs/synthetic.yaml で合成データを使う）:
uv run tias-dataset build-manifest --config configs/newton.yaml
uv run tias-dataset build-manifest --config configs/newton.yaml --overwrite
uv run tias-dataset survey --config configs/newton.yaml
uv run tias-remote run --name survey -- tias-dataset survey --config configs/newton.yaml
環境変数:
TIAS_DATA_ROOT data.root を上書き（tias-remote run が各サーバの data_dir を渡す）
TIAS_GIT_COMMIT 実験 ID に入れる git の短縮ハッシュ（tias-remote run が渡す．無ければ nogit）
"""
def resolve_git_commit(environ: Mapping[str, str]) -> str:
"""実験 ID に入れる git の短縮ハッシュを環境変数から決める.
Args:
environ: 参照する環境変数．
Returns:
`TIAS_GIT_COMMIT` の値．未設定・空なら `nogit`．
Raises:
ValueError: 値が実験 ID に使えない文字を含む場合．
"""
value = environ.get(GIT_COMMIT_ENV_VAR, "").strip()
if not value:
return NO_GIT
if not _GIT_COMMIT_RE.match(value):
raise ValueError(
f"環境変数 {GIT_COMMIT_ENV_VAR} の値が不正である: {value!r}"
"（英数字で始まり，英数字・_・- のみ，64 文字以内）"
)
return value
def make_run_id(now: datetime, name: str, git_commit: str) -> str:
"""実験 ID `YYYYmmdd-HHMMSS_<名前>_<git短縮ハッシュ>` を作る（SPEC_01「実験の出力」）.
Args:
now: 実行日時．
name: 名前（データセット集計なら `survey`）．
git_commit: git の短縮ハッシュ（または `nogit`）．
Returns:
実験 ID．
"""
return f"{now:%Y%m%d-%H%M%S}_{name}_{git_commit}"
def resolved_config_document(
config: ExperimentConfig, *, run_id: str, git_commit: str, command: str
) -> dict[str, Any]:
"""`config.yaml` に書く内容（解決済み設定 ＋ 実験 ID ＋ git ハッシュ）を作る.
Args:
config: 解決済みの設定（環境変数の上書き適用後）．
run_id: 実験 ID．
git_commit: git の短縮ハッシュ（または `nogit`）．
command: 実行したサブコマンド名．
Returns:
YAML に書き出せる dict．
"""
return {
"run": {"id": run_id, "command": command, "git_commit": git_commit},
"config": config.model_dump(mode="json"),
}
# ---------------------------------------------------------------------------
# build-manifest
# ---------------------------------------------------------------------------
def _cmd_build_manifest(args: argparse.Namespace) -> int:
config = load_config(args.config)
manifest_path = config.data.manifest_path
if manifest_path.exists() and not args.overwrite:
logger.error("manifest が既にある: %s（作り直すには --overwrite を付ける）", manifest_path)
return 1
records = build_manifest_records(config.data.images_dir)
write_manifest(records, manifest_path)
logger.info("manifest を書き出した: %s（%d 件）", manifest_path, len(records))
return 0
# ---------------------------------------------------------------------------
# survey
# ---------------------------------------------------------------------------
def _cmd_survey(args: argparse.Namespace) -> int:
config = load_config(args.config)
git_commit = resolve_git_commit(os.environ)
run_id = make_run_id(datetime.now(), SURVEY_RUN_NAME, git_commit)
run_dir = args.out_root / run_id
if run_dir.exists():
logger.error("実験ディレクトリが既にある: %s（時間をおいて再実行する）", run_dir)
return 1
run_dir.mkdir(parents=True)
file_handler = logging.FileHandler(run_dir / LOG_FILENAME, encoding="utf-8")
file_handler.setFormatter(logging.Formatter(_LOG_FORMAT))
root = logging.getLogger()
root.addHandler(file_handler)
try:
_run_survey(config, run_dir, run_id=run_id, git_commit=git_commit)
except (ValueError, OSError) as exc:
# log.txt にも失敗の理由を残すため，ハンドラを外す前にここで記録する．
logger.error("%s", exc)
return 1
finally:
root.removeHandler(file_handler)
file_handler.close()
sys.stdout.write(f"{run_id}\n")
sys.stdout.flush()
return 0
def _run_survey(config: ExperimentConfig, run_dir: Path, *, run_id: str, git_commit: str) -> None:
started = time.monotonic()
logger.info("データセット集計を開始する（実験 ID %s）", run_id)
_write_yaml(
run_dir / CONFIG_FILENAME,
resolved_config_document(config, run_id=run_id, git_commit=git_commit, command="survey"),
)
records = read_manifest(config.data.manifest_path)
stats = collect_image_stats(records, config.data.images_dir)
summary = summarize_dataset(
records,
stats,
seed=config.split.seed,
val_ratio=config.split.val_ratio,
test_ratio=config.split.test_ratio,
n_folds=config.split.n_folds,
configured_severities=config.data.severities,
)
summary = {"run_id": run_id, "git_commit": git_commit, **summary}
_write_json(run_dir / SUMMARY_FILENAME, summary)
figures_dir = run_dir / FIGURES_DIRNAME
plot_lab_by_group(lab_values_by_group(records, stats), figures_dir / LAB_BOXPLOT_FILENAME)
groups = summary["lab"]["groups"]
plot_group_means_ab(
{g: _channel_values(s, "mean") for g, s in groups.items()},
{g: _channel_values(s, "std") for g, s in groups.items()},
figures_dir / LAB_MEANS_AB_FILENAME,
)
loading = summary["loading"]
counts = summary["counts"]["total"]
logger.info(
"集計した: 画像 %d 件・患者 %d 人（読めない %d 件・RGB 以外 %d 件・アルファ付き %d 件）",
counts["images"],
counts["patients"],
loading["unreadable"],
loading["non_rgb_mode"],
loading["with_alpha"],
)
for item in summary["delta_e"]["mild_comparison"]:
logger.info(
"%s: ΔE（対 %s）= %s・ΔE（対 %s）= %s → %s に近い",
item["group"],
item["normal_group"],
item["delta_e_to_normal"],
item["strong_group"],
item["delta_e_to_strong"],
item["closer_to"],
)
logger.info("出力先: %s（所要 %.1f 秒）", run_dir, time.monotonic() - started)
def _channel_values(stats: Mapping[str, Any], key: str) -> tuple[float, ...] | None:
values = [stats[c][key] for c in LAB_CHANNELS]
if key == "mean" and any(v is None for v in values):
return None
return tuple(values)
def _write_json(path: Path, data: Mapping[str, Any]) -> None:
with path.open("w", encoding="utf-8", newline="\n") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
def _write_yaml(path: Path, data: Mapping[str, Any]) -> None:
with path.open("w", encoding="utf-8", newline="\n") as f:
yaml.safe_dump(dict(data), f, allow_unicode=True, sort_keys=False)
# ---------------------------------------------------------------------------
# エントリポイント
# ---------------------------------------------------------------------------
def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="tias-dataset",
description="manifest の生成とデータセットの集計（集計値だけを書き出す）",
epilog=_EPILOG,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
sub = parser.add_subparsers(dest="subcommand", required=True)
p_build = sub.add_parser(
"build-manifest",
help="images/ を走査して manifest を書き出す",
description=(
"<data.root>/images/<label>_<severity>/*.png を走査して manifest を書き出す．"
"既存の manifest があれば --overwrite が無い限り止まる．"
),
)
p_build.add_argument("--config", type=Path, required=True, help="実験設定 YAML．")
p_build.add_argument("--overwrite", action="store_true", help="既存の manifest を上書きする．")
p_survey = sub.add_parser(
"survey",
help="データセットを集計して runs/<実験ID>/ に集計値だけを書き出す",
description=(
"manifest と画像を読み，件数・読み込み・サイズ・舌領域の割合・Lab 色・グループ間の"
"色差・分割の件数を集計する．画像単位の値・患者 ID・ファイル名は出力しない．"
"最後に実験 ID を標準出力へ出す．"
),
)
p_survey.add_argument("--config", type=Path, required=True, help="実験設定 YAML．")
p_survey.add_argument(
"--out-root", type=Path, default=Path("runs"), help="実験ディレクトリの親（既定: runs）．"
)
return parser.parse_args(argv)
_HANDLERS: dict[str, Callable[[argparse.Namespace], int]] = {
"build-manifest": _cmd_build_manifest,
"survey": _cmd_survey,
}
def _configure_logging() -> None:
# ログは標準エラーへ（標準出力は実験 ID の受け渡しに使うため汚さない）．
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter(_LOG_FORMAT))
logging.basicConfig(level=logging.INFO, handlers=[handler])
def main(argv: Sequence[str] | None = None) -> int:
"""tias-dataset コマンドのメインエントリポイント.
Args:
argv: 引数リスト．None のとき sys.argv[1:] を使用．
Returns:
終了コード（成功 0・入力の問題 1）．
"""
args = _parse_args(argv)
_configure_logging()
try:
return _HANDLERS[args.subcommand](args)
except ValidationError as exc:
logger.error("設定が不正である:\n%s", exc)
return 1
except (ValueError, OSError) as exc:
logger.error("%s", exc)
return 1
if __name__ == "__main__":
sys.exit(main())
