import os import re import csv import json import datetime from glob import glob import argparse # ============================================================================== # スクリプト概要 # ============================================================================== """ このスクリプトは、複数のVASP計算結果ディレクトリを自動的に巡回し、 指定した形式（CSV, HTML, JSON）で集約ファイルを出力します。 他の研究者が再利用しやすいよう詳細コメントを記載しています。 """ """ このスクリプトは、複数のサブディレクトリに保存されたVASPなどの計算結果ファイルを読み込み、 主要なデータを抽出して、単一のCSVファイルおよびHTMLファイルに集約します。 主な機能: - 各サブディレクトリを自動的に巡回し、結果ファイルを探索します。 - 以下のファイルから情報を抽出します: - symmetrized.cif: 最適化後の格子定数、体積、角度、空間群、化学式 - Summarize.txt: バンドギャップ(Eg)、SCFエネルギー(E_scf) - {INITIAL_STRUCTURE_DIR}/*.cif: 計算前の初期格子定数 - *.out.txt: 計算を実行したサーバー名、MPIランク - DoVASP.log: エラーの有無 - `E_scf`の代替取得: Summarize.txtからE_scfが取得できない場合、'SCF/OUTCAR'を探索し、 最終的な"energy without entropy"の値をE_scfとして取得します。 - 参照ファイルが存在しない場合、その項目を'N/A'として処理を続行します。 - 最終結果をタイムスタンプ付きのCSVとHTMLに出力し、処理ログも保存します。 """ # ------------------------------------------------------------------------------ # 定数定義（必要に応じて変更） # ------------------------------------------------------------------------------ TIMESTAMP = datetime.datetime.now().strftime("%Y%m%d%H%M") OUTPUT_CSV = f'summary_{TIMESTAMP}.csv' OUTPUT_HTML = f'summary_{TIMESTAMP}.html' OUTPUT_JSON = f'summary_{TIMESTAMP}.json' LOG_FILE = f'summarize_log_{TIMESTAMP}.txt' INITIAL_STRUCTURE_DIR = "Original" SUMMARIZE_FILENAME = "Summarize.txt" SYM_CIF_FILENAME = "symmetrized.cif" LOG_FILENAME = "DoVASP.log" SCF_DIR = "SCF" OUTCAR_FILENAME = "OUTCAR" EXCLUDE_DIRS = ["cif"] HEADERS = [ "dir_name", "server", "mpirank", "space_group", "space_group_num", "chemcal_formula", "length_a_1", "length_b_1", "length_c_1", "length_a_2", "length_b_2", "length_c_2", "Volume", "alpha", "beta", "gamma", "E_scf", "Eg", "Traceback", "ERROR" ] # ------------------------------------------------------------------------------ # データ抽出関数 # ------------------------------------------------------------------------------ def extract_numeric(value): """CIFファイルの値（例: '3.2518(3)'）から数値部分を抽出。""" match = re.match(r"([-+]?[0-9]*\.?[0-9]+)", value) return float(match.group(1)) if match else None def parse_summarize_txt(file_path): """Summarize.txt を解析して Eg と E_scf を取得。""" data = {'Eg': None, 'E_scf': None} with open(file_path, 'r', encoding='utf-8') as f: for line in f: if "Eg=" in line: match = re.search(r"Eg=\s*([-+]?[0-9]*\.?[0-9]+)", line) if match: data['Eg'] = float(match.group(1)) if "FreeEnergy: free energy" in line: match = re.search(r"TOTEN\s*=\s*([-+]?[0-9]*\.?[0-9]+)", line) if match: data['E_scf'] = float(match.group(1)) return data def get_e_scf_from_outcar(base_dir_path): """SCF/OUTCAR から 'energy without entropy' の最終値を取得。""" outcar_path = os.path.join(base_dir_path, SCF_DIR, OUTCAR_FILENAME) if not os.path.exists(outcar_path): return None last_energy_line = None with open(outcar_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: if "energy without entropy=" in line: last_energy_line = line if last_energy_line: try: return float(last_energy_line.split()[-1]) except (ValueError, IndexError): return None return None def parse_symmetrized_cif(file_path): """symmetrized.cif を解析し、空間群、格子情報などを取得。""" data = { 'space_group': None, 'space_group_num': None, 'chemcal_formula': None, 'length_a_2': None, 'length_b_2': None, 'length_c_2': None, 'Volume': None, 'alpha': None, 'beta': None, 'gamma': None } key_map = { "_symmetry_space_group_name_H-M": "space_group", "_symmetry_Int_Tables_number": "space_group_num", "_chemical_formula_sum": "chemcal_formula", "_cell_length_a": "length_a_2", "_cell_length_b": "length_b_2", "_cell_length_c": "length_c_2", "_cell_volume": "Volume", "_cell_angle_alpha": "alpha", "_cell_angle_beta": "beta", "_cell_angle_gamma": "gamma" } with open(file_path, 'r', encoding='utf-8') as f: for line in f: parts = line.split() if not parts: continue key = parts[0] if key in key_map: value_str = parts[-1] data_key = key_map[key] if data_key in ["space_group", "chemcal_formula"]: data[data_key] = line.split(None, 1)[-1].strip().strip('"\'') elif data_key == "space_group_num": data[data_key] = value_str else: data[data_key] = extract_numeric(value_str) return data def parse_original_cif(file_path): """初期構造の CIF ファイルから格子定数を取得。""" data = {'length_a_1': None, 'length_b_1': None, 'length_c_1': None} with open(file_path, 'r', encoding='utf-8') as f: for line in f: if "_cell_length_a" in line: data['length_a_1'] = extract_numeric(line.split()[-1]) elif "_cell_length_b" in line: data['length_b_1'] = extract_numeric(line.split()[-1]) elif "_cell_length_c" in line: data['length_c_1'] = extract_numeric(line.split()[-1]) return data def parse_out_txt(file_path): """*.out.txt を解析してサーバー名と MPI ランクを取得。""" server = os.path.basename(file_path).replace('.out.txt', '') mpirank = None try: with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: if "running" in line: parts = line.split() if len(parts) > 1: mpirank = parts[1] break except FileNotFoundError: pass return {'server': server, 'mpirank': mpirank} def check_log_for_errors(log_path): """DoVASP.log を確認し、Traceback や ERROR を検出。""" traceback_flag, error_flag = 0, 0 if os.path.exists(log_path): with open(log_path, 'r', encoding='utf-8', errors='ignore') as f: text = f.read() if "Traceback" in text: traceback_flag = 1 if "ERROR" in text: error_flag = 1 return {'Traceback': traceback_flag, 'ERROR': error_flag} # ------------------------------------------------------------------------------ # ディレクトリ単位処理 # ------------------------------------------------------------------------------ def process_directory(dir_name, log_lines): """ディレクトリを処理し、各種ファイルから情報を抽出して辞書にまとめる。""" data = {header: None for header in HEADERS} data['dir_name'] = dir_name sym_cif_path = os.path.join(dir_name, SYM_CIF_FILENAME) if os.path.exists(sym_cif_path): data.update(parse_symmetrized_cif(sym_cif_path)) else: log_lines.append(f"[INFO] {dir_name}: '{SYM_CIF_FILENAME}' が見つかりません。") initial_cif_glob = glob(os.path.join(dir_name, INITIAL_STRUCTURE_DIR, "*.cif")) if initial_cif_glob: data.update(parse_original_cif(initial_cif_glob[0])) else: log_lines.append(f"[INFO] {dir_name}: '{INITIAL_STRUCTURE_DIR}/*.cif' が見つかりません。") summarize_path = os.path.join(dir_name, SUMMARIZE_FILENAME) if os.path.exists(summarize_path): data.update(parse_summarize_txt(summarize_path)) else: log_lines.append(f"[INFO] {dir_name}: '{SUMMARIZE_FILENAME}' が見つかりません。") if data['E_scf'] is None: e_scf_fallback = get_e_scf_from_outcar(dir_name) if e_scf_fallback is not None: data['E_scf'] = e_scf_fallback log_lines.append(f"[INFO] {dir_name}: E_scf を '{SCF_DIR}/{OUTCAR_FILENAME}' から取得しました。") else: log_lines.append(f"[INFO] {dir_name}: E_scf は代替ファイルからも取得できませんでした。") out_txt_glob = glob(os.path.join(dir_name, "*.out.txt")) if out_txt_glob: data.update(parse_out_txt(out_txt_glob[0])) log_path = os.path.join(dir_name, LOG_FILENAME) data.update(check_log_for_errors(log_path)) return data # ------------------------------------------------------------------------------ # ファイル出力関数 # ------------------------------------------------------------------------------ def write_summary_files(results_data, output_formats): """指定された形式で結果ファイルを出力する。""" if 'csv' in output_formats: with open(OUTPUT_CSV, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f) writer.writerow(HEADERS) for data_dict in results_data: row = [data_dict.get(h) for h in HEADERS] row_na = ["N/A" if v is None else v for v in row] writer.writerow(row_na) print(f"✅ CSVファイルを {OUTPUT_CSV} に保存しました。") if 'html' in output_formats: with open(OUTPUT_HTML, "w", encoding="utf-8") as f: f.write("\n計算結果サマリー") f.write("") f.write("\n") f.write(f"

