"
+ name += ''
for p in n.input_params:
# some of our input param values are '{}' which mermaid doesn't like
- convert_input_params = n.input_params[p].replace("{", "{").replace("}", "{")
+ convert_input_params = (
+ n.input_params[p]
+ .replace("{", "{")
+ .replace("}", "{")
+ )
name += f"{p}={convert_input_params}
"
name += ""
- bases.append(" "+n.id+"{{"+name+"}}")
- bases.append(" style "+n.id+" fill:lightgray,stroke:#aaa;")
+ bases.append(" " + n.id + "{{" + name + "}}")
+ bases.append(" style " + n.id + " fill:lightgray,stroke:#aaa;")
# render output artifacts
if SHOW_ARTIFACTS:
for a in n.input_artifacts:
a_id = n.id + "__" + a
- name = f"{a}"
+ name = f'{a}'
bases.append(f" {a_id}({name})")
- bases.append(" style "+a_id+" fill:gold,stroke:#222;")
+ bases.append(" style " + a_id + " fill:gold,stroke:#222;")
for t in n.tasks:
# render the task
- name = f"{t.name}"
+ name = f'{t.name}'
if t.when:
when = t.when.replace("{{", "").replace("}}", "")
- name += f"when: {when}"
- if t.ref_node and nodes.get(t.ref_node) and not show_node(nodes[t.ref_node]):
+ name += f'when: {when}'
+ if (
+ t.ref_node
+ and nodes.get(t.ref_node)
+ and not show_node(nodes[t.ref_node])
+ ):
if nodes[t.ref_node].image:
- name += f"image: {nodes[t.ref_node].image}"
+ name += f'image: {nodes[t.ref_node].image}'
if nodes[t.ref_node].script:
- name += f"script: {nodes[t.ref_node].script}"
+ name += f'script: {nodes[t.ref_node].script}'
if t.input_params:
- name += f""
+ name += ''
for p in t.input_params:
name += f"{p}={t.input_params[p]}
"
name += ""
bases.append(f" {t.id}[{name}]")
- bases.append(" style "+t.id+" fill:white;")
+ bases.append(" style " + t.id + " fill:white;")
# render artifacts
if SHOW_ARTIFACTS:
for a in t.input_artifacts:
- name = f"{a}"
+ name = f'{a}'
# bases.append(f" {a_id}({name})")
# bases.append(" style "+a_id+" fill:gold,stroke:#222;")
if SHOW_ARTIFACTS and t.ref_node and nodes.get(t.ref_node):
for a in nodes[t.ref_node].output_artifacts:
a_id = t.id + "__" + a
- name = f"{a}"
+ name = f'{a}'
bases.append(f" {a_id}({name})")
- bases.append(" style "+a_id+" fill:gold,stroke:#222;")
+ bases.append(" style " + a_id + " fill:gold,stroke:#222;")
for t in n.steps:
log.debug(f"rendering step: {t}")
# render the step
- name = f"{t.name}"
+ name = f'{t.name}'
if t.when:
when = t.when.replace("{{", "").replace("}}", "")
- name += f"when: {when}"
- if t.ref_node and nodes.get(t.ref_node) and not show_node(nodes[t.ref_node]):
+ name += f'when: {when}'
+ if (
+ t.ref_node
+ and nodes.get(t.ref_node)
+ and not show_node(nodes[t.ref_node])
+ ):
if nodes[t.ref_node].image:
- name += f"image: {nodes[t.ref_node].image}"
+ name += f'image: {nodes[t.ref_node].image}'
if nodes[t.ref_node].script:
- name += f"script: {nodes[t.ref_node].script}"
+ name += f'script: {nodes[t.ref_node].script}'
if t.input_params:
- name += f""
+ name += ''
for p in t.input_params:
name += f"{p}={t.input_params[p]}
"
name += ""
bases.append(f" {t.id}[{name}]")
- bases.append(" style "+t.id+" fill:white;")
+ bases.append(" style " + t.id + " fill:white;")
# render artifacts
if SHOW_ARTIFACTS:
for a in t.input_artifacts:
- name = f"{a}"
+ name = f'{a}'
# bases.append(f" {a_id}({name})")
# bases.append(" style "+a_id+" fill:gold,stroke:#222;")
if SHOW_ARTIFACTS and t.ref_node and nodes.get(t.ref_node):
for a in nodes[t.ref_node].output_artifacts:
a_id = t.id + "__" + a
- name = f"{a}"
+ name = f'{a}'
bases.append(f" {a_id}({name})")
- bases.append(" style "+a_id+" fill:gold,stroke:#222;")
+ bases.append(" style " + a_id + " fill:gold,stroke:#222;")
bases.append("end")
@@ -532,9 +604,13 @@ def generate_mermaid(workflows, nodes, output_name, output_file, workflow_readme
if flow_lines:
out.append(f"linkStyle {','.join(flow_lines)} stroke:#888,stroke-width:2px;")
if interpackage_lines:
- out.append(f"linkStyle {','.join(interpackage_lines)} stroke:#888,stroke-width:2px;")
+ out.append(
+ f"linkStyle {','.join(interpackage_lines)} stroke:#888,stroke-width:2px;"
+ )
if artifact_lines:
- out.append(f"linkStyle {','.join(artifact_lines)} stroke:#fa0,stroke-width:2px;")
+ out.append(
+ f"linkStyle {','.join(artifact_lines)} stroke:#fa0,stroke-width:2px;"
+ )
mermaid_output = "\n".join(out)
with open(output_file, "w") as f:
@@ -563,8 +639,8 @@ def parse_depends(depends):
class Tokenizer:
def __init__(self, input_string):
- for op in ('||', '&&', '!', '(', ')'):
- input_string = input_string.replace(op, ' ' + op + ' ')
+ for op in ("||", "&&", "!", "(", ")"):
+ input_string = input_string.replace(op, " " + op + " ")
self.tokens = input_string.split()
self.position = 0
@@ -592,16 +668,16 @@ def pr(self, *args):
def parse_expression(self):
token = self.tokenizer.peek()
- if token == '(':
+ if token == "(":
self.parse_parentheses()
- elif token == '!':
+ elif token == "!":
self.parse_not()
else:
self.parse_term()
def parse_term(self):
token = self.tokenizer.consume()
- if token is None or token in ('&&', '||', '!', ')'):
+ if token is None or token in ("&&", "||", "!", ")"):
raise ValueError("Unexpected token: " + str(token))
self.term_prefixes.append(token.split(".")[0])
self.pr(f"term({token})")
@@ -610,7 +686,7 @@ def parse_parentheses(self):
self.pr("(")
self.tokenizer.consume() # Consume '('
self.parse_or()
- if self.tokenizer.consume() != ')':
+ if self.tokenizer.consume() != ")":
raise ValueError("Missing closing parenthesis")
self.pr(")")
@@ -620,15 +696,17 @@ def parse_not(self):
self.parse_expression()
def parse_and(self):
- left = self.parse_expression()
- while self.tokenizer.peek() == '&&':
+ # TODO: this doesn't make sense, the variable was unused
+ _ = self.parse_expression()
+ while self.tokenizer.peek() == "&&":
self.pr(" and ")
self.tokenizer.consume() # Consume 'AND'
self.parse_expression()
def parse_or(self):
- left = self.parse_and()
- while self.tokenizer.peek() == '||':
+ # TODO: this doesn't make sense, the variable was unused
+ _ = self.parse_and()
+ while self.tokenizer.peek() == "||":
self.pr(" or ")
self.tokenizer.consume() # Consume 'OR'
self.parse_and()
@@ -641,7 +719,7 @@ def parse_or(self):
def load_readme(file_path):
try:
- with open(file_path, 'r') as stream:
+ with open(file_path, "r") as stream:
return stream.read()
except Exception:
log.exception("Problem loading README.md file")
@@ -649,7 +727,7 @@ def load_readme(file_path):
def parse_yaml(file_path):
- with open(file_path, 'r') as stream:
+ with open(file_path, "r") as stream:
try:
return yaml.safe_load(stream)
except yaml.YAMLError as exc: