python 脚本

python 自动化

DeePMD install

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
from json import tool
from optparse import Option
import subprocess
import os
import sys
from numpy import str_
from sty import fg, bg, ef, rs
import time

def commandline(command_line):
    # 用来输入命令行
    # para
    # @command_line -> list[string]
    # @return stderr -> string 
    #   标准错误
    # @return stdout -> string
    #   标准输出
    try:
        process = subprocess.Popen(command_line,
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE)
        stdout, stderr = process.communicate()
        stderr = stderr.decode('utf-8')
        stdout = stdout.decode('utf-8')   
    except:
        #stderr = " commandline is not avilable!"
        stderr = "what are you talking about?"
        stdout = ""

    return stderr,stdout

def split_str_by_blank(str_com,split_char = ""):
    # str 是字符串，默认用空格分割
    # para
    # @str_com -> string 待分割的字符串
    # @split_char -> string 分割标志
    # @return -> list 分割输出
    if not split_char:
        str_list = str_com.split()
    else:
        str_list = str_com.split(split_char)
    return str_list

def command_print_out_err(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        print("ERROR ! : \n",err)
    elif out :
        print("OUTPUT : \n",out)
def command_print_out_err_for_deepmd(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        wall_time_index = err.index("wall time")
        second_unit_index = err.index("s",wall_time_index)
        total_time = float(err[wall_time_index+11:second_unit_index])
        return total_time
    elif out :
        print("OUTPUT : \n",out)
if __name__ == "__main__":
    os.environ.setdefault('DP_VARIANT', 'cuda')
    command_print_out_err("pip install /home/xwy/asc/pack-1/deepmd-kit/")

运行测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from json import tool
from optparse import Option
import subprocess
import os
import sys
from numpy import str_
from sty import fg, bg, ef, rs
import time

def commandline(command_line):
    # 用来输入命令行
    # para
    # @command_line -> list[string]
    # @return stderr -> string 
    #   标准错误
    # @return stdout -> string
    #   标准输出
    try:
        process = subprocess.Popen(command_line,
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE)
        stdout, stderr = process.communicate()
        stderr = stderr.decode('utf-8')
        stdout = stdout.decode('utf-8')   
    except:
        #stderr = " commandline is not avilable!"
        stderr = "what are you talking about?"
        stdout = ""

    return stderr,stdout

def split_str_by_blank(str_com,split_char = ""):
    # str 是字符串，默认用空格分割
    # para
    # @str_com -> string 待分割的字符串
    # @split_char -> string 分割标志
    # @return -> list 分割输出
    if not split_char:
        str_list = str_com.split()
    else:
        str_list = str_com.split(split_char)
    return str_list

def command_print_out_err(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        print("ERROR ! : \n",err)
    elif out :
        print("OUTPUT : \n",out)
def command_print_out_err_for_deepmd(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        wall_time_index = err.index("wall time")
        second_unit_index = err.index("s",wall_time_index)
        total_time = float(err[wall_time_index+11:second_unit_index])
        return total_time
    elif out :
        print("OUTPUT : \n",out)
if __name__ == "__main__":
        root_dir = "/home/xwy/asc/pack-1/"
        current_time = time.strftime("%m-%d_%H-%M-%S", time.localtime())
        os.chdir(root_dir+ "data/water/")
        command_print_out_err("dp train input.json --skip-neighbor-stat")
        # with open("/home/xwy/asc/pack-1/data/water/lcurve.out","a") as f:
        #     f.write("\n"+str(time_str))
        # print("time !!!-> ",time_str)
        # command_print_out_err("mv lcurve.out"+" "+"lcurve.out-"+current_time)

测试50次计时

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from json import tool
from optparse import Option
import subprocess
import os
import sys
from numpy import str_
from sty import fg, bg, ef, rs
import time

def commandline(command_line):
    # 用来输入命令行
    # para
    # @command_line -> list[string]
    # @return stderr -> string 
    #   标准错误
    # @return stdout -> string
    #   标准输出
    try:
        process = subprocess.Popen(command_line,
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE)
        stdout, stderr = process.communicate()
        stderr = stderr.decode('utf-8')
        stdout = stdout.decode('utf-8')   
    except:
        #stderr = " commandline is not avilable!"
        stderr = "what are you talking about?"
        stdout = ""

    return stderr,stdout

def split_str_by_blank(str_com,split_char = ""):
    # str 是字符串，默认用空格分割
    # para
    # @str_com -> string 待分割的字符串
    # @split_char -> string 分割标志
    # @return -> list 分割输出
    if not split_char:
        str_list = str_com.split()
    else:
        str_list = str_com.split(split_char)
    return str_list

def command_print_out_err(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        print("ERROR ! : \n",err)
    elif out :
        print("OUTPUT : \n",out)
def command_print_out_err_for_deepmd(str_com):
    # 用来执行命令，并且根据是否报错输出err 或者 out
    # para
    # @str_com -> string 输入命令
    # @print 输出 err 或者 out
    str_list = split_str_by_blank(str_com)
    err , out = commandline(str_list)
    if err :
        wall_time_index = err.index("wall time")
        second_unit_index = err.index("s",wall_time_index)
        total_time = float(err[wall_time_index+11:second_unit_index])
        return total_time
    elif out :
        print("OUTPUT : \n",out)
if __name__ == "__main__":
    for i in range (50):
        current_time = time.strftime("%m-%d_%H-%M-%S", time.localtime()) 
        time_str = command_print_out_err_for_deepmd("dp train input.json --skip-neighbor-stat")
        with open("lcurve.out","a") as f:
            f.write("\n"+str(time_str))
        print("time !!!-> ",time_str)
        command_print_out_err("mv lcurve.out"+" "+"lcurve.out-"+current_time)

运行时间的统计量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import os 
import glob
import numpy as np
from numpy import size
out_list = [f for f in glob.glob("lcurve.out-*")]
time_list= []
for file in out_list:
    time_in_file = .0
    with open(file,"r") as f:
        last_line = f.readlines()[-1]
        time_in_file = float(last_line[0:-2])
        time_list.append(time_in_file)
print("  mean = ",np.mean(time_list))
print("   var = ",np.var(time_list))
print("median = ",np.median(time_list))
#python
python 脚本
http://home.ustc.edu.cn/~ustcxwy0271/2022/04/06/python-script-1/
作者
Xu Weiye
发布于
2022年4月6日
许可协议