Skip to content
GitLab
Menu
Projects
Groups
Snippets
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
genotoul-bioinfo
jflow
Commits
58773fa7
Commit
58773fa7
authored
Oct 25, 2016
by
Floreal Cabanettes
Browse files
Add rules to multiple parameters + refactor check rules
parent
8b8be33e
Changes
5
Hide whitespace changes
Inline
Side-by-side
bin/jflow_cli.py
View file @
58773fa7
...
...
@@ -29,6 +29,7 @@ from jflow.workflows_manager import WorkflowsManager
from
jflow.workflow
import
Workflow
import
jflow.utils
as
utils
class
JflowArgumentParser
(
argparse
.
ArgumentParser
):
def
_read_args_from_files
(
self
,
arg_strings
):
# expand arguments referencing files
...
...
@@ -56,129 +57,22 @@ class JflowArgumentParser (argparse.ArgumentParser):
return
new_arg_strings
def
_check_rule
(
rule
:
str
,
src_arg
:
str
,
user_args
:
dict
,
parameters
:
dict
,
all_files
:
list
,
wf_sub_parser
):
"""
Check a parameter rule
@param rule: string describing the rule
@param src_arg: the argument containing the rule (string)
@param user_args: all arguments given by the user, with their values: {parameter_name1: value, ...}
@param parameters: all workflow parameters objects as dict: {parameter_name1: parameter_object1, ...}
@param all_files: all files given by the user
"""
from
jflow.parameter
import
InputFileList
,
InputFile
,
InputDirectory
print
(
src_arg
)
print
(
user_args
[
src_arg
])
# Rule: Unique
if
rule
==
"unique"
:
if
isinstance
(
parameters
[
src_arg
],
InputFileList
):
for
file
in
user_args
[
src_arg
]:
if
user_args
[
src_arg
].
count
(
file
)
>
1
:
wf_sub_parser
.
error
(
"the file
\"
"
+
file
+
"
\"
in parameter
\"
"
+
src_arg
+
"
\"
is duplicated"
)
else
:
print
(
"
\033
[93mWarning: rule unique is ignored: "
+
src_arg
+
" is not an input file list
\033
[0m"
)
# Rule: Unique_all
elif
rule
==
"unique_all"
:
if
isinstance
(
parameters
[
src_arg
],
InputFile
)
or
isinstance
(
parameters
[
src_arg
],
InputFileList
):
files
=
user_args
[
src_arg
]
if
type
(
files
)
==
str
:
files
=
[
files
]
for
file
in
files
:
if
all_files
.
count
(
file
)
>
1
:
wf_sub_parser
.
error
(
"the file
\"
"
+
file
+
"
\"
from parameter
\"
"
+
src_arg
+
"
\"
is given "
"several times (on this or another argument)"
)
else
:
print
(
"
\033
[93mWarning: rule unique_all is ignored: "
+
src_arg
+
" is not an input file or input file "
"list
\033
[0m"
)
# Rule: Exclude
elif
rule
.
startswith
(
"exclude="
):
match
=
re
.
match
(
r
"exclude=(.+)"
,
rule
)
if
match
:
excludes
=
match
.
group
(
1
).
split
(
","
)
for
exclude
in
excludes
:
if
exclude
in
user_args
and
user_args
[
exclude
]
is
not
None
:
wf_sub_parser
.
error
(
"arguments "
+
src_arg
+
" and "
+
exclude
+
" are mutually excluded"
)
else
:
print
(
"
\033
[93mWarning: unrecognized exclude list: "
+
rule
+
"
\033
[0m"
)
def
_load_sub_parameters
(
items
,
parameter
):
parameters_used
=
{}
for
item
in
items
:
p_item
=
item
[
0
].
replace
(
"-"
,
"_"
)
param_name
=
parameter
.
name
+
">"
+
p_item
if
param_name
not
in
parameters_used
:
parameters_used
[
param_name
]
=
[]
parameters_used
[
param_name
].
append
(
item
[
1
])
return
parameters_used
# Rule: Exclude if
elif
rule
.
startswith
(
"exclude_if"
):
if
not
isinstance
(
parameters
[
src_arg
],
InputFile
)
and
not
isinstance
(
parameters
[
src_arg
],
InputFileList
)
and
\
not
isinstance
(
parameters
[
src_arg
],
InputDirectory
):
match
=
re
.
match
(
r
"exclude_if\[([^\]]+)\]=(.+)"
,
rule
)
if
match
:
conditions
=
match
.
group
(
1
).
split
(
","
)
excludes
=
match
.
group
(
2
).
split
(
","
)
# Check condition is raised:
condition_raised
=
None
for
condition
in
conditions
:
if
(
condition
[
0
]
==
"!"
and
user_args
[
src_arg
]
!=
condition
[
1
:])
or
(
condition
[
0
]
!=
"!"
and
str
(
user_args
[
src_arg
])
==
str
(
condition
)):
condition_raised
=
condition
break
# Apply rule if true:
if
condition_raised
is
not
None
:
for
exclude
in
excludes
:
if
exclude
in
user_args
and
user_args
[
exclude
]
is
not
None
:
if
not
condition_raised
.
startswith
(
"!"
):
wf_sub_parser
.
error
(
"argument "
+
src_arg
+
" is set to "
+
condition_raised
+
", so the "
+
\
exclude
+
" argument is not allowed"
)
else
:
wf_sub_parser
.
error
(
"argument "
+
src_arg
+
" is not set to "
+
condition_raised
[
1
:]
+
\
", so the "
+
exclude
+
" argument is not allowed"
)
else
:
print
(
"
\033
[93mWarning: unrecognized conditional exclude list: "
+
rule
+
"
\033
[0m"
)
else
:
print
(
"
\033
[93mWarning: rule exclude_if is not accepted for inputs files/directories. "
"The rule has been ignored
\033
[0m"
)
# Rule: To be required
elif
rule
.
startswith
(
"to_be_required="
):
match
=
re
.
match
(
r
"to_be_required=(.+)"
,
rule
)
if
match
:
requires
=
match
.
group
(
1
).
split
(
","
)
for
require
in
requires
:
if
require
not
in
user_args
or
(
require
in
user_args
and
user_args
[
require
]
is
None
):
wf_sub_parser
.
error
(
"the argument "
+
src_arg
+
" require argument "
+
require
+
" to be defined"
)
else
:
print
(
"
\033
[93mWarning: unrecognized to_be_required list: "
+
rule
+
"
\033
[0m"
)
# Rule: To be required if:
elif
rule
.
startswith
(
"to_be_required_if"
):
match
=
re
.
match
(
r
"to_be_required_if\[([^\]]+)\]=(.+)"
,
rule
)
if
match
:
conditions
=
match
.
group
(
1
).
split
(
","
)
requires
=
match
.
group
(
2
).
split
(
","
)
# Check condition is raised:
condition_raised
=
None
for
condition
in
conditions
:
if
(
condition
[
0
]
==
"!"
and
user_args
[
src_arg
]
!=
condition
[
1
:])
or
(
condition
[
0
]
!=
"!"
and
str
(
user_args
[
src_arg
])
==
str
(
condition
)):
condition_raised
=
condition
break
# Apply rule if true:
if
condition_raised
is
not
None
:
for
require
in
requires
:
if
require
not
in
user_args
or
(
require
in
user_args
and
user_args
[
require
]
is
None
):
if
not
condition_raised
.
startswith
(
"!"
):
wf_sub_parser
.
error
(
"the argument "
+
src_arg
+
" is set to "
+
condition_raised
+
", so it require argument "
+
require
+
" to be defined"
)
else
:
wf_sub_parser
.
error
(
"the argument "
+
src_arg
+
" is not set to "
+
condition_raised
[
1
:]
+
", so it require argument "
+
require
+
" to be defined"
)
else
:
print
(
"
\033
[93mWarning: unrecognized rule: "
+
rule
+
"
\033
[0m"
)
def
parse_parameters_rules
(
wf_class
,
wf_manager
):
from
jflow.parameter
import
InputFile
,
InputFileList
,
MultiParameterList
def
parse_parameters_rules
(
wf_instance
):
from
jflow.parameter
import
InputFile
,
InputFileList
,
MultiParameterList
,
MultiParameter
# Get users arguments:
arg_parser
=
JflowArgumentParser
()
wf_instance
=
wf_manager
.
get_workflow_by_class
(
wf_class
)
arg_subparsers
=
arg_parser
.
add_subparsers
(
title
=
'Available sub commands'
)
arg_sub_parser
=
arg_subparsers
.
add_parser
(
wf_instance
.
name
,
help
=
wf_instance
.
description
,
fromfile_prefix_chars
=
'@'
)
arg_sub_parser
.
convert_arg_line_to_args
=
wf_instance
.
__class__
.
config_parser
...
...
@@ -191,30 +85,29 @@ def parse_parameters_rules(wf_class, wf_manager):
rules
=
{}
all_files
=
[]
# Store all given files, used later
parameters_used
=
{}
# Parameters given by the user
p
arameters_dict
=
{}
# All parameters of the workflow
p
rint
(
user_params
)
for
parameter
in
parameters
:
parameters_dict
[
parameter
.
name
]
=
parameter
# Get sub-parameters of multiparameter lists:
if
isinstance
(
parameter
,
MultiParameterList
):
# Get sub-parameters of MultiParameter or MultiParameterLists:
if
isinstance
(
parameter
,
MultiParameterList
)
or
isinstance
(
parameter
,
MultiParameter
):
parameters_used
[
parameter
.
name
]
=
{}
# Get parameters given by the user:
for
rows
in
user_params
[
parameter
.
name
]:
for
item
in
rows
:
p_item
=
item
[
0
].
replace
(
"-"
,
"_"
)
parameter_name
=
parameter
.
name
+
">"
+
p_item
# Name of the parameter to store in parameters_used
if
parameter_name
not
in
parameters_used
:
parameters_used
[
parameter_name
]
=
[]
parameters_used
[
parameter_name
].
append
(
item
[
1
])
if
isinstance
(
parameter
,
MultiParameterList
):
for
rows
in
user_params
[
parameter
.
name
]:
parameters_used
=
{
**
parameters_used
,
**
_load_sub_parameters
(
rows
,
parameter
)}
else
:
# MultiParameter
parameters_used
=
{
**
parameters_used
,
**
_load_sub_parameters
(
user_params
[
parameter
.
name
],
parameter
)}
# Get all parameters and check rules for the ones given by the user:
for
sub_param
in
parameter
.
sub_parameters
:
parameter_name
=
parameter
.
name
+
">"
+
sub_param
.
name
parameters_dict
[
parameter_name
]
=
sub_param
if
parameter_name
in
parameters_used
:
param_name
=
parameter
.
name
+
">"
+
sub_param
.
name
if
param_name
in
parameters_used
:
if
sub_param
.
rules
is
not
None
:
rules
[
param
eter
_name
]
=
sub_param
.
rules
.
split
(
";"
)
rules
[
param_name
]
=
sub_param
.
rules
.
split
(
";"
)
if
isinstance
(
sub_param
,
InputFile
):
all_files
+=
parameters_used
[
parameter_name
]
all_files
+=
parameters_used
[
param_name
]
if
parameter
.
rules
is
not
None
:
rules
[
parameter
.
name
]
=
parameter
.
rules
.
split
(
";"
)
# Other parameters:
elif
parameter
.
name
in
user_params
and
user_params
[
parameter
.
name
]
is
not
None
:
...
...
@@ -229,7 +122,10 @@ def parse_parameters_rules(wf_class, wf_manager):
# Check rules:
for
src_arg
,
arg_rules
in
rules
.
items
():
for
arg_rule
in
arg_rules
:
_check_rule
(
arg_rule
,
src_arg
,
parameters_used
,
parameters_dict
,
all_files
,
arg_sub_parser
)
try
:
wf_instance
.
check_parameter_rule
(
arg_rule
,
src_arg
,
parameters_used
,
all_files
)
except
Exception
as
e
:
arg_sub_parser
.
error
(
e
)
if
__name__
==
'__main__'
:
...
...
@@ -294,7 +190,6 @@ if __name__ == '__main__':
sub_parser
=
subparsers
.
add_parser
(
instance
.
name
,
help
=
instance
.
description
,
fromfile_prefix_chars
=
'@'
)
sub_parser
.
convert_arg_line_to_args
=
instance
.
__class__
.
config_parser
[
parameters_groups
,
parameters_order
]
=
instance
.
get_parameters_per_groups
()
print
(
parameters_order
)
for
group
in
parameters_order
:
if
group
==
"default"
:
for
param
in
parameters_groups
[
group
]:
...
...
@@ -321,7 +216,8 @@ if __name__ == '__main__':
parser
.
exit
(
0
,
""
)
if
args
[
"cmd_object"
]
in
wf_classes
:
parse_parameters_rules
(
args
[
"cmd_object"
],
wfmanager
)
workflow
=
wfmanager
.
get_workflow_by_class
(
args
[
"cmd_object"
])
parse_parameters_rules
(
workflow
)
wfmanager
.
run_workflow
(
args
[
"cmd_object"
],
args
)
elif
args
[
"cmd_object"
]
==
"rerun"
:
wfmanager
.
rerun_workflow
(
args
[
"workflow_id"
])
...
...
src/jflow/parameter.py
View file @
58773fa7
...
...
@@ -426,9 +426,9 @@ class IOFile(str, AbstractIOFile):
class
MultiParameter
(
dict
,
AbstractParameter
):
def
__init__
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
cmd_format
=
""
,
argpos
=-
1
):
def
__init__
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
cmd_format
=
""
,
argpos
=-
1
,
rules
=
None
):
AbstractParameter
.
__init__
(
self
,
name
,
help
,
required
=
required
,
type
=
"multiple"
,
flag
=
flag
,
group
=
group
,
display_name
=
display_name
,
cmd_format
=
cmd_format
,
argpos
=
argpos
)
display_name
=
display_name
,
cmd_format
=
cmd_format
,
argpos
=
argpos
,
rules
=
rules
)
return
dict
.
__init__
(
self
,
{})
def
add_sub_parameter
(
self
,
param
):
...
...
@@ -457,14 +457,22 @@ class MultiParameter(dict, AbstractParameter):
self
.
help
=
self
.
global_help
+
self
.
type
.
get_help
()
param
.
flag
=
param_flag
self
.
sub_parameters
.
append
(
param
)
def
get_sub_parameters_by_name
(
self
):
sub_params
=
{}
for
sub_parameter
in
self
.
sub_parameters
:
if
sub_parameter
.
name
not
in
sub_params
:
sub_params
[
sub_parameter
.
name
]
=
[]
sub_params
[
sub_parameter
.
name
].
append
(
sub_parameter
)
return
sub_params
class
MultiParameterList
(
list
,
AbstractParameter
):
def
__init__
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
cmd_format
=
""
,
argpos
=-
1
):
def
__init__
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
cmd_format
=
""
,
argpos
=-
1
,
rules
=
None
):
AbstractParameter
.
__init__
(
self
,
name
,
help
,
required
=
required
,
type
=
"multiple"
,
flag
=
flag
,
action
=
"append"
,
group
=
group
,
display_name
=
display_name
,
cmd_format
=
cmd_format
,
argpos
=
argpos
)
cmd_format
=
cmd_format
,
argpos
=
argpos
,
rules
=
rules
)
return
list
.
__init__
(
self
,
[])
def
add_sub_parameter
(
self
,
param
):
...
...
@@ -493,7 +501,15 @@ class MultiParameterList(list, AbstractParameter):
self
.
help
=
self
.
global_help
+
self
.
type
.
get_help
()
param
.
flag
=
param_flag
self
.
sub_parameters
.
append
(
param
)
def
get_sub_parameters_by_name
(
self
):
sub_params
=
{}
for
sub_parameter
in
self
.
sub_parameters
:
if
sub_parameter
.
name
not
in
sub_params
:
sub_params
[
sub_parameter
.
name
]
=
[]
sub_params
[
sub_parameter
.
name
].
append
(
sub_parameter
)
return
sub_params
def
__getitem__
(
self
,
key
):
getitem
=
self
.
__dict__
.
get
(
"__getitem__"
,
list
.
__getitem__
)
if
isinstance
(
key
,
int
):
...
...
src/jflow/rules.py
0 → 100644
View file @
58773fa7
#
# Copyright (C) 2015 INRA
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from
abc
import
ABC
,
abstractmethod
class
SimpleRule
(
ABC
):
def
__init__
(
self
,
user_args
,
wf_instance
,
src_arg
,
all_files
=
None
):
self
.
user_args
=
user_args
self
.
wf_instance
=
wf_instance
self
.
src_arg
=
src_arg
self
.
all_files
=
all_files
@
abstractmethod
def
check
(
self
):
pass
class
LinkRule
(
SimpleRule
):
def
__init__
(
self
,
user_args
,
wf_instance
,
src_arg
,
targets_args
,
all_files
=
None
):
SimpleRule
.
__init__
(
self
,
user_args
,
wf_instance
,
src_arg
,
all_files
)
self
.
targets_args
=
targets_args
@
abstractmethod
def
check
(
self
):
pass
class
ConditionalLinkRule
(
LinkRule
):
def
__init__
(
self
,
user_args
,
wf_instance
,
src_arg
,
targets_args
,
conditions
,
all_files
=
None
):
LinkRule
.
__init__
(
self
,
user_args
,
wf_instance
,
src_arg
,
targets_args
,
all_files
)
self
.
conditions
=
conditions
@
abstractmethod
def
check
(
self
):
pass
src/jflow/workflow.py
View file @
58773fa7
...
...
@@ -46,6 +46,8 @@ from jflow.utils import get_octet_string_representation, get_nb_octet
from
jflow.parameter
import
*
from
cctools.util
import
time_format
from
jflow.rules
import
*
from
weaver.script
import
ABSTRACTIONS
from
weaver.script
import
DATASETS
from
weaver.script
import
FUNCTIONS
...
...
@@ -57,6 +59,8 @@ from weaver.options import Options
from
cctools.makeflow
import
MakeflowLog
from
cctools.makeflow.log
import
Node
from
workflows
import
rules
as
wf_rules
class
MINIWorkflow
(
object
):
...
...
@@ -233,14 +237,17 @@ class Workflow(threading.Thread):
self
.
params_order
.
append
(
name
)
self
.
__setattr__
(
name
,
new_param
)
def
add_multiple_parameter
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
):
def
add_multiple_parameter
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
rules
=
None
):
self
.
params_order
.
append
(
name
)
new_param
=
MultiParameter
(
name
,
help
,
flag
=
flag
,
required
=
required
,
group
=
group
,
display_name
=
display_name
)
new_param
=
MultiParameter
(
name
,
help
,
flag
=
flag
,
required
=
required
,
group
=
group
,
display_name
=
display_name
,
rules
=
rules
)
self
.
__setattr__
(
name
,
new_param
)
def
add_multiple_parameter_list
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
):
def
add_multiple_parameter_list
(
self
,
name
,
help
,
required
=
False
,
flag
=
None
,
group
=
"default"
,
display_name
=
None
,
rules
=
None
):
self
.
params_order
.
append
(
name
)
new_param
=
MultiParameterList
(
name
,
help
,
flag
=
flag
,
required
=
required
,
group
=
group
,
display_name
=
display_name
)
new_param
=
MultiParameterList
(
name
,
help
,
flag
=
flag
,
required
=
required
,
group
=
group
,
display_name
=
display_name
,
rules
=
rules
)
self
.
__setattr__
(
name
,
new_param
)
def
add_parameter
(
self
,
name
,
help
,
default
=
None
,
type
=
str
,
choices
=
None
,
...
...
@@ -1231,4 +1238,71 @@ class Workflow(threading.Thread):
# Import symbols from module into global namespace, which we store as
# an attribute for later use (i.e. during compile)
for
symbol
in
symbols
:
self
.
globals
[
symbol
]
=
getattr
(
m
,
symbol
)
\ No newline at end of file
self
.
globals
[
symbol
]
=
getattr
(
m
,
symbol
)
def
check_parameter_rule
(
self
,
rule
:
str
,
src_arg
:
str
,
user_args
:
dict
,
all_files
:
list
):
"""
Check a parameter rule
@param rule: string describing the rule
@param src_arg: the argument containing the rule (string)
@param user_args: all arguments given by the user, with their values: {parameter_name1: value, ...}
@param parameters: all workflow parameters objects as dict: {parameter_name1: parameter_object1, ...}
@param all_files: all files given by the user
"""
###############################################################
# We have to find which type is the rule #
# There is 3 types of rules: #
# - SimpleRule: <name> #
# - LinkRule: <name>=<targets> #
# - ConditionalLinkRule: <name>?[<conditions>]=<targets> #
# With: #
# - <targets> = target1,target2,... #
# - <conditions> = condition1,condition2,... ("," is an "or") #
###############################################################
# Regexp for special rules:
conditional_link_rule
=
r
"(.+)\?\[(.+)\]=(.+)"
link_rule
=
r
"(.+)=(.+)"
is_conditional_link_rule
=
re
.
match
(
conditional_link_rule
,
rule
)
# Check first special rule
if
is_conditional_link_rule
:
# Check this special rule:
name
=
is_conditional_link_rule
.
group
(
1
)
conditions
=
is_conditional_link_rule
.
group
(
2
).
split
(
","
)
targets
=
is_conditional_link_rule
.
group
(
3
).
split
(
","
)
if
hasattr
(
wf_rules
,
name
):
# Check the rule exists
validator_instance
=
getattr
(
wf_rules
,
name
)
if
issubclass
(
validator_instance
,
ConditionalLinkRule
):
# Check the rule is the same special rule
validator
=
getattr
(
wf_rules
,
name
)(
user_args
,
self
,
src_arg
,
targets
,
conditions
,
all_files
)
validator
.
check
()
else
:
raise
Exception
(
"Rule is not a conditional link rule: "
+
name
)
else
:
raise
Exception
(
"Rule is not defined: "
+
name
)
else
:
is_link_rule
=
re
.
match
(
link_rule
,
rule
)
# Check the second special rule
if
is_link_rule
:
# Check this special rule:
name
=
is_link_rule
.
group
(
1
)
targets
=
is_link_rule
.
group
(
2
).
split
(
","
)
if
hasattr
(
wf_rules
,
name
):
# Check the rule exists
validator_class
=
getattr
(
wf_rules
,
name
)
if
issubclass
(
validator_class
,
LinkRule
):
# Check the rule is the same special rule
validator
=
validator_class
(
user_args
,
self
,
src_arg
,
targets
,
all_files
)
validator
.
check
()
else
:
raise
Exception
(
"Rule is not a link rule: "
+
name
)
else
:
raise
Exception
(
"Rule is not defined: "
+
name
)
else
:
# It's a simple rule
name
=
rule
if
hasattr
(
wf_rules
,
name
):
# Check the rule exists
validator_class
=
getattr
(
wf_rules
,
name
)
if
issubclass
(
validator_class
,
SimpleRule
):
# Check it's a simple rule
validator
=
validator_class
(
user_args
,
self
,
src_arg
,
all_files
)
validator
.
check
()
else
:
raise
Exception
(
"Rule is not a simple rule: "
+
name
)
else
:
raise
Exception
(
"Rule is not defined: "
+
name
)
workflows/rules.py
0 → 100644
View file @
58773fa7
#
# Copyright (C) 2015 INRA
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from
jflow.rules
import
SimpleRule
,
LinkRule
,
ConditionalLinkRule
from
jflow.parameter
import
InputFile
,
InputFileList
,
InputDirectory
,
MultiParameter
,
MultiParameterList
# Define your rules check below
class
Unique
(
SimpleRule
):
def
check
(
self
):
if
">"
not
in
self
.
src_arg
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
)
else
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
[:
self
.
src_arg
.
index
(
">"
)]).
\
get_sub_parameters_by_name
(
self
.
src_arg
[
self
.
src_arg
.
index
(
">"
)
+
1
:])
if
isinstance
(
wf_parameter
,
InputFileList
):
for
file
in
self
.
user_args
[
self
.
src_arg
]:
if
self
.
user_args
[
self
.
src_arg
].
count
(
file
)
>
1
:
raise
Exception
(
"the file
\"
"
+
file
+
"
\"
in parameter
\"
"
+
self
.
src_arg
+
"
\"
is duplicated"
)
else
:
print
(
"
\033
[93mWarning: rule unique is ignored: "
+
self
.
src_arg
+
" is not an input file list
\033
[0m"
)
class
UniqueAll
(
SimpleRule
):
def
check
(
self
):
if
">"
not
in
self
.
src_arg
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
)
else
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
[:
self
.
src_arg
.
index
(
">"
)]).
\
get_sub_parameters_by_name
()[
self
.
src_arg
[
self
.
src_arg
.
index
(
">"
)
+
1
:]][
0
]
if
isinstance
(
wf_parameter
,
InputFile
)
or
isinstance
(
wf_parameter
,
InputFileList
):
files
=
self
.
user_args
[
self
.
src_arg
]
if
type
(
files
)
==
str
:
files
=
[
files
]
for
file
in
files
:
if
self
.
all_files
.
count
(
file
)
>
1
:
raise
Exception
(
"the file
\"
"
+
file
+
"
\"
from parameter
\"
"
+
self
.
src_arg
+
"
\"
is given "
"several times (on this or another argument)"
)
else
:
print
(
"
\033
[93mWarning: rule unique_all is ignored: "
+
self
.
src_arg
+
" is not an input file or input file list
\033
[0m"
)
class
Exclude
(
LinkRule
):
def
check
(
self
):
for
exclude
in
self
.
targets_args
:
if
exclude
in
self
.
user_args
and
self
.
user_args
[
exclude
]
is
not
None
:
raise
Exception
(
"arguments "
+
self
.
src_arg
+
" and "
+
exclude
+
" are mutually excluded"
)
class
ExcludeIf
(
ConditionalLinkRule
):
def
check
(
self
):
# Rule valid anony for non input files/directories:
if
">"
not
in
self
.
src_arg
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
)
else
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
[:
self
.
src_arg
.
index
(
">"
)]).
\
get_sub_parameters_by_name
(
self
.
src_arg
[
self
.
src_arg
.
index
(
">"
)
+
1
:])
if
not
isinstance
(
wf_parameter
,
InputFile
)
and
\
not
isinstance
(
wf_parameter
,
InputFileList
)
and
not
isinstance
(
wf_parameter
,
InputDirectory
)
and
\
not
isinstance
(
wf_parameter
,
MultiParameter
)
and
not
isinstance
(
wf_parameter
,
MultiParameterList
):
condition_raised
=
None
for
condition
in
self
.
conditions
:
if
(
condition
[
0
]
==
"!"
and
self
.
user_args
[
self
.
src_arg
]
!=
condition
[
1
:])
or
(
condition
[
0
]
!=
"!"
and
str
(
self
.
user_args
[
self
.
src_arg
])
==
str
(
condition
)):
condition_raised
=
condition
break
if
condition_raised
:
validator
=
Exclude
(
self
.
user_args
,
self
.
wf_instance
,
self
.
src_arg
,
self
.
targets_args
,
self
.
all_files
)
try
:
validator
.
check
()
except
Exception
as
e
:
raise
Exception
(
"Argument "
+
self
.
src_arg
+
" is set to "
+
condition_raised
+
": "
+
str
(
e
))
else
:
raise
Exception
(
"Rule ExcludeIf is not available on files and directories parameters "
"or multiples parameters"
)
class
ToBeRequired
(
LinkRule
):
def
check
(
self
):
for
require
in
self
.
targets_args
:
if
require
not
in
self
.
user_args
or
(
require
in
self
.
user_args
and
self
.
user_args
[
require
]
is
None
):
raise
Exception
(
"the argument "
+
self
.
src_arg
+
" require argument "
+
require
+
" to be defined"
)
class
ToBeRequiredIf
(
ConditionalLinkRule
):
def
check
(
self
):
if
">"
not
in
self
.
src_arg
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
)
else
:
wf_parameter
=
getattr
(
self
.
wf_instance
,
self
.
src_arg
[:
self
.
src_arg
.
index
(
">"
)]).
\
get_sub_parameters_by_name
(
self
.
src_arg
[
self
.
src_arg
.
index
(
">"
)
+
1
:])
# Rule valid only for non input files/directories:
if
not
isinstance
(
wf_parameter
,
InputFile
)
and
\
not
isinstance
(
wf_parameter
,
InputFileList
)
and
not
isinstance
(
wf_parameter
,
InputDirectory
)
and
\
not
isinstance
(
wf_parameter
,
MultiParameter
)
and
not
isinstance
(
wf_parameter
,
MultiParameterList
):
condition_raised
=
None
for
condition
in
self
.
conditions
:
if
(
condition
[
0
]
==
"!"
and
self
.
user_args
[
self
.
src_arg
]
!=
condition
[
1
:])
or
(
condition
[
0
]
!=
"!"
and
str
(
self
.
user_args
[
self
.
src_arg
])
==
str
(
condition
)):
condition_raised
=
condition
break
if
condition_raised
:
validator
=
ToBeRequired
(
self
.
user_args
,
self
.
wf_instance
,
self
.
src_arg
,
self
.
targets_args
,
self
.
all_files
)
try
:
validator
.
check
()
except
Exception
as
e
:
raise
Exception
(
"Argument "
+
self
.
src_arg
+
" is set to "
+
condition_raised
+
": "
+
str
(
e
))
else
:
raise
Exception
(
"Rule ToBeRequiredIf is not available on files and directories parameters or multiple "
"parameters"
)
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment