File indexing completed on 2026-04-10 08:39:07
0001 __author__ = "retmas"
0002
0003 from snakemake.parser import Rule, RuleKeywordState
0004 from snakemake.workflow import Workflow
0005
0006
0007 def scatter(self, value):
0008 def decorate(rule_info):
0009 rule_info.scatter = value
0010 return rule_info
0011
0012 return decorate
0013
0014
0015 class Scatter(RuleKeywordState):
0016 @property
0017 def keyword(self):
0018 return "scatter"
0019
0020
0021 def loop(self, value):
0022 def decorate(rule_info):
0023 rule_info.loop = value
0024 return rule_info
0025
0026 return decorate
0027
0028
0029 class Loop(RuleKeywordState):
0030 @property
0031 def keyword(self):
0032 return "loop"
0033
0034
0035 def condition(self, value):
0036 def decorate(rule_info):
0037 rule_info.condition = value
0038 return rule_info
0039
0040 return decorate
0041
0042
0043 class Condition(RuleKeywordState):
0044 @property
0045 def keyword(self):
0046 return "when"
0047
0048
0049 _rule_properties = dict(scatter=Scatter, loop=Loop, when=Condition)
0050
0051
0052 def inject():
0053 Rule.subautomata.update(**_rule_properties)
0054 setattr(Workflow, "scatter", scatter)
0055 setattr(Workflow, "loop", loop)
0056 setattr(Workflow, "when", condition)
0057 rule = getattr(Workflow, "rule")
0058
0059 def rule_new(self, name=None, lineno=None, snakefile=None, checkpoint=False):
0060 decorate = rule(self, name, lineno, snakefile, checkpoint)
0061
0062 def decorate_new(rule_info):
0063 rule_obj = self.get_rule(name)
0064 if getattr(rule_info, "scatter", None):
0065 rule_obj.scatter = rule_info.scatter
0066 if getattr(rule_info, "loop", None):
0067 rule_obj.loop = rule_info.loop
0068 if getattr(rule_info, "condition", None):
0069 rule_obj.condition = rule_info.condition
0070 return decorate(rule_info)
0071
0072 return decorate_new
0073
0074 setattr(Workflow, "rule", rule_new)
