diff --git a/CLAUDE.md b/CLAUDE.md index 7794bb8..c0f363a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,28 +15,35 @@ - **Data Storage**: CSV (Local file system) ## Architecture & File Structure -- `app.py`: アプリケーションのメインロジック。UI描画、データ処理を全て担当。 -- `data/`: 画像データ格納ディレクトリ。 - - 構造: `data/{case_id}/*.jpg` (各症例フォルダ内に連番JPEG画像) -- `annotations.csv`: アノテーション結果の保存先。 +- `app.py`: アプリケーションのメインロジック。 +- **User Selected Directory**: ユーザーが起動時に指定するルートディレクトリ。 + - 構造: `{root_dir}/{case_id}/images/*.png` (指定フォルダ内のサブフォルダを症例として認識) +- `annotations_{annotater}.csv`: アノテーション結果の保存先。 + - Location: **ユーザーが選択したルートディレクトリの直下**に保存する。 - Encoding: `utf-8-sig` (Excelでの日本語文字化け防止のため必須) ## Code Style & Conventions - **Naming**: 変数名・関数名は英語（スネークケース）、UIの表示ラベルは**日本語**とする。 - **Type Hinting**: 可能な限りPythonの型ヒントを使用する。 -- **Error Handling**: 画像読み込みエラーやファイルアクセス権限エラーを適切にハンドリングする。 +- **Error Handling**: ディレクトリが存在しない場合や、画像が含まれていない場合の警告を表示する。 -## Annotation Specifications (Important) -アプリ内で実装すべきアノテーション項目と選択肢は以下の通り固定する。 +## Application Flow & UI Specifications +アプリの動作フローは以下の順序とする。 + +### Phase 1: Session Setup (Sidebar or Top) +アプリ起動時、以下の情報をユーザーに入力させる。情報が確定するまで画像表示画面には進まない。 +1. **Annotator Name**: 作業者の名前（必須入力）。 +2. **Data Directory Path**: 画像データが格納されているローカルフォルダのパス（テキスト入力またはダイアログ）。 + +### Phase 2: Annotation (Main Area) +Phase 1の設定完了後、フォルダ内の症例をロードして表示する。 1. **合併症予測 (Prediction)** - - UI: Radio Button - - Options: "なし", "あり" + - UI: Radio Button ("なし", "あり") 2. **確信度 (Confidence)** - - UI: Slider - - Range: 0 - 100 (%) + - UI: Slider (0 - 100 %) 3. **判断根拠 (Reasons)** - - UI: Checkboxes (Multiple selection) + - UI: Checkboxes - Options: - "石灰化プラークが多い" - "石灰化プラークが少ない" @@ -45,8 +52,8 @@ - "その他（記述）" 4. **自由記述 (Comments)** - UI: Text Area - - Input: 自由入力 ## Data Management Rules -- CSV保存時は「追記モード」を使用し、既存データを消さないこと。 +- CSVファイル名は `annotations_{annotater}.csv` とし、**選択されたデータディレクトリ直下**に保存または読み込みを行う。 +- 保存時は「追記モード」を使用。 - 保存カラム順序: `timestamp`, `case_id`, `prediction`, `confidence`, `reasons`, `comment`, `annotator` \ No newline at end of file diff --git a/README.md b/README.md index e69de29..717a863 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,302 @@ +# IVUS合併症アノテーションツール + +循環器内科医がIVUS（血管内超音波）画像を確認し、No reflow / Slow flow の合併症リスクを予測・アノテーションするためのStreamlitベースのWebアプリケーション。 + +## 目次 + +- [概要](#概要) +- [機能](#機能) +- [必要環境](#必要環境) +- [インストール](#インストール) +- [使い方](#使い方) +- [データ形式](#データ形式) +- [トラブルシューティング](#トラブルシューティング) + +--- + +## 概要 + +このツールは、医師がIVUS画像（1症例あたり約20フレーム）を確認し、以下の情報をアノテーションできます： + +- **合併症予測**：No reflow/Slow flowの発生有無 +- **確信度**：予測の確信度（0-100%） +- **判断根拠**：石灰化プラーク、減衰プラークなどの所見 +- **自由記述**：追加コメント + +アノテーション結果はCSVファイルとして保存され、Excelで開いて分析できます。 + +--- + +## 機能 + +### 実装済み機能 + +- **GUIベースの設定**：起動時にフォルダ/ファイルをWindows標準ダイアログで選択 +- **画像ビューアー**：スライダーでフレームを前後にスクロール +- **症例ナビゲーション**：前へ/次へボタン、ドロップダウン選択 +- **進捗管理**：完了済み症例に自動でチェックマーク表示 +- **自動保存後進行**：保存後、未完了の次の症例へ自動移動 +- **Excel互換CSV出力**：日本語が正しく表示される utf-8-sig 形式 +- **Ground Truth記録**：正解ラベルをCSVに記録（UI非表示） + +--- + +## 必要環境 + +- **OS**: Windows 10/11（推奨） +- **Python**: 3.10以上 +- **必要なPythonパッケージ**: requirements.txt 参照 + +--- + +## インストール + +### 1. リポジトリのクローン + +```bash +git clone <このリポジトリのURL> +cd ivus-complication-annotation-tool +``` + +### 2. 依存パッケージのインストール + +```bash +pip install -r requirements.txt +``` + +**インストールされるパッケージ**: +- streamlit: Webアプリケーションフレームワーク +- pandas: CSV/Excelデータ処理 +- Pillow: 画像読み込み・表示 +- openpyxl: Excelファイル読み込み（pandasが使用） +- watchdog: 開発時の自動リロード + +--- + +## 使い方 + +### 起動方法 + +```bash +streamlit run app.py +``` + +起動すると、自動的にブラウザが開き、設定画面が表示されます。 + +### ステップ1: 設定画面 + +アプリケーション起動時、以下の3つの情報を入力します： + +#### 1. アノテーター名（必須） +- あなたの名前を入力してください +- 例: 田中 + +#### 2. データディレクトリパス（必須） +- 症例フォルダ（CHIBAMI_xxx_pre）が含まれるディレクトリを選択 +- 📁 参照 ボタンでWindows標準のフォルダ選択ダイアログで選択してください + +#### 3. Excelラベルファイルパス（必須） +- 正解ラベルが記載されたExcelファイル（CHIBAMI_case_list_xxx.xlsx）を選択 +- 📁 参照 ボタンでWindows標準のファイル選択ダイアログで選択してください + +**注意**: 全ての項目を正しく入力しないと、アノテーション開始ボタンが有効になりません。 + +### ステップ2: アノテーション画面 + +設定完了後、メインのアノテーション画面が表示されます。 + +#### 画像ビューアー（中央） +- **フレーム番号スライダー**: マウスでスライダーを動かして、画像を前後にスクロール +- 1症例あたり約20フレームの画像を確認できます + +#### 症例ナビゲーション（サイドバー上部） +- **症例を選択**: ドロップダウンから任意の症例を選択 + - 完了済み症例には ✓ マークが表示されます +- **← 前へ / 次へ →**: ボタンで前後の症例に移動 + +#### アノテーション入力（サイドバー下部） + +**Q1: 合併症予測 (No reflow/Slow flow)** +- ラジオボタンで なし または あり を選択 + +**Q2: 確信度 (%)** +- スライダーで0〜100%の間で確信度を設定（デフォルト: 50%） + +**Q3: 判断根拠**（複数選択可） +- 以下の選択肢から、該当するものを全てチェック: + - 石灰化プラークが多い + - 石灰化プラークが少ない + - 減衰プラークが多い + - 減衰プラークが少ない + - その他（記述） + +**Q4: 自由記述欄** +- 追加のコメントや所見を自由に記述 + +#### 保存 + +- **💾 保存して次へ ボタン**をクリック +- **バリデーション**: Q3（判断根拠）が最低1つ選択されていない場合、エラーメッセージが表示されます +- 保存成功後、自動的に次の未完了症例へ移動します + +--- + +## データ形式 + +### 入力データ構造 + +``` +データディレクトリ/ +├── CHIBAMI_134_pre/ +│ ├── images/ +│ │ ├── frame_134_4440.png +│ │ ├── frame_134_4500.png +│ │ └── ... (約20枚) +│ └── ... (その他のマスクデータなど) +├── CHIBAMI_135_pre/ +│ └── images/ +│ └── ... +└── ... +``` + +**重要**: +- フォルダ名: CHIBAMI_{症例番号}_pre +- 画像形式: PNG +- 画像ファイル名: frame_{症例番号}_{フレーム番号}.png + +### 出力CSV形式 + +保存されるCSVファイル名: annotations_{アノテーター名}.csv + +**保存場所**: データディレクトリ直下 + +**CSVカラム**: + +| カラム名 | 説明 | 例 | +|---------|------|-----| +| timestamp | 保存日時 | 2025-12-08 19:30:45 | +| case_id | 症例番号 | 134 または 134.1 | +| prediction | 合併症予測 | あり または なし | +| confidence | 確信度(%) | 75 | +| reasons | 判断根拠（セミコロン区切り） | 石灰化プラークが多い; 減衰プラークが多い | +| comment | 自由記述 | 明確な所見あり | +| annotator | アノテーター名 | 田中 | +| ground_truth | 正解ラベル | True / False / (空=不明) | + +**文字エンコーディング**: utf-8-sig（Excelで開いても文字化けしません） + +--- + +## トラブルシューティング + +### Q1: アプリが起動しない + +**A1**: 必要なパッケージがインストールされているか確認してください。 + +```bash +pip install -r requirements.txt +``` + +### Q2: 📁 参照 ボタンを押してもダイアログが開かない + +**A2**: WSL（Linux環境）で実行している可能性があります。このツールはWindows環境での使用を前提としています。 + +- **対処法**: Windows上で直接Pythonを実行してください。 +- WSL環境では、パスを手入力（コピー&ペースト）してください。 + +### Q3: フォルダ/ファイルが見つからないとエラーが出る + +**A3**: パスが正しいか確認してください。 + +- **Windowsパスの例**: C:\Users\YourName\Documents\data +- **WSLパスの例**: /mnt/d/Research/data/...（Windows環境では使えません） + +### Q4: CSVをExcelで開くと文字化けする + +**A4**: 通常は発生しませんが、もし発生した場合: + +1. Excelでデータ→テキストファイルから を選択 +2. エンコーディングをUTF-8に指定して開く + +### Q5: 画像が表示されない + +**A5**: 以下を確認してください: + +1. データディレクトリ内に CHIBAMI_{番号}_pre/images/ フォルダが存在するか +2. images/ フォルダ内に frame_*.png ファイルが存在するか +3. PNGファイルが破損していないか + +### Q6: 進捗が保存されない + +**A6**: CSVファイルの保存先を確認してください。 + +- 保存先: {データディレクトリ}/annotations_{アノテーター名}.csv +- 例: C:\Users\YourName\Documents\data\annotations_tanaka.csv + +ファイルの書き込み権限があるか確認してください。 + +--- + +## プロジェクト構造 + +``` +ivus-complication-annotation-tool/ +├── app.py # メインアプリケーション +├── requirements.txt # 依存パッケージリスト +├── README.md # このファイル +├── CLAUDE.md # プロジェクトガイドライン +└── utils/ # ユーティリティモジュール + ├── __init__.py # パッケージ初期化 + ├── excel.py # Excelラベル読み込み + ├── data_loader.py # 症例データ読み込み + └── annotation_saver.py # アノテーション保存 +``` + +--- + +## 開発者向け情報 + +### コーディング規約 + +- **変数名・関数名**: 英語（スネークケース） +- **UIラベル**: 日本語 +- **型ヒント**: 可能な限り使用 +- 詳細は CLAUDE.md を参照 + +### 起動コマンド（開発用） + +```bash +# 自動リロード有効 +streamlit run app.py + +# ポート指定 +streamlit run app.py --server.port 8501 + +# ブラウザを自動で開かない +streamlit run app.py --server.headless true +``` + +--- + +## ライセンス + +研究用途のみ。商用利用不可。 + +--- + +## サポート + +不具合や質問がある場合は、プロジェクト管理者にご連絡ください。 + +--- + +## 更新履歴 + +### v1.0.0 (2025-12-08) +- 初回リリース +- GUIベースの設定画面 +- Windows標準フォルダ選択ダイアログ +- 画像ビューアー +- アノテーション入力フォーム +- CSV保存機能 +- 進捗管理機能 diff --git a/app.py b/app.py new file mode 100644 index 0000000..c803e6f --- /dev/null +++ b/app.py @@ -0,0 +1,551 @@ +""" +IVUS Complication Annotation Tool +Streamlit-based application for annotating IVUS images with complication predictions. +""" + +import streamlit as st +from PIL import Image +import os +from typing import Optional, Union + +# Tkinter import for folder/file dialog (Windows native dialog) +try: + import tkinter as tk + from tkinter import filedialog + TKINTER_AVAILABLE = True +except ImportError: + TKINTER_AVAILABLE = False + +from utils import ( + get_all_case_ids, + get_case_images, + load_ground_truth_labels, + validate_case_directory, + save_annotation, + get_annotated_cases +) + +# UI Constants +PREDICTION_OPTIONS = ["なし", "あり"] +REASONS_OPTIONS = [ + "石灰化プラークが多い", + "石灰化プラークが少ない", + "減衰プラークが多い", + "減衰プラークが少ない", + "その他（記述）" +] + +MIN_CONFIDENCE = 0 +MAX_CONFIDENCE = 100 +DEFAULT_CONFIDENCE = 50 + +# Page configuration +st.set_page_config( + page_title="IVUS合併症アノテーションツール", + page_icon="🫀", + layout="wide" +) + + +def open_folder_dialog(initial_dir: str = ".") -> str: + """ + Windows標準のフォルダ選択ダイアログを開き、選択されたパスを返す。 + + Args: + initial_dir: 初期表示ディレクトリ + + Returns: + 選択されたフォルダパス（キャンセル時は空文字列） + + Note: + WSL環境では動作しない可能性があります。Windows環境での使用を想定。 + """ + if not TKINTER_AVAILABLE: + return "" + + try: + root = tk.Tk() + root.withdraw() # メインウィンドウを隠す + root.wm_attributes('-topmost', 1) # ダイアログを最前面に表示 + folder_path = filedialog.askdirectory(initialdir=initial_dir) + root.destroy() + return folder_path if folder_path else "" + except Exception as e: + print(f"フォルダ選択ダイアログエラー: {e}") + return "" + + +def open_file_dialog(initial_dir: str = ".", file_types=None) -> str: + """ + Windows標準のファイル選択ダイアログを開き、選択されたパスを返す。 + + Args: + initial_dir: 初期表示ディレクトリ + file_types: ファイル種類のフィルタ + + Returns: + 選択されたファイルパス（キャンセル時は空文字列） + + Note: + WSL環境では動作しない可能性があります。Windows環境での使用を想定。 + """ + if not TKINTER_AVAILABLE: + return "" + + if file_types is None: + file_types = [("Excel files", "*.xlsx;*.xls"), ("All files", "*.*")] + + try: + root = tk.Tk() + root.withdraw() + root.wm_attributes('-topmost', 1) + file_path = filedialog.askopenfilename( + initialdir=initial_dir, + filetypes=file_types + ) + root.destroy() + return file_path if file_path else "" + except Exception as e: + print(f"ファイル選択ダイアログエラー: {e}") + return "" + + +def render_configuration_form(): + """ + 起動時の設定フォームを表示。 + アノテーター名とデータパスを入力してから進む。 + フォルダ/ファイル選択ボタンも提供。 + """ + st.title("IVUS合併症アノテーションツール - 設定") + st.markdown("---") + + # Initialize session state for paths if not exists (空の初期値) + if 'temp_data_root' not in st.session_state: + st.session_state.temp_data_root = "" + if 'temp_excel_path' not in st.session_state: + st.session_state.temp_excel_path = "" + + st.subheader("設定") + + # Annotator name input + annotator = st.text_input( + "アノテーター名 *", + value="", + help="あなたの名前を入力してください（必須）", + placeholder="例: 田中" + ) + + st.markdown("---") + + # Data Directory Path with folder picker + st.markdown("**データディレクトリパス** *") + st.caption("症例フォルダ（CHIBAMI_xxx_pre）が含まれるディレクトリを選択してください") + + col1, col2 = st.columns([4, 1]) + with col1: + data_root = st.text_input( + "データディレクトリパス", + value=st.session_state.temp_data_root, + help="「📁 参照」ボタンでフォルダを選択してください", + placeholder="「📁 参照」ボタンでフォルダを選択", + label_visibility="collapsed" + ) + # Update session state when user types + if data_root != st.session_state.temp_data_root: + st.session_state.temp_data_root = data_root + + with col2: + if st.button("📁 参照", key="btn_browse_data", use_container_width=True): + if TKINTER_AVAILABLE: + # Use current path as initial dir if exists, otherwise use home directory + initial = st.session_state.temp_data_root if st.session_state.temp_data_root else os.path.expanduser("~") + selected_path = open_folder_dialog(initial_dir=initial) + if selected_path: + st.session_state.temp_data_root = selected_path + st.rerun() + else: + st.warning("フォルダ選択機能はWindows環境でのみ利用可能です") + + # Show current path validation + if data_root and os.path.exists(data_root): + st.success(f"✓ フォルダが見つかりました: {data_root}") + elif data_root: + st.error(f"✗ フォルダが見つかりません: {data_root}") + + st.markdown("---") + + # Excel Label Path with file picker + st.markdown("**Excelラベルファイルパス** *") + st.caption("正解ラベルが記載されたExcelファイル（CHIBAMI_case_list_xxx.xlsx）を選択してください") + + col1, col2 = st.columns([4, 1]) + with col1: + excel_path = st.text_input( + "Excelラベルファイルパス", + value=st.session_state.temp_excel_path, + help="「📁 参照」ボタンでExcelファイルを選択してください", + placeholder="「📁 参照」ボタンでファイルを選択", + label_visibility="collapsed" + ) + # Update session state when user types + if excel_path != st.session_state.temp_excel_path: + st.session_state.temp_excel_path = excel_path + + with col2: + if st.button("📁 参照", key="btn_browse_excel", use_container_width=True): + if TKINTER_AVAILABLE: + # Use directory of current path as initial dir if exists + initial = os.path.dirname(st.session_state.temp_excel_path) if st.session_state.temp_excel_path else os.path.expanduser("~") + selected_path = open_file_dialog(initial_dir=initial) + if selected_path: + st.session_state.temp_excel_path = selected_path + st.rerun() + else: + st.warning("ファイル選択機能はWindows環境でのみ利用可能です") + + # Show current file validation + if excel_path and os.path.exists(excel_path): + st.success(f"✓ ファイルが見つかりました: {excel_path}") + elif excel_path: + st.error(f"✗ ファイルが見つかりません: {excel_path}") + + st.markdown("---") + + # Submit button + if st.button("アノテーション開始", type="primary", use_container_width=True): + # Validation + errors = [] + if not annotator or annotator.strip() == "": + errors.append("❌ アノテーター名は必須です") + if not data_root or data_root.strip() == "": + errors.append("❌ データディレクトリパスは必須です - 「📁 参照」ボタンでフォルダを選択してください") + elif not os.path.exists(data_root): + errors.append(f"❌ データディレクトリが見つかりません: {data_root}") + if not excel_path or excel_path.strip() == "": + errors.append("❌ Excelラベルファイルパスは必須です - 「📁 参照」ボタンでファイルを選択してください") + elif not os.path.exists(excel_path): + errors.append(f"❌ Excelファイルが見つかりません: {excel_path}") + + if errors: + for error in errors: + st.error(error) + else: + # Save configuration to session state + st.session_state.config_complete = True + st.session_state.annotator = annotator.strip() + st.session_state.data_root = data_root.strip() + st.session_state.excel_path = excel_path.strip() + st.session_state.csv_path = os.path.join( + data_root.strip(), + f"annotations_{annotator.strip()}.csv" + ) + st.rerun() + + +def initialize_session_state(): + """設定完了後、セッション状態を初期化。""" + if 'config_complete' not in st.session_state: + st.session_state.config_complete = False + + if not st.session_state.config_complete: + return + + # Load case IDs + if 'case_ids' not in st.session_state: + st.session_state.case_ids = get_all_case_ids(st.session_state.data_root) + + # Initialize current case index + if 'current_case_idx' not in st.session_state: + st.session_state.current_case_idx = 0 + + # Load ground truth labels (not displayed, only for CSV saving) + if 'ground_truth_labels' not in st.session_state: + st.session_state.ground_truth_labels = load_ground_truth_labels( + st.session_state.excel_path + ) + + # Load annotated cases + if 'annotated_cases' not in st.session_state: + st.session_state.annotated_cases = get_annotated_cases( + st.session_state.csv_path, + st.session_state.annotator + ) + + +def get_current_case_id() -> Optional[Union[int, float]]: + """現在の症例IDを取得。""" + if (st.session_state.current_case_idx >= 0 and + st.session_state.current_case_idx < len(st.session_state.case_ids)): + return st.session_state.case_ids[st.session_state.current_case_idx] + return None + + +def render_header(): + """進捗情報を含むヘッダーを表示。""" + st.title("IVUS合併症アノテーションツール") + + total_cases = len(st.session_state.case_ids) + annotated_count = len(st.session_state.annotated_cases) + + col1, col2, col3, col4 = st.columns(4) + with col1: + st.metric("アノテーター", st.session_state.annotator) + with col2: + st.metric("全症例数", total_cases) + with col3: + st.metric("完了数", annotated_count) + with col4: + progress = annotated_count / total_cases if total_cases > 0 else 0 + st.metric("進捗", f"{progress:.1%}") + + +def render_case_selector(): + """サイドバーに症例ナビゲーションを表示。""" + st.sidebar.header("症例ナビゲーション") + + current_case_id = get_current_case_id() + + # Case selector dropdown + case_display = [ + f"症例 {cid}" + (" ✓" if cid in st.session_state.annotated_cases else "") + for cid in st.session_state.case_ids + ] + + selected_idx = st.sidebar.selectbox( + "症例を選択", + range(len(st.session_state.case_ids)), + index=st.session_state.current_case_idx, + format_func=lambda i: case_display[i] + ) + + if selected_idx != st.session_state.current_case_idx: + st.session_state.current_case_idx = selected_idx + st.rerun() + + # Previous/Next buttons + col1, col2 = st.sidebar.columns(2) + with col1: + if st.button("← 前へ", disabled=st.session_state.current_case_idx == 0): + st.session_state.current_case_idx -= 1 + st.rerun() + with col2: + if st.button( + "次へ →", + disabled=st.session_state.current_case_idx >= len(st.session_state.case_ids) - 1 + ): + st.session_state.current_case_idx += 1 + st.rerun() + + # Display current case info + if current_case_id: + st.sidebar.info(f"**現在の症例:** {current_case_id}") + + +def render_image_viewer(case_id: Union[int, float]): + """ + スライダー付き画像ビューアーを表示。 + + Args: + case_id: 現在の症例ID + """ + images = get_case_images(st.session_state.data_root, case_id) + + if len(images) == 0: + st.error(f"症例 {case_id} の画像が見つかりません") + return + + # Image slider + frame_idx = st.slider( + "フレーム番号", + min_value=0, + max_value=len(images) - 1, + value=0, + key=f"frame_slider_{case_id}" + ) + + # Display image + try: + img = Image.open(images[frame_idx]) + st.image( + img, + caption=f"症例 {case_id} - フレーム {frame_idx + 1}/{len(images)}", + use_container_width=True + ) + except Exception as e: + st.error(f"画像読み込みエラー: {e}") + + +def render_annotation_form(case_id: Union[int, float]) -> dict: + """ + アノテーション入力フォームを表示。 + + Args: + case_id: 現在の症例ID + + Returns: + アノテーションデータの辞書 + """ + st.sidebar.markdown("---") + st.sidebar.header("アノテーション") + + # Display annotator name (read-only) + st.sidebar.text(f"アノテーター: {st.session_state.annotator}") + st.sidebar.markdown("---") + + # Q1: Prediction + prediction = st.sidebar.radio( + "Q1: 合併症予測 (No reflow/Slow flow)", + PREDICTION_OPTIONS, + key=f"prediction_{case_id}" + ) + + # Q2: Confidence + confidence = st.sidebar.slider( + "Q2: 確信度 (%)", + min_value=MIN_CONFIDENCE, + max_value=MAX_CONFIDENCE, + value=DEFAULT_CONFIDENCE, + key=f"confidence_{case_id}" + ) + + # Q3: Reasons + st.sidebar.markdown("**Q3: 判断根拠** (複数選択可)") + reasons = [] + for reason in REASONS_OPTIONS: + if st.sidebar.checkbox(reason, key=f"reason_{case_id}_{reason}"): + reasons.append(reason) + + # Q4: Free text comment + comment = st.sidebar.text_area( + "Q4: 自由記述欄", + value="", + key=f"comment_{case_id}", + height=100 + ) + + return { + 'prediction': prediction, + 'confidence': confidence, + 'reasons': reasons, + 'comment': comment + } + + +def validate_annotation(annotation_data: dict) -> tuple[bool, str]: + """ + 保存前にアノテーションデータを検証。 + + Args: + annotation_data: アノテーションフィールドの辞書 + + Returns: + (有効かどうか, エラーメッセージ) のタプル + """ + if len(annotation_data['reasons']) == 0: + return False, "判断根拠を最低1つ選択してください (Q3)" + + return True, "" + + +def save_and_advance(case_id: Union[int, float], annotation_data: dict): + """ + アノテーションを保存して次の症例へ進む。 + + Args: + case_id: 現在の症例ID + annotation_data: アノテーションフィールドの辞書 + """ + is_valid, error_msg = validate_annotation(annotation_data) + + if not is_valid: + st.sidebar.error(error_msg) + return + + try: + # Get ground truth label for this case (may be None) + ground_truth = st.session_state.ground_truth_labels.get(case_id, None) + + save_annotation( + csv_path=st.session_state.csv_path, + case_id=case_id, + prediction=annotation_data['prediction'], + confidence=annotation_data['confidence'], + reasons=annotation_data['reasons'], + comment=annotation_data['comment'], + annotator=st.session_state.annotator, + ground_truth=ground_truth + ) + + # Update annotated cases + st.session_state.annotated_cases.add(case_id) + + # Show success message + st.sidebar.success(f"✓ 症例 {case_id} を保存しました！") + + # Auto-advance to next unannotated case + next_idx = st.session_state.current_case_idx + 1 + while next_idx < len(st.session_state.case_ids): + next_case_id = st.session_state.case_ids[next_idx] + if next_case_id not in st.session_state.annotated_cases: + st.session_state.current_case_idx = next_idx + st.rerun() + return + next_idx += 1 + + # If no more unannotated cases, stay on current or show completion + if next_idx < len(st.session_state.case_ids): + st.session_state.current_case_idx = next_idx + st.rerun() + else: + st.sidebar.info("全症例のアノテーションが完了しました！ 🎉") + + except Exception as e: + st.sidebar.error(f"保存エラー: {e}") + + +def main(): + """メインアプリケーションエントリーポイント。""" + initialize_session_state() + + # Show configuration form if not completed + if not st.session_state.config_complete: + render_configuration_form() + return + + # Check if there are any cases + if len(st.session_state.case_ids) == 0: + st.error(f"{st.session_state.data_root} に症例が見つかりません") + st.info("アプリケーションを再起動して、データディレクトリパスを確認してください。") + return + + render_header() + render_case_selector() + + current_case_id = get_current_case_id() + + if current_case_id is None: + st.error("利用可能な症例がありません") + return + + # Validate case directory + is_valid, error_msg = validate_case_directory( + st.session_state.data_root, + current_case_id + ) + if not is_valid: + st.error(error_msg) + return + + # Main layout: image viewer in main area + render_image_viewer(current_case_id) + + # Annotation form in sidebar + annotation_data = render_annotation_form(current_case_id) + + # Save button + if st.sidebar.button("💾 保存して次へ", type="primary", use_container_width=True): + save_and_advance(current_case_id, annotation_data) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4a15dc1 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +streamlit>=1.28.0 +pandas>=2.0.0 +Pillow>=10.0.0 +openpyxl>=3.1.0 +watchdog>=3.0.0 diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..e39e90c --- /dev/null +++ b/utils/__init__.py @@ -0,0 +1,27 @@ +""" +Utility modules for IVUS Complication Annotation Tool. +""" + +from .excel import Excel +from .data_loader import ( + get_all_case_ids, + get_case_images, + load_ground_truth_labels, + validate_case_directory +) +from .annotation_saver import ( + initialize_csv, + save_annotation, + get_annotated_cases +) + +__all__ = [ + 'Excel', + 'get_all_case_ids', + 'get_case_images', + 'load_ground_truth_labels', + 'validate_case_directory', + 'initialize_csv', + 'save_annotation', + 'get_annotated_cases' +] diff --git a/utils/annotation_saver.py b/utils/annotation_saver.py new file mode 100644 index 0000000..6206659 --- /dev/null +++ b/utils/annotation_saver.py @@ -0,0 +1,148 @@ +""" +Annotation saving utilities for IVUS annotation tool. +Handles CSV file creation, saving annotations, and tracking progress. +""" + +import os +import csv +import pandas as pd +from datetime import datetime +from typing import List, Set, Union, Optional + + +def initialize_csv(csv_path: str) -> None: + """ + Create annotation CSV with headers if it doesn't exist. + + Args: + csv_path: Path to CSV file + + CSV Columns: + - timestamp: When annotation was saved (YYYY-MM-DD HH:MM:SS) + - case_id: Case number (int or float) + - prediction: Complication prediction ("あり" or "なし") + - confidence: Confidence level (0-100) + - reasons: Selected reasons (semicolon-separated) + - comment: Free text comment + - annotator: Name of annotator + - ground_truth: Ground truth label from Excel (True/False/None) + """ + if not os.path.exists(csv_path): + with open(csv_path, 'w', encoding='utf-8-sig', newline='') as f: + writer = csv.writer(f) + writer.writerow([ + 'timestamp', + 'case_id', + 'prediction', + 'confidence', + 'reasons', + 'comment', + 'annotator', + 'ground_truth' + ]) + + +def save_annotation( + csv_path: str, + case_id: Union[int, float], + prediction: str, + confidence: int, + reasons: List[str], + comment: str, + annotator: str, + ground_truth: Optional[bool] +) -> None: + """ + Save annotation to CSV file in append mode. + + Args: + csv_path: Path to CSV file + case_id: Case number + prediction: "あり" or "なし" + confidence: 0-100 + reasons: List of selected reasons + comment: Free text comment + annotator: Annotator name + ground_truth: Ground truth label (True/False/None if not available) + + Example: + >>> save_annotation( + ... "/path/to/annotations_tanaka.csv", + ... 134, + ... "あり", + ... 75, + ... ["石灰化プラークが多い", "減衰プラークが多い"], + ... "明確な所見あり", + ... "tanaka", + ... True + ... ) + """ + initialize_csv(csv_path) + + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + reasons_str = "; ".join(reasons) # Join multiple reasons with semicolon + + # Convert ground_truth to string representation + if ground_truth is None: + gt_str = "" + elif ground_truth: + gt_str = "True" + else: + gt_str = "False" + + with open(csv_path, 'a', encoding='utf-8-sig', newline='') as f: + writer = csv.writer(f) + writer.writerow([ + timestamp, + case_id, + prediction, + confidence, + reasons_str, + comment, + annotator, + gt_str + ]) + + +def get_annotated_cases(csv_path: str, annotator: str = None) -> Set[Union[int, float]]: + """ + Get set of case IDs that have been annotated. + + Args: + csv_path: Path to CSV file + annotator: Optional filter by specific annotator + + Returns: + Set of case IDs that have annotations + + Example: + >>> get_annotated_cases("/path/to/annotations_tanaka.csv", "tanaka") + {134, 134.1, 135, 136, ...} + """ + if not os.path.exists(csv_path): + return set() + + try: + df = pd.read_csv(csv_path, encoding='utf-8-sig') + + if annotator: + df = df[df['annotator'] == annotator] + + # Convert case_ids to appropriate type (int or float) + case_ids = set() + for case_id in df['case_id'].unique(): + try: + # Try to convert to float first + case_id_float = float(case_id) + # If it's a whole number, store as int, otherwise as float + if case_id_float.is_integer(): + case_ids.add(int(case_id_float)) + else: + case_ids.add(case_id_float) + except (ValueError, TypeError): + continue + + return case_ids + except Exception as e: + print(f"Warning: Could not read annotations from {csv_path}: {e}") + return set() diff --git a/utils/data_loader.py b/utils/data_loader.py new file mode 100644 index 0000000..7f82089 --- /dev/null +++ b/utils/data_loader.py @@ -0,0 +1,188 @@ +""" +Data loading utilities for IVUS annotation tool. +Handles case discovery, image loading, and ground truth labels. +""" + +import os +import glob +from typing import List, Dict, Tuple, Union +from pathlib import Path +from .excel import Excel + + +def get_all_case_ids(data_root: str) -> List[Union[int, float]]: + """ + Scan data directory and extract case numbers from folder names. + Handles both integer case IDs (134) and decimal case IDs (134.1, 134.2). + + Args: + data_root: Root directory containing case folders + + Returns: + List of case numbers (int or float) sorted in ascending order + + Example: + >>> get_all_case_ids("/mnt/d/Research/data/2025_10_30_updated") + [134, 134.1, 134.2, 135, 135.1, ..., 795] + """ + if not os.path.exists(data_root): + return [] + + try: + case_dirs = [ + d for d in os.listdir(data_root) + if os.path.isdir(os.path.join(data_root, d)) + and d.startswith("CHIBAMI_") + and d.endswith("_pre") + ] + + case_ids = [] + for case_dir in case_dirs: + try: + # Extract number from "CHIBAMI_134_pre" or "CHIBAMI_134.1_pre" + # Split by underscore: ["CHIBAMI", "134", "pre"] or ["CHIBAMI", "134.1", "pre"] + parts = case_dir.split("_") + if len(parts) >= 3: + # Parse as float to preserve decimals, then convert to int if whole number + case_num_float = float(parts[1]) + # If it's a whole number, store as int, otherwise as float + if case_num_float.is_integer(): + case_ids.append(int(case_num_float)) + else: + case_ids.append(case_num_float) + except (IndexError, ValueError): + # Skip folders that don't match expected pattern + continue + + return sorted(case_ids) + except Exception as e: + print(f"Error scanning directory {data_root}: {e}") + return [] + + +def get_case_images(data_root: str, case_id: Union[int, float]) -> List[str]: + """ + Get all image paths for a given case, sorted by frame index. + + Args: + data_root: Root directory containing case folders + case_id: Case number (int or float, e.g., 134 or 134.1) + + Returns: + List of absolute paths to PNG files, sorted by frame index + + Example: + >>> get_case_images("/mnt/d/Research/data/2025_10_30_updated", 134) + ['/mnt/d/.../CHIBAMI_134_pre/images/frame_134_4440.png', ...] + >>> get_case_images("/mnt/d/Research/data/2025_10_30_updated", 134.1) + ['/mnt/d/.../CHIBAMI_134.1_pre/images/frame_134_4440.png', ...] + """ + # Construct folder name based on case_id type + if isinstance(case_id, float) and not case_id.is_integer(): + # Decimal case: CHIBAMI_134.1_pre + case_dir = os.path.join(data_root, f"CHIBAMI_{case_id}_pre") + else: + # Integer case: CHIBAMI_134_pre + case_dir = os.path.join(data_root, f"CHIBAMI_{int(case_id)}_pre") + + if not os.path.exists(case_dir): + return [] + + images_dir = os.path.join(case_dir, "images") + + if not os.path.exists(images_dir): + return [] + + # Get all PNG files matching the frame pattern + image_files = glob.glob(os.path.join(images_dir, "frame_*.png")) + + # Sort by frame index (extract from filename) + def extract_frame_idx(filepath): + """Extract frame index from filename like 'frame_134_4440.png'.""" + filename = os.path.basename(filepath) + try: + # Remove extension: "frame_134_4440.png" -> "frame_134_4440" + rest_name = filename.replace(".png", "") + # Split: ["frame", "134", "4440"] + parts = rest_name.split("_") + # Return the last part as integer + return int(parts[-1]) + except (IndexError, ValueError): + return 0 + + return sorted(image_files, key=extract_frame_idx) + + +def load_ground_truth_labels(excel_path: str) -> Dict[Union[int, float], bool]: + """ + Load ground truth labels from Excel file. + + Args: + excel_path: Path to Excel file containing ground truth labels + + Returns: + Dictionary mapping case_id to has_complication boolean + - True: Complication present (No/Slow flow) + - False: No complication + + Example: + >>> load_ground_truth_labels("/path/to/labels.xlsx") + {134: True, 134.1: False, 135: False, ...} + """ + try: + excel = Excel(excel_path) + return excel.extract_data() + except FileNotFoundError: + print(f"Warning: Excel file not found: {excel_path}") + return {} + except Exception as e: + print(f"Warning: Could not load Excel file: {e}") + return {} + + +def validate_case_directory(data_root: str, case_id: Union[int, float]) -> Tuple[bool, str]: + """ + Validate that a case has accessible images. + + Args: + data_root: Root directory containing case folders + case_id: Case number to validate (int or float) + + Returns: + Tuple of (is_valid, error_message) + - If valid: (True, "") + - If invalid: (False, "Error description") + + Example: + >>> validate_case_directory("/mnt/d/Research/data/2025_10_30_updated", 134) + (True, "") + >>> validate_case_directory("/mnt/d/Research/data/2025_10_30_updated", 999) + (False, "Case directory not found: ...") + """ + # Construct folder name based on case_id type + if isinstance(case_id, float) and not case_id.is_integer(): + case_dir = os.path.join(data_root, f"CHIBAMI_{case_id}_pre") + case_name = f"CHIBAMI_{case_id}_pre" + else: + case_dir = os.path.join(data_root, f"CHIBAMI_{int(case_id)}_pre") + case_name = f"CHIBAMI_{int(case_id)}_pre" + + if not os.path.exists(case_dir): + return False, f"Case directory not found: {case_name}" + + if not os.path.isdir(case_dir): + return False, f"Path is not a directory: {case_dir}" + + images_dir = os.path.join(case_dir, "images") + + if not os.path.exists(images_dir): + return False, f"Images directory not found: {images_dir}" + + if not os.path.isdir(images_dir): + return False, f"Images path is not a directory: {images_dir}" + + images = get_case_images(data_root, case_id) + if len(images) == 0: + return False, f"No PNG images found in: {images_dir}" + + return True, "" diff --git a/utils/excel.py b/utils/excel.py new file mode 100644 index 0000000..7cc1945 --- /dev/null +++ b/utils/excel.py @@ -0,0 +1,209 @@ +import os +import re +import pandas as pd +from typing import Dict, Optional, Tuple, Union + + +class Excel: + def __init__(self, exel_path): + self.path = exel_path + + def extract_data(self): + df = pd.read_excel(self.path) + + #　key : Case Numner, value : 合併症の有無（True : 合併症あり，False : 合併症なし） + data = {} + + for _, row in df.iterrows(): + # if row["除外理由"]: + # continue + if pd.notna(row["No"]) and pd.notna(row["No/Slow flow during procedure"]) and pd.notna(row["IVUS解析除外"]): + case_num = int(row["No"]) + + if row["No/Slow flow during procedure"] == 1: + data[int(case_num)] = True + elif row["No/Slow flow during procedure"] == 0: + data[int(case_num)] = False + + # for k, v in data.items(): + # print(f"Case Number : {k}, Complication : {v}") + + return data + + def rename(self, root_dir): + data_dict = self.extract_data() + No_data = [] + + for case in os.listdir(root_dir): + case_dir = os.path.join(root_dir, case) + if os.path.isdir(case_dir): + images_dir = os.path.join(case_dir, "images") # IVUS元画像用のディレクトリ + + for image in os.listdir(images_dir): + image_path = os.path.join(images_dir, image) + rest_path, ext = os.path.splitext(image) + parts = rest_path.split("_") + case_num = int(float(parts[1])) + + # ファイル名に "_T" または "_F" が含まれている画像ファイルはリネーム処理の対象外 + if image.endswith("_T.png") or image.endswith("_F.png"): + continue + + if case_num not in data_dict: + No_data.append(case_num) + continue + + """ + 合併症の有無によってファイル名を変更 + 合併症の場合 : frame_49_4020_T.png, frame_49_4020_F.png + """ + if data_dict[case_num]: + new_image_name = f"{parts[0]}_{case_num}_{parts[-1]}_T{ext}" + else: + new_image_name = f"{parts[0]}_{case_num}_{parts[-1]}_F{ext}" + + new_image_path = os.path.join(images_dir, new_image_name) + os.rename(image_path, new_image_path) + + No_data = sorted(set(No_data)) + for case_num in No_data: + print(f"Case Number : {case_num} is not found in the excel file.") + print(f"Total {len(No_data)} cases are not found in the excel file.") + + def improve_name(self, root_dir): + for case in os.listdir(root_dir): + case_dir = os.path.join(root_dir, case) + if os.path.isdir(case_dir): + images_dir = os.path.join(case_dir, "images") + + for image in os.listdir(images_dir): + image_path = os.path.join(images_dir, image) + + rest_name, ext = os.path.splitext(image) + parts = rest_name.split("_") + + if parts[-2] == "F" or parts[-2] == "T": + new_image_name = f"{parts[0]}_{parts[1]}_{parts[-1]}_{parts[-2]}{ext}" + new_image_path = os.path.join(images_dir, new_image_name) + os.rename(image_path, new_image_path) + + def restore_name(self, root_dir): + for case in os.listdir(root_dir): + case_dir = os.path.join(root_dir, case) + if os.path.isdir(case_dir): + images_dir = os.path.join(case_dir, 'images') + + for image in os.listdir(images_dir): + image_path = os.path.join(images_dir, image) + + rest_name, ext = os.path.splitext(image) + parts = rest_name.split("_") # frame_66_60_F + + if parts[-1] == "F" or parts[-1] == "T": + new_image_name = f"{parts[0]}_{parts[1]}_{parts[2]}{ext}" + new_image_path = os.path.join(images_dir, new_image_name) + os.rename(image_path, new_image_path) + + +class DICOMTagCSVController: + """ + DICOMタグ抽出で作成されたCSVファイルを制御するクラス + """ + + def __init__(self, csv_path: str): + """ + 初期化 + Args: + csv_path (str): DICOMタグ抽出で作成されたCSVファイルのパス + """ + self.csv_path = csv_path + self.df = None + self._load_csv() + + def _load_csv(self): + """ + CSVファイルを読み込む + """ + try: + if not os.path.exists(self.csv_path): + raise FileNotFoundError(f"CSVファイルが見つかりません: {self.csv_path}") + + self.df = pd.read_csv(self.csv_path) + except Exception as e: + raise ValueError(f"CSVファイルの読み込みエラー: {e}") + + def get_pixel_spacing(self, case_id: Union[int, str]) -> Optional[Tuple[float, float]]: + """ + 指定された症例番号のPixel Spacingを取得("pre"がついているもので，最大の数値のもの) + """ + try: + # case_idを文字列または数値として検索 + if isinstance(case_id, str): + mask = self.df['case_id'].astype(str) == str(case_id) + else: + mask = self.df['case_id'] == case_id + + matching_rows = self.df[mask] + + if len(matching_rows) == 0: + raise ValueError(f"症例番号 {case_id} がDICOMタグ抽出で作成されたCSVファイルに見つかりません") + + # "pre"が含まれている行のみをフィルタリング + pre_rows = matching_rows[matching_rows['case_name'].str.contains("pre", na=False)] + + if len(pre_rows) == 0: + raise ValueError(f"症例番号 {case_id} に対して，'pre'が含まれている行が見つかりません") + + # 複数の"pre"が含まれている行が存在する場合，最大の数値を持つものを選択 + if len(pre_rows) > 1: + max_pre_number = -1 + selected_row = None + + for idx, row in pre_rows.iterrows(): + case_name = row['case_name'] + + # "pre"の後の数字を抽出 + match = re.search(r'pre(\d+)', case_name) + if match: + pre_number = int(match.group(1)) + if pre_number > max_pre_number: + max_pre_number = pre_number + selected_row = row + else: + # "pre"のみで数字がない場合 + if max_pre_number == -1: + selected_row = row + + if selected_row is None: + raise ValueError(f"症例番号 {case_id} に対して，'pre'のみで数字がない行が見つかりました") + + row_data = selected_row + else: + # pre_rowsが1行の場合 + row_data = pre_rows.iloc[0] + + # pixel spacingを取得 + pixel_spacing_row = row_data['pixel_spacing_row'] + pixel_spacing_col = row_data['pixel_spacing_col'] + + if pd.isna(pixel_spacing_row) or pd.isna(pixel_spacing_col): + raise ValueError(f"症例番号 {case_id} に対して，Pixel Spacingが見つかりません") + + return float(pixel_spacing_row), float(pixel_spacing_col) + + except Exception as e: + raise ValueError(f"Pixel Spacingの取得エラー: {e}") + +if __name__ == '__main__': + excel_path = "../../../data/CHIBAMI_case_list_2023rev20240523.xlsx" + root_path = "../../../data/list" + # excel_path = "../../Data/CHIBAMI_case_list.xlsx" + # root_path = "../../Data/list" + + excel = Excel(excel_path) + data = excel.extract_data() + # print(data) + + # excel.rename(root_path) + # excel.improve_name(root_path) + # excel.restore_name(root_path) \ No newline at end of file