[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

dpti/__init__.py

@@ -6,10 +6,10 @@
     "equi",
     "gdi",
     "hti",
-    "hti_liq",
     "hti_ice",
+    "hti_liq",
     "hti_water",
+    "lib",
     "ti",
     "ti_water",
-    "lib",
 ]

dpti/dags/dp_ti_gdi.py

@@ -2,7 +2,7 @@
 import os
 import time
 from datetime import datetime, timedelta
-from typing import ClassVar, Dict
+from typing import ClassVar
 
 from airflow import DAG
 from airflow.api.client.local_client import Client
@@ -30,12 +30,12 @@
 
 
 class GDIDAGFactory:
-    default_args: ClassVar[Dict[str, object]] = {
+    default_args: ClassVar[dict[str, object]] = {
         "owner": "airflow",
         "start_date": datetime(2018, 1, 1),
     }
 
-    dagargs: ClassVar[Dict[str, object]] = {
+    dagargs: ClassVar[dict[str, object]] = {
         "default_args": default_args,
         "schedule_interval": None,
     }

dpti/equi.py

@@ -312,12 +312,12 @@ def npt_equi_conf(npt_dir):
     stat_bsize = jdata["stat_bsize"]
 
     data = get_thermo(thermo_file)
-    lx, lxe = block_avg(data[:, 8], skip=stat_skip, block_size=stat_bsize)
-    ly, lye = block_avg(data[:, 9], skip=stat_skip, block_size=stat_bsize)
-    lz, lze = block_avg(data[:, 10], skip=stat_skip, block_size=stat_bsize)
-    xy, xye = block_avg(data[:, 11], skip=stat_skip, block_size=stat_bsize)
-    xz, xze = block_avg(data[:, 12], skip=stat_skip, block_size=stat_bsize)
-    yz, yze = block_avg(data[:, 13], skip=stat_skip, block_size=stat_bsize)
+    lx, _lxe = block_avg(data[:, 8], skip=stat_skip, block_size=stat_bsize)
+    ly, _lye = block_avg(data[:, 9], skip=stat_skip, block_size=stat_bsize)
+    lz, _lze = block_avg(data[:, 10], skip=stat_skip, block_size=stat_bsize)
+    xy, _xye = block_avg(data[:, 11], skip=stat_skip, block_size=stat_bsize)
+    xz, _xze = block_avg(data[:, 12], skip=stat_skip, block_size=stat_bsize)
+    yz, _yze = block_avg(data[:, 13], skip=stat_skip, block_size=stat_bsize)
     # print('~~~', lx , ly, lz , xy, xz, yz)
 
     last_dump = get_last_dump(dump_file).split("\n")

dpti/hti.py

@@ -86,7 +86,7 @@ def _ff_deep_on(lamb, model, sparam, if_meam=False, meam_model=None, append=None
     #     ret += 'pair_coeff      * * meam /home/fengbo/4_Sn/meam_files/library_18Metal.meam Sn /home/fengbo/4_Sn/meam_files/Sn_18Metal.meam Sn \n'
     if if_meam:
         ret += f"pair_style      hybrid/overlay meam lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
-        ret += f'pair_coeff      * * meam {meam_model["library"]} {meam_model["element"]} {meam_model["potential"]} {meam_model["element"]}\n'
+        ret += f"pair_coeff      * * meam {meam_model['library']} {meam_model['element']} {meam_model['potential']} {meam_model['element']}\n"
     else:
         if append:
             ret += f"pair_style      hybrid/overlay deepmd {model:s} {append:s} lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
@@ -175,7 +175,7 @@ def _ff_lj_off(lamb, model, sparam, if_meam=False, meam_model=None, append=None)
     #     ret += 'pair_coeff      * * meam /home/fengbo/4_Sn/meam_files/library_18Metal.meam Sn /home/fengbo/4_Sn/meam_files/Sn_18Metal.meam Sn\n'
     if if_meam:
         ret += f"pair_style      hybrid/overlay meam lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
-        ret += f'pair_coeff      * * meam {meam_model["library"]} {meam_model["element"]} {meam_model["potential"]} {meam_model["element"]}\n'
+        ret += f"pair_coeff      * * meam {meam_model['library']} {meam_model['element']} {meam_model['potential']} {meam_model['element']}\n"
         # ret += f'pair_coeff      * * meam {meam_model[0]} {meam_model[2]} {meam_model[1]} {meam_model[2]}\n'
     else:
         if append:
@@ -709,9 +709,9 @@ def _make_tasks(
             "sigma_" + str(t[0]) + "_" + str(t[1]) for t in sigma_key_index
         ]
         for sigma_key_name in sigma_key_name_list:
-            assert sparam.get(
-                sigma_key_name, None
-            ), f"there must be key-value for {sigma_key_name} in soft_param"
+            assert sparam.get(sigma_key_name, None), (
+                f"there must be key-value for {sigma_key_name} in soft_param"
+            )
 
     if crystal == "frenkel":
         m_spring_k = []
@@ -983,7 +983,7 @@ def post_tasks(iter_name, jdata, natoms=None, method="inte", scheme="s"):
                 step="lj_on",
             )
         elif method == "mbar":
-            e0, err0, tinfo0 = _post_tasks_mbar(
+            e0, err0, _tinfo0 = _post_tasks_mbar(
                 subtask_name, jdata, natoms=natoms, switch=switch, step="lj_on"
             )
         else:
@@ -1412,9 +1412,9 @@ def hti_phase_trans_analyze(job, jdata=None):
 def run_task(task_dir, machine_file, task_name, no_dp=False):
     if task_name == "00" or task_name == "01" or task_name == "02":
         job_work_dir_ = glob.glob(os.path.join(task_dir, task_name + "*"))
-        assert (
-            len(job_work_dir_) == 1
-        ), f"The task_name you entered is {task_name}. It indicates that you want to run tasks for step {task_name} of the two-step or three-step HTI. Please make sure that there is one and only one {task_name}.* directory in the hti task directory."
+        assert len(job_work_dir_) == 1, (
+            f"The task_name you entered is {task_name}. It indicates that you want to run tasks for step {task_name} of the two-step or three-step HTI. Please make sure that there is one and only one {task_name}.* directory in the hti task directory."
+        )
         job_work_dir = job_work_dir_[0]
     elif task_name == "one-step":
         job_work_dir = task_dir

dpti/hti_liq.py

@@ -48,9 +48,9 @@ def parse_lj_sigma_epsilon(ret, sparam, hybrid=False):
     activation_ij = np.zeros((element_num, element_num))
 
     if epsilon is not None:
-        assert (
-            epsilon_0_0 is None
-        ), "epsilon and epsilon_0_0 cannot be set at the same time"
+        assert epsilon_0_0 is None, (
+            "epsilon and epsilon_0_0 cannot be set at the same time"
+        )
         for i, j in sigma_key_index:
             epsilon_ij[i, j] = epsilon
     else:
@@ -68,9 +68,9 @@ def parse_lj_sigma_epsilon(ret, sparam, hybrid=False):
             sigma_ij[i, j] = sparam["sigma_" + str(i) + "_" + str(j)]
 
     if activation is not None:
-        assert (
-            activation_0_0 is None
-        ), "activation and activation_0_0 cannot be set at the same time"
+        assert activation_0_0 is None, (
+            "activation and activation_0_0 cannot be set at the same time"
+        )
         for i, j in sigma_key_index:
             activation_ij[i, j] = activation
     else:
@@ -110,7 +110,7 @@ def _ff_deep_on(lamb, sparam, model, if_meam=False, meam_model=None):
     ret += "variable        ONE equal 1\n"
     if if_meam:
         ret += f"pair_style      hybrid/overlay meam lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
-        ret += f'pair_coeff      * * meam {meam_model["library"]} {meam_model["element"]} {meam_model["potential"]} {meam_model["element"]}\n'
+        ret += f"pair_coeff      * * meam {meam_model['library']} {meam_model['element']} {meam_model['potential']} {meam_model['element']}\n"
         # ret += f'pair_coeff      * * meam {meam_model[0]} {meam_model[2]} {meam_model[1]} {meam_model[2]}\n'
     else:
         ret += f"pair_style      hybrid/overlay deepmd {model} lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
@@ -138,7 +138,7 @@ def _ff_soft_off(lamb, sparam, model, if_meam=False, meam_model=None):
     ret += "variable        INV_LAMBDA equal 1-${LAMBDA}\n"
     if if_meam:
         ret += f"pair_style      hybrid/overlay meam lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
-        ret += f'pair_coeff      * * meam {meam_model["library"]} {meam_model["element"]} {meam_model["potential"]} {meam_model["element"]}\n'
+        ret += f"pair_coeff      * * meam {meam_model['library']} {meam_model['element']} {meam_model['potential']} {meam_model['element']}\n"
         # ret += f'pair_coeff      * * meam {meam_model[0]} {meam_model[2]} {meam_model[1} {meam_model[2]} \n'
     else:
         ret += f"pair_style      hybrid/overlay deepmd {model} lj/cut/soft {nn:f} {alpha_lj:f} {rcut:f}\n"
@@ -289,9 +289,9 @@ def _make_tasks(iter_name, jdata, step, if_meam=False, meam_model=None):
                 "sigma_" + str(t[0]) + "_" + str(t[1]) for t in sigma_key_index
             ]
             for sigma_key_name in sigma_key_name_list:
-                assert sparam.get(
-                    sigma_key_name, None
-                ), f"there must be key-value for sigma or {sigma_key_name} in soft_param"
+                assert sparam.get(sigma_key_name, None), (
+                    f"there must be key-value for sigma or {sigma_key_name} in soft_param"
+                )
 
     job_abs_dir = create_path(iter_name)
 
@@ -464,9 +464,9 @@ def _post_tasks(iter_name, step, natoms):
 def post_tasks(iter_name, natoms):
     fe = einstein.ideal_gas_fe(iter_name)
     subtask_name = os.path.join(iter_name, "00.soft_on")
-    e0, err0, tinfo0 = _post_tasks(subtask_name, "soft_on", natoms)
+    e0, err0, _tinfo0 = _post_tasks(subtask_name, "soft_on", natoms)
     subtask_name = os.path.join(iter_name, "01.deep_on")
-    e1, err1, tinfo1 = _post_tasks(subtask_name, "deep_on", natoms)
+    e1, err1, _tinfo1 = _post_tasks(subtask_name, "deep_on", natoms)
     subtask_name = os.path.join(iter_name, "02.soft_off")
     e2, err2, tinfo2 = _post_tasks(subtask_name, "soft_off", natoms)
     fe = fe + e0 + e1 + e2

dpti/hti_water.py

@@ -583,7 +583,7 @@ def _post_tasks_mbar(iter_name, step, natoms):
 
     nk = np.array(nk)
     mbar = pymbar.MBAR(ukn, nk)
-    Deltaf_ij, dDeltaf_ij, Theta_ij = mbar.getFreeEnergyDifferences()
+    Deltaf_ij, dDeltaf_ij, _Theta_ij = mbar.getFreeEnergyDifferences()
     Deltaf_ij = Deltaf_ij / natoms
     dDeltaf_ij = dDeltaf_ij / natoms
 
@@ -672,15 +672,15 @@ def post_tasks(iter_name, natoms, method="inte", scheme="s"):
     if method == "inte":
         e0, err0, tinfo0 = _post_tasks(subtask_name, "angle_on", natoms, scheme=scheme)
     elif method == "mbar":
-        e0, err0, tinfo0 = _post_tasks_mbar(subtask_name, "angle_on", natoms)
+        e0, err0, _tinfo0 = _post_tasks_mbar(subtask_name, "angle_on", natoms)
     print(f"# fe of angle_on : {e0:20.12f}  {err0[0]:10.3e} {err0[1]:10.3e}")
     # _print_thermo_info(tinfo)
     # print(e, err)
     subtask_name = os.path.join(iter_name, "01.deep_on")
     if method == "inte":
         e1, err1, tinfo1 = _post_tasks(subtask_name, "deep_on", natoms, scheme=scheme)
     elif method == "mbar":
-        e1, err1, tinfo1 = _post_tasks_mbar(subtask_name, "deep_on", natoms)
+        e1, err1, _tinfo1 = _post_tasks_mbar(subtask_name, "deep_on", natoms)
     print(f"# fe of deep_on  : {e1:20.12f}  {err1[0]:10.3e} {err1[1]:10.3e}")
     # _print_thermo_info(tinfo)
     # print(e, err)

dpti/lib/RemoteJob.py

@@ -274,7 +274,7 @@ def _make_script(self, job_dirs, cmd, args=None, resources=None):
 class SlurmJob(RemoteJob):
     def submit(self, job_dirs, cmd, args=None, resources=None):
         script_name = self._make_script(job_dirs, cmd, args, res=resources)
-        stdin, stdout, stderr = self.block_checkcall(
+        _stdin, stdout, _stderr = self.block_checkcall(
             f"cd {self.remote_root}; sbatch {script_name}"
         )
         subret = stdout.readlines()
@@ -288,7 +288,7 @@ def check_status(self):
         job_id = self._get_job_id()
         if job_id == "":
             raise RuntimeError(f"job {self.remote_root} is has not been submitted")
-        ret, stdin, stdout, stderr = self.block_call("squeue --job " + job_id)
+        ret, _stdin, stdout, stderr = self.block_call("squeue --job " + job_id)
         err_str = stderr.read().decode("utf-8")
         if ret != 0:
             if "Invalid job id specified" in err_str:
@@ -413,7 +413,7 @@ def _make_script(self, job_dirs, cmd, args=None, res=None):
 class PBSJob(RemoteJob):
     def submit(self, job_dirs, cmd, args=None, resources=None):
         script_name = self._make_script(job_dirs, cmd, args, res=resources)
-        stdin, stdout, stderr = self.block_checkcall(
+        _stdin, stdout, _stderr = self.block_checkcall(
             f"cd {self.remote_root}; qsub {script_name}"
         )
         subret = stdout.readlines()
@@ -427,7 +427,7 @@ def check_status(self):
         job_id = self._get_job_id()
         if job_id == "":
             raise RuntimeError(f"job {self.remote_root} is has not been submitted")
-        ret, stdin, stdout, stderr = self.block_call("qstat -x " + job_id)
+        ret, _stdin, stdout, stderr = self.block_call("qstat -x " + job_id)
         err_str = stderr.read().decode("utf-8")
         if ret != 0:
             if "qstat: Unknown Job Id" in err_str:

dpti/lib/dump.py

@@ -38,7 +38,7 @@ def get_atype(lines):
 
 
 def get_natoms(lines):
-    blk, head = _get_block(lines, "NUMBER OF ATOMS")
+    blk, _head = _get_block(lines, "NUMBER OF ATOMS")
     return int(blk[0])
 
 
@@ -82,7 +82,7 @@ def get_posi(lines):
 
 
 def get_dumpbox(lines):
-    blk, h = _get_block(lines, "BOX BOUNDS")
+    blk, _h = _get_block(lines, "BOX BOUNDS")
     bounds = np.zeros([3, 2])
     tilt = np.zeros([3])
     for dd in range(3):

dpti/lib/lmp.py

@@ -104,7 +104,7 @@ def get_id(lines):
     alines = get_atoms(lines)
     idx_list = []
     for ii in alines:
-        idx, at, x, y, z = _atom_info_atom(ii)
+        idx, _at, _x, _y, _z = _atom_info_atom(ii)
         idx_list.append(idx)
     return np.array(idx_list, dtype=int)
 
@@ -114,7 +114,7 @@ def get_atype(lines):
     atype = []
     for ii in alines:
         # idx, mt, at, q, x, y, z = _atom_info_mol(ii)
-        idx, at, x, y, z = _atom_info_atom(ii)
+        _idx, at, _x, _y, _z = _atom_info_atom(ii)
         atype.append(at)
     return np.array(atype, dtype=int)
 

dpti/lib/utils.py

@@ -389,12 +389,14 @@ def integrate_range_simpson(xx, yy, ye):
     xx0, inte0, stat_err0 = _integrate_range_simpson_inner(xx, yy, ye)
     if len(xx) < 5:
         return xx0, inte0, stat_err0, np.zeros(xx0.shape)
-    xx1, inte1, stat_err1 = _integrate_range_simpson_inner(xx[::2], yy[::2], ye[::2])
+    xx1, inte1, _stat_err1 = _integrate_range_simpson_inner(xx[::2], yy[::2], ye[::2])
     diff1 = np.abs(inte1 - inte0[::2]) / 16.0
     assert np.linalg.norm(xx1 - xx0[::2]) < 1e-10
     # error esti series 1, shifted from series 0
-    xx2, inte2, stat_err2 = _integrate_range_simpson_inner(xx[2:], yy[2:], ye[2:])
-    xx3, inte3, stat_err3 = _integrate_range_simpson_inner(xx[2::2], yy[2::2], ye[2::2])
+    xx2, inte2, _stat_err2 = _integrate_range_simpson_inner(xx[2:], yy[2:], ye[2:])
+    xx3, inte3, _stat_err3 = _integrate_range_simpson_inner(
+        xx[2::2], yy[2::2], ye[2::2]
+    )
     from scipy.interpolate import interp1d
 
     f = interp1d(xx1, diff1)