-
Notifications
You must be signed in to change notification settings - Fork 0
/
parseinput.py
50 lines (41 loc) · 1.49 KB
/
parseinput.py
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
import sys
import os
import logging
from copy import deepcopy
from TDAS.combine_modules import Intermedia
DEFAULT_CONFIG_ID="DEFAULT"
def parse_inputfile(config:dict,seqs:str):
seqinfo_file=os.path.abspath(seqs)
logging.info(f"input seq info file is {seqinfo_file}")
names=config["seq_info_format"][1:]
sep=config["seq_info_format"][0]
if sep=="":
sep=None
f=open(seqinfo_file,'r')
part=Intermedia.RawPartID
seq_order=1
for line in f:
if line.strip().startswith("#"):
continue ## pass comment lines
lineSplit=line.split(sep)
if not len(lineSplit):
continue
project=lineSplit[0]
for i in range(0,len(names)):
term=names[i]
value=lineSplit[i].strip()
Intermedia.add_term(part,project,term,str(value))
cmd=f'echo "###### start process pipline ######"'
cmd_part="1"
Intermedia.add_term(part,project,term="command",value=cmd)
Intermedia.add_term(part,project,term="command_part",value=cmd_part)
if "seq_order" not in names:
Intermedia.add_term(part,project,term="seq_order",value=seq_order)
seq_order =1
if "config_id" not in names:
Intermedia.add_term(part,project,term="config_id",value=DEFAULT_CONFIG_ID)
f.close()
logging.info(Intermedia.get_str())
return Intermedia
def parse_dirs(config:dict,dir:str):
logging.info("you are using a testing function!")