計算結果サマリー ({TIMESTAMP})

\n") f.write("\n") f.write("" + "".join(f"" for h in HEADERS) + "\n") for data_dict in results_data: row = [data_dict.get(h) for h in HEADERS] row_na = ["N/A" if v is None else v for v in row] f.write("" + "".join(f"" for v in row_na) + "\n") f.write("
{h}
{v}
\n") print(f"✅ HTMLファイルを {OUTPUT_HTML} に保存しました。") if 'json' in output_formats: with open(OUTPUT_JSON, 'w', encoding='utf-8') as f: json.dump(results_data, f, ensure_ascii=False, indent=2) print(f"✅ JSONファイルを {OUTPUT_JSON} に保存しました。") # ------------------------------------------------------------------------------ # メイン処理 # ------------------------------------------------------------------------------ def main(): parser = argparse.ArgumentParser(description="VASP計算結果をまとめて出力するツール") parser.add_argument('--formats', nargs='+', default=['csv', 'html'], choices=['csv', 'html', 'json'], help="出力するファイル形式 (複数選択可)。例: --formats csv html json") args = parser.parse_args() all_results = [] log_lines = [] for item_name in sorted(os.listdir()): if not os.path.isdir(item_name): continue if item_name in EXCLUDE_DIRS: log_lines.append(f"[INFO] ディレクトリ '{item_name}' は除外設定されているためスキップします。") continue try: log_lines.append(f"--- Processing directory: {item_name} ---") dir_data = process_directory(item_name, log_lines) all_results.append(dir_data) except Exception as e: log_lines.append(f"[CRITICAL_ERROR] {item_name} の処理中に予期せぬエラーが発生しました: {e}") if all_results: write_summary_files(all_results, args.formats) else: log_lines.append("[WARNING] 有効なデータが1件も見つかりませんでした。ファイルは出力されません。") print("⚠️ データが見つかりませんでした。") with open(LOG_FILE, 'w', encoding='utf-8') as log_f: log_f.write("\n".join(log_lines)) print(f"📄 ログを {LOG_FILE} に書き込みました（{len(log_lines)}件の記録あり）。") if __name__ == "__main__": main()