[WIP]rewrite workflow implementation. Add prefect framework. prepare for paper.

.cursor/rules/dpti-default-ai-rules.mdc

@@ -0,0 +1,61 @@
+---
+description: General Rules
+globs: *py,*json,*yaml,*ipynb,*sh,*md,*txt,*.lammps,*tex,*toml,*yml
+alwaysApply: false
+---
+
+<!-- - You can @ files here
+- You can use markdown but dont have to -->
+[base.py](mdc:dpti/workflows/simulations/base.py) [equi.py](mdc:dpti/equi.py) [equi_sim.py](mdc:dpti/workflows/simulations/equi_sim.py) [ti.py](mdc:dpti/ti.py) [mti.py](mdc:dpti/mti.py) [ti_sim.py](mdc:dpti/workflows/simulations/ti_sim.py) [hti.py](mdc:dpti/hti.py) [hti_liq.py](mdc:dpti/hti_liq.py) [hti_ice.py](mdc:dpti/hti_ice.py) [hti_water.py](mdc:dpti/hti_water.py) [hti_sim.py](mdc:dpti/workflows/simulations/hti_sim.py) [einstein.py](mdc:dpti/einstein.py) [di.py](mdc:dpti/workflows/service/di.py) [file_handler.py](mdc:dpti/workflows/service/file_handler.py) [job_executor.py](mdc:dpti/workflows/service/job_executor.py) [ti_line_flow.py](mdc:dpti/workflows/flows/ti_line_flow.py)
+
+
+# 角色
+你是一名精通科学计算和分子动力学模拟的高级工程师，拥有10年以上的Python科学计算应用开发经验，熟悉LAMMPS、DeepMD-kit、Prefect等开发工具和技术栈。你的任务是帮助用户设计和开发易用且易于维护的分子动力学自由能计算应用。始终遵循科学计算领域的最佳实践，并坚持代码可重复性和计算效率的原则。
+# 目标
+你的目标是以用户容易理解的方式帮助他们完成分子动力学自由能计算应用的设计和开发工作，确保计算结果准确可靠、性能高#效、用户使用便捷。
+# 要求
+在理解用户需求、设计计算流程、编写代码、解决问题和项目迭代优化时，你应该始终遵循以下原则：
+项目初始化
+仔细阅读项目的README.md，理解自由能计算的基本原理、计算流程和技术实现方案
+如果没有README.md，创建一个包含计算理论基础、输入输出格式、依赖环境等信息的文档
+# 需求理解
+理解用户的具体计算需求，包括体系类型、温度压力条件、所需精度等
+选择合适的热力学积分方法，避免不必要的计算
+# 计算流程设计
+使用Prefect设计可重复、可追踪的计算工作流
+确保计算任务在不同计算环境下都能正确运行
+# 代码编写
+技术选型：
+Python科学计算库(numpy, scipy)用于数据处理和分析
+LAMMPS用于分子动力学模拟
+Prefect 框架用于工作流管理
+DeepMD-kit用于机器学习势能计算
+# 代码结构：
+模块化设计，将计算、分析、可视化等功能分离
+使用面向对象编程，提高代码复用性
+遵循Python科学计算代码规范
+# 计算安全性：
+检查输入参数的合法性
+保存计算中间结果，支持断点续算
+验证计算结果的物理合理性
+# 性能优化：
+优化并行计算策略
+减少不必要的IO操作
+合理使用内存缓存
+# 测试与文档：
+编写单元测试验证计算正确性
+提供详细的API文档和使用示例
+添加计算原理和方法的说明
+# 问题解决
+全面理解分子动力学模拟和自由能计算的原理
+根据计算结果的异常分析问题原因
+确保代码修改不影响计算的物理正确性
+# 迭代优化
+收集用户反馈，优化计算流程和接口设计
+及时更新文档，包括新增功能和优化建议
+持续改进计算性能和稳定性
+# 方法论
+1.Unix编程风格,开源软件设计风格,恰当根据项目或者模块需要,使用设计模式
+2.坚持dont repeat your self 原则
+3.在软件架构上给出成熟考虑。
+2.回答问题时,也给出关于设计模式上的思考,提示用户相关联的最佳实践(如果有)。

dpti/aitools/codebase/collect_py_files_context.py

@@ -0,0 +1,163 @@
+# %%
+
+import argparse
+import json
+from pathlib import Path
+from typing import Dict
+
+
+def collect_python_files(directory: str) -> Dict[str, str]:
+    """
+    Collect paths and contents of all Python files in the specified directory.
+
+    Args:
+        directory: Directory path to scan
+
+    Returns
+    -------
+    Dictionary containing file paths and their contents
+    """
+    python_files = {}
+    base_dir = Path(directory).name
+
+    # Use Path object for directory traversal
+    for file_path in Path(directory).rglob("*.py"):
+        # Skip __pycache__ directory
+        if "__pycache__" in str(file_path):
+            continue
+
+        try:
+            # Read file content
+            with open(file_path, encoding="utf-8") as f:
+                content = f.read()
+
+            # Convert to relative path and include base directory
+            relative_path = str(file_path.relative_to(directory))
+            full_path = f"{base_dir}/{relative_path}"
+            python_files[full_path] = content
+
+        except Exception as e:
+            print(f"Error reading file {file_path}: {e!s}")
+
+    return python_files
+
+
+def group_by_directory(
+    files: Dict[str, str], max_files_per_group: int = 20
+) -> Dict[str, Dict[str, str]]:
+    """
+    Group files by their directory structure.
+
+    Args:
+        files: Dictionary of file paths and contents
+        max_files_per_group: Maximum number of files per group
+
+    Returns
+    -------
+    Dictionary of grouped files
+    """
+    # First group by directory
+    dir_groups = {}
+    for file_path, content in files.items():
+        dir_name = str(Path(file_path).parent)
+        if dir_name not in dir_groups:
+            dir_groups[dir_name] = {}
+        dir_groups[dir_name][file_path] = content
+
+    # Merge small groups and split large groups
+    final_groups = {}
+    current_group = {}
+    current_group_size = 0
+    group_index = 1
+
+    for dir_name, dir_files in dir_groups.items():
+        # If adding this directory's files would exceed the limit
+        if (
+            current_group_size + len(dir_files) > max_files_per_group
+            and current_group_size > 0
+        ):
+            # Save current group and start a new one
+            final_groups[f"group_{group_index}"] = current_group
+            current_group = {}
+            current_group_size = 0
+            group_index += 1
+
+        # Add files to current group
+        current_group.update(dir_files)
+        current_group_size += len(dir_files)
+
+    # Don't forget to save the last group
+    if current_group:
+        final_groups[f"group_{group_index}"] = current_group
+
+    return final_groups
+
+
+def save_grouped_json(groups: Dict[str, Dict[str, str]], output_base: str) -> None:
+    """
+    Save each group to a separate JSON file.
+
+    Args:
+        groups: Grouped files dictionary
+        output_base: Base name for output files
+    """
+    output_base = Path(output_base)
+    base_name = output_base.stem
+    parent_dir = output_base.parent
+
+    for group_name, group_files in groups.items():
+        output_file = parent_dir / f"{base_name}_{group_name}.json"
+        wrapper = {
+            "project_files": {
+                "description": f"Python source files collection - {group_name}",
+                "base_directory": Path().absolute().name,
+                "files": group_files,
+            }
+        }
+
+        with open(output_file, "w", encoding="utf-8") as f:
+            json.dump(wrapper, f, ensure_ascii=False, indent=2)
+        print(f"Saved {len(group_files)} files to {output_file}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Collect Python files content into JSON for context"
+    )
+    parser.add_argument(
+        "-d",
+        "--directory",
+        default="../../../dpti/",
+        help="Directory to scan (default: current directory)",
+    )
+    parser.add_argument(
+        "-o",
+        "--output",
+        default="build/python_files_context.json",
+        help="Output JSON file base name (default: python_files_context.json)",
+    )
+    parser.add_argument(
+        "-n",
+        "--num-files",
+        type=int,
+        default=20,
+        help="Maximum number of files per group (default: 20)",
+    )
+
+    args = parser.parse_args()
+
+    # Collect Python files
+    python_files = collect_python_files(args.directory)
+
+    # Group files
+    groups = group_by_directory(python_files, args.num_files)
+
+    # Save grouped files
+    save_grouped_json(groups, args.output)
+
+    print(f"\nTotal files collected: {len(python_files)}")
+    print(f"Split into {len(groups)} groups")
+
+
+if __name__ == "__main__":
+    main()

dpti/hti_liq.py

@@ -503,6 +503,9 @@ def compute_task(
     if "copies" in jdata:
         natoms *= np.prod(jdata["copies"])
     fe, fe_err, thermo_info = post_tasks(job, natoms)
+    print(
+        f"hti_liq.compute_task: ideal gas fe = {fe:.68}eV, err = {fe_err[0]:.6f}eV, {fe_err[1]:.6f}eV per atom"
+    )
     _print_thermo_info(thermo_info)
 
     info = thermo_info.copy()
@@ -555,6 +558,8 @@ def compute_task(
     info["pv_err"] = pv_err
     # info['de'] = de
     # info['de_err'] = de_err
+    info["e0"] = fe
+    info["e0_err"] = fe_err
     info["e1"] = e1
     info["e1_err"] = e1_err
     with open(os.path.join(job, "result.json"), "w") as result:

dpti/ti.py

@@ -108,10 +108,10 @@ def _gen_lammps_input(
     ret += "compute         allmsd all msd\n"
     if ens == "nvt":
         ret += "thermo_style    custom step ke pe etotal enthalpy temp press vol c_allmsd[*]\n"
-        ret += "thermo_modify   format 4*8 %20.6f\n"
+        ret += "thermo_modify   format float %20.6f\n"
     elif "npt" in ens:
         ret += "thermo_style    custom step ke pe etotal enthalpy temp press vol c_allmsd[*]\n"
-        ret += "thermo_modify   format 4*8 %20.6f\n"
+        ret += "thermo_modify   format float %20.6f\n"
     else:
         raise RuntimeError(f"unknow ensemble {ens}\n")
     ret += "dump            1 all custom ${DUMP_FREQ} traj.dump id type x y z\n"
@@ -669,7 +669,8 @@ def post_tasks(
         "all_fe": all_fe.tolist(),
         "all_fe_stat_err": all_fe_err.tolist(),
         "all_fe_inte_err": all_fe_sys_err.tolist(),
-        "all_fe_tot_err": np.linalg.norm([all_fe_err[ii], all_fe_sys_err[ii]]).tolist(),
+        # "all_fe_tot_err": np.linalg.norm([all_fe_err[ii], all_fe_sys_err[ii]]).tolist(),
+        "all_fe_tot_err": np.linalg.norm([all_fe_err, all_fe_sys_err]).tolist(),
     }
 
     # data = [all_temps.tolist(), all_press.tolist(),

dpti/ti_water.py

@@ -160,7 +160,7 @@ def handle_compute(args):
         elif path == "p":
             args.To = jdata_hti_in["pres"]
     if args.inte_method == "inte":
-        ti.post_tasks(
+        ti_info = ti.post_tasks(
             job,
             jdata,
             args.Eo,
@@ -170,6 +170,7 @@ def handle_compute(args):
             scheme=args.scheme,
             shift=args.shift,
         )
+        return ti_info
     elif args.inte_method == "mbar":
         ti.post_tasks_mbar(job, jdata, args.Eo, natoms=nmols)
     else:

dpti/workflows/__init__.py

@@ -0,0 +1,18 @@
+pass
+# import importlib
+
+# def _import_module(module_name):
+#     try:
+#         return importlib.import_module(f'dpti.workflows.{module_name}')
+#     except ImportError:
+#         return importlib.import_module(f'.{module_name}', package='dpti.workflows')
+
+# MODULES_TO_IMPORT = [
+#     'job_executor',
+#     # 'another_module',
+#     # 'yet_another_module',
+# ]
+
+# # dynamically import sub modules
+# for module_name in MODULES_TO_IMPORT:
+#     globals()[module_name] = _import_module(module_name